Telegram MCP Server

Instructions

Implementation Reference

• src/mcp_telegram/tools.py:120-157 (handler)

  The main handler function for executing the ListMessages tool. It fetches messages from the specified Telegram dialog_id, optionally filtering by unread status and limiting the number, and returns text contents of messages.
  @tool_runner.register async def list_messages( args: ListMessages, ) -> t.Sequence[TextContent | ImageContent | EmbeddedResource]: client: TelegramClient logger.info("method[ListMessages] args[%s]", args) response: list[TextContent] = [] async with create_client() as client: result = await client(functions.messages.GetPeerDialogsRequest(peers=[args.dialog_id])) if not result: raise ValueError(f"Channel not found: {args.dialog_id}") if not isinstance(result, types.messages.PeerDialogs): raise TypeError(f"Unexpected result: {type(result)}") for dialog in result.dialogs: logger.debug("dialog: %s", dialog) for message in result.messages: logger.debug("message: %s", message) iter_messages_args: dict[str, t.Any] = { "entity": args.dialog_id, "reverse": False, } if args.unread: iter_messages_args["limit"] = min(dialog.unread_count, args.limit) else: iter_messages_args["limit"] = args.limit logger.debug("iter_messages_args: %s", iter_messages_args) async for message in client.iter_messages(**iter_messages_args): logger.debug("message: %s", type(message)) if isinstance(message, custom.Message) and message.text: logger.debug("message: %s", message.text) response.append(TextContent(type="text", text=message.text)) return response
• src/mcp_telegram/tools.py:104-118 (schema)

  Pydantic model defining the input parameters for the ListMessages tool: dialog_id (required int), unread (bool default False), limit (int default 100). The docstring provides the tool description.
  class ListMessages(ToolArgs): """ List messages in a given dialog, chat or channel. The messages are listed in order from newest to oldest. If `unread` is set to `True`, only unread messages will be listed. Once a message is read, it will not be listed again. If `limit` is set, only the last `limit` messages will be listed. If `unread` is set, the limit will be the minimum between the unread messages and the limit. """ dialog_id: int unread: bool = False limit: int = 100
• src/mcp_telegram/server.py:29-37 (registration)

  Dynamic registration of tools by inspecting subclasses of ToolArgs in the tools module, including ListMessages, and creating the tool mapping used by list_tools and call_tool.
  def enumerate_available_tools() -> t.Generator[tuple[str, Tool], t.Any, None]: for _, tool_args in inspect.getmembers(tools, inspect.isclass): if issubclass(tool_args, tools.ToolArgs) and tool_args != tools.ToolArgs: logger.debug("Found tool: %s", tool_args) description = tools.tool_description(tool_args) yield description.name, description mapping: dict[str, Tool] = dict(enumerate_available_tools())
• src/mcp_telegram/tools.py:56-62 (helper)

  Helper function that generates the Tool object from a ToolArgs class, used for registration of ListMessages.
  def tool_description(args: type[ToolArgs]) -> Tool: return Tool( name=args.__name__, description=args.__doc__, inputSchema=args.model_json_schema(), )
• src/mcp_telegram/server.py:70-86 (helper)

  MCP server handler for tool calls that dispatches to the specific tool_runner for ListMessages via the dynamic mapping.
  async def call_tool(name: str, arguments: t.Any) -> Sequence[TextContent | ImageContent | EmbeddedResource]: # noqa: ANN401 """Handle tool calls for command line run.""" if not isinstance(arguments, dict): raise TypeError("arguments must be dictionary") tool = mapping.get(name) if not tool: raise ValueError(f"Unknown tool: {name}") try: args = tools.tool_args(tool, **arguments) return await tools.tool_runner(args) except Exception as e: logger.exception("Error running tool: %s", name) raise RuntimeError(f"Caught Exception. Error: {e}") from e