mikl/telegram-mcp
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor: Remove file handling functions from main.py and update README.md

Browse source 
Removed several functions that required direct file path access, including `send_file`, `download_media`, `set_profile_photo`, `edit_chat_photo`, `send_voice`, `send_sticker`, and `upload_file`, due to limitations in the current MCP environment. Updated the README to reflect these changes and added a section on removed functionality.
This commit is contained in:
anonim 2025-04-22 18:24:46 +03:00
parent 0833e51f48
commit d562cf61fb
2 changed files with 7 additions and 223 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

17
README.md
View file

@ -41,7 +41,6 @@ This MCP server exposes a huge suite of Telegram tools. **Every major Telegram/T
- **create_group(title, user_ids)**: Create a new group
- **create_channel(title, about, megagroup)**: Create a channel or supergroup
- **edit_chat_title(chat_id, title)**: Change chat/group/channel title
- **edit_chat_photo(chat_id, file_path)**: Set chat/group/channel photo
- **delete_chat_photo(chat_id)**: Remove chat/group/channel photo
- **leave_chat(chat_id)**: Leave a group or channel
- **get_participants(chat_id)**: List all participants
@ -70,6 +69,7 @@ This MCP server exposes a huge suite of Telegram tools. **Every major Telegram/T
- **get_message_context(chat_id, message_id, context_size)**: Context around a message
- **get_history(chat_id, limit)**: Full chat history
- **get_pinned_messages(chat_id)**: List pinned messages
- **get_last_interaction(contact_id)**: Most recent message with a contact
### Contact Management
- **list_contacts()**: List all contacts
@ -84,21 +84,15 @@ This MCP server exposes a huge suite of Telegram tools. **Every major Telegram/T
- **get_contact_ids()**: List all contact IDs
- **get_direct_chat_by_contact(contact_query)**: Find direct chat with a contact
- **get_contact_chats(contact_id)**: List all chats with a contact
- **get_last_interaction(contact_id)**: Most recent message with a contact
### User & Profile
- **get_me()**: Get your user info
- **update_profile(first_name, last_name, about)**: Update your profile
- **set_profile_photo(file_path)**: Set your profile photo
- **delete_profile_photo()**: Remove your profile photo
- **get_user_photos(user_id, limit)**: Get a user's profile photos
- **get_user_status(user_id)**: Get a user's online status
### Media
- **send_file(chat_id, file_path, caption)**: Send a file
- **send_voice(chat_id, file_path)**: Send a voice message
- **download_media(chat_id, message_id, file_path)**: Download media
- **upload_file(file_path)**: Upload a file to Telegram servers
- **get_media_info(chat_id, message_id)**: Get info about media in a message
### Search & Discovery
@ -108,9 +102,6 @@ This MCP server exposes a huge suite of Telegram tools. **Every major Telegram/T
### Stickers, GIFs, Bots
- **get_sticker_sets()**: List sticker sets
- **send_sticker(chat_id, file_path)**: Send a sticker
- **get_gif_search(query, limit)**: Search for GIFs
- **send_gif(chat_id, gif_id)**: Send a GIF
- **get_bot_info(bot_username)**: Get info about a bot
- **set_bot_commands(bot_username, commands)**: Set bot commands (bot accounts only)
@ -123,6 +114,12 @@ This MCP server exposes a huge suite of Telegram tools. **Every major Telegram/T
- **unarchive_chat(chat_id)**: Unarchive a chat
- **get_recent_actions(chat_id)**: Get recent admin actions
## Removed Functionality
Please note that tools requiring direct file path access on the server (`send_file`, `download_media`, `set_profile_photo`, `edit_chat_photo`, `send_voice`, `send_sticker`, `upload_file`) have been removed from `main.py`. This is due to limitations in the current MCP environment regarding handling file attachments and local file system paths.
Additionally, GIF-related tools (`get_gif_search`, `get_saved_gifs`, `send_gif`) have been removed due to ongoing issues with reliability in the Telethon library or Telegram API interactions.
---
## 📋 Requirements

213
main.py
View file

@ -996,53 +996,6 @@ async def get_participants(chat_id: int) -> str:
return log_and_format_error("get_participants", e, chat_id=chat_id)
@mcp.tool()
async def send_file(chat_id: int, file_path: str, caption: str = None) -> str:
"""
Send a file to a chat.
Args:
chat_id: The chat ID.
file_path: Absolute path to the file to send (must exist and be readable).
caption: Optional caption for the file.
"""
try:
if not os.path.isfile(file_path):
return f"File not found: {file_path}"
if not os.access(file_path, os.R_OK):
return f"File is not readable: {file_path}"
entity = await client.get_entity(chat_id)
await client.send_file(entity, file_path, caption=caption)
return f"File sent to chat {chat_id}."
except Exception as e:
return log_and_format_error("send_file", e, chat_id=chat_id, file_path=file_path, caption=caption)
@mcp.tool()
async def download_media(chat_id: int, message_id: int, file_path: str) -> str:
"""
Download media from a message in a chat.
Args:
chat_id: The chat ID.
message_id: The message ID containing the media.
file_path: Absolute path to save the downloaded file (must be writable).
"""
try:
entity = await client.get_entity(chat_id)
msg = await client.get_messages(entity, ids=message_id)
if not msg or not msg.media:
return "No media found in the specified message."
# Check if directory is writable
dir_path = os.path.dirname(file_path) or '.'
if not os.access(dir_path, os.W_OK):
return f"Directory not writable: {dir_path}"
await client.download_media(msg, file=file_path)
if not os.path.isfile(file_path):
return f"Download failed: file not created at {file_path}"
return f"Media downloaded to {file_path}."
except Exception as e:
return log_and_format_error("download_media", e, chat_id=chat_id, message_id=message_id, file_path=file_path)
@mcp.tool()
async def update_profile(first_name: str = None, last_name: str = None, about: str = None) -> str:
"""
@ -1059,20 +1012,6 @@ async def update_profile(first_name: str = None, last_name: str = None, about: s
return log_and_format_error("update_profile", e, first_name=first_name, last_name=last_name, about=about)
@mcp.tool()
async def set_profile_photo(file_path: str) -> str:
"""
Set a new profile photo.
"""
try:
await client(functions.photos.UploadProfilePhotoRequest(
file=await client.upload_file(file_path)
))
return "Profile photo updated."
except Exception as e:
return log_and_format_error("set_profile_photo", e, file_path=file_path)
@mcp.tool()
async def delete_profile_photo() -> str:
"""
@ -1278,37 +1217,6 @@ async def edit_chat_title(chat_id: int, title: str) -> str:
return log_and_format_error("edit_chat_title", e, chat_id=chat_id, title=title)
@mcp.tool()
async def edit_chat_photo(chat_id: int, file_path: str) -> str:
"""
Edit the photo of a chat, group, or channel. Requires a file path to an image.
"""
try:
if not os.path.isfile(file_path):
return f"Photo file not found: {file_path}"
if not os.access(file_path, os.R_OK):
return f"Photo file not readable: {file_path}"
entity = await client.get_entity(chat_id)
uploaded_file = await client.upload_file(file_path)
if isinstance(entity, Channel):
# For channels/supergroups, use EditPhotoRequest with InputChatUploadedPhoto
input_photo = InputChatUploadedPhoto(file=uploaded_file)
await client(functions.channels.EditPhotoRequest(channel=entity, photo=input_photo))
elif isinstance(entity, Chat):
# For basic groups, use EditChatPhotoRequest with InputChatUploadedPhoto
input_photo = InputChatUploadedPhoto(file=uploaded_file)
await client(functions.messages.EditChatPhotoRequest(chat_id=chat_id, photo=input_photo))
else:
return f"Cannot edit photo for this entity type ({type(entity)})."
return f"Chat {chat_id} photo updated."
except Exception as e:
logger.exception(f"edit_chat_photo failed (chat_id={chat_id}, file_path='{file_path}')")
return log_and_format_error("edit_chat_photo", e, chat_id=chat_id, file_path=file_path)
@mcp.tool()
async def delete_chat_photo(chat_id: int) -> str:
"""
@ -1749,29 +1657,6 @@ async def import_chat_invite(hash: str) -> str:
return log_and_format_error("import_chat_invite", e, hash=hash)
@mcp.tool()
async def send_voice(chat_id: int, file_path: str) -> str:
"""
Send a voice message to a chat. File must be an OGG/OPUS voice note.
Args:
chat_id: The chat ID.
file_path: Absolute path to the OGG/OPUS file.
"""
try:
if not os.path.isfile(file_path):
return f"File not found: {file_path}"
if not os.access(file_path, os.R_OK):
return f"File is not readable: {file_path}"
mime, _ = mimetypes.guess_type(file_path)
if not (mime and (mime == 'audio/ogg' or file_path.lower().endswith('.ogg') or file_path.lower().endswith('.opus'))):
return "Voice file must be .ogg or .opus format."
entity = await client.get_entity(chat_id)
await client.send_file(entity, file_path, voice_note=True)
return f"Voice message sent to chat {chat_id}."
except Exception as e:
return log_and_format_error("send_voice", e, chat_id=chat_id, file_path=file_path)
@mcp.tool()
async def forward_message(from_chat_id: int, message_id: int, to_chat_id: int) -> str:
"""
@ -1864,24 +1749,6 @@ async def reply_to_message(chat_id: int, message_id: int, text: str) -> str:
return log_and_format_error("reply_to_message", e, chat_id=chat_id, message_id=message_id, text=text)
@mcp.tool()
async def upload_file(file_path: str) -> str:
"""
Upload a file to Telegram servers (returns file handle as string, not a file path).
Args:
file_path: Absolute path to the file to upload (must exist and be readable).
"""
try:
if not os.path.isfile(file_path):
return f"File not found: {file_path}"
if not os.access(file_path, os.R_OK):
return f"File is not readable: {file_path}"
file = await client.upload_file(file_path)
return str(file)
except Exception as e:
return log_and_format_error("upload_file", e, file_path=file_path)
@mcp.tool()
async def get_media_info(chat_id: int, message_id: int) -> str:
"""
@ -2049,86 +1916,6 @@ async def get_sticker_sets() -> str:
return log_and_format_error("get_sticker_sets", e)
@mcp.tool()
async def send_sticker(chat_id: int, file_path: str) -> str:
"""
Send a sticker to a chat. File must be a valid .webp sticker file.
Args:
chat_id: The chat ID.
file_path: Absolute path to the .webp sticker file.
"""
try:
if not os.path.isfile(file_path):
return f"Sticker file not found: {file_path}"
if not os.access(file_path, os.R_OK):
return f"Sticker file is not readable: {file_path}"
if not file_path.lower().endswith('.webp'):
return "Sticker file must be a .webp file."
entity = await client.get_entity(chat_id)
await client.send_file(entity, file_path, force_document=False)
return f"Sticker sent to chat {chat_id}."
except Exception as e:
return log_and_format_error("send_sticker", e, chat_id=chat_id, file_path=file_path)
@mcp.tool()
async def get_gif_search(query: str, limit: int = 10) -> str:
"""
Search for GIFs by query. Returns a list of Telegram document IDs (not file paths).
Args:
query: Search term for GIFs.
limit: Max number of GIFs to return.
"""
try:
# Try approach 1: SearchGifsRequest
try:
result = await client(functions.messages.SearchGifsRequest(q=query, offset_id=0, limit=limit))
if not result.gifs:
return "[]"
return json.dumps([g.document.id for g in result.gifs], indent=2, default=json_serializer)
except (AttributeError, ImportError):
# Fallback approach: Use SearchRequest with GIF filter
try:
from telethon.tl.types import InputMessagesFilterGif
result = await client(functions.messages.SearchRequest(
peer="gif", q=query, filter=InputMessagesFilterGif(),
min_date=None, max_date=None, offset_id=0, add_offset=0,
limit=limit, max_id=0, min_id=0, hash=0
))
if not result or not hasattr(result, 'messages') or not result.messages:
return "[]"
# Extract document IDs from any messages with media
gif_ids = []
for msg in result.messages:
if hasattr(msg, 'media') and msg.media and hasattr(msg.media, 'document'):
gif_ids.append(msg.media.document.id)
return json.dumps(gif_ids, default=json_serializer)
except Exception as inner_e:
# Last resort: Try to fetch from a public bot
return f"Could not search GIFs using available methods: {inner_e}"
except Exception as e:
logger.exception(f"get_gif_search failed (query={query}, limit={limit})")
return log_and_format_error("get_gif_search", e, query=query, limit=limit)
@mcp.tool()
async def send_gif(chat_id: int, gif_id: int) -> str:
"""
Send a GIF to a chat by Telegram GIF document ID (not a file path).
Args:
chat_id: The chat ID.
gif_id: Telegram document ID for the GIF (from get_gif_search).
"""
try:
if not isinstance(gif_id, int):
return "gif_id must be a Telegram document ID (integer), not a file path. Use get_gif_search to find IDs."
entity = await client.get_entity(chat_id)
await client.send_file(entity, gif_id)
return f"GIF sent to chat {chat_id}."
except Exception as e:
return log_and_format_error("send_gif", e, chat_id=chat_id, gif_id=gif_id)
@mcp.tool()
async def get_bot_info(bot_username: str) -> str:
"""