telegram_notify
Send structured notifications to Telegram with event types including completed, error, question, and progress. Provide concise summaries and optional details for remote task monitoring and decision-making.
Instructions
发送结构化通知到 Telegram
参数：
- event: 事件类型（completed/error/question/progress）
- summary: 简短总结，必填，限制200字以内
- details: 详细信息，可选
最佳实践：
1. summary 必须简洁明了（1-2句话），说明做了什么、结果如何
2. 不要包含思考过程、不要包含代码片段
3. 需要用户决策时，清晰说明选项
示例：
telegram_notify(
event="completed",
summary="修复了 auth.py:45 的空指针异常，所有测试通过",
details="修改文件: auth.py, test_auth.py\n测试: 12/12 passed"
)
telegram_notify(
event="question",
summary="发现3种修复方案：1)添加空值检查 2)使用Optional类型 3)重构整个模块。推荐方案1，是否继续？"
)
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| details | No | 详细信息（可选） | |
| event | Yes | 事件类型 | |
| summary | Yes | 简短总结（必填，200字以内） |
Implementation Reference
- telegram_mcp_server/server.py:693-736 (handler)The core handler function for the 'telegram_notify' tool. It validates input, formats the notification message with event emoji and session ID, optionally includes details, updates the session state, sends the message to Telegram, and returns a success or error response.async def handle_telegram_notify(session, arguments: dict) -> list[TextContent]: """Handle telegram_notify tool""" event = arguments.get("event") summary = arguments.get("summary", "") details = arguments.get("details", "") # Validate summary length if len(summary) > 200: return [TextContent( type="text", text="错误: summary 过长，请精炼到200字以内" )] # Format message emoji = { "completed": "✅", "error": "❌", "question": "❓", "progress": "⏳" } message = f"{emoji.get(event, '🔔')} [`{session.session_id}`]\n{summary}" if details: message += f"\n\n━━━━━━━━━━━━\n📝 详情:\n{details}" # Update session session.last_message = summary session.update_activity() registry.update_session(session) # Save to shared storage # Send to Telegram try: await send_telegram_message(session.chat_id, message) return [TextContent( type="text", text=f"✅ 已发送通知到 Telegram (会话: {session.session_id})" )] except Exception as e: return [TextContent( type="text", text=f"❌ 发送失败: {str(e)}" )]
- telegram_mcp_server/server.py:132-183 (registration)Registration of the 'telegram_notify' tool in the MCP server's list_tools() method, defining the tool name, detailed description, and input schema with required event and summary parameters.Tool( name="telegram_notify", description=""" ⚠️ 这是 Telegram MCP Server 的通知工具 用于向用户的 Telegram Bot 发送任务进度通知 ❌ 这不是通用的 Telegram 消息发送工具 ❌ 不能发送消息到任意 Telegram 用户或群组 ✅ 只能发送通知到配置的 Telegram Bot（用户会在 Telegram 中收到） 💡 推荐使用 telegram_notify_with_actions 代替此工具 telegram_notify_with_actions 提供动态按钮，用户体验更好 此工具适用于： - 简单的状态更新（不需要用户交互） - 快速通知（无需提供下一步建议） - 向后兼容旧代码 参数： - event: 事件类型（completed/error/question/progress） - summary: 简短总结（必填，200字以内） - details: 详细信息（可选） 示例： telegram_notify( event="completed", summary="修复了 auth.py:45 的空指针异常，所有测试通过" ) 💡 更好的选择：使用 telegram_notify_with_actions 提供智能建议按钮 """, inputSchema={ "type": "object", "properties": { "event": { "type": "string", "enum": ["completed", "error", "question", "progress"], "description": "事件类型" }, "summary": { "type": "string", "description": "简短总结（必填，200字以内）", "maxLength": 200 }, "details": { "type": "string", "description": "详细信息（可选）" } }, "required": ["event", "summary"] } ),
- Input schema definition for the 'telegram_notify' tool, specifying the object structure with event (enum), summary (required, max 200 chars), and optional details.inputSchema={ "type": "object", "properties": { "event": { "type": "string", "enum": ["completed", "error", "question", "progress"], "description": "事件类型" }, "summary": { "type": "string", "description": "简短总结（必填，200字以内）", "maxLength": 200 }, "details": { "type": "string", "description": "详细信息（可选）" } }, "required": ["event", "summary"] }
- telegram_mcp_server/server.py:76-109 (helper)Helper function to send messages to Telegram via HTTP API, with Markdown support, automatic retry on parse errors, and comprehensive error handling. Used by the telegram_notify handler.async def send_telegram_message(chat_id: str, message: str, parse_mode: str = "Markdown") -> None: """Send message to Telegram (async) using HTTP API""" import httpx url = f"https://api.telegram.org/bot{config.TELEGRAM_BOT_TOKEN}/sendMessage" payload = { "chat_id": chat_id, "text": message } # Only add parse_mode if it's not None if parse_mode: payload["parse_mode"] = parse_mode try: async with httpx.AsyncClient() as client: response = await client.post(url, json=payload, timeout=10.0) response.raise_for_status() except httpx.HTTPStatusError as e: if e.response.status_code == 400 and parse_mode: # Markdown parsing failed, retry without parse_mode logger.warning(f"Markdown parsing failed (400 Bad Request), retrying as plain text") payload.pop("parse_mode", None) async with httpx.AsyncClient() as client: response = await client.post(url, json=payload, timeout=10.0) response.raise_for_status() else: logger.error(f"Failed to send Telegram message: {e}") raise except Exception as e: logger.error(f"Failed to send Telegram message: {e}") raise
- telegram_mcp_server/server.py:671-672 (handler)Dispatch logic in the main call_tool handler that routes 'telegram_notify' calls to the specific handle_telegram_notify function.if name == "telegram_notify": return await handle_telegram_notify(session, arguments)