内容
# aiogram-mcp
[](https://github.com/Py2755/aiogram-mcp/actions/workflows/ci.yml)
[](https://www.python.org/downloads/)
[](LICENSE)
[](https://pypi.org/project/aiogram-mcp/)
[](https://registry.modelcontextprotocol.io/)
<!-- mcp-name: io.github.Py2755/aiogram-mcp -->
**将您的 Telegram 机器人连接到 AI 代理,借助模型上下文协议。**
`aiogram-mcp` 将任何 [aiogram](https://github.com/aiogram/aiogram) 机器人转变为 [MCP](https://modelcontextprotocol.io/) 服务器。这样,Claude Desktop 等 AI 客户端就可以通过您的现有机器人发送消息、读取聊天记录、构建交互式菜单,并实时响应事件,无需重写任何处理程序。
## 为什么选择 aiogram-mcp?
大多数 Telegram MCP 服务器只是薄薄的封装,带有 3-5 个工具。而 `aiogram-mcp` 则更进一步:
- **30 个工具** — 消息传递、富媒体、审核、交互式键盘、事件订阅、广播
- **7 种资源** — 机器人信息、配置、聊天列表、消息历史、事件队列、文件元数据、审核日志
- **3 种提示** — 预制好的审核、公告和用户报告工作流
- **结构化输出** — 每个工具都返回带有 `outputSchema` 的类型化 Pydantic 模型,以便于程序化解析
- **实时事件** — 机器人通过 MCP 通知将 Telegram 事件推送给 AI 客户端(无需轮询)
- **交互式消息** — AI 代理创建内联键盘菜单、处理按钮按下、编辑消息
- **速率限制** — 内置令牌桶防止 Telegram 429 错误
- **权限级别** — 限制 AI 代理的访问权限为只读、消息传递、审核或完全管理员权限
- **审核日志** — 跟踪每个工具调用的时间戳和参数
- **零改写** — 在现有的机器人中添加 5 行代码即可使用,保持所有处理程序不变
## 工作原理
```
Telegram 用户 您的 aiogram 机器人 AI 代理(Claude Desktop)
| | |
| 发送消息、点击按钮 | |
| --------------------------> | |
| | MCP 服务器(stdio 或 SSE) |
| | <-------------------------> |
| | 工具 / 资源 / 事件 |
| | |
| 机器人回复、显示菜单 | send_message, edit, ban |
| <-------------------------- | <--------------------------- |
```
机器人正常为 Telegram 用户运行。MCP 服务器与其一同运行,通过工具和资源为 AI 代理提供对机器人的访问。
## 安装
```bash
pip install aiogram-mcp
```
需要 Python 3.10+ 和 aiogram 3.20+。
## 快速入门
### 1. 将 aiogram-mcp 添加到您的机器人
```python
import asyncio
from aiogram import Bot, Dispatcher
from aiogram_mcp import AiogramMCP, EventManager, MCPMiddleware
bot = Bot(token="YOUR_BOT_TOKEN")
dp = Dispatcher()
# 中间件跟踪聊天、用户、消息历史和事件
event_manager = EventManager()
middleware = MCPMiddleware(event_manager=event_manager)
dp.message.middleware(middleware)
dp.callback_query.middleware(middleware) # 用于交互式按钮
# 在此处注册您的普通处理程序
# @dp.message(...)
# async def my_handler(message): ...
# 创建 MCP 服务器
mcp = AiogramMCP(
bot=bot,
dp=dp,
name="my-bot",
middleware=middleware,
event_manager=event_manager,
allowed_chat_ids=[123456789], # 可选:限制 AI 可以访问的聊天
)
async def main():
await mcp.run_alongside_bot(transport="stdio")
asyncio.run(main())
```
### 2. 连接 Claude Desktop
将以下内容添加到您的 `claude_desktop_config.json` 中:
```json
{
"mcpServers": {
"my-telegram-bot": {
"command": "python",
"args": ["path/to/your/bot.py"],
"env": {
"BOT_TOKEN": "123456:ABC-DEF..."
}
}
}
}
```
现在,Claude 可以在您的 Telegram 机器人中发送消息、读取历史记录、创建按钮菜单,并响应事件。
## 内置工具
### 消息传递(5 个工具)
| 工具 | 描述 |
|------|-------------|
| `send_message` | 发送带有 HTML/Markdown 格式的文本 |
| `send_photo` | 通过 URL 发送照片,附带可选标题 |
| `forward_message` | 在聊天之间转发消息 |
| `delete_message` | 删除消息 |
| `pin_message` | 在聊天中固定消息 |
### 交互式消息(3 个工具)
| 工具 | 描述 |
|------|-------------|
| `send_interactive_message` | 发送带有内联键盘按钮的消息(回调或 URL) |
| `edit_message` | 编辑现有消息的文本和/或键盘 |
| `answer_callback_query` | 通过 toast 或 alert 响应按钮按下 |
### 用户(3 个工具)
| 工具 | 描述 |
|------|-------------|
| `get_bot_info` | 获取机器人元数据(用户名、功能) |
| `get_chat_member_info` | 获取用户在聊天中的角色和个人资料 |
| `get_user_profile_photos` | 获取用户的个人资料照片 |
### 聊天(6 个工具)
| 工具 | 描述 |
|------|-------------|
| `get_chat_info` | 获取聊天元数据(标题、类型、描述) |
| `get_chat_members_count` | 获取聊天中的成员数量 |
| `ban_user` | 禁止用户(永久或临时) |
| `unban_user` | 解禁用户 |
| `set_chat_title` | 更改聊天标题 |
| `set_chat_description` | 更改聊天描述 |
### 富媒体(10 个工具)
| 工具 | 描述 |
|------|-------------|
| `send_document` | 通过 URL 发送文件/文档,附带可选标题 |
| `send_voice` | 通过 URL 发送语音消息 |
| `send_video` | 通过 URL 发送视频,附带可选标题 |
| `send_animation` | 通过 URL 发送 GIF/动画 |
| `send_audio` | 通过 URL 发送音频/音乐,附带表演者和标题 |
| `send_sticker` | 通过 file_id 或 URL 发送贴纸 |
| `send_video_note` | 通过 URL 发送圆形视频笔记 |
| `send_contact` | 发送带有电话号码和姓名的联系人 |
| `send_location` | 发送地理位置图钉 |
| `send_poll` | 创建带有多个选项的投票 |
### 事件(2 个工具)
| 工具 | 描述 |
|------|-------------|
| `subscribe_events` | 订阅带有聊天/类型过滤器的实时事件 |
| `unsubscribe_events` | 删除订阅 |
### 广播(1 个工具,需选择加入)
| 工具 | 描述 |
|------|-------------|
| `broadcast` | 向多个聊天发送消息(需要 `enable_broadcast=True`) |
## MCP 资源
AI 代理无需调用工具即可访问的只读数据:
| URI | 描述 |
|-----|-------------|
| `telegram://bot/info` | 机器人用户名、ID 和功能 |
| `telegram://config` | 服务器名称和允许的聊天 ID |
| `telegram://chats` | 带有元数据的活动聊天列表 |
| `telegram://chats/{chat_id}/history` | 聊天中的最后 50 条消息 |
| `telegram://events/queue` | 带有自动递增 ID 的事件队列 |
| `telegram://files/{file_id}` | 文件元数据(大小、路径、唯一 ID) |
| `telegram://audit/log` | 工具调用的审核日志(需选择加入) |
## MCP 提示
为 AI 代理提供结构化上下文的预制工作流:
| 提示 | 参数 | 功能 |
|--------|-----------|-------------|
| `moderation_prompt` | `chat_id`, `user_id`, `reason` | 获取用户信息 + 消息历史,建议警告/禁言/封禁 |
| `announcement_prompt` | `topic`, `audience?`, `tone?` | 起草格式化的 Telegram 公告 |
| `user_report_prompt` | `chat_id`, `user_id` | 编制完整的用户活动报告 |
## 实时事件流
AI 代理无需轮询。机器人会自动推送事件:
```
Telegram 消息到达
→ MCPMiddleware 捕获它
→ EventManager 存储它(类型:“message”,“command”或“callback_query”)
→ MCP 通知发送到订阅的客户端
→ AI 代理读取 telegram://events/queue
```
AI 代理一次性调用 `subscribe_events`,然后在新事件与过滤器匹配时接收推送通知。
## 交互式消息
AI 代理可以构建完整的交互式 UI,在 Telegram 中创建菜单、确认、 多步骤向导:
**AI 代理发送带有按钮的消息:**
```
┌─────────────────────────┐
│ 确认部署? │
│ │
│ [✅ 是] [❌ 否] │
│ [📖 查看文档] │
└─────────────────────────┘
```
**用户点击按钮 → 事件出现在队列中 → AI 代理反应:**
```
┌─────────────────────────┐
│ ✅ 已部署! │
│ │
│ [📋 查看日志] │
└─────────────────────────┘
```
机器人需要 `dp.callback_query.middleware(middleware)` 来捕获按钮按下。
## 安全控制
```python
mcp = AiogramMCP(
bot=bot,
dp=dp,
allowed_chat_ids=[123456789, -1001234567890], # 限制 AI 访问
enable_broadcast=True, # 选择加入广播工具
max_broadcast_recipients=500, # 安全限制
)
```
- **`allowed_chat_ids`** — AI 只能与列出的聊天互动。默认:所有聊天。
- **`enable_broadcast`** — 默认情况下,广播工具作为安全措施被禁用。
- **`max_broadcast_recipients`** — 限制单个广播中的聊天数量。
## 高级配置
### 速率限制
```python
mcp = AiogramMCP(
bot=bot, dp=dp,
rate_limit=30, # 请求/秒(默认),0 禁用
)
```
内置令牌桶速率限制器可防止 Telegram 429 错误。所有发出的 API 调用都会自动控制速率。
### 权限级别
```python
mcp = AiogramMCP(
bot=bot, dp=dp,
permission_level="messaging", # 仅限读取 + 消息传递工具
)
```
| 级别 | 访问权限 |
|-------|--------|
| `read` | 机器人信息、聊天信息、用户个人资料 |
| `messaging` | 读取 + 发送消息、照片、媒体、交互式消息 |
| `moderation` | 消息传递 + 删除、固定、封禁、解除封禁、聊天设置 |
| `admin` | 包括广播和事件订阅在内的完全访问权限 |
### 审核日志
```python
mcp = AiogramMCP(
bot=bot, dp=dp,
enable_audit=True,
audit_log_size=1000,
)
```
每个工具调用都会被记录。通过 `telegram://audit/log` 资源访问。
## 示例
| 示例 | 传输方式 | 功能 |
|---------|-----------|----------|
| [basic_bot.py](examples/basic_bot.py) | stdio | 带有中间件、事件和回调跟踪的完整设置 |
| [incident_alert_bot.py](examples/incident_alert_bot.py) | SSE | 启用广播的 Ops 机器人,用于事件通知 |
## 开发
```bash
git clone https://github.com/Py2755/aiogram-mcp.git
cd aiogram-mcp
pip install -e ".[dev]"
pytest -v # ~228 测试
ruff check aiogram_mcp tests examples
mypy aiogram_mcp # 严格模式
```
## 许可证
MIT。参见 [LICENSE](LICENSE)。