refactor telegram for mcp
This commit is contained in:
Muhammad18557
parent
5bc8ff4c70
commit
094ed9b4be
4 changed files with 158 additions and 180 deletions
29
telegram-mcp-server/telegram/__init__.py
Normal file
29
telegram-mcp-server/telegram/__init__.py
Normal file
|
|
@ -0,0 +1,29 @@
|
|||
"""
|
||||
This module implements data models and functions for retrieving and sending
|
||||
Telegram messages, managing chats, and working with contacts.
|
||||
|
||||
The module connects to a SQLite database that stores all Telegram messages and chat data,
|
||||
which is maintained by the Telegram Bridge. It also provides an HTTP client for sending
|
||||
messages via the Bridge's API endpoint.
|
||||
|
||||
Main features:
|
||||
- Data models for messages, chats, contacts, and message context
|
||||
- Database access functions for retrieving messages, chats, and contacts
|
||||
- HTTP client for sending messages through the Telegram Bridge
|
||||
- Helper functions for displaying formatted messages and chats
|
||||
"""
|
||||
|
||||
import os.path
|
||||
|
||||
# Database path
|
||||
MESSAGES_DB_PATH = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), '..', 'telegram-bridge', 'store', 'messages.db')
|
||||
TELEGRAM_API_BASE_URL = "http://localhost:8081/api"
|
||||
|
||||
# Import all components to make them available at the module level
|
||||
from .models import Message, Chat, Contact, MessageContext
|
||||
from .display import print_message, print_messages_list, print_chat, print_chats_list
|
||||
from .api import send_message
|
||||
from .database import (
|
||||
search_contacts, list_messages, get_message_context, list_chats,
|
||||
get_chat, get_direct_chat_by_contact, get_contact_chats, get_last_interaction
|
||||
)
|
||||
47
telegram-mcp-server/telegram/api.py
Normal file
47
telegram-mcp-server/telegram/api.py
Normal file
|
|
@ -0,0 +1,47 @@
|
|||
"""
|
||||
API client for interacting with the Telegram Bridge API.
|
||||
"""
|
||||
|
||||
import requests
|
||||
import json
|
||||
from typing import Tuple
|
||||
|
||||
from . import TELEGRAM_API_BASE_URL
|
||||
|
||||
def send_message(recipient: str, message: str) -> Tuple[bool, str]:
|
||||
"""Send a Telegram message to the specified recipient.
|
||||
|
||||
Args:
|
||||
recipient: The recipient - either a username (with or without @),
|
||||
or a chat ID as a string or integer
|
||||
message: The message text to send
|
||||
|
||||
Returns:
|
||||
Tuple[bool, str]: A tuple containing success status and a status message
|
||||
"""
|
||||
try:
|
||||
# Validate input
|
||||
if not recipient:
|
||||
return False, "Recipient must be provided"
|
||||
|
||||
url = f"{TELEGRAM_API_BASE_URL}/send"
|
||||
payload = {
|
||||
"recipient": recipient,
|
||||
"message": message
|
||||
}
|
||||
|
||||
response = requests.post(url, json=payload)
|
||||
|
||||
# Check if the request was successful
|
||||
if response.status_code == 200:
|
||||
result = response.json()
|
||||
return result.get("success", False), result.get("message", "Unknown response")
|
||||
else:
|
||||
return False, f"Error: HTTP {response.status_code} - {response.text}"
|
||||
|
||||
except requests.RequestException as e:
|
||||
return False, f"Request error: {str(e)}"
|
||||
except json.JSONDecodeError:
|
||||
return False, f"Error parsing response: Unknown"
|
||||
except Exception as e:
|
||||
return False, f"Unexpected error: {str(e)}"
|
||||
|
|
@ -1,150 +1,13 @@
|
|||
"""
|
||||
This module implements data models and functions for retrieving and sending
|
||||
Telegram messages, managing chats, and working with contacts.
|
||||
|
||||
The module connects to a SQLite database that stores all Telegram messages and chat data,
|
||||
which is maintained by the Telegram Bridge. It also provides an HTTP client for sending
|
||||
messages via the Bridge's API endpoint.
|
||||
|
||||
Main features:
|
||||
- Data models for messages, chats, contacts, and message context
|
||||
- Database access functions for retrieving messages, chats, and contacts
|
||||
- HTTP client for sending messages through the Telegram Bridge
|
||||
- Helper functions for displaying formatted messages and chats
|
||||
|
||||
All database operations use parameterized queries to prevent SQL injection.
|
||||
Database operations for retrieving and managing Telegram data.
|
||||
"""
|
||||
|
||||
import sqlite3
|
||||
import requests
|
||||
import json
|
||||
import os.path
|
||||
from datetime import datetime
|
||||
from dataclasses import dataclass
|
||||
from typing import Optional, List, Tuple, Dict, Any
|
||||
from typing import Optional, List, Tuple
|
||||
|
||||
# Database path
|
||||
MESSAGES_DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', 'telegram-bridge', 'store', 'messages.db')
|
||||
TELEGRAM_API_BASE_URL = "http://localhost:8081/api"
|
||||
|
||||
@dataclass
|
||||
class Message:
|
||||
"""
|
||||
Represents a Telegram message with all its metadata.
|
||||
|
||||
Attributes:
|
||||
id: Unique message identifier
|
||||
chat_id: ID of the chat the message belongs to
|
||||
chat_title: Title of the chat (user name, group name, etc.)
|
||||
sender_name: Name of the message sender
|
||||
content: Text content of the message
|
||||
timestamp: Date and time when the message was sent
|
||||
is_from_me: Boolean indicating if the message was sent by the user
|
||||
sender_id: ID of the message sender
|
||||
"""
|
||||
id: int
|
||||
chat_id: int
|
||||
chat_title: str
|
||||
sender_name: str
|
||||
content: str
|
||||
timestamp: datetime
|
||||
is_from_me: bool
|
||||
sender_id: int
|
||||
|
||||
@dataclass
|
||||
class Chat:
|
||||
"""
|
||||
Represents a Telegram chat (direct message, group, channel, etc.).
|
||||
|
||||
Attributes:
|
||||
id: Unique chat identifier
|
||||
title: Name of the chat (user name, group name, etc.)
|
||||
username: Optional Telegram username (without @)
|
||||
type: Type of chat ('user', 'group', 'channel', 'supergroup')
|
||||
last_message_time: Timestamp of the most recent message in the chat
|
||||
"""
|
||||
id: int
|
||||
title: str
|
||||
username: Optional[str]
|
||||
type: str
|
||||
last_message_time: Optional[datetime]
|
||||
|
||||
@dataclass
|
||||
class Contact:
|
||||
"""
|
||||
Represents a Telegram contact.
|
||||
|
||||
Attributes:
|
||||
id: Unique contact identifier
|
||||
username: Optional Telegram username (without @)
|
||||
name: Display name of the contact
|
||||
"""
|
||||
id: int
|
||||
username: Optional[str]
|
||||
name: str
|
||||
|
||||
@dataclass
|
||||
class MessageContext:
|
||||
"""
|
||||
Provides context around a specific message.
|
||||
|
||||
Attributes:
|
||||
message: The target message
|
||||
before: List of messages that came before the target message
|
||||
after: List of messages that came after the target message
|
||||
"""
|
||||
message: Message
|
||||
before: List[Message]
|
||||
after: List[Message]
|
||||
|
||||
def print_message(message: Message, show_chat_info: bool = True) -> None:
|
||||
"""Print a single message with consistent formatting."""
|
||||
direction = "→" if message.is_from_me else "←"
|
||||
|
||||
if show_chat_info:
|
||||
print(f"[{message.timestamp:%Y-%m-%d %H:%M:%S}] {direction} Chat: {message.chat_title} (ID: {message.chat_id})")
|
||||
else:
|
||||
print(f"[{message.timestamp:%Y-%m-%d %H:%M:%S}] {direction}")
|
||||
|
||||
print(f"From: {'Me' if message.is_from_me else message.sender_name}")
|
||||
print(f"Message: {message.content}")
|
||||
print("-" * 100)
|
||||
|
||||
def print_messages_list(messages: List[Message], title: str = "", show_chat_info: bool = True) -> None:
|
||||
"""Print a list of messages with a title and consistent formatting."""
|
||||
if not messages:
|
||||
print("No messages to display.")
|
||||
return
|
||||
|
||||
if title:
|
||||
print(f"\n{title}")
|
||||
print("-" * 100)
|
||||
|
||||
for message in messages:
|
||||
print_message(message, show_chat_info)
|
||||
|
||||
def print_chat(chat: Chat) -> None:
|
||||
"""Print a single chat with consistent formatting."""
|
||||
print(f"Chat: {chat.title} (ID: {chat.id})")
|
||||
print(f"Type: {chat.type}")
|
||||
if chat.username:
|
||||
print(f"Username: @{chat.username}")
|
||||
if chat.last_message_time:
|
||||
print(f"Last active: {chat.last_message_time:%Y-%m-%d %H:%M:%S}")
|
||||
print("-" * 100)
|
||||
|
||||
def print_chats_list(chats: List[Chat], title: str = "") -> None:
|
||||
"""Print a list of chats with a title and consistent formatting."""
|
||||
if not chats:
|
||||
print("No chats to display.")
|
||||
return
|
||||
|
||||
if title:
|
||||
print(f"\n{title}")
|
||||
print("-" * 100)
|
||||
|
||||
for chat in chats:
|
||||
print_chat(chat)
|
||||
from . import MESSAGES_DB_PATH
|
||||
from .models import Message, Chat, Contact, MessageContext
|
||||
|
||||
def search_contacts(query: str) -> List[Contact]:
|
||||
"""Search contacts by name or username."""
|
||||
|
|
@ -579,42 +442,4 @@ def get_last_interaction(contact_id: int) -> Optional[Message]:
|
|||
return None
|
||||
finally:
|
||||
if 'conn' in locals():
|
||||
conn.close()
|
||||
|
||||
def send_message(recipient: str, message: str) -> Tuple[bool, str]:
|
||||
"""Send a Telegram message to the specified recipient.
|
||||
|
||||
Args:
|
||||
recipient: The recipient - either a username (with or without @),
|
||||
or a chat ID as a string or integer
|
||||
message: The message text to send
|
||||
|
||||
Returns:
|
||||
Tuple[bool, str]: A tuple containing success status and a status message
|
||||
"""
|
||||
try:
|
||||
# Validate input
|
||||
if not recipient:
|
||||
return False, "Recipient must be provided"
|
||||
|
||||
url = f"{TELEGRAM_API_BASE_URL}/send"
|
||||
payload = {
|
||||
"recipient": recipient,
|
||||
"message": message
|
||||
}
|
||||
|
||||
response = requests.post(url, json=payload)
|
||||
|
||||
# Check if the request was successful
|
||||
if response.status_code == 200:
|
||||
result = response.json()
|
||||
return result.get("success", False), result.get("message", "Unknown response")
|
||||
else:
|
||||
return False, f"Error: HTTP {response.status_code} - {response.text}"
|
||||
|
||||
except requests.RequestException as e:
|
||||
return False, f"Request error: {str(e)}"
|
||||
except json.JSONDecodeError:
|
||||
return False, f"Error parsing response: Unknown"
|
||||
except Exception as e:
|
||||
return False, f"Unexpected error: {str(e)}"
|
||||
conn.close()
|
||||
77
telegram-mcp-server/telegram/models.py
Normal file
77
telegram-mcp-server/telegram/models.py
Normal file
|
|
@ -0,0 +1,77 @@
|
|||
"""
|
||||
Data models for Telegram entities.
|
||||
"""
|
||||
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime
|
||||
from typing import Optional, List
|
||||
|
||||
@dataclass
|
||||
class Message:
|
||||
"""
|
||||
Represents a Telegram message with all its metadata.
|
||||
|
||||
Attributes:
|
||||
id: Unique message identifier
|
||||
chat_id: ID of the chat the message belongs to
|
||||
chat_title: Title of the chat (user name, group name, etc.)
|
||||
sender_name: Name of the message sender
|
||||
content: Text content of the message
|
||||
timestamp: Date and time when the message was sent
|
||||
is_from_me: Boolean indicating if the message was sent by the user
|
||||
sender_id: ID of the message sender
|
||||
"""
|
||||
id: int
|
||||
chat_id: int
|
||||
chat_title: str
|
||||
sender_name: str
|
||||
content: str
|
||||
timestamp: datetime
|
||||
is_from_me: bool
|
||||
sender_id: int
|
||||
|
||||
@dataclass
|
||||
class Chat:
|
||||
"""
|
||||
Represents a Telegram chat (direct message, group, channel, etc.).
|
||||
|
||||
Attributes:
|
||||
id: Unique chat identifier
|
||||
title: Name of the chat (user name, group name, etc.)
|
||||
username: Optional Telegram username (without @)
|
||||
type: Type of chat ('user', 'group', 'channel', 'supergroup')
|
||||
last_message_time: Timestamp of the most recent message in the chat
|
||||
"""
|
||||
id: int
|
||||
title: str
|
||||
username: Optional[str]
|
||||
type: str
|
||||
last_message_time: Optional[datetime]
|
||||
|
||||
@dataclass
|
||||
class Contact:
|
||||
"""
|
||||
Represents a Telegram contact.
|
||||
|
||||
Attributes:
|
||||
id: Unique contact identifier
|
||||
username: Optional Telegram username (without @)
|
||||
name: Display name of the contact
|
||||
"""
|
||||
id: int
|
||||
username: Optional[str]
|
||||
name: str
|
||||
|
||||
@dataclass
|
||||
class MessageContext:
|
||||
"""
|
||||
Provides context around a specific message.
|
||||
|
||||
Attributes:
|
||||
message: The target message
|
||||
before: List of messages that came before the target message
|
||||
after: List of messages that came after the target message
|
||||
"""
|
||||
message: Message
|
||||
before: List[Message]
|
||||
after: List[Message]
|
||||
Loading…
Add table
Reference in a new issue