telegram-mcp/telegram-mcp-server/main.py

from typing import List, Dict, Any, Optional, Tuple
from mcp.server.fastmcp import FastMCP
from datetime import datetime
from telegram import (
    search_contacts as telegram_search_contacts,
    list_messages as telegram_list_messages,
    list_chats as telegram_list_chats,
    get_chat as telegram_get_chat,
    get_direct_chat_by_contact as telegram_get_direct_chat_by_contact,
    get_contact_chats as telegram_get_contact_chats,
    get_last_interaction as telegram_get_last_interaction,
    get_message_context as telegram_get_message_context,
    send_message as telegram_send_message
)

# Initialize FastMCP server
mcp = FastMCP("telegram")

@mcp.tool()
def search_contacts(query: str) -> List[Dict[str, Any]]:
    """Search Telegram contacts by name or username.
    
    Args:
        query: Search term to match against contact names or usernames
    """
    contacts = telegram_search_contacts(query)
    return contacts

@mcp.tool()
def list_messages(
    date_range: Optional[Tuple[datetime, datetime]] = None,
    sender_id: Optional[int] = None,
    chat_id: Optional[int] = None,
    query: Optional[str] = None,
    limit: int = 20,
    page: int = 0,
    include_context: bool = True,
    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
        sender_id: Optional sender ID to filter messages by sender
        chat_id: Optional chat ID to filter messages by chat
        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)
        context_before: Number of messages to include before each match (default 1)
        context_after: Number of messages to include after each match (default 1)
    """
    messages = telegram_list_messages(
        date_range=date_range,
        sender_id=sender_id,
        chat_id=chat_id,
        query=query,
        limit=limit,
        page=page,
        include_context=include_context,
        context_before=context_before,
        context_after=context_after
    )
    return messages

@mcp.tool()
def list_chats(
    query: Optional[str] = None,
    limit: int = 20,
    page: int = 0,
    chat_type: Optional[str] = None,
    sort_by: str = "last_active"
) -> List[Dict[str, Any]]:
    """Get Telegram chats matching specified criteria.
    
    Args:
        query: Optional search term to filter chats by name or username
        limit: Maximum number of chats to return (default 20)
        page: Page number for pagination (default 0)
        chat_type: Optional chat type filter ("user", "group", "channel", or "supergroup")
        sort_by: Field to sort results by, either "last_active" or "title" (default "last_active")
    """
    chats = telegram_list_chats(
        query=query,
        limit=limit,
        page=page,
        chat_type=chat_type,
        sort_by=sort_by
    )
    return chats

@mcp.tool()
def get_chat(chat_id: int) -> Dict[str, Any]:
    """Get Telegram chat metadata by ID.
    
    Args:
        chat_id: The ID of the chat to retrieve
    """
    chat = telegram_get_chat(chat_id)
    return chat

@mcp.tool()
def get_direct_chat_by_contact(contact_id: int) -> Dict[str, Any]:
    """Get Telegram chat metadata by contact ID.
    
    Args:
        contact_id: The contact ID to search for
    """
    chat = telegram_get_direct_chat_by_contact(contact_id)
    return chat

@mcp.tool()
def get_contact_chats(contact_id: int, limit: int = 20, page: int = 0) -> List[Dict[str, Any]]:
    """Get all Telegram chats involving the contact.
    
    Args:
        contact_id: The contact's ID to search for
        limit: Maximum number of chats to return (default 20)
        page: Page number for pagination (default 0)
    """
    chats = telegram_get_contact_chats(contact_id, limit, page)
    return chats

@mcp.tool()
def get_last_interaction(contact_id: int) -> Dict[str, Any]:
    """Get most recent Telegram message involving the contact.
    
    Args:
        contact_id: The ID of the contact to search for
    """
    message = telegram_get_last_interaction(contact_id)
    return message

@mcp.tool()
def get_message_context(
    message_id: int,
    chat_id: int,
    before: int = 5,
    after: int = 5
) -> Dict[str, Any]:
    """Get context around a specific Telegram message.
    
    Args:
        message_id: The ID of the message to get context for
        chat_id: The ID of the chat containing the message
        before: Number of messages to include before the target message (default 5)
        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

@mcp.tool()
def send_message(
    recipient: str,
    message: str
) -> Dict[str, Any]:
    """Send a Telegram message to a person or group.

    Args:
        recipient: The recipient - either a username (with or without @) or a chat ID
        message: The message text to send
    
    Returns:
        A dictionary containing success status and a status message
    """
    # Validate input
    if not recipient:
        return {
            "success": False,
            "message": "Recipient must be provided"
        }
    
    # Call the telegram_send_message function
    success, status_message = telegram_send_message(recipient, message)
    return {
        "success": success,
        "message": status_message
    }

if __name__ == "__main__":
    # Redirect stdout/stderr to suppress initial output that might confuse Claude
    import sys
    import os
    
    # Create logs directory
    os.makedirs(os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 'logs'), exist_ok=True)
    
    # Redirect stderr to log file
    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')