telegram_send
Send free-form messages to Telegram with automatic handling of long content by truncating messages over 300 characters. Use for direct communication through the Telegram MCP Server.
Instructions
发送自由格式消息到 Telegram（不推荐，请优先使用 telegram_notify）
自动处理：
- 超过300字自动截断
- 会提示使用 telegram_notify 发送结构化消息
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| message | Yes | 消息内容 |
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"] } ),
- 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