Telegram MCP Server

Instructions

Implementation Reference

• src/mcp_telegram/tools.py:79-98 (handler)

  The main handler function for the ListDialogs tool. It connects to Telegram, iterates over dialogs with specified filters (unread, archived, ignore_pinned), and returns formatted text contents listing dialog details.
  @tool_runner.register async def list_dialogs( args: ListDialogs, ) -> t.Sequence[TextContent | ImageContent | EmbeddedResource]: client: TelegramClient logger.info("method[ListDialogs] args[%s]", args) response: list[TextContent] = [] async with create_client() as client: dialog: custom.dialog.Dialog async for dialog in client.iter_dialogs(archived=args.archived, ignore_pinned=args.ignore_pinned): if args.unread and dialog.unread_count == 0: continue msg = ( f"name='{dialog.name}' id={dialog.id} " f"unread={dialog.unread_count} mentions={dialog.unread_mentions_count}" ) response.append(TextContent(type="text", text=msg)) return response
• src/mcp_telegram/tools.py:71-77 (schema)

  Pydantic model defining the input schema for the ListDialogs tool, including optional filters for unread messages, archived dialogs, and ignoring pinned dialogs.
  class ListDialogs(ToolArgs): """List available dialogs, chats and channels.""" unread: bool = False archived: bool = False ignore_pinned: bool = False
• src/mcp_telegram/server.py:28-37 (registration)

  Dynamically discovers and registers all tools (including ListDialogs) by inspecting subclasses of ToolArgs in the tools module, creating Tool descriptions used in list_tools().
  @cache 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/server.py:52-56 (registration)

  MCP server endpoint that lists all registered tools, including ListDialogs, from the mapping.
  @app.list_tools() async def list_tools() -> list[Tool]: """List available tools.""" return list(mapping.values())
• src/mcp_telegram/server.py:69-86 (registration)

  MCP server endpoint that handles calls to tools like ListDialogs by parsing arguments, instantiating the schema, and dispatching to the tool_runner.
  @app.call_tool() 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