Skip to content

Base Client

llm_expose.clients.base

Abstract base class for messaging client adapters.

BaseClient

Bases: ABC


              flowchart TD
              llm_expose.clients.base.BaseClient[BaseClient]

              

              click llm_expose.clients.base.BaseClient href "" "llm_expose.clients.base.BaseClient"

Common interface that all messaging client adapters must implement.

A client is responsible for:

  1. Connecting to the external messaging platform.
  2. Receiving user messages.
  3. Delegating them to the registered :data:MessageHandler.
  4. Sending the handler's reply back to the user.
Source code in llm_expose/clients/base.py 
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
class BaseClient(ABC):
    """Common interface that all messaging client adapters must implement.

    A client is responsible for:

    1. Connecting to the external messaging platform.
    2. Receiving user messages.
    3. Delegating them to the registered :data:`MessageHandler`.
    4. Sending the handler's reply back to the user.
    """

    def __init__(self, handler: MessageHandler) -> None:
        """Initialise the client with an async *handler*.

        Args:
            handler: Async callable that receives a user message string and
                returns the model's reply string.
        """
        self._handler = handler

    @abstractmethod
    async def start(self) -> None:
        """Connect to the platform and begin receiving messages.

        This method should block (run the event loop) until :meth:`stop` is
        called or an unrecoverable error occurs.
        """

    @abstractmethod
    async def stop(self) -> None:
        """Gracefully disconnect from the platform and release resources."""

    @abstractmethod
    async def send_message(self, user_id: str, text: str) -> dict:
        """Send a direct message to a specific user.

        This method sends a plain text message directly to a user, bypassing
        the regular message handler. It is intended for programmatic message
        delivery (e.g., cron jobs, scheduled notifications).

        Args:
            user_id: The user/chat ID to send the message to.
            text: The message text to send. May include Markdown formatting
                if the client supports it.

        Returns:
            A dict with keys:
                - message_id: The platform-specific message ID (str)
                - timestamp: ISO8601 timestamp when message was sent (str)
                - status: Status message (str, e.g. \"sent\")
                - user_id: Echo of the user_id parameter (str)

        Raises:
            Exception: Platform-specific exceptions on send failures (network,
                authentication, rate limits, invalid user_id).
        """

    @abstractmethod
    async def send_images(self, user_id: str, image_urls: list[str]) -> dict:
        """Send one or more images directly to a specific user.

        Args:
            user_id: The user/chat ID to send images to.
            image_urls: Image references as remote URLs or data URLs.

        Returns:
            A dict with provider-specific send metadata.
        """

    @abstractmethod
    async def send_file(self, user_id: str, file_path: str) -> dict:
        """Send a local file directly to a specific user.

        Args:
            user_id: The user/chat ID to send the file to.
            file_path: Path to a local file.

        Returns:
            A dict with provider-specific send metadata.
        """

    async def notify_tool_status(
        self,
        user_id: str,
        status: str,
        tool_name: str,
        *,
        approval_id: str | None = None,
        detail: str | None = None,
    ) -> None:
        """Publish tool lifecycle status feedback to end users.

        Clients may choose how to present updates (new messages, message edits,
        inline UI updates, etc.). The default implementation is a no-op to keep
        backward compatibility for clients that do not support interim feedback.

        Args:
            user_id: Platform-specific chat/user identifier.
            status: Lifecycle state (for example: "running", "failed").
            tool_name: Name of the tool being executed.
            approval_id: Optional approval identifier when this status relates
                to an approval workflow.
            detail: Optional human-readable extra detail for the status.
        """
        return None

    def set_handler(self, handler: MessageHandler) -> None:
        """Replace the current message handler with *handler*.

        Args:
            handler: Async callable that receives a user message string and
                returns the model's reply string.
        """
        self._handler = handler

__init__(handler)

Initialise the client with an async handler.

Parameters:

Name Type Description Default
handler MessageHandler

Async callable that receives a user message string and returns the model's reply string.

 required
Source code in llm_expose/clients/base.py 
47
48
49
50
51
52
53
54
def __init__(self, handler: MessageHandler) -> None:
    """Initialise the client with an async *handler*.

    Args:
        handler: Async callable that receives a user message string and
            returns the model's reply string.
    """
    self._handler = handler

notify_tool_status(user_id, status, tool_name, *, approval_id=None, detail=None) async

Publish tool lifecycle status feedback to end users.

Clients may choose how to present updates (new messages, message edits, inline UI updates, etc.). The default implementation is a no-op to keep backward compatibility for clients that do not support interim feedback.

Parameters:

Name Type Description Default
user_id str

Platform-specific chat/user identifier.

 required
status str

Lifecycle state (for example: "running", "failed").

 required
tool_name str

Name of the tool being executed.

 required
approval_id str | None

Optional approval identifier when this status relates to an approval workflow.

 None
detail str | None

Optional human-readable extra detail for the status.

 None
Source code in llm_expose/clients/base.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
async def notify_tool_status(
    self,
    user_id: str,
    status: str,
    tool_name: str,
    *,
    approval_id: str | None = None,
    detail: str | None = None,
) -> None:
    """Publish tool lifecycle status feedback to end users.

    Clients may choose how to present updates (new messages, message edits,
    inline UI updates, etc.). The default implementation is a no-op to keep
    backward compatibility for clients that do not support interim feedback.

    Args:
        user_id: Platform-specific chat/user identifier.
        status: Lifecycle state (for example: "running", "failed").
        tool_name: Name of the tool being executed.
        approval_id: Optional approval identifier when this status relates
            to an approval workflow.
        detail: Optional human-readable extra detail for the status.
    """
    return None

send_file(user_id, file_path) abstractmethod async

Send a local file directly to a specific user.

Parameters:

Name Type Description Default
user_id str

The user/chat ID to send the file to.

 required
file_path str

Path to a local file.

 required

Returns:

Type Description
dict

A dict with provider-specific send metadata.
Source code in llm_expose/clients/base.py 
105
106
107
108
109
110
111
112
113
114
115
@abstractmethod
async def send_file(self, user_id: str, file_path: str) -> dict:
    """Send a local file directly to a specific user.

    Args:
        user_id: The user/chat ID to send the file to.
        file_path: Path to a local file.

    Returns:
        A dict with provider-specific send metadata.
    """

send_images(user_id, image_urls) abstractmethod async

Send one or more images directly to a specific user.

Parameters:

Name Type Description Default
user_id str

The user/chat ID to send images to.

 required
image_urls list[str]

Image references as remote URLs or data URLs.

 required

Returns:

Type Description
dict

A dict with provider-specific send metadata.
Source code in llm_expose/clients/base.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
@abstractmethod
async def send_images(self, user_id: str, image_urls: list[str]) -> dict:
    """Send one or more images directly to a specific user.

    Args:
        user_id: The user/chat ID to send images to.
        image_urls: Image references as remote URLs or data URLs.

    Returns:
        A dict with provider-specific send metadata.
    """

send_message(user_id, text) abstractmethod async

Send a direct message to a specific user.

This method sends a plain text message directly to a user, bypassing the regular message handler. It is intended for programmatic message delivery (e.g., cron jobs, scheduled notifications).

Parameters:

Name Type Description Default
user_id str

The user/chat ID to send the message to.

 required
text str

The message text to send. May include Markdown formatting if the client supports it.

 required

Returns:

Type Description
dict

A dict with keys: - message_id: The platform-specific message ID (str) - timestamp: ISO8601 timestamp when message was sent (str) - status: Status message (str, e.g. "sent") - user_id: Echo of the user_id parameter (str)

Raises:

Type Description
Exception

Platform-specific exceptions on send failures (network, authentication, rate limits, invalid user_id).
Source code in llm_expose/clients/base.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
@abstractmethod
async def send_message(self, user_id: str, text: str) -> dict:
    """Send a direct message to a specific user.

    This method sends a plain text message directly to a user, bypassing
    the regular message handler. It is intended for programmatic message
    delivery (e.g., cron jobs, scheduled notifications).

    Args:
        user_id: The user/chat ID to send the message to.
        text: The message text to send. May include Markdown formatting
            if the client supports it.

    Returns:
        A dict with keys:
            - message_id: The platform-specific message ID (str)
            - timestamp: ISO8601 timestamp when message was sent (str)
            - status: Status message (str, e.g. \"sent\")
            - user_id: Echo of the user_id parameter (str)

    Raises:
        Exception: Platform-specific exceptions on send failures (network,
            authentication, rate limits, invalid user_id).
    """

set_handler(handler)

Replace the current message handler with handler.

Parameters:

Name Type Description Default
handler MessageHandler

Async callable that receives a user message string and returns the model's reply string.

 required
Source code in llm_expose/clients/base.py 
142
143
144
145
146
147
148
149
def set_handler(self, handler: MessageHandler) -> None:
    """Replace the current message handler with *handler*.

    Args:
        handler: Async callable that receives a user message string and
            returns the model's reply string.
    """
    self._handler = handler

start() abstractmethod async

Connect to the platform and begin receiving messages.

This method should block (run the event loop) until :meth:stop is called or an unrecoverable error occurs.

Source code in llm_expose/clients/base.py 
56
57
58
59
60
61
62
@abstractmethod
async def start(self) -> None:
    """Connect to the platform and begin receiving messages.

    This method should block (run the event loop) until :meth:`stop` is
    called or an unrecoverable error occurs.
    """

stop() abstractmethod async

Gracefully disconnect from the platform and release resources.

Source code in llm_expose/clients/base.py 
64
65
66
@abstractmethod
async def stop(self) -> None:
    """Gracefully disconnect from the platform and release resources."""

MessageResponse dataclass

Structured response that can include approval metadata for interactive UI.

Attributes:

Name Type Description
content str

The text message to display to the user.
images list[str] | None

Optional list of image URLs/data URLs to send as references.
approval_id str | None

Optional approval ID for tool execution confirmation.
tool_names list[str] | None

Optional list of tool names requiring approval.
server_names dict[str, str] | None

Optional mapping of tool names to server names.
Source code in llm_expose/clients/base.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
@dataclass
class MessageResponse:
    """Structured response that can include approval metadata for interactive UI.

    Attributes:
        content: The text message to display to the user.
        images: Optional list of image URLs/data URLs to send as references.
        approval_id: Optional approval ID for tool execution confirmation.
        tool_names: Optional list of tool names requiring approval.
        server_names: Optional mapping of tool names to server names.
    """

    content: str
    images: list[str] | None = None
    approval_id: str | None = None
    tool_names: list[str] | None = None
    server_names: dict[str, str] | None = None