Telegram MCP Server

Instructions

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.

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