Telegram MCP Server

Instructions

        等待用户回复（阻塞式轮询）

        参数：
        - max_wait: 最长等待时间（秒），默认604800（7天/1周）

        行为：
        - 前10分钟：每30秒检查一次
        - 10分钟-1小时：每60秒检查一次
        - 1小时以上：每120秒检查一次
        - 用户可以按 Ctrl+C 中断等待
        - 超时返回 timeout: true

        返回：
        - reply: 用户回复内容
        - timeout: 是否超时
        - interrupted: 是否被用户中断
        

Implementation Reference

• telegram_mcp_server/server.py:879-929 (handler)

  Core implementation of the telegram_wait_reply tool handler. Polls message_queue for user replies with progressive backoff (30s/60s/120s based on elapsed time), handles timeout, interruption (Ctrl+C), and returns the reply or status.

  async def handle_telegram_wait_reply(session, arguments: dict) -> list[TextContent]:
      """Handle telegram_wait_reply tool"""
      max_wait = arguments.get("max_wait", config.TELEGRAM_MAX_WAIT)
  
      logger.info(f"Session {session.session_id} waiting for reply (max {max_wait}s)")
  
      # Mark session as waiting
      session.set_waiting()
      registry.update_session(session)  # Save to shared storage
  
      # Poll for messages
      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} wait timeout")
                  return [TextContent(
                      type="text",
                      text=f"超时: 等待了 {int(elapsed)} 秒未收到回复"
                  )]
  
              # 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 reply: {reply}")
                  return [TextContent(
                      type="text",
                      text=f"用户回复: {reply}"
                  )]
  
              # Progressive polling
              interval = get_poll_interval(elapsed)
              logger.debug(f"Session {session.session_id} 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} wait 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:331-340 (schema)

  Input schema definition for the telegram_wait_reply tool, specifying the optional max_wait parameter.

  inputSchema={
      "type": "object",
      "properties": {
          "max_wait": {
              "type": "integer",
              "description": "最长等待时间（秒），默认604800（7天）",
              "default": 604800
          }
      }
  }

• telegram_mcp_server/server.py:311-341 (registration)

  Registration of the telegram_wait_reply tool in the list_tools() function, including name, description, and input schema.

  Tool(
      name="telegram_wait_reply",
      description="""
      等待用户回复（阻塞式轮询）
  
      参数：
      - max_wait: 最长等待时间（秒），默认604800（7天/1周）
  
      行为：
      - 前10分钟：每30秒检查一次
      - 10分钟-1小时：每60秒检查一次
      - 1小时以上：每120秒检查一次
      - 用户可以按 Ctrl+C 中断等待
      - 超时返回 timeout: true
  
      返回：
      - reply: 用户回复内容
      - timeout: 是否超时
      - interrupted: 是否被用户中断
      """,
      inputSchema={
          "type": "object",
          "properties": {
              "max_wait": {
                  "type": "integer",
                  "description": "最长等待时间（秒），默认604800（7天）",
                  "default": 604800
              }
          }
      }
  ),

• telegram_mcp_server/server.py:111-122 (helper)

  Helper function used by the handler to determine progressive polling intervals based on elapsed waiting time.

  def get_poll_interval(elapsed_seconds: float) -> int:
      """
      Get polling interval based on elapsed time
      Progressive slowdown: 30s -> 60s -> 120s
      """
      if elapsed_seconds < config.POLL_THRESHOLDS[0]:  # < 10 minutes
          return config.POLL_INTERVALS[0]  # 30 seconds
      elif elapsed_seconds < config.POLL_THRESHOLDS[1]:  # < 1 hour
          return config.POLL_INTERVALS[1]  # 60 seconds
      else:
          return config.POLL_INTERVALS[2]  # 120 seconds