Telegram MCP Server

Instructions

Implementation Reference

• telegram_mcp_server/server.py:1203-1266 (handler)

  Main handler function that implements the telegram_unattended_mode tool logic: enters silent waiting mode, polls for Telegram messages with progressive intervals, handles timeouts, user exits, and returns received instructions.
  async def handle_telegram_unattended_mode(session, arguments: dict) -> list[TextContent]: """Handle telegram_unattended_mode tool""" current_status = arguments.get("current_status", "") max_wait = arguments.get("max_wait", config.TELEGRAM_MAX_WAIT) # Update session state session.last_message = current_status session.update_activity() session.set_waiting() registry.update_session(session) # Save to shared storage # Silent waiting - no notification sent # User should call telegram_notify before calling this tool logger.info(f"Session {session.session_id} in unattended mode, waiting for instruction (silent)") start_time = time.time() try: while True: elapsed = time.time() - start_time # Check timeout if elapsed >= max_wait: session.set_running() registry.update_session(session) # Save to shared storage logger.info(f"Session {session.session_id} unattended mode timeout") return [TextContent( type="text", text=f"⏱️ 超时: 等待了 {int(elapsed)} 秒未收到指令\n\n建议：可以继续调用此工具重新进入等待，或者退出无人值守模式。" )] # Check message queue if message_queue.has_messages(session.session_id): reply = message_queue.pop(session.session_id) session.set_running() registry.update_session(session) # Save to shared storage logger.info(f"Session {session.session_id} received instruction: {reply}") # Check if user wants to exit if reply.lower() in ['退出', 'exit', 'quit', '结束']: return [TextContent( type="text", text=f"🚪 已退出无人值守模式\n\n用户指令: {reply}\n\n你可以继续正常对话，不再自动循环。" )] # Return the instruction return [TextContent( type="text", text=f"📨 收到新指令: {reply}\n\n请执行此指令，完成后再次调用 telegram_unattended_mode 继续循环。" )] # Progressive polling interval = get_poll_interval(elapsed) logger.debug(f"Session {session.session_id} unattended mode polling (interval={interval}s, elapsed={int(elapsed)}s)") await asyncio.sleep(interval) except (KeyboardInterrupt, asyncio.CancelledError): session.set_running() registry.update_session(session) # Save to shared storage logger.info(f"Session {session.session_id} unattended mode interrupted by user") return [TextContent( type="text", text=f"⚠️ 无人值守模式被用户中断 (Ctrl+C)\n\n已运行: {int(time.time() - start_time)} 秒\n\n已退出无人值守模式，你可以继续正常对话。" )]
• telegram_mcp_server/server.py:544-650 (schema)

  Tool schema definition including name, detailed description for usage in unattended mode loops, and input schema for current_status and max_wait parameters.
  Tool( name="telegram_unattended_mode", description=""" ⚠️ 这是 Telegram MCP Server 的无人值守模式工具 用于等待用户通过 Telegram Bot 发送的下一步指令 ❌ 这不是通用的 Telegram 操作工具 ❌ 不用于发送 Telegram 消息（使用 telegram_notify 系列工具） ❌ 不用于管理 Telegram 群组或频道 ✅ 正确用途：远程任务循环 - 等待用户通过 Telegram 发送指令 工作流程： 1. 执行当前任务 2. 使用 telegram_notify_with_actions 发送结果（带智能按钮） 3. 调用 telegram_unattended_mode 等待用户通过 Telegram 发送的下一步指令 4. 收到指令后执行，重复循环 示例场景： 用户说："进入无人值守模式，任务：分析项目结构" 你应该： 1. 分析项目结构 2. 调用 telegram_notify_with_actions 发送分析结果 3. 调用 telegram_unattended_mode 等待下一步指令 4. 用户在 Telegram 中发送"优化性能" 5. 你收到指令，执行优化 6. 重复步骤 2-5 ⚠️ 重要： - 完成任务后必须调用通知工具发送结果 - telegram_unattended_mode 本身不发送消息，只等待 - 这样用户每次只收到任务结果，不会有重复的等待提示 📋 推荐使用 telegram_notify_with_actions 发送结果： ⭐ 最佳实践（带智能按钮）： telegram_notify_with_actions( event="completed", summary="✅ 完成代码审查\\n- 发现 3 个可优化点\\n- 代码质量：B+", actions=[ {"text": "💡 优化这 3 处", "action": "自动优化发现的问题"}, {"text": "📊 查看详情", "action": "显示详细的优化建议"} ] ) ✅ 简单通知（无按钮）： telegram_notify_with_actions( event="completed", summary="修复了 auth.py 的空指针异常，测试通过", actions=[] # 不提供按钮 ) 或使用基础版本： telegram_notify( event="completed", summary="创建了 3 个文件：main.py, utils.py, test.py" ) ⚠️ 仅在必要时发送代码/文件： - 遇到无法自动修复的错误 → telegram_send_code 展示错误代码 - 用户明确要求 → telegram_send_file 发送文件 - 修复关键 bug → telegram_send_code 展示修复对比 🎯 智能判断示例： - 任务完成 → telegram_notify_with_actions（带下一步建议按钮） - 遇到错误 → telegram_notify_with_actions（带修复方案按钮） - 需要决策 → telegram_notify_with_actions（带选项按钮） - 简单更新 → telegram_notify（无按钮） 退出方式： - Telegram 发送 "退出" 或 "exit" - Claude Code 按 Ctrl+C 或 ESC 轮询策略： - 前10分钟：每30秒检查一次 - 10分钟-1小时：每60秒检查一次 - 1小时以上：每120秒检查一次 参数： - current_status: 当前任务状态的简短总结（1-2句话） - max_wait: 每次等待的最长时间（秒），默认604800（7天） - silent: 静默模式（不发送等待提示，默认 false） - 首次进入时使用 false（发送提示） - 后续循环使用 true（减少噪音） 返回： - next_instruction: 用户的下一步指令 - should_exit: 是否应该退出无人值守模式 - interrupted: 是否被用户中断（Ctrl+C/ESC） """, inputSchema={ "type": "object", "properties": { "current_status": { "type": "string", "description": "当前任务状态描述" }, "max_wait": { "type": "integer", "description": "最长等待时间（秒），默认604800（7天）", "default": 604800 } }, "required": [] } )
• telegram_mcp_server/server.py:654-690 (registration)

  Tool registration and dispatching logic in the main call_tool handler, which routes calls to telegram_unattended_mode to its specific handler function.
  @server.call_tool() async def call_tool(name: str, arguments: dict) -> list[TextContent]: """Handle tool calls""" # Validate configuration try: config.validate_config() except ValueError as e: return [TextContent(type="text", text=f"配置错误: {str(e)}")] session_id = get_session_id() # Ensure session is registered (lazy registration) await ensure_session_registered(session_id) session = registry.get(session_id) if name == "telegram_notify": return await handle_telegram_notify(session, arguments) elif name == "telegram_notify_with_actions": return await handle_telegram_notify_with_actions(session, arguments) elif name == "telegram_wait_reply": return await handle_telegram_wait_reply(session, arguments) elif name == "telegram_send": return await handle_telegram_send(session, arguments) elif name == "telegram_send_code": return await handle_telegram_send_code(session, arguments) elif name == "telegram_send_image": return await handle_telegram_send_image(session, arguments) elif name == "telegram_send_file": return await handle_telegram_send_file(session, arguments) elif name == "telegram_get_context_info": return await handle_telegram_get_context_info(session, arguments) elif name == "telegram_unattended_mode": return await handle_telegram_unattended_mode(session, arguments) else: return [TextContent(type="text", text=f"Unknown tool: {name}")]