miti-agent-sdk

Connect third-party agents (Hermes, OpenClaw, Claude Code, etc.) to the Miti IM platform.

Installation

pip install miti-agent-sdk

Quick Start

from miti_agent_sdk import MitiAgent

agent = MitiAgent(
    app_id="cli_xxx",
    app_secret="secret_xxx",
)

# 单聊消息
@agent.on_message
async def handle(event):
    user_text = event.event.message.content.get("text", "")
    await event.reply(f"收到：{user_text}")

# 群 @ 消息
@agent.on_group_at
async def handle_group(event):
    text = event.event.message.content.get("text", "")
    await event.reply(f"群收到：{text}")

agent.run()

SDK 内部自动处理：Token 获取/刷新、WebSocket 保活、指数退避重连。

过载回调

普通消息处理达到并发上限时，可注册 on_overload 返回明确的繁忙响应：

@agent.on_overload
async def handle_overload(event):
    await event.reply("系统繁忙，请稍后重试", use_markdown=False)

过载回调使用独立且严格有界的执行槽。回调必须快速发送 busy/failed 响应并立即返回，不应执行原消息的耗时工作。若过载槽本身也已满，SDK 会记录 warning 并执行硬过载丢弃，以保证内存和任务数量有界。

取消与流式可靠性

assistant.cancel 使用独立的有界队列，不会阻塞 WebSocket 消息读取：

@agent.on_cancel
async def handle_cancel(event):
    # 这里只设置取消标记，不要等待模型调用或执行网络 I/O。
    cancel_flags[event.event.run_id].set()

取消回调必须快速返回。队列达到硬上限时，SDK 会记录 warning 并丢弃后续 取消事件，以避免任务和内存无界增长。

流式回复完成时，可同时提交完整回答和完整 reasoning：

await event.reply_completed(
    answer_markdown,
    seq=seq,
    run_id=run_id,
    assistant_msg_id=assistant_msg_id,
    reasoning=reasoning,
)

流式增量帧不等待 ACK；completed / failed 终态帧会在 ACK 超时后使用 相同的 event_id 重试。Gateway 服务端必须按 event_id 幂等去重，避免 终态消息重复投递。

环境配置

base_url 按以下优先级决定（先找到先用）：

优先级	来源	示例
1（最高）	构造函数 base_url= 参数	MitiAgent(base_url="https://...")
2	环境变量 MITI_API_BASE_URL	export MITI_API_BASE_URL="https://..."
3（默认）	正式环境	https://www.miti.chat/chat

app_id / app_secret 可在 Miti App 的 我的 → 设置 → 智能体管理 → 连接智能体 创建后获取。

大多数场景下不需要配置任何东西，SDK 自动连接正式环境。

本地开发调试时，通过环境变量指向本地 appserver 即可，无需改业务代码：

export MITI_API_BASE_URL="http://localhost:10006/chat"

也可以在初始化时显式传入：

agent = MitiAgent(
    app_id="cli_xxx",
    app_secret="secret_xxx",
    base_url="http://localhost:10006/chat",
)

拉取历史记录

@agent.on_message
async def handle(event):
    history = await agent.get_history(
        event.event.message.conversation_id, limit=10
    )
    # history: list[HistoryMessage]
    # 每条消息包含 .role、.content 和可选的 .sender_name

群聊中可以使用当前消息作为锚点，只获取它之前的历史：

@agent.on_group_at
async def handle_group_at(event):
    history = await agent.get_history(
        event.event.message.conversation_id,
        limit=20,
        before_message_id=event.event.message.message_id,
    )

    # 群发送者名称；旧事件没有该字段时为空字符串
    sender_name = event.event.sender.name

    for message in history:
        # 群历史发送者名称；旧响应没有该字段时为空字符串
        print(message.sender_name, message.content)

before_message_id 为空时不会发送该查询参数，行为与旧版本一致。

日志 & 问题排查

SDK 使用 Python 标准 logging 模块，logger 层级为 miti_agent_sdk.*。

快速开启日志

import miti_agent_sdk

# INFO 级别：token 获取/刷新、WebSocket 连接/断开/重连、事件收发、消息发送
miti_agent_sdk.enable_logging()

# DEBUG 级别：额外输出每次 HTTP 请求 URL/参数、自动刷新 sleep 倒计时
miti_agent_sdk.enable_logging(logging.DEBUG)

日志输出到 stderr（默认），格式示例：

2026-06-18 18:00:01 [miti_agent_sdk.auth] INFO token acquired, expires_in=7200s
2026-06-18 18:00:01 [miti_agent_sdk.gateway] INFO connecting to wss://www.miti.chat/chat/agent/gateway
2026-06-18 18:00:02 [miti_agent_sdk.gateway] INFO gateway connected
2026-06-18 18:00:05 [miti_agent_sdk.gateway] INFO event received: type=im.message.receive id=evt-xxx
2026-06-18 18:00:05 [miti_agent_sdk.message] INFO send_message success: target=user-abc message_id=msg-xxx

日志存储

SDK 本身不写日志文件。日志的最终存储取决于宿主进程：

方式	配置	适用场景
控制台	miti_agent_sdk.enable_logging()	本地开发调试
文件	logging.basicConfig(filename="agent.log", level=logging.INFO)	单机部署
结构化日志	宿主框架的 JSON handler（如 python-json-logger）	生产环境 / 接入 ELK 等
容器 stdout	不做额外配置，enable_logging() 输出到 stderr 即可	Docker / K8s

排查常见问题

现象	排查方向	关键日志
启动后无反应	Token 获取失败	auth token API error: code=... msg=...
连接后收不到消息	WebSocket 断开或无事件推送	gateway connection lost / no handler for event_type
回复消息失败	权限不足或目标不存在	send_message API error: code=10012 msg=no permission
Token 过期后断连	自动刷新失败	auto-refresh failed, will retry in ...s
WebSocket 频繁重连	网络不稳或服务端问题	reconnecting in Xs（观察退避时间是否递增）

安全注意事项

日志安全

SDK 自身不会在日志中输出 app_secret 或 access_token。但如果你开启了 aiohttp 的 trace/debug 日志（如 AIOHTTP_TRACE=1），HTTP 请求体会被底层框架 打印出来，其中包含 app_secret。生产环境请勿开启 aiohttp trace 日志。

凭证内存存储

app_secret 和 access_token 在 SDK 运行期间以明文存储在进程内存中。 这是 Python SDK 的通用模式（与飞书 SDK、Slack SDK 一致），无法避免。 请确保：

• 不要在多租户共享环境中运行 agent 进程
• 不要将进程 memory dump 输出到不安全的位置
• 进程退出后凭证自动释放，无需手动清理

TLS 要求

SDK 强制要求 base_url 使用 https://（对应 WebSocket wss://）。 本地开发调试时 localhost / 127.0.0.1 可使用 http://，其他地址会被拒绝。

Requirements

• Python 3.9+
• aiohttp >= 3.9
• websockets >= 12.0