telegram_unattended_mode

Enables continuous remote task execution by waiting for user instructions between automated operations, sending only task results to reduce notification noise while maintaining persistent workflow control.

Instructions

        进入无人值守模式 - 智能远程任务循环

        工作流程：
        1. 执行当前任务
        2. 根据情况智能选择通知方式：
           - 默认：使用 telegram_notify 发送总结
           - 遇到关键问题/错误：使用 telegram_send_code 展示问题代码
           - 用户明确要求：使用 telegram_send_file 发送文件
        3. 调用 telegram_unattended_mode 等待下一步指令（静默等待，不发送额外提示）
        4. 收到指令后执行，重复循环

        ⚠️ 重要：
        - 完成任务后必须调用 telegram_notify 发送结果
        - telegram_unattended_mode 本身不发送消息，只等待
        - 这样用户每次只收到任务结果，不会有重复的等待提示

        📋 通知内容最佳实践：
        ✅ 优先发送总结：
        - "修复了 auth.py 的空指针异常，测试通过"
        - "创建了 3 个文件：main.py, utils.py, test.py"
        - "代码重构完成，性能提升 30%"

        ⚠️ 仅在必要时发送代码：
        - 遇到无法自动修复的错误，需要展示错误代码
        - 修复了关键 bug，展示修复前后对比
        - 用户明确要求："查看 main.py"、"发送代码给我"

        🎯 智能判断示例：
        - 创建新文件 → telegram_notify("创建了 config.json")
        - 修复 bug → telegram_notify("修复了登录异常") + 如果复杂就 telegram_send_code
        - 用户问"文件内容是什么" → telegram_send_file

        退出方式：
        - 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）

Input Schema

TableJSON Schema

Name	Required	Description	Default
`current_status`	No	当前任务状态描述
`max_wait`	No	最长等待时间（秒），默认604800（7天）

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}")]

Telegram MCP Server