Telegram MCP Server

Instructions

        发送自由格式消息到 Telegram（不推荐，请优先使用 telegram_notify）

        自动处理：
        - 超过300字自动截断
        - 会提示使用 telegram_notify 发送结构化消息
        

Implementation Reference

• telegram_mcp_server/server.py:931-959 (handler)

  Main handler function that executes the 'telegram_send' tool logic: extracts message, truncates if >300 chars, formats with session ID, updates session state, sends to Telegram via send_telegram_message, returns success/error content.

  async def handle_telegram_send(session, arguments: dict) -> list[TextContent]:
      """Handle telegram_send tool"""
      message = arguments.get("message", "")
  
      # Auto-truncate if too long
      if len(message) > 300:
          message = message[:280] + "\n\n... [消息过长已截断，建议使用 telegram_notify]"
  
      # Format message
      formatted = f"🤖 [`{session.session_id}`]\n{message}"
  
      # Update session
      session.last_message = message
      session.update_activity()
      registry.update_session(session)  # Save to shared storage
  
      # Send to Telegram
      try:
          await send_telegram_message(session.chat_id, formatted)
          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:342-361 (registration)

  Tool registration in list_tools(): defines name 'telegram_send', description, and inputSchema requiring 'message' string.

  Tool(
      name="telegram_send",
      description="""
      发送自由格式消息到 Telegram（不推荐，请优先使用 telegram_notify）
  
      自动处理：
      - 超过300字自动截断
      - 会提示使用 telegram_notify 发送结构化消息
      """,
      inputSchema={
          "type": "object",
          "properties": {
              "message": {
                  "type": "string",
                  "description": "消息内容"
              }
          },
          "required": ["message"]
      }
  ),

• telegram_mcp_server/server.py:351-360 (schema)

  Input schema for 'telegram_send' tool: object with required 'message' property (string).

  inputSchema={
      "type": "object",
      "properties": {
          "message": {
              "type": "string",
              "description": "消息内容"
          }
      },
      "required": ["message"]
  }

• telegram_mcp_server/server.py:677-678 (handler)

  Dispatch logic in call_tool() that routes 'telegram_send' calls to the handle_telegram_send function.

  elif name == "telegram_send":
      return await handle_telegram_send(session, arguments)

• telegram_mcp_server/server.py:76-109 (helper)

  Helper function used by telegram_send to actually send the HTTP request to Telegram API, with Markdown fallback.

  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