diff --git a/telegram-mcp-server/main.py b/telegram-mcp-server/main.py index d0e5964..478cfb5 100644 --- a/telegram-mcp-server/main.py +++ b/telegram-mcp-server/main.py @@ -14,6 +14,7 @@ and also communicates with the Bridge's HTTP API for sending messages. from typing import List, Dict, Any, Optional, Tuple from mcp.server.fastmcp import FastMCP from datetime import datetime +from dataclasses import asdict from telegram import ( search_contacts as telegram_search_contacts, list_messages as telegram_list_messages, @@ -29,6 +30,21 @@ from telegram import ( # Initialize FastMCP server mcp = FastMCP("telegram") +# Author: Danny Stocker (fix for datetime serialization bug) +# Date: 2025-10-26 +def serialize_dataclass(obj: Any) -> Dict[str, Any]: + """Convert dataclass to dictionary with JSON-serializable values.""" + if hasattr(obj, '__dataclass_fields__'): + result = asdict(obj) + # Convert datetime objects to ISO format strings + for key, value in result.items(): + if isinstance(value, datetime): + result[key] = value.isoformat() + elif isinstance(value, list): + result[key] = [serialize_dataclass(item) if hasattr(item, '__dataclass_fields__') else item for item in value] + return result + return obj + @mcp.tool() def search_contacts(query: str) -> List[Dict[str, Any]]: """Search Telegram contacts by name or username. @@ -37,35 +53,53 @@ def search_contacts(query: str) -> List[Dict[str, Any]]: query: Search term to match against contact names or usernames """ contacts = telegram_search_contacts(query) - return contacts + return [serialize_dataclass(c) for c in contacts] @mcp.tool() def list_messages( - date_range: Optional[Tuple[datetime, datetime]] = None, + date_range: Optional[Tuple[str, str]] = None, sender_id: Optional[int] = None, chat_id: Optional[int] = None, + title: Optional[str] = None, query: Optional[str] = None, limit: int = 20, page: int = 0, - include_context: bool = True, + include_context: bool = False, context_before: int = 1, context_after: int = 1 ) -> List[Dict[str, Any]]: """Get Telegram messages matching specified criteria with optional context. Args: - date_range: Optional tuple of (start_date, end_date) to filter messages by date + date_range: Optional tuple of (start_date, end_date) ISO strings to filter messages by date sender_id: Optional sender ID to filter messages by sender chat_id: Optional chat ID to filter messages by chat + title: Optional chat title to search for (will find chat_id automatically) query: Optional search term to filter messages by content limit: Maximum number of messages to return (default 20) page: Page number for pagination (default 0) - include_context: Whether to include messages before and after matches (default True) + include_context: Whether to include messages before and after matches (default False) context_before: Number of messages to include before each match (default 1) context_after: Number of messages to include after each match (default 1) """ + # Convert date strings to datetime if provided + parsed_date_range = None + if date_range: + parsed_date_range = ( + datetime.fromisoformat(date_range[0]), + datetime.fromisoformat(date_range[1]) + ) + + # If title is provided, find the chat_id + if title and not chat_id: + from telegram import list_chats as find_chats + chats = find_chats(query=title, limit=5) + if chats: + # Use first matching chat + chat_id = chats[0].id + messages = telegram_list_messages( - date_range=date_range, + date_range=parsed_date_range, sender_id=sender_id, chat_id=chat_id, query=query, @@ -75,7 +109,7 @@ def list_messages( context_before=context_before, context_after=context_after ) - return messages + return [serialize_dataclass(m) for m in messages] @mcp.tool() def list_chats( @@ -101,7 +135,7 @@ def list_chats( chat_type=chat_type, sort_by=sort_by ) - return chats + return [serialize_dataclass(c) for c in chats] @mcp.tool() def get_chat(chat_id: int) -> Dict[str, Any]: @@ -111,7 +145,7 @@ def get_chat(chat_id: int) -> Dict[str, Any]: chat_id: The ID of the chat to retrieve """ chat = telegram_get_chat(chat_id) - return chat + return serialize_dataclass(chat) @mcp.tool() def get_direct_chat_by_contact(contact_id: int) -> Dict[str, Any]: @@ -121,7 +155,7 @@ def get_direct_chat_by_contact(contact_id: int) -> Dict[str, Any]: contact_id: The contact ID to search for """ chat = telegram_get_direct_chat_by_contact(contact_id) - return chat + return serialize_dataclass(chat) @mcp.tool() def get_contact_chats(contact_id: int, limit: int = 20, page: int = 0) -> List[Dict[str, Any]]: @@ -133,7 +167,7 @@ def get_contact_chats(contact_id: int, limit: int = 20, page: int = 0) -> List[D page: Page number for pagination (default 0) """ chats = telegram_get_contact_chats(contact_id, limit, page) - return chats + return [serialize_dataclass(c) for c in chats] @mcp.tool() def get_last_interaction(contact_id: int) -> Dict[str, Any]: @@ -143,7 +177,7 @@ def get_last_interaction(contact_id: int) -> Dict[str, Any]: contact_id: The ID of the contact to search for """ message = telegram_get_last_interaction(contact_id) - return message + return serialize_dataclass(message) @mcp.tool() def get_message_context( @@ -161,7 +195,7 @@ def get_message_context( after: Number of messages to include after the target message (default 5) """ context = telegram_get_message_context(message_id, chat_id, before, after) - return context + return serialize_dataclass(context) @mcp.tool() def send_message( @@ -171,7 +205,7 @@ def send_message( """Send a Telegram message to a person or group. Args: - recipient: The recipient - either a username (with or without @) or a chat ID + recipient: The recipient - either a username (with or without @), chat title, or a chat ID message: The message text to send Returns: @@ -203,4 +237,4 @@ if __name__ == "__main__": sys.stderr = open(os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 'logs', 'mcp_error.log'), 'w') # Initialize and run the server - mcp.run(transport='stdio') \ No newline at end of file + mcp.run(transport='stdio')