Wecom Aibot Python SDK

https://github.com/wecomteam/wecom-aibot-python-sdk
Admin
The Wecom Aibot Python SDK provides core capabilities for intelligent robots on WeCom, including...
Tokens:29,696
Snippets:197
Trust Score:9
License:MIT
Update:3 weeks ago
Show doc for...
Tokens:
Raw
### Basic download example

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/wecom-api-client.md

Basic usage example: create a WeComApiClient with a custom timeout and download a file, printing the filename and data length.

```python
from aibot import WeComApiClient

api = WeComApiClient(logger, timeout=15000)
data, filename = await api.download_file_raw("https://example.com/pic.jpg")
print(filename, len(data))
```

--------------------------------

### reply_welcome example with error handling

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/wsclient.md

Example of using reply_welcome in an event handler for 'event.enter_chat'. It catches TimeoutError to handle the 5-second welcome window expiration.

```python
@ws_client.on("event.enter_chat")
async def on_enter_chat(frame):
    try:
        await ws_client.reply_welcome(frame, {
            "msgtype": "text",
            "text": {"content": "您好！有什么可以帮您的吗？"},
        })
    except TimeoutError:
        print("welcome too late (5s window expired)")
```

--------------------------------

### WSClient instantiation example

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/wsclient.md

Example of creating a WSClient instance with WSClientOptions. Requires bot_id and secret from WeCom admin. Sets max_reconnect_attempts to -1 for infinite reconnection.

```python
from aibot import WSClient, WSClientOptions

ws_client = WSClient(
    WSClientOptions(
        bot_id="bot-id-from-wecom-admin",
        secret="bot-secret-from-wecom-admin",
        max_reconnect_attempts=-1,   # reconnect forever
    )
)
```

--------------------------------

### Example: download_file in image message handler

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/wsclient.md

Example of using download_file within a message.image event handler to download and print the filename and size of an image.

```python
@ws_client.on("message.image")
async def on_image(frame):
    image = frame.get("body", {}).get("image", {})
    try:
        buffer, filename = await ws_client.download_file(image.get("url"), image.get("aeskey"))
        print(f"filename={filename}, size={len(buffer)}")
    except Exception as e:
        print(f"下载失败: {e}")
```

--------------------------------

### WSClient.run() usage example

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/wsclient.md

Example of using WSClient with the run() method. Creates a WSClient with WSClientOptions, registers an event handler for 'event.enter_chat', and calls run() which blocks. The handler is async and receives a frame argument.

```python
from aibot import WSClient, WSClientOptions

ws_client = WSClient(WSClientOptions(bot_id="...", secret="..."))

@ws_client.on("event.enter_chat")
async def on_enter_chat(frame): ...

ws_client.run()  # blocking
```

--------------------------------

### Example usage

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/message-handler.md

Shows how to embed the handler directly with a custom emitter and how to listen to emitted events in the WSClient integration.

```APIDOC
## Example usage

### Direct embedding

```python
from aibot import MessageHandler

handler = MessageHandler(logger)

def my_emit(event, *args):
    print(f"event={event}", args[0].get("body") if args else None)

handler.handle_frame(
    {"cmd": "aibot_event_callback", "headers": {"req_id": "r1"},
     "body": {"event": {"eventtype": "enter_chat"}}},
    MyEmitter(),  # any object with .emit(name, *args)
)
```

### WSClient integration

The normal integration is via WSClient, which passes itself as emitter:

```python
@ws_client.on("message.text")
async def on_text(frame):          # emitted by MessageHandler
    content = frame.get("body", {}).get("text", {}).get("content", "")
    print(content)
```

### Inbound body shapes

| msgtype | Body keys |
| --- | --- |
| text | body.text.content: str |
| image | body.image.url: str, body.image.aeskey: str |
| mixed | body.mixed.msg_item: List[{msgtype, text?, image?}] |
| voice | body.voice.content: str |
| file | body.file.url: str, body.file.aeskey: str |
| event callbacks | body.event.eventtype: str, body.event.event_key: str |
```

--------------------------------

### send_reply example

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/wsconnection-manager.md

Example usage of send_reply. It connects to the WeCom WebSocket, authenticates, and sends a reply with a request id and text body. The ack is awaited and a TimeoutError is caught if the ack does not arrive within 5 seconds.

```python
from aibot.ws import WsConnectionManager, WsCmd

manager = WsConnectionManager(logger, ws_url="wss://openws.work.weixin.qq.com")
manager.set_credentials("bot-id", "bot-secret")
manager.on_authenticated = lambda: print("authed")
await manager.connect()

try:
    ack = await manager.send_reply(
        "req_123",
        {"msgtype": "text", "text": {"content": "hi"}},
        cmd=WsCmd.RESPONSE,
    )
except TimeoutError:
    print("ack 5s timeout")
```

--------------------------------

### run()

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/wsclient.md

Creates and installs a new asyncio event loop, runs connect() to completion, then runs loop.run_forever(). On KeyboardInterrupt it calls disconnect(); loop.close() runs in finally. Blocking — intended to be called from a synchronous main().

```APIDOC
## `run` (convenience bootstrap)

### Description
Creates and installs a new asyncio event loop, runs `connect()` to completion, then runs `loop.run_forever()`. On `KeyboardInterrupt` it calls `disconnect()`; `loop.close()` runs in `finally`. Blocking — intended to be called from a synchronous `main()`.

### Method
run

### Signature
```python
def run(self) -> None
```

### Parameters
None

### Returns
None

### Example
```python
from aibot import WSClient, WSClientOptions

ws_client = WSClient(WSClientOptions(bot_id="...", secret="..."))

@ws_client.on("event.enter_chat")
async def on_enter_chat(frame): ...

ws_client.run()  # blocking
```

### Notes
- Blocking call, intended for synchronous `main()`.
- On `KeyboardInterrupt`, calls `disconnect()`.
- `loop.close()` runs in `finally`.
```

--------------------------------

### Basic decrypt_file usage

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/crypto-utils.md

Basic usage example. Import decrypt_file from aibot and call it with encrypted bytes and a base64-encoded AES key.

```python
from aibot import decrypt_file

plain = decrypt_file(encrypted_bytes, aes_key="base64...key...")
```

--------------------------------

### send_message usage example

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/wsclient.md

Example call to send_message with a markdown message body. Uses a user or group chat ID and a dict with msgtype markdown. The message content is a proactive push in Chinese.

```python
ack = await ws_client.send_message("user-or-chat-id", {
    "msgtype": "markdown",
    "markdown": {"content": "这是一条**主动推送**的消息"},
})
```

--------------------------------

### Direct MessageHandler usage example

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/message-handler.md

Example of embedding the MessageHandler directly. Creates a handler with a logger, defines a custom emit function, and calls handle_frame with a sample event callback frame. The emitter must support .emit(name, *args).

```python
from aibot import MessageHandler

handler = MessageHandler(logger)

def my_emit(event, *args):
    print(f"event={event}", args[0].get("body") if args else None)

handler.handle_frame(
    {"cmd": "aibot_event_callback", "headers": {"req_id": "r1"},
     "body": {"event": {"eventtype": "enter_chat"}}},
    MyEmitter(),  # any object with .emit(name, *args)
)
```

--------------------------------

### Connect to WebSocket and check connection status

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/wsclient.md

Opens the WebSocket connection and returns self for chaining. Sets _started = True and delegates to WsConnectionManager.connect(). If already started, logs a warning and returns self without reconnecting.

```python
async def main():
    await ws_client.connect()
    print("connected:", ws_client.is_connected)
```

--------------------------------

### reply_stream

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/wsclient.md

Documentation for the reply_stream async method, including parameters, return type, and usage examples.

```APIDOC
## reply_stream (streaming text reply)

### Description
Sends a streaming text reply through the WebSocket client. It builds a `{"msgtype": "stream", "stream": {"id", "finish", "content"[, "msg_item"][, "feedback"]}}` message and sends it via `reply()` with default `cmd` (`WsCmd.RESPONSE`).

### Method
`reply_stream` (async method)

### Signature
```python
async def reply_stream(
    self,
    frame: WsFrameHeaders,
    stream_id: str,
    content: str,
    finish: bool = False,
    msg_item: Optional[List[Dict[str, Any]]] = None,
    feedback: Optional[Dict[str, Any]] = None,
) -> WsFrame
```

### Parameters
- **frame** (`WsFrameHeaders`) - Required - Received frame; `headers.req_id` is passed through.
- **stream_id** (`str`) - Required - Stream message ID; the README/examples generate it with `generate_req_id("stream")`.
- **content** (`str`) - Required - Reply content; the SDK passes it verbatim, documented as supporting Markdown.
- **finish** (`bool`) - Optional, default `False` - `True` marks the final chunk of the stream.
- **msg_item** (`Optional[List[Dict[str, Any]]]`) - Optional, default `None` - Rich-text/图文混排 items; **only** attached to the body when `finish is True` and the list is non-empty.
- **feedback** (`Optional[Dict[str, Any]]`) - Optional, default `None` - Feedback info; attached only when truthy, documented as "only on the first reply".

### Return Value
- **WsFrame** - ack `WsFrame`.

### Throws
Same as `reply`.

### Example (basic + non-trivial)
```python
from aibot import generate_req_id

async def on_text(frame):
    stream_id = generate_req_id("stream")
    await ws_client.reply_stream(frame, stream_id, "正在思考中...", False)
    await asyncio.sleep(1)
    # finish=True with mixed text+image items
    await ws_client.reply_stream(
        frame, stream_id, "结果如下：", True,
        msg_item=[
            {"msgtype": "text", "text": {"content": "第 1 步"}},
            {"msgtype": "image", "image": {"url": "https://..."}},
        ],
        feedback={"feedback_type": "end", "is_helpful": True},
    )
```
```

--------------------------------

### run() method signature

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/wsclient.md

Signature of the run() method. It creates and installs a new asyncio event loop, runs connect() to completion, then runs loop.run_forever(). On KeyboardInterrupt it calls disconnect(); loop.close() runs in finally. Blocking — intended to be called from a synchronous main().

```python
def run(self) -> None
```

--------------------------------

### reply_stream usage example

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/wsclient.md

Shows how to use reply_stream for a basic streaming reply and a final chunk with mixed text and image items. The stream_id is generated with generate_req_id("stream").

```python
from aibot import generate_req_id

async def on_text(frame):
    stream_id = generate_req_id("stream")
    await ws_client.reply_stream(frame, stream_id, "正在思考中...", False)
    await asyncio.sleep(1)
    # finish=True with mixed text+image items
    await ws_client.reply_stream(
        frame, stream_id, "结果如下：", True,
        msg_item=[
            {"msgtype": "text", "text": {"content": "第 1 步"}},
            {"msgtype": "image", "image": {"url": "https://..."}},
        ],
        feedback={"feedback_type": "end", "is_helpful": True},
    )
```

--------------------------------

### Error handling for download_file_raw

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/wecom-api-client.md

Error handling example: catches aiohttp.ClientResponseError for HTTP error statuses and asyncio.TimeoutError for timeouts.

```python
try:
    data, filename = await api.download_file_raw(url)
except aiohttp.ClientResponseError as e:
    print(f"HTTP {e.status}: {e.message}")
except asyncio.TimeoutError:
    print("download timed out")
```

--------------------------------

### decrypt_file error handling

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/crypto-utils.md

Error handling example. Catches ValueError for invalid input (empty data or bad key) and RuntimeError for decryption failures (corrupted data or wrong key).

```python
try:
    plain = decrypt_file(encrypted, aes_key)
except ValueError as e:      # invalid input (empty data / bad key)
    print("bad input:", e)
except RuntimeError as e:    # decryption failure (corrupted data / wrong key)
    print("decrypt failed:", e)
```

--------------------------------

### WSClient constructor

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/wsclient.md

Creates a new WSClient instance with the provided configuration options. The constructor does not open a connection; you must call connect() or run() afterwards.

```APIDOC
## WSClient(options)

### Description
Creates a new WSClient instance. The constructor does not open a connection; call `await connect()` or `run()` to start.

### Method
Constructor

### Signature
```python
def __init__(self, options: WSClientOptions) -> None
```

### Parameters
- **options** (WSClientOptions) - Required - Configuration object. `bot_id` and `secret` are required fields; `logger` defaults to `DefaultLogger()`; `request_timeout`, `heartbeat_interval`, `reconnect_interval`, `max_reconnect_attempts`, `ws_url` are forwarded to the internal `WsConnectionManager` / `WeComApiClient`.

### Request Example
```python
from aibot import WSClient, WSClientOptions

ws_client = WSClient(
    WSClientOptions(
        bot_id="bot-id-from-wecom-admin",
        secret="bot-secret-from-wecom-admin",
        max_reconnect_attempts=-1,   # reconnect forever
    )
)
```

### Response
No return value.
```

--------------------------------

### reply_welcome

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/wsclient.md

Sends a welcome message in response to an `event.enter_chat` event. Must be called with the corresponding event's `req_id` and within 5 seconds of receiving the event callback.

```APIDOC
## reply_welcome

### Description
Sends a welcome message in response to an `event.enter_chat` event. Must be called with the corresponding event's `req_id` and within 5 seconds of receiving the event callback.

### Method
async def reply_welcome(self, frame: WsFrameHeaders, body: Dict[str, Any]) -> WsFrame

### Parameters
- **frame** (WsFrameHeaders) - Required - The frame of the triggering event (e.g. `event.enter_chat`); its `req_id` is required.
- **body** (Dict[str, Any]) - Required - Welcome body; documented/supported as `{"msgtype": "text", ...}` or `{"msgtype": "template_card", ...}`.

### Request Example
```python
await ws_client.reply_welcome(frame, {
    "msgtype": "text",
    "text": {"content": "您好！有什么可以帮您的吗？"},
})
```

### Response
Returns an ack `WsFrame`. Throws same as `reply` (e.g., `TimeoutError` if the 5-second window expires).
```

--------------------------------

### WeComApiClient.__init__

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/wecom-api-client.md

Constructor for WeComApiClient. Initializes the client with a logger and optional timeout in milliseconds.

```APIDOC
## WeComApiClient.__init__

### Description
Initializes a new WeComApiClient instance with the provided logger and timeout.

### Method
Constructor

### Signature
```python
def __init__(self, logger: Any, timeout: int = 10000):
```

### Parameters
- **logger** (Any) - Required - Log target (see Logger protocol).
- **timeout** (int) - Optional, default `10000` - Request timeout in milliseconds; converted to seconds internally.

### Return Value
None

### Example
```python
api = WeComApiClient(logger, timeout=15000)
```
```

--------------------------------

### WSClient Methods

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/wsconnection-manager.md

API documentation for the WSClient class methods set_credentials and connect.

```APIDOC
## set_credentials

### Description
Stores the bot credentials (bot_id and bot_secret) that will be sent in the authentication frame when `connect()` is next called. This method is called automatically by `WSClient.__init__` immediately after construction.

### Method
set_credentials

### Signature
```python
def set_credentials(self, bot_id: str, bot_secret: str) -> None
```

### Parameters
- **bot_id** (str) - Required - Bot ID, sent in the auth frame body.
- **bot_secret** (str) - Required - Bot secret, sent in the auth frame body.

### Returns
None
```

```APIDOC
## connect

### Description
Establishes the WebSocket connection to the server. It performs cleanup of any previous socket, connects using `websockets.connect` with the library's own application-level heartbeats (protocol-level ping is disabled), resets reconnect counters, fires the `on_connected` callback, sends the authentication frame, and starts the receive loop task. On failure, it logs the error, fires `on_error`, and schedules a reconnect if possible; it does not raise to the caller when reconnection is still possible.

### Method
connect

### Signature
```python
async def connect(self) -> None
```

### Parameters
None

### Returns
None

### Auth Frame Payload
```json
{
  "cmd": "aibot_subscribe",
  "headers": {"req_id": "aibot_subscribe_<ts>_<rand>"},
  "body": {"bot_id": "<bot_id>", "secret": "<bot_secret>"}
}
```
```

--------------------------------

### Create WSClientOptions with all constructor options

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/configuration.md

Constructs a WSClientOptions dataclass instance with all available constructor options. bot_id and secret are required; other fields have defaults. reconnect_interval uses exponential backoff with a 30-second cap, max_reconnect_attempts of -1 means infinite reconnection, request_timeout is converted to seconds for aiohttp, and an empty ws_url falls back to the default WebSocket URL.

```python
from aibot import WSClientOptions

options = WSClientOptions(
    bot_id="...",
    secret="...",
    reconnect_interval=1000,
    max_reconnect_attempts=10,
    heartbeat_interval=30000,
    request_timeout=10000,
    ws_url="",
    logger=None,
)
```

--------------------------------

### reply_welcome method signature

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/wsclient.md

Signature for the reply_welcome method. It takes a WsFrameHeaders frame and a body dictionary, returning a WsFrame. The frame's req_id is required, and the body supports text or template_card message types.

```python
async def reply_welcome(
    self,
    frame: WsFrameHeaders,
    body: Dict[str, Any],
) -> WsFrame
```

--------------------------------

### Using DefaultLogger with WSClient

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/default-logger.md

Passes a DefaultLogger instance to WSClientOptions.logger, demonstrating that DefaultLogger can be used wherever a Logger is accepted.

```python
from aibot import WSClient, WSClientOptions, DefaultLogger

ws_client = WSClient(WSClientOptions(bot_id="...", secret="...", logger=DefaultLogger("MyBot")))
```

--------------------------------

### DefaultLogger.__init__

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/default-logger.md

Constructor for the DefaultLogger class. Accepts an optional prefix string used in log output formatting.

```APIDOC
## DefaultLogger.__init__

### Description
Initializes a new DefaultLogger instance with an optional prefix that is embedded in every log line between the timestamp and the level tag.

### Method
Constructor

### Signature
```python
def __init__(self, prefix: str = "AiBotSDK")
```

### Parameters
- **prefix** (str) - Optional - Default: `"AiBotSDK"` - Prefix embedded in every line between the timestamp and the level, e.g. `[...] [AiBotSDK] [INFO] ...`.

### Return Value
None
```

--------------------------------

### MessageHandler.__init__

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/message-handler.md

Constructor for MessageHandler, which stores a logger for output.

```APIDOC
## Constructor

### Description
Initializes a new MessageHandler instance with the provided logger.

### Signature
`def __init__(self, logger: Any) -> None`

### Parameters
- **logger** (Any) - Required - Log target used for warn/error/debug output.

### Return Value
None
```

--------------------------------

### WSClient constructor signature

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/wsclient.md

Constructor signature for WSClient. Takes a WSClientOptions object. Does not open a connection; call await connect() or run() after construction.

```python
def __init__(self, options: WSClientOptions) -> None
```

--------------------------------

### Basic DefaultLogger usage

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/default-logger.md

Creates a DefaultLogger with a custom prefix and logs an info and an error message. The output goes to sys.stderr with the format shown in the comments.

```python
from aibot import DefaultLogger

logger = DefaultLogger(prefix="MyBot")
logger.info("Connected")           # -> [...] [MyBot] [INFO] Connected
logger.error("Auth failed", 401)  # -> [...] [MyBot] [ERROR] Auth failed 401
```

--------------------------------

### WSClient Methods

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/wsclient.md

Documentation for the WSClient class methods: connect, disconnect, and reply. These are the public methods users can call directly on the WebSocket client.

```APIDOC
## `connect`

```python
async def connect(self) -> "WSClient"
```

| Parameter | Type | Required | Default | Description |
|-----------|------|----------|---------|-------------|
| — | — | — | — | No parameters. |

Opens the WebSocket connection and returns `self` for chaining. Internally sets `self._started = True` and delegates to `WsConnectionManager.connect()`, which performs: socket open → `on_connected` callback → authentication frame (`aibot_subscribe` with `bot_id`/`secret`) → receive loop. If already started, logs a warning (`"Client already connected"`) and returns `self` immediately without reconnecting.

**Return**: the `WSClient` instance (chainable).

**Throws**: propagates `WsConnectionManager.connect()` failures as the low-level `websockets` exception; the manager catches them and schedules a reconnect instead of raising in most cases.

### Example (basic)

```python
async def main():
    await ws_client.connect()
    print("connected:", ws_client.is_connected)
```
```

```APIDOC
## `disconnect`

```python
def disconnect(self) -> None
```

Synchronous. If not started, logs a warning (`"Client not connected"`) and returns. Otherwise sets `_started = False` and calls `WsConnectionManager.disconnect()`, which marks the close as manual (so no reconnect is scheduled), stops heartbeat, fails all pending replies with `RuntimeError("Connection manually closed, ...")`, and schedules an async close (`code=1000`, reason `"Manual disconnect"`). Does not await the async close.
```

```APIDOC
## `reply` (generic reply)

```python
async def reply(
    self,
    frame: WsFrameHeaders,
    body: Dict[str, Any],
    cmd: Optional[str] = None,
) -> WsFrame
```

| Parameter | Type | Required | Default | Description |
|-----------|------|----------|---------|-------------|
| `frame` | `WsFrameHeaders` | ✅ | — | The received frame (or any dict with `headers.req_id`); only `headers.req_id` is used. |
| `body` | `Dict[str, Any]` | ✅ | — | Reply message body, e.g. `{"msgtype": "stream", "stream": {...}}`. |
| `cmd` | `Optional[str]` | — | `None` (→ `WsCmd.RESPONSE` = `"aibot_respond_msg"`) | Wire command name. The expression is `cmd or WsCmd.RESPONSE`, so `None` or `""` both become `WsCmd.RESPONSE`. |

**Return**: the ack `WsFrame` received from the server (resolved only after the serialized per-`req_id` queue has sent the message and the ack frame with `errcode == 0` arrived). Note: the frame is a `Dict[str, Any]` alias.

**Throws**: `TimeoutError` (ack not received within 5 s), `RuntimeError` (ack `errcode != 0`, send failure, queue overflow, connection closed).

### Example

```python
ack = await ws_client.reply(
    frame,
    {"msgtype": "text", "text": {"content": "plain text reply"}},
    cmd="aibot_respond_msg",
)
```
```

--------------------------------

### connect method signature

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/wsconnection-manager.md

Defines the async method that establishes the WebSocket connection. On failure it logs, fires on_error(e), and awaits _schedule_reconnect(); it does not raise to the caller when reconnect is still possible.

```python
async def connect(self) -> None
```

--------------------------------

### WsConnectionManager constructor

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/wsconnection-manager.md

Constructor signature for WsConnectionManager. Parameters: logger (required), heartbeat_interval (default 30000 ms), reconnect_base_delay (default 1000 ms, exponential backoff capped at 30000 ms), max_reconnect_attempts (default 10, -1 for indefinite), ws_url (defaults to DEFAULT_WS_URL).

```python
def __init__(
    self,
    logger: Any,
    heartbeat_interval: int = 30000,
    reconnect_base_delay: int = 1000,
    max_reconnect_attempts: int = 10,
    ws_url: Optional[str] = None,
)
```

--------------------------------

### download_file_raw method signature

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/wecom-api-client.md

Method signature for download_file_raw. It downloads the file from the given URL and returns a tuple of (file content bytes, filename).

```python
async def download_file_raw(self, url: str) -> Tuple[bytes, Optional[str]]
```

--------------------------------

### reply_template_card

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/wsclient.md

Sends a template card message via `reply()` with default `cmd`. The card content is specified by `template_card`, and optional `feedback` is merged into the card copy as a new `"feedback"` key.

```APIDOC
## reply_template_card

### Description
Sends a template card message via `reply()` with default `cmd`. The card content is specified by `template_card`, and optional `feedback` is merged into the card copy as a new `"feedback"` key.

### Method
async def reply_template_card(self, frame: WsFrameHeaders, template_card: Dict[str, Any], feedback: Optional[Dict[str, Any]] = None) -> WsFrame

### Parameters
- **frame** (WsFrameHeaders) - Required - Received frame; `req_id` passed through.
- **template_card** (Dict[str, Any]) - Required - Card content; `card_type` values are enumerated by `TemplateCardType` (`text_notice`, `news_notice`, `button_interaction`, `vote_interaction`, `multiple_interaction`).
- **feedback** (Optional[Dict[str, Any]]) - Optional - When truthy, merged into the card copy as a new `"feedback"` key (the source `template_card` dict is not mutated).

### Request Example
```python
await ws_client.reply_template_card(frame, {
    "card_type": "text_notice",
    "source": {"icon_url": "...", "desc": "..."},
    "main_title": {"title": "..."},
    "sub_title_text": "...",
    "card_action": {"type": 1, "url": "..."}
})
```

### Response
Returns an ack `WsFrame`. Throws same as `reply`.
```

--------------------------------

### set_credentials method signature

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/wsconnection-manager.md

Defines the method for storing bot credentials. The bot_id and bot_secret are sent in the auth frame body when connect() is next called. WSClient.__init__ calls this immediately after construction.

```python
def set_credentials(self, bot_id: str, bot_secret: str) -> None
```

--------------------------------

### Event emission summary

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/wsclient.md

WSClient._setup_ws_events wires WsConnectionManager callbacks to self.emit, emitting events such as connected, authenticated, disconnected, reconnecting, error, and message.

```APIDOC
## Event emission summary

### Description
`WSClient._setup_ws_events` (`client.py:60-79`) wires `WsConnectionManager` callbacks to `self.emit`.

### Events

| Event | Emitted value |
| --- | --- |
| `connected` | — |
| `authenticated` | — (after logging `"Authenticated"`) |
| `disconnected` | `reason: str` |
| `reconnecting` | `attempt: int` |
| `error` | `error: Exception` |
| `message` → `MessageHandler.handle_frame(frame, self)` | dispatches `message.*` / `event.*` events (see [MessageHandler](message-handler.md)) |

### Source reference
`aibot/client.py` (constructor `client.py:29`, `connect` `client.py:81`, `disconnect` `client.py:100`, `reply` `client.py:111`, `reply_stream` `client.py:129`, `reply_welcome` `client.py:171`, `reply_template_card` `client.py:188`, `reply_stream_with_card` `client.py:209`, `update_template_card` `client.py:260`, `send_message` `client.py:286`, `download_file` `client.py:304`, `is_connected` `client.py:336`, `api` `client.py:341`, `run` `client.py:345`).
```

--------------------------------

### update_template_card method signature

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/wsclient.md

Defines the update_template_card async method. Requires the frame from the template_card_event callback and a replacement template_card dict with matching task_id. Optional userids list replaces specific user copies. Must be called within 5 seconds of the event callback.

```python
async def update_template_card(
    self,
    frame: WsFrameHeaders,
    template_card: Dict[str, Any],
    userids: Optional[List[str]] = None,
) -> WsFrame
```

--------------------------------

### DefaultLogger constructor signature

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/default-logger.md

Constructor for DefaultLogger with an optional prefix parameter that defaults to "AiBotSDK". The prefix is embedded in every log line between the timestamp and the level.

```python
def __init__(self, prefix: str = "AiBotSDK"):
```

--------------------------------

### api

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/wsclient.md

Provides access to the internal WeComApiClient instance for advanced use cases.

```APIDOC
## `api`

```python
@property
def api(self) -> WeComApiClient
```

Returns the internal `WeComApiClient` instance "for advanced use" (e.g. calling `download_file_raw` directly). See [WeComApiClient](wecom-api-client.md).
```

--------------------------------

### WSClient event handler for message.text

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/message-handler.md

Shows the normal integration via WSClient, which passes itself as the emitter. This async handler listens for the 'message.text' event emitted by MessageHandler and prints the text content from the frame body.

```python
@ws_client.on("message.text")
async def on_text(frame):          # emitted by MessageHandler
    content = frame.get("body", {}).get("text", {}).get("content", "")
    print(content)
```

--------------------------------

### Basic usage of generate_req_id

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/utils.md

Shows how to generate a stream ID and use it in a reply_stream call. The prefix is conventionally the command name, e.g., "stream".

```python
from aibot import generate_req_id

stream_id = generate_req_id("stream")   # e.g. "stream_1720000000000_4f9a2c81"
await ws_client.reply_stream(frame, stream_id, "hello", True)
```

--------------------------------

### WSClientOptions dataclass definition

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/types.md

Dataclass for WSClient configuration options. Requires bot_id and secret; defaults include reconnect_interval=1000, max_reconnect_attempts=10, heartbeat_interval=30000, request_timeout=10000, ws_url="", and logger=None. Accepted by WSClient.__init__(options).

```python
@dataclass
class WSClientOptions:
    """WSClient 配置选项"""

    bot_id: str
    secret: str
    reconnect_interval: int = 1000
    max_reconnect_attempts: int = 10
    heartbeat_interval: int = 30000
    request_timeout: int = 10000
    ws_url: str = ""
    logger: Optional[Any] = None
```

--------------------------------

### DefaultLogger.info

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/default-logger.md

Logs a message at INFO level to sys.stderr. The output includes a UTC ISO8601 timestamp, the configured prefix, and the INFO level tag.

```APIDOC
## DefaultLogger.info

### Description
Prints an info-level log message to sys.stderr with the format `[{iso8601 UTC}] [{prefix}] [INFO] {message} {*args}`.

### Method
Method

### Signature
```python
def info(self, message: str, *args: object) -> None
```

### Parameters
- **message** (str) - Required - The log message.
- **args** (object) - Optional - Additional arguments passed to print, matching built-in print(*args) semantics.

### Return Value
None
```

--------------------------------

### download_file method signature

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/wsclient.md

Signature of the download_file method. It takes a URL and an optional AES key, returning a tuple of decrypted bytes and filename.

```python
async def download_file(
    self, url: str, aes_key: Optional[str] = None
) -> Tuple[bytes, Optional[str]]
```

--------------------------------

### decrypt_file used by WSClient.download_file

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/crypto-utils.md

Usage within WSClient.download_file. Downloads raw encrypted data and then decrypts it using decrypt_file.

```python
encrypted_data, filename = await ws_client.download_file_raw(url)  # via api client
decrypted_data = decrypt_file(encrypted_data, aes_key)
```

--------------------------------

### WeComApiClient.download_file_raw

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/wecom-api-client.md

Asynchronously downloads a file from the given URL and returns the raw bytes and filename. Raises exceptions on HTTP errors or network failures.

```APIDOC
## WeComApiClient.download_file_raw

### Description
Downloads a file from the provided absolute URL using an HTTP GET request. Parses the filename from the Content-Disposition header and returns the file content as bytes along with the filename (or None if not available).

### Method
async def

### Signature
```python
async def download_file_raw(self, url: str) -> Tuple[bytes, Optional[str]]:
```

### Parameters
- **url** (str) - Required - Absolute file URL to download.

### Behavior
1. Logs "Downloading file..." at info level.
2. Opens an aiohttp ClientSession with the configured timeout and TLS context.
3. Sends a GET request to the URL.
4. Calls `response.raise_for_status()` to raise on non-2xx status codes.
5. Reads the raw response body as bytes.
6. Parses the filename from the `Content-Disposition` header, preferring RFC 5987 `filename*=UTF-8''<value>` and falling back to `filename="..."` or `filename=...`.
7. Logs "File downloaded successfully" at info level.

### Return Value
- **Tuple[bytes, Optional[str]]** - A tuple containing the file content as bytes and the filename as a string, or None if the header is missing or cannot be parsed.

### Throws
- **aiohttp.ClientError** - For network errors.
- **asyncio.TimeoutError** - If the request times out.
- **aiohttp.ClientResponseError** - For HTTP error status codes (non-2xx).

All exceptions are logged at error level and re-raised to the caller.

### Example (basic)
```python
from aibot import WeComApiClient

api = WeComApiClient(logger, timeout=15000)
data, filename = await api.download_file_raw("https://example.com/pic.jpg")
print(filename, len(data))
```

### Example (error handling)
```python
try:
    data, filename = await api.download_file_raw(url)
except aiohttp.ClientResponseError as e:
    print(f"HTTP {e.status}: {e.message}")
except asyncio.TimeoutError:
    print("download timed out")
```
```

--------------------------------

### Log line format

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/default-logger.md

Format of each log line produced by DefaultLogger methods: timestamp, prefix, level, message, and optional arguments.

```text
[{iso8601 UTC}] [{prefix}] [{LEVEL}] {message} {*args}
```

--------------------------------

### reply_stream_with_card method signature

Source: https://github.com/wecomteam/wecom-aibot-python-sdk/blob/master/_autodocs/api-reference/wsclient.md

Defines the reply_stream_with_card method that sends a combined stream and template card message. The template_card is omitted from the body when None, and card_feedback is merged into template_card as 'feedback' when truthy. Returns an ack WsFrame and throws the same exceptions as reply.

```python
async def reply_stream_with_card(
    self,
    frame: WsFrameHeaders,
    stream_id: str,
    content: str,
    finish: bool = False,
    msg_item: Optional[List[Dict[str, Any]]] = None,
    stream_feedback: Optional[Dict[str, Any]] = None,
    template_card: Optional[Dict[str, Any]] = None,
    card_feedback: Optional[Dict[str, Any]] = None,
) -> WsFrame
```
Tokens:
Raw