get_messages
Retrieve a specified number of messages from a Telegram entity to monitor conversations or extract information for analysis.
Instructions
Get messages limited by a count from an entity
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| id | Yes | ||
| count | No |
Implementation Reference
- mcp-server/mcp-server.py:28-30 (registration)Registration of the MCP tool 'get_messages' using FastMCP decorator.
@mcp.tool( name="get_messages", description="Get messages limited by a count from an entity" ) - mcp-server/mcp-server.py:31-32 (handler)Handler function for the 'get_messages' MCP tool, which proxies the request to the HTTP API endpoint.
async def get_messages(id: int, count: int = 0) -> list[dict]: return get(f"{api_endpoint}get_messages/{id}", params={"count": count}).json() - http-server/http-api.py:58-95 (helper)Backend HTTP API endpoint that implements the core logic for fetching and formatting Telegram messages using Telethon client.
@app.get("/get_messages/{chat_id}") async def get_messages(chat_id: int, count: int = 0): entities = {} messages = [] e: Entity = await get_entity(entity=chat_id, raw=True) async for message in client.iter_messages(entity=e, reverse=False, limit=count): e_id: int = None if message.from_id is not None: e_id = message.from_id.user_id elif message.peer_id is not None: e_id = message.peer_id.user_id if e_id not in entities: entity = await client.get_entity(e_id) entities[e_id] = entity # Handle message text that might be binary message_text = message.message if isinstance(message_text, bytes): try: message_text = message_text.decode('utf-8', errors='replace') except Exception: message_text = '[Binary content]' messages += [ { "text": message_text, "date": message.date, "id": message.id, "from": await format_entity(entity=entities[e_id]), } ] await client.send_read_acknowledge(entity=chat_id, message=message) return messages