图形-mcp
   ](https://pypi.org/project/aiogram-mcp/) 
通过模型上下文协议将您的Telegram机器人连接到AI代理。
为什么选择aiogram mcp?
大多数Telegram MCP服务器都是带有3-5个工具的瘦包装器。 aiogram-mcp 更进一步:
- 30工具 --消息传递、富媒体、审核、交互式键盘、活动订阅、广播
- 7资源 --机器人信息、配置、聊天列表、消息历史、事件队列、文件元数据、审计日志
- 3个提示 --现成的审核、公告和用户报告工作流
- 结构化输出 --每个工具都返回键入的Pydantic模型
outputSchema用于程序化解析 - 实时事件 --机器人通过MCP通知将Telegram事件推送到AI客户端(无轮询)
- 交互式消息 --AI代理创建内联键盘菜单、处理按钮按下、编辑消息
- 速率限制 --内置令牌桶可防止Telegram 429错误
- 权限级别 --将AI代理限制为只读、消息传递、审核或完全管理员访问权限
- 审计日志 --使用时间戳和参数跟踪每次工具调用
- 零重写 --在现有的bot中添加5行,保留所有处理程序
运作原理
Telegram users Your aiogram bot AI agent (Claude Desktop)
| | |
| send messages, tap buttons | |
| --------------------------> | |
| | MCP server (stdio or SSE) |
| | |
| | tools / resources / events |
| | |
| bot replies, shows menus | send_message, edit, ban |
| <-------------------------- | <--------------------------- |该机器人对Telegram用户正常运行。MCP服务器与之并行运行,使AI代理能够通过工具和资源访问同一机器人。
安装
pip install aiogram-mcp需要Python 3.10+和aiogram 3.20+。
快速入门
1.将aiogram mcp添加到您的机器人中
import asyncio
from aiogram import Bot, Dispatcher
from aiogram_mcp import AiogramMCP, EventManager, MCPMiddleware
bot = Bot(token="YOUR_BOT_TOKEN")
dp = Dispatcher()
# Middleware tracks chats, users, message history, and events
event_manager = EventManager()
middleware = MCPMiddleware(event_manager=event_manager)
dp.message.middleware(middleware)
dp.callback_query.middleware(middleware) # for interactive buttons
# Register your normal handlers here
# @dp.message(...)
# async def my_handler(message): ...
# Create the MCP server
mcp = AiogramMCP(
bot=bot,
dp=dp,
name="my-bot",
middleware=middleware,
event_manager=event_manager,
allowed_chat_ids=[123456789], # optional: restrict which chats AI can access
)
async def main():
await mcp.run_alongside_bot(transport="stdio")
asyncio.run(main())2.连接克劳德桌面
添加到您的 claude_desktop_config.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 | 用祝酒词或警报回应按钮按下 |
用户(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? | 起草一份格式化的电报公告 |
user_report_prompt | chat_id, user_id | 编译完整的用户活动报告 |
实时事件流
AI代理不需要投票。机器人会自动推送事件:
Telegram message arrives
→ MCPMiddleware captures it
→ EventManager stores it (type: "message", "command", or "callback_query")
→ MCP notification sent to subscribed clients
→ AI agent reads telegram://events/queueAI代理调用 subscribe_events 一旦有新事件与其过滤器匹配,就会接收推送通知。
交互式消息
AI代理可以在Telegram中构建完整的交互式UI——菜单、确认、多步骤向导:
AI代理通过按钮发送消息:
┌─────────────────────────┐
│ Confirm deployment? │
│ │
│ [✅ Yes] [❌ No] │
│ [📖 View docs] │
└─────────────────────────┘用户点击按钮→ 事件出现在队列中→ AI代理反应:
┌─────────────────────────┐
│ ✅ Deployed! │
│ │
│ [📋 View logs] │
└─────────────────────────┘机器人需要 dp.callback_query.middleware(middleware) 捕捉按钮按下。
安全控制
mcp = AiogramMCP(
bot=bot,
dp=dp,
allowed_chat_ids=[123456789, -1001234567890], # restrict AI access
enable_broadcast=True, # opt-in for broadcast tool
max_broadcast_recipients=500, # safety limit
)allowed_chat_ids--AI只能与列出的聊天记录进行交互。默认值:所有聊天记录。enable_broadcast--作为安全措施,默认情况下禁用广播工具。max_broadcast_recipients--限制单个广播中的聊天次数。
高级配置
速率限制
mcp = AiogramMCP(
bot=bot, dp=dp,
rate_limit=30, # requests/sec (default), 0 to disable
)内置令牌桶速率限制器可防止Telegram 429错误。所有传出的API调用都是自动定速的。
权限级别
mcp = AiogramMCP(
bot=bot, dp=dp,
permission_level="messaging", # read + messaging tools only
)| 级别 | 访问 |
|---|---|
read | 机器人信息、聊天信息、用户资料 |
messaging | 阅读+发送消息、照片、媒体、互动消息 |
moderation | 消息+删除、pin、禁令、取消禁令、聊天设置 |
admin | 完全访问权限,包括广播和活动订阅 |
审计日志
mcp = AiogramMCP(
bot=bot, dp=dp,
enable_audit=True,
audit_log_size=1000,
)每次工具调用都会被记录下来。通过以下方式访问 telegram://audit/log 资源。
示例
| 示例 | 运输 | 功能 |
|---|---|---|
| basic_bot.py | stdio | 完整设置中间件、事件和回调跟踪 |
| incident_alert_bot.py | SSE | 用于事件通知的广播式操作机器人 |
发展
git clone https://github.com/Py2755/aiogram-mcp.git
cd aiogram-mcp
pip install -e ".[dev]"
pytest -v # ~228 tests
ruff check aiogram_mcp tests examples
mypy aiogram_mcp # strict mode许可证
MIT。看 许可证.
