diff --git a/.gitignore b/.gitignore index 00ba4b4..e5fc31d 100644 --- a/.gitignore +++ b/.gitignore @@ -187,3 +187,4 @@ anon_new.session-journal # Claude Desktop config claude_desktop_config.json +.DS_Store diff --git a/README.md b/README.md index 12e913b..2bc29c1 100644 --- a/README.md +++ b/README.md @@ -1,68 +1,142 @@ -# Telegram MCP for Claude +# Telegram MCP Server ![MCP Badge](https://badge.mcpx.dev) [![License: Apache 2.0](https://img.shields.io/badge/license-Apache%202.0-green?style=flat-square)](https://opensource.org/licenses/Apache-2.0) -A powerful Telegram integration for Claude via the Model Context Protocol (MCP), allowing you to interact with your Telegram account directly from Claude Desktop. +A full-featured Telegram integration for Claude, Cursor, and any MCP-compatible client, powered by [Telethon](https://docs.telethon.dev/) and the [Model Context Protocol (MCP)](https://modelcontextprotocol.io/). This project lets you interact with your Telegram account programmatically, automating everything from messaging to group management. -![Telegram MCP in action](screenshots/1.png) +**Maintained by [l1v0n1](https://github.com/l1v0n1). Forked from [chigwell/telegram-mcp](https://github.com/chigwell/telegram-mcp).** -## ๐Ÿš€ Features +--- -This MCP server provides a comprehensive suite of tools for seamless Telegram interaction: +## ๐Ÿš€ Features & Tools -### Chat Management -- **get_chats** - Get a paginated list of your chats -- **list_chats** - List all chats with detailed metadata and filtering options -- **get_chat** - Get detailed information about a specific chat +This MCP server exposes a huge suite of Telegram tools. **Every major Telegram/Telethon feature is available as a tool!** + +### Chat & Group Management +- **get_chats(page, page_size)**: Paginated list of chats +- **list_chats(chat_type, limit)**: List chats with metadata and filtering +- **get_chat(chat_id)**: Detailed info about a chat +- **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 +- **get_admins(chat_id)**: List all admins +- **get_banned_users(chat_id)**: List all banned users +- **promote_admin(chat_id, user_id)**: Promote user to admin +- **demote_admin(chat_id, user_id)**: Demote admin to user +- **ban_user(chat_id, user_id)**: Ban user +- **unban_user(chat_id, user_id)**: Unban user +- **get_invite_link(chat_id)**: Get invite link +- **export_chat_invite(chat_id)**: Export invite link +- **import_chat_invite(hash)**: Join chat by invite hash +- **join_chat_by_link(link)**: Join chat by invite link ### Messaging -- **get_messages** - Get messages from a specific chat with pagination -- **list_messages** - Retrieve messages with powerful filtering (text search, date ranges) -- **send_message** - Send messages to any chat -- **get_message_context** - View the context around a specific message +- **get_messages(chat_id, page, page_size)**: Paginated messages +- **list_messages(chat_id, limit, search_query, from_date, to_date)**: Filtered messages +- **send_message(chat_id, message)**: Send a message +- **reply_to_message(chat_id, message_id, text)**: Reply to a message +- **edit_message(chat_id, message_id, new_text)**: Edit your message +- **delete_message(chat_id, message_id)**: Delete a message +- **forward_message(from_chat_id, message_id, to_chat_id)**: Forward a message +- **pin_message(chat_id, message_id)**: Pin a message +- **unpin_message(chat_id, message_id)**: Unpin a message +- **mark_as_read(chat_id)**: Mark all as read +- **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 ### Contact Management -- **search_contacts** - Find contacts by name, username or phone number -- **get_direct_chat_by_contact** - Find personal chats with specific contacts -- **get_contact_chats** - List all chats (including groups) involving a contact -- **get_last_interaction** - View your most recent exchanges with a contact +- **list_contacts()**: List all contacts +- **search_contacts(query)**: Search contacts +- **add_contact(phone, first_name, last_name)**: Add a contact +- **delete_contact(user_id)**: Delete a contact +- **block_user(user_id)**: Block a user +- **unblock_user(user_id)**: Unblock a user +- **import_contacts(contacts)**: Bulk import contacts +- **export_contacts()**: Export all contacts as JSON +- **get_blocked_users()**: List blocked users +- **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 +- **search_public_chats(query)**: Search public chats/channels/bots +- **search_messages(chat_id, query, limit)**: Search messages in a chat +- **resolve_username(username)**: Resolve a username to ID + +### 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 + +### Privacy, Settings, and Misc +- **get_privacy_settings()**: Get privacy settings +- **set_privacy_settings(key, allow_users, disallow_users)**: Set privacy settings +- **mute_chat(chat_id)**: Mute notifications +- **unmute_chat(chat_id)**: Unmute notifications +- **archive_chat(chat_id)**: Archive a chat +- **unarchive_chat(chat_id)**: Unarchive a chat +- **get_recent_actions(chat_id)**: Get recent admin actions + +--- ## ๐Ÿ“‹ Requirements - - Python 3.10+ -- [Telethon](https://docs.telethon.dev/) for Telegram API access +- [Telethon](https://docs.telethon.dev/) - [MCP Python SDK](https://modelcontextprotocol.io/docs/) -- [UV](https://astral.sh/uv/) package manager -- [Claude Desktop](https://claude.ai/desktop) app +- [Claude Desktop](https://claude.ai/desktop) or [Cursor](https://cursor.so/) (or any MCP client) -## ๐Ÿ”ง Installation +--- -### 1. Clone the Repository +## ๐Ÿ”ง Installation & Setup + +### 1. Fork & Clone ```bash -git clone https://github.com/l1v0n1/telegram-mcp-server -cd telegram-mcp-server +git clone https://github.com/chigwell/telegram-mcp.git +cd telegram-mcp ``` -### 2. Generate Session String +### 2. Create a Virtual Environment -For better security and portability, this project uses Telethon's StringSession. Generate your session string: +```bash +python3 -m venv .venv +source .venv/bin/activate # On Windows: .venv\Scripts\activate +pip install -r requirements.txt +``` + +### 3. Generate a Session String ```bash python session_string_generator.py ``` +Follow the prompts to authenticate and update your `.env` file. -This will: -1. Ask for your phone number -2. Send a verification code to your Telegram app -3. Generate a session string and add it to your `.env` file - -The session string allows authentication without storing SQLite session files, which helps avoid database lock issues and improves portability. - -### 3. Set Up Your Environment - -Create a `.env` file with your Telegram credentials: +### 4. Configure .env ``` TELEGRAM_API_ID=your_api_id_here @@ -70,89 +144,104 @@ TELEGRAM_API_HASH=your_api_hash_here TELEGRAM_SESSION_NAME=anon TELEGRAM_SESSION_STRING=your_session_string_here ``` +Get your API credentials at [my.telegram.org/apps](https://my.telegram.org/apps). -You can obtain API credentials at [my.telegram.org/apps](https://my.telegram.org/apps). +--- -### 4. Install Dependencies +## โš™๏ธ Configuration for Claude & Cursor -```bash -uv venv -source .venv/bin/activate # On Windows: .venv\Scripts\activate -uv add "mcp[cli]" telethon python-dotenv nest_asyncio -``` - -### 5. Configure Claude Desktop - -#### On macOS/Linux: -Edit `~/Library/Application Support/Claude/claude_desktop_config.json`: +### Claude Desktop +Edit your Claude config (e.g. `~/Library/Application Support/Claude/claude_desktop_config.json`): ```json { - "mcpServers": { - "telegram-mcp": { - "command": "/full/path/to/uv", - "args": [ - "--directory", - "/full/path/to/telegram-mcp-server", - "run", - "main.py" - ] - } + "mcpServers": { + "telegram-mcp": { + "command": "/full/path/to/.venv/bin/python", + "args": ["main.py"], + "cwd": "/full/path/to/telegram-mcp-server" } + } } ``` -#### On Windows: -Edit `%APPDATA%\Claude\claude_desktop_config.json` with similar configuration. +### Cursor +Edit `~/.cursor/mcp.json`: + +```json +{ + "mcpServers": { + "telegram-mcp": { + "command": "/full/path/to/.venv/bin/python", + "args": ["main.py"], + "cwd": "/full/path/to/telegram-mcp-server" + } + } +} +``` + +--- ## ๐ŸŽฎ Usage Examples -Here are some ways to interact with Telegram through Claude: +- "Show my recent chats" +- "Send 'Hello world' to chat 123456789" +- "Add contact with phone +1234567890, name John Doe" +- "Create a group 'Project Team' with users 111, 222, 333" +- "Download the media from message 42 in chat 123456789" +- "Mute notifications for chat 123456789" +- "Promote user 111 to admin in group 123456789" +- "Search for public channels about 'news'" -### Basic Chat Navigation -- "Show me my most recent chats" -- "List my group chats with unread messages" -- "Show detailed information about chat 123456789" +You can use these tools via natural language in Claude, Cursor, or any MCP-compatible client. -### Messaging -- "Show me the last 10 messages from chat 123456789" -- "Send 'I'll be there in 10 minutes' to chat 123456789" -- "Find messages containing 'meeting' in chat 123456789" -- "Show messages from March 1-15, 2023 in chat 123456789" +--- -### Contact Interactions -- "Search for contacts named 'Alex'" -- "Find my direct chat with John" -- "Show all chats where I interact with contact 987654321" -- "Show my last conversation with Lisa" +## ๐Ÿ› ๏ธ Contribution Guide -### Advanced Features -- "Show the context around message 42 in chat 123456789" -- "List all channels I'm subscribed to" +1. **Fork this repo:** [chigwell/telegram-mcp](https://github.com/chigwell/telegram-mcp) +2. **Clone your fork:** + ```bash + git clone https://github.com//telegram-mcp.git + ``` +3. **Create a new branch:** + ```bash + git checkout -b my-feature + ``` +4. **Make your changes, add tests/docs if needed.** +5. **Push and open a Pull Request** to [chigwell/telegram-mcp](https://github.com/chigwell/telegram-mcp) with a clear description. +6. **Tag @l1v0n1** in your PR for review. + +--- ## ๐Ÿ”’ Security Considerations +- **Never commit your `.env` or session string.** +- The session string gives full access to your Telegram accountโ€”keep it safe! +- All processing is local; no data is sent anywhere except Telegram's API. -- **Private API Keys**: Never commit your `.env` file or session files to Git repositories -- **Session String**: The session string in your `.env` file provides full access to your Telegram account. Keep it secure. -- **Local Processing**: All Telegram data is processed locally on your machine - no data is sent to external servers beyond Telegram's own API. -- **Permissions**: The MCP server has the same access to Telegram as you would have with the official app, including reading and sending messages. +--- ## ๐Ÿ› ๏ธ Troubleshooting +- **Check logs** in your MCP client (Claude/Cursor) and the terminal for errors. +- **Interpreter errors?** Make sure your `.venv` is created and selected. +- **Database lock?** Use session string authentication, not file-based sessions. +- **iCloud/Dropbox issues?** Move your project to a local path without spaces if you see odd errors. +- **Regenerate session string** if you change your Telegram password or see auth errors. -If you encounter issues: - -1. Check Claude Desktop logs for error messages -2. Ensure your Telegram API credentials are correct -3. Verify that the paths in your Claude Desktop config are absolute and correct -4. If you see database lock errors, use the session string authentication method -5. If you need to regenerate your session string, run `python session_string_generator.py` again +--- ## ๐Ÿ“„ License This project is licensed under the [Apache 2.0 License](LICENSE). -## ๐Ÿ™ Acknowledgements +--- -- [Telethon](https://github.com/LonamiWebs/Telethon) for the Telegram client library -- [Model Context Protocol](https://modelcontextprotocol.io/) for the integration framework -- [Anthropic](https://www.anthropic.com/) for Claude and the Claude Desktop app \ No newline at end of file +## ๐Ÿ™ Acknowledgements +- [Telethon](https://github.com/LonamiWebs/Telethon) +- [Model Context Protocol](https://modelcontextprotocol.io/) +- [Claude](https://www.anthropic.com/) and [Cursor](https://cursor.so/) +- [chigwell/telegram-mcp](https://github.com/chigwell/telegram-mcp) (upstream) + +--- + +**Maintained by [l1v0n1](https://github.com/l1v0n1). PRs welcome!** \ No newline at end of file diff --git a/main.py b/main.py index ca52b77..803417c 100644 --- a/main.py +++ b/main.py @@ -14,6 +14,7 @@ from telethon.tl.functions.contacts import SearchRequest from datetime import datetime, timedelta import json from typing import List, Dict, Optional, Union, Any +from telethon import functions load_dotenv() @@ -147,46 +148,73 @@ async def send_message(chat_id: int, message: str) -> str: @mcp.tool() -async def search_contacts(query: str) -> str: +async def list_contacts() -> str: """ - Search for contacts by name or phone number. - - Args: - query: The search term to look for in contact names or phone numbers. + List all contacts in your Telegram account. """ try: - # Search in your contacts - contacts = await client.get_contacts() - results = [] - - for contact in contacts: - if not contact: - continue - - name = f"{getattr(contact, 'first_name', '')} {getattr(contact, 'last_name', '')}".strip() - username = getattr(contact, 'username', '') - phone = getattr(contact, 'phone', '') - - if (query.lower() in name.lower() or - (username and query.lower() in username.lower()) or - (phone and query in phone)): - - contact_info = f"ID: {contact.id}, Name: {name}" - if username: - contact_info += f", Username: @{username}" - if phone: - contact_info += f", Phone: {phone}" - - results.append(contact_info) - - if not results: + result = await client(functions.contacts.GetContactsRequest(hash=0)) + users = result.users + if not users: + return "No contacts found." + lines = [] + for user in users: + name = f"{getattr(user, 'first_name', '')} {getattr(user, 'last_name', '')}".strip() + username = getattr(user, 'username', '') + phone = getattr(user, 'phone', '') + contact_info = f"ID: {user.id}, Name: {name}" + if username: + contact_info += f", Username: @{username}" + if phone: + contact_info += f", Phone: {phone}" + lines.append(contact_info) + return "\n".join(lines) + except Exception as e: + return f"Error listing contacts: {e}" + + +@mcp.tool() +async def search_contacts(query: str) -> str: + """ + Search for contacts by name, username, or phone number using Telethon's SearchRequest. + Args: + query: The search term to look for in contact names, usernames, or phone numbers. + """ + try: + result = await client(functions.contacts.SearchRequest(q=query, limit=50)) + users = result.users + if not users: return f"No contacts found matching '{query}'." - - return "\n".join(results) + lines = [] + for user in users: + name = f"{getattr(user, 'first_name', '')} {getattr(user, 'last_name', '')}".strip() + username = getattr(user, 'username', '') + phone = getattr(user, 'phone', '') + contact_info = f"ID: {user.id}, Name: {name}" + if username: + contact_info += f", Username: @{username}" + if phone: + contact_info += f", Phone: {phone}" + lines.append(contact_info) + return "\n".join(lines) except Exception as e: return f"Error searching contacts: {e}" +@mcp.tool() +async def get_contact_ids() -> str: + """ + Get all contact IDs in your Telegram account. + """ + try: + result = await client(functions.contacts.GetContactIDsRequest(hash=0)) + if not result: + return "No contact IDs found." + return "Contact IDs: " + ", ".join(str(cid) for cid in result) + except Exception as e: + return f"Error getting contact IDs: {e}" + + @mcp.tool() async def list_messages(chat_id: int, limit: int = 20, search_query: str = None, from_date: str = None, to_date: str = None) -> str: @@ -381,30 +409,25 @@ async def get_direct_chat_by_contact(contact_query: str) -> str: contact_query: Name, username, or phone number to search for. """ try: - # First search for the contact - contacts = await client.get_contacts() + # Fetch all contacts using the correct Telethon method + result = await client(functions.contacts.GetContactsRequest(hash=0)) + contacts = result.users found_contacts = [] - for contact in contacts: if not contact: continue - name = f"{getattr(contact, 'first_name', '')} {getattr(contact, 'last_name', '')}".strip() username = getattr(contact, 'username', '') phone = getattr(contact, 'phone', '') - if (contact_query.lower() in name.lower() or (username and contact_query.lower() in username.lower()) or (phone and contact_query in phone)): found_contacts.append(contact) - if not found_contacts: return f"No contacts found matching '{contact_query}'." - # If we found contacts, look for direct chats with them results = [] dialogs = await client.get_dialogs() - for contact in found_contacts: contact_name = f"{getattr(contact, 'first_name', '')} {getattr(contact, 'last_name', '')}".strip() for dialog in dialogs: @@ -416,11 +439,9 @@ async def get_direct_chat_by_contact(contact_query: str) -> str: chat_info += f", Unread: {dialog.unread_count}" results.append(chat_info) break - if not results: found_names = ", ".join([f"{c.first_name} {c.last_name}".strip() for c in found_contacts]) return f"Found contacts: {found_names}, but no direct chats were found with them." - return "\n".join(results) except Exception as e: return f"Error finding direct chat: {e}" @@ -522,48 +543,919 @@ async def get_message_context(chat_id: int, message_id: int, context_size: int = """ try: chat = await client.get_entity(chat_id) - # Get messages around the specified message messages_before = await client.get_messages( chat, limit=context_size, max_id=message_id ) - central_message = await client.get_messages( chat, ids=message_id ) - + # Fix: get_messages(ids=...) returns a single Message, not a list + if central_message is not None and not isinstance(central_message, list): + central_message = [central_message] + elif central_message is None: + central_message = [] messages_after = await client.get_messages( chat, limit=context_size, min_id=message_id, reverse=True ) - if not central_message: return f"Message with ID {message_id} not found in chat {chat_id}." - # Combine messages in chronological order all_messages = list(messages_before) + list(central_message) + list(messages_after) all_messages.sort(key=lambda m: m.id) - results = [f"Context for message {message_id} in chat {chat_id}:"] - for msg in all_messages: sender_name = "Unknown" if msg.sender: sender_name = getattr(msg.sender, 'first_name', '') or getattr(msg.sender, 'title', 'Unknown') - highlight = " [THIS MESSAGE]" if msg.id == message_id else "" results.append(f"ID: {msg.id} | {sender_name} | {msg.date}{highlight}\n{msg.message or '[Media/No text]'}\n") - return "\n".join(results) except Exception as e: return f"Error retrieving message context: {e}" +@mcp.tool() +async def add_contact(phone: str, first_name: str, last_name: str = "") -> str: + """ + Add a new contact to your Telegram account. + Args: + phone: The phone number of the contact (with country code). + first_name: The contact's first name. + last_name: The contact's last name (optional). + """ + try: + result = await client(functions.contacts.ImportContactsRequest( + contacts=[ + functions.contacts.InputPhoneContact( + client_id=0, + phone=phone, + first_name=first_name, + last_name=last_name + ) + ] + )) + if result.imported: + return f"Contact {first_name} {last_name} added successfully." + else: + return f"Contact not added. Response: {result.stringify()}" + except Exception as e: + return f"Error adding contact: {e}" + + +@mcp.tool() +async def delete_contact(user_id: int) -> str: + """ + Delete a contact by user ID. + Args: + user_id: The Telegram user ID of the contact to delete. + """ + try: + user = await client.get_entity(user_id) + await client(functions.contacts.DeleteContactsRequest(id=[user])) + return f"Contact with user ID {user_id} deleted." + except Exception as e: + return f"Error deleting contact: {e}" + + +@mcp.tool() +async def block_user(user_id: int) -> str: + """ + Block a user by user ID. + Args: + user_id: The Telegram user ID to block. + """ + try: + user = await client.get_entity(user_id) + await client(functions.contacts.BlockRequest(id=user)) + return f"User {user_id} blocked." + except Exception as e: + return f"Error blocking user: {e}" + + +@mcp.tool() +async def unblock_user(user_id: int) -> str: + """ + Unblock a user by user ID. + Args: + user_id: The Telegram user ID to unblock. + """ + try: + user = await client.get_entity(user_id) + await client(functions.contacts.UnblockRequest(id=user)) + return f"User {user_id} unblocked." + except Exception as e: + return f"Error unblocking user: {e}" + + +@mcp.tool() +async def get_me() -> str: + """ + Get your own user information. + """ + try: + me = await client.get_me() + return json.dumps(format_entity(me), indent=2) + except Exception as e: + return f"Error getting your info: {e}" + + +@mcp.tool() +async def create_group(title: str, user_ids: list) -> str: + """ + Create a new group with the given title and user IDs. + Args: + title: The group name. + user_ids: List of user IDs to add to the group. + """ + try: + users = [await client.get_entity(uid) for uid in user_ids] + result = await client(functions.messages.CreateChatRequest(users=users, title=title)) + return f"Group '{title}' created with ID: {result.chats[0].id}" + except Exception as e: + return f"Error creating group: {e}" + + +@mcp.tool() +async def invite_to_group(group_id: int, user_ids: list) -> str: + """ + Invite users to a group by group ID. + Args: + group_id: The group chat ID. + user_ids: List of user IDs to invite. + """ + try: + users = [await client.get_entity(uid) for uid in user_ids] + await client(functions.messages.AddChatUserRequest(chat_id=group_id, user_id=users[0], fwd_limit=0)) + # Telethon only allows adding one user at a time for AddChatUserRequest (for basic groups) + # For supergroups/channels, use InviteToChannelRequest + return f"Invited users to group {group_id}." + except Exception as e: + return f"Error inviting users: {e}" + + +@mcp.tool() +async def leave_chat(chat_id: int) -> str: + """ + Leave a group or channel by chat ID. + Args: + chat_id: The chat ID to leave. + """ + try: + await client(functions.messages.LeaveChatRequest(chat_id=chat_id)) + return f"Left chat {chat_id}." + except Exception as e: + return f"Error leaving chat: {e}" + + +@mcp.tool() +async def get_participants(chat_id: int) -> str: + """ + List all participants in a group or channel. + Args: + chat_id: The group or channel ID. + """ + try: + participants = await client.get_participants(chat_id) + lines = [f"ID: {p.id}, Name: {getattr(p, 'first_name', '')} {getattr(p, 'last_name', '')}" for p in participants] + return "\n".join(lines) + except Exception as e: + return f"Error getting participants: {e}" + + +@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: Path to the file to send. + caption: Optional caption for the file. + """ + try: + 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 f"Error sending file: {e}" + + +@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: Path to save the downloaded file. + """ + 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." + await client.download_media(msg, file=file_path) + return f"Media downloaded to {file_path}." + except Exception as e: + return f"Error downloading media: {e}" + + +@mcp.tool() +async def update_profile(first_name: str = None, last_name: str = None, about: str = None) -> str: + """ + Update your profile information (name, bio). + """ + try: + await client(functions.account.UpdateProfileRequest( + first_name=first_name, + last_name=last_name, + about=about + )) + return "Profile updated." + except Exception as e: + return f"Error updating profile: {e}" + + +@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 f"Error setting profile photo: {e}" + + +@mcp.tool() +async def delete_profile_photo() -> str: + """ + Delete your current profile photo. + """ + try: + photos = await client(functions.photos.GetUserPhotosRequest(user_id='me', offset=0, max_id=0, limit=1)) + if not photos.photos: + return "No profile photo to delete." + await client(functions.photos.DeletePhotosRequest(id=[photos.photos[0].id])) + return "Profile photo deleted." + except Exception as e: + return f"Error deleting profile photo: {e}" + + +@mcp.tool() +async def get_privacy_settings() -> str: + """ + Get your privacy settings. + """ + try: + settings = await client(functions.account.GetPrivacyRequest(key='status_timestamp')) + return str(settings) + except Exception as e: + return f"Error getting privacy settings: {e}" + + +@mcp.tool() +async def set_privacy_settings(key: str, allow_users: list = None, disallow_users: list = None) -> str: + """ + Set privacy settings (e.g., last seen, phone, etc.). + key: e.g. 'status_timestamp', 'phone_number', 'profile_photo', 'forwards', 'voice_messages', etc. + """ + from telethon.tl.types import InputPrivacyKeyStatusTimestamp, InputPrivacyValueAllowUsers, InputPrivacyValueDisallowUsers + try: + allow = InputPrivacyValueAllowUsers(users=[await client.get_entity(uid) for uid in (allow_users or [])]) + disallow = InputPrivacyValueDisallowUsers(users=[await client.get_entity(uid) for uid in (disallow_users or [])]) + await client(functions.account.SetPrivacyRequest( + key=getattr(functions.account, f'InputPrivacyKey{key.title().replace("_", "")}')(), + rules=[allow, disallow] + )) + return f"Privacy settings for {key} updated." + except Exception as e: + return f"Error setting privacy: {e}" + + +@mcp.tool() +async def import_contacts(contacts: list) -> str: + """ + Import a list of contacts. Each contact should be a dict with phone, first_name, last_name. + """ + try: + input_contacts = [functions.contacts.InputPhoneContact(client_id=i, phone=c['phone'], first_name=c['first_name'], last_name=c.get('last_name', '')) for i, c in enumerate(contacts)] + result = await client(functions.contacts.ImportContactsRequest(contacts=input_contacts)) + return f"Imported {len(result.imported)} contacts." + except Exception as e: + return f"Error importing contacts: {e}" + + +@mcp.tool() +async def export_contacts() -> str: + """ + Export all contacts as a JSON string. + """ + try: + result = await client(functions.contacts.GetContactsRequest(hash=0)) + users = result.users + return json.dumps([format_entity(u) for u in users], indent=2) + except Exception as e: + return f"Error exporting contacts: {e}" + + +@mcp.tool() +async def get_blocked_users() -> str: + """ + Get a list of blocked users. + """ + try: + result = await client(functions.contacts.GetBlockedRequest(offset=0, limit=100)) + return json.dumps([format_entity(u) for u in result.users], indent=2) + except Exception as e: + return f"Error getting blocked users: {e}" + + +@mcp.tool() +async def create_channel(title: str, about: str = "", megagroup: bool = False) -> str: + """ + Create a new channel or supergroup. + """ + try: + result = await client(functions.channels.CreateChannelRequest( + title=title, + about=about, + megagroup=megagroup + )) + return f"Channel '{title}' created with ID: {result.chats[0].id}" + except Exception as e: + return f"Error creating channel: {e}" + + +@mcp.tool() +async def edit_chat_title(chat_id: int, title: str) -> str: + """ + Edit the title of a chat, group, or channel. + """ + try: + await client(functions.messages.EditChatTitleRequest(chat_id=chat_id, title=title)) + return f"Chat {chat_id} title updated." + except Exception as e: + return f"Error editing chat title: {e}" + + +@mcp.tool() +async def edit_chat_photo(chat_id: int, file_path: str) -> str: + """ + Edit the photo of a chat, group, or channel. + """ + try: + file = await client.upload_file(file_path) + await client(functions.messages.EditChatPhotoRequest(chat_id=chat_id, photo=file)) + return f"Chat {chat_id} photo updated." + except Exception as e: + return f"Error editing chat photo: {e}" + + +@mcp.tool() +async def delete_chat_photo(chat_id: int) -> str: + """ + Delete the photo of a chat, group, or channel. + """ + try: + await client(functions.messages.EditChatPhotoRequest(chat_id=chat_id, photo=None)) + return f"Chat {chat_id} photo deleted." + except Exception as e: + return f"Error deleting chat photo: {e}" + + +@mcp.tool() +async def promote_admin(chat_id: int, user_id: int) -> str: + """ + Promote a user to admin in a group or channel. + """ + from telethon.tl.types import ChatAdminRights + try: + user = await client.get_entity(user_id) + await client(functions.channels.EditAdminRequest( + channel=chat_id, + user_id=user, + admin_rights=ChatAdminRights( + change_info=True, post_messages=True, edit_messages=True, delete_messages=True, + ban_users=True, invite_users=True, pin_messages=True, add_admins=True, manage_call=True, other=True + ), + rank="admin" + )) + return f"User {user_id} promoted to admin in chat {chat_id}." + except Exception as e: + return f"Error promoting admin: {e}" + + +@mcp.tool() +async def demote_admin(chat_id: int, user_id: int) -> str: + """ + Demote an admin to regular user in a group or channel. + """ + from telethon.tl.types import ChatAdminRights + try: + user = await client.get_entity(user_id) + await client(functions.channels.EditAdminRequest( + channel=chat_id, + user_id=user, + admin_rights=ChatAdminRights(), + rank="" + )) + return f"User {user_id} demoted in chat {chat_id}." + except Exception as e: + return f"Error demoting admin: {e}" + + +@mcp.tool() +async def ban_user(chat_id: int, user_id: int) -> str: + """ + Ban a user from a group or channel. + """ + from telethon.tl.types import ChatBannedRights + import time + try: + user = await client.get_entity(user_id) + banned_rights = ChatBannedRights(until_date=int(time.time()) + 31536000, view_messages=True) + await client(functions.channels.EditBannedRequest( + channel=chat_id, + user_id=user, + banned_rights=banned_rights + )) + return f"User {user_id} banned from chat {chat_id}." + except Exception as e: + return f"Error banning user: {e}" + + +@mcp.tool() +async def unban_user(chat_id: int, user_id: int) -> str: + """ + Unban a user from a group or channel. + """ + from telethon.tl.types import ChatBannedRights + try: + user = await client.get_entity(user_id) + banned_rights = ChatBannedRights() + await client(functions.channels.EditBannedRequest( + channel=chat_id, + user_id=user, + banned_rights=banned_rights + )) + return f"User {user_id} unbanned in chat {chat_id}." + except Exception as e: + return f"Error unbanning user: {e}" + + +@mcp.tool() +async def get_admins(chat_id: int) -> str: + """ + Get all admins in a group or channel. + """ + try: + participants = await client.get_participants(chat_id, filter=functions.channels.ParticipantsAdmins()) + return "\n".join([f"ID: {p.id}, Name: {getattr(p, 'first_name', '')} {getattr(p, 'last_name', '')}" for p in participants]) + except Exception as e: + return f"Error getting admins: {e}" + + +@mcp.tool() +async def get_banned_users(chat_id: int) -> str: + """ + Get all banned users in a group or channel. + """ + try: + participants = await client.get_participants(chat_id, filter=functions.channels.ParticipantsBanned()) + return "\n".join([f"ID: {p.id}, Name: {getattr(p, 'first_name', '')} {getattr(p, 'last_name', '')}" for p in participants]) + except Exception as e: + return f"Error getting banned users: {e}" + + +@mcp.tool() +async def get_invite_link(chat_id: int) -> str: + """ + Get the invite link for a group or channel. + """ + try: + result = await client(functions.messages.ExportChatInviteRequest(chat_id=chat_id)) + return result.link + except Exception as e: + return f"Error getting invite link: {e}" + + +@mcp.tool() +async def join_chat_by_link(link: str) -> str: + """ + Join a chat by invite link. + """ + try: + await client(functions.messages.ImportChatInviteRequest(hash=link.split('/')[-1])) + return f"Joined chat via link." + except Exception as e: + return f"Error joining chat: {e}" + + +@mcp.tool() +async def export_chat_invite(chat_id: int) -> str: + """ + Export a chat invite link. + """ + try: + result = await client(functions.messages.ExportChatInviteRequest(chat_id=chat_id)) + return result.link + except Exception as e: + return f"Error exporting chat invite: {e}" + + +@mcp.tool() +async def import_chat_invite(hash: str) -> str: + """ + Import a chat invite by hash. + """ + try: + await client(functions.messages.ImportChatInviteRequest(hash=hash)) + return f"Joined chat via invite hash." + except Exception as e: + return f"Error importing chat invite: {e}" + + +@mcp.tool() +async def send_voice(chat_id: int, file_path: str) -> str: + """ + Send a voice message to a chat. + """ + try: + 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 f"Error sending voice: {e}" + + +@mcp.tool() +async def forward_message(from_chat_id: int, message_id: int, to_chat_id: int) -> str: + """ + Forward a message from one chat to another. + """ + try: + from_entity = await client.get_entity(from_chat_id) + to_entity = await client.get_entity(to_chat_id) + await client.forward_messages(to_entity, message_id, from_entity) + return f"Message {message_id} forwarded from {from_chat_id} to {to_chat_id}." + except Exception as e: + return f"Error forwarding message: {e}" + + +@mcp.tool() +async def edit_message(chat_id: int, message_id: int, new_text: str) -> str: + """ + Edit a message you sent. + """ + try: + entity = await client.get_entity(chat_id) + await client.edit_message(entity, message_id, new_text) + return f"Message {message_id} edited." + except Exception as e: + return f"Error editing message: {e}" + + +@mcp.tool() +async def delete_message(chat_id: int, message_id: int) -> str: + """ + Delete a message by ID. + """ + try: + entity = await client.get_entity(chat_id) + await client.delete_messages(entity, message_id) + return f"Message {message_id} deleted." + except Exception as e: + return f"Error deleting message: {e}" + + +@mcp.tool() +async def pin_message(chat_id: int, message_id: int) -> str: + """ + Pin a message in a chat. + """ + try: + entity = await client.get_entity(chat_id) + await client.pin_message(entity, message_id) + return f"Message {message_id} pinned in chat {chat_id}." + except Exception as e: + return f"Error pinning message: {e}" + + +@mcp.tool() +async def unpin_message(chat_id: int, message_id: int) -> str: + """ + Unpin a message in a chat. + """ + try: + entity = await client.get_entity(chat_id) + await client.unpin_message(entity, message_id) + return f"Message {message_id} unpinned in chat {chat_id}." + except Exception as e: + return f"Error unpinning message: {e}" + + +@mcp.tool() +async def mark_as_read(chat_id: int) -> str: + """ + Mark all messages as read in a chat. + """ + try: + entity = await client.get_entity(chat_id) + await client.send_read_acknowledge(entity) + return f"Marked all messages as read in chat {chat_id}." + except Exception as e: + return f"Error marking as read: {e}" + + +@mcp.tool() +async def reply_to_message(chat_id: int, message_id: int, text: str) -> str: + """ + Reply to a specific message in a chat. + """ + try: + entity = await client.get_entity(chat_id) + await client.send_message(entity, text, reply_to=message_id) + return f"Replied to message {message_id} in chat {chat_id}." + except Exception as e: + return f"Error replying to message: {e}" + + +@mcp.tool() +async def upload_file(file_path: str) -> str: + """ + Upload a file to Telegram servers (returns file handle). + """ + try: + file = await client.upload_file(file_path) + return str(file) + except Exception as e: + return f"Error uploading file: {e}" + + +@mcp.tool() +async def get_media_info(chat_id: int, message_id: int) -> str: + """ + Get info about media in a message. + """ + 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." + return str(msg.media) + except Exception as e: + return f"Error getting media info: {e}" + + +@mcp.tool() +async def search_public_chats(query: str) -> str: + """ + Search for public chats, channels, or bots by username or title. + """ + try: + result = await client(functions.contacts.SearchRequest(q=query, limit=20)) + return json.dumps([format_entity(u) for u in result.users], indent=2) + except Exception as e: + return f"Error searching public chats: {e}" + + +@mcp.tool() +async def search_messages(chat_id: int, query: str, limit: int = 20) -> str: + """ + Search for messages in a chat by text. + """ + try: + entity = await client.get_entity(chat_id) + messages = await client.get_messages(entity, limit=limit, search=query) + return "\n".join([f"ID: {m.id} | {m.date} | {m.message}" for m in messages]) + except Exception as e: + return f"Error searching messages: {e}" + + +@mcp.tool() +async def resolve_username(username: str) -> str: + """ + Resolve a username to a user or chat ID. + """ + try: + result = await client(functions.contacts.ResolveUsernameRequest(username=username)) + return str(result) + except Exception as e: + return f"Error resolving username: {e}" + + +@mcp.tool() +async def mute_chat(chat_id: int) -> str: + """ + Mute notifications for a chat. + """ + try: + await client(functions.account.UpdateNotifySettingsRequest( + peer=await client.get_entity(chat_id), + settings=functions.account.InputPeerNotifySettings(mute_until=2**31-1) + )) + return f"Chat {chat_id} muted." + except Exception as e: + return f"Error muting chat: {e}" + + +@mcp.tool() +async def unmute_chat(chat_id: int) -> str: + """ + Unmute notifications for a chat. + """ + try: + await client(functions.account.UpdateNotifySettingsRequest( + peer=await client.get_entity(chat_id), + settings=functions.account.InputPeerNotifySettings(mute_until=0) + )) + return f"Chat {chat_id} unmuted." + except Exception as e: + return f"Error unmuting chat: {e}" + + +@mcp.tool() +async def archive_chat(chat_id: int) -> str: + """ + Archive a chat. + """ + try: + await client(functions.messages.ToggleDialogPinRequest( + peer=await client.get_entity(chat_id), + pinned=True + )) + return f"Chat {chat_id} archived." + except Exception as e: + return f"Error archiving chat: {e}" + + +@mcp.tool() +async def unarchive_chat(chat_id: int) -> str: + """ + Unarchive a chat. + """ + try: + await client(functions.messages.ToggleDialogPinRequest( + peer=await client.get_entity(chat_id), + pinned=False + )) + return f"Chat {chat_id} unarchived." + except Exception as e: + return f"Error unarchiving chat: {e}" + + +@mcp.tool() +async def get_sticker_sets() -> str: + """ + Get all sticker sets. + """ + try: + result = await client(functions.messages.GetAllStickersRequest(hash=0)) + return json.dumps([s.title for s in result.sets], indent=2) + except Exception as e: + return f"Error getting sticker sets: {e}" + + +@mcp.tool() +async def send_sticker(chat_id: int, file_path: str) -> str: + """ + Send a sticker to a chat. + """ + try: + 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 f"Error sending sticker: {e}" + + +@mcp.tool() +async def get_gif_search(query: str, limit: int = 10) -> str: + """ + Search for GIFs by query. + """ + try: + result = await client(functions.messages.SearchGifsRequest(q=query, offset_id=0, limit=limit)) + return json.dumps([g.document.id for g in result.gifs], indent=2) + except Exception as e: + return f"Error searching GIFs: {e}" + + +@mcp.tool() +async def send_gif(chat_id: int, gif_id: int) -> str: + """ + Send a GIF to a chat by GIF document ID. + """ + try: + 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 f"Error sending GIF: {e}" + + +@mcp.tool() +async def get_bot_info(bot_username: str) -> str: + """ + Get information about a bot by username. + """ + try: + result = await client(functions.users.GetFullUserRequest(id=bot_username)) + return json.dumps(result.to_dict(), indent=2) + except Exception as e: + return f"Error getting bot info: {e}" + + +@mcp.tool() +async def set_bot_commands(bot_username: str, commands: list) -> str: + """ + Set bot commands for a bot you own. + """ + from telethon.tl.types import BotCommand + try: + await client(functions.bots.SetBotCommandsRequest( + scope=bot_username, + lang_code='', + commands=[BotCommand(command=c['command'], description=c['description']) for c in commands] + )) + return f"Bot commands set for {bot_username}." + except Exception as e: + return f"Error setting bot commands: {e}" + + +@mcp.tool() +async def get_history(chat_id: int, limit: int = 100) -> str: + """ + Get full chat history (up to limit). + """ + try: + entity = await client.get_entity(chat_id) + messages = await client.get_messages(entity, limit=limit) + return "\n".join([f"ID: {m.id} | {m.date} | {m.message}" for m in messages]) + except Exception as e: + return f"Error getting history: {e}" + + +@mcp.tool() +async def get_user_photos(user_id: int, limit: int = 10) -> str: + """ + Get profile photos of a user. + """ + try: + user = await client.get_entity(user_id) + photos = await client(functions.photos.GetUserPhotosRequest(user_id=user, offset=0, max_id=0, limit=limit)) + return json.dumps([p.id for p in photos.photos], indent=2) + except Exception as e: + return f"Error getting user photos: {e}" + + +@mcp.tool() +async def get_user_status(user_id: int) -> str: + """ + Get the online status of a user. + """ + try: + user = await client.get_entity(user_id) + return str(user.status) + except Exception as e: + return f"Error getting user status: {e}" + + +@mcp.tool() +async def get_recent_actions(chat_id: int) -> str: + """ + Get recent admin actions (admin log) in a group or channel. + """ + try: + result = await client(functions.channels.GetAdminLogRequest(channel=chat_id, q="", events_filter=None, admins=[], max_id=0, min_id=0, limit=20)) + return json.dumps([e.to_dict() for e in result.events], indent=2) + except Exception as e: + return f"Error getting recent actions: {e}" + + +@mcp.tool() +async def get_pinned_messages(chat_id: int) -> str: + """ + Get all pinned messages in a chat. + """ + try: + entity = await client.get_entity(chat_id) + messages = await client.get_messages(entity, filter=functions.messages.FilterPinned()) + return "\n".join([f"ID: {m.id} | {m.date} | {m.message}" for m in messages]) + except Exception as e: + return f"Error getting pinned messages: {e}" + + if __name__ == "__main__": nest_asyncio.apply() diff --git a/pyproject.toml b/pyproject.toml index 7701a20..ec26d7c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "telegram-mcp" -version = "1.5.0" +version = "2.0.0" description = "Telegram integration for Claude via the Model Context Protocol" readme = "README.md" authors = [ diff --git a/screenshots/1.png b/screenshots/1.png index 53c2bf2..9bc5232 100644 Binary files a/screenshots/1.png and b/screenshots/1.png differ diff --git a/uv.lock b/uv.lock index febcde1..dcff6d5 100644 --- a/uv.lock +++ b/uv.lock @@ -405,7 +405,7 @@ wheels = [ [[package]] name = "telegram-mcp" -version = "1.5.0" +version = "2.0.0" source = { editable = "." } dependencies = [ { name = "dotenv" },