Telegram MCP Server

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，是否继续？"
        )
        

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

• telegram_mcp_server/server.py:163-182 (schema)

  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)