77b4106fbe
- Add serialize_dataclass() helper to convert dataclass instances to JSON-compatible dicts - Convert datetime fields to ISO format strings for JSON serialization - Fix all MCP tool return values (list_messages, list_chats, search_contacts, etc.) - Add 'title' parameter to list_messages for easier chat lookup by name - Change date_range parameter to accept ISO date strings instead of datetime objects - Set include_context default to False for better performance Fixes compatibility issues where MCP tools were failing with 'tool compatibility issue' error because Python dataclasses with datetime fields cannot be directly JSON-serialized. All tools now return properly serialized dictionaries that work with the MCP protocol. Author: Danny Stocker Date: 2025-10-26
240 lines
8.1 KiB
Python
240 lines
8.1 KiB
Python
"""
|
|
Telegram MCP Server
|
|
|
|
The server uses FastMCP to expose Telegram functionality in a standardized way,
|
|
with each tool implemented as a decorated function. This allows Claude to:
|
|
1. Search contacts and chats
|
|
2. Retrieve message history with optional context
|
|
3. Send messages to individuals or groups
|
|
|
|
The server connects to the local SQLite database maintained by the Telegram Bridge,
|
|
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,
|
|
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")
|
|
|
|
# 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.
|
|
|
|
Args:
|
|
query: Search term to match against contact names or usernames
|
|
"""
|
|
contacts = telegram_search_contacts(query)
|
|
return [serialize_dataclass(c) for c in contacts]
|
|
|
|
@mcp.tool()
|
|
def list_messages(
|
|
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 = 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) 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 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=parsed_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 [serialize_dataclass(m) for m in 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 [serialize_dataclass(c) for c in 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 serialize_dataclass(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 serialize_dataclass(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 [serialize_dataclass(c) for c in 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 serialize_dataclass(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 serialize_dataclass(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 @), chat title, 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')
|