telegram_unattended_mode

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）
        

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