Telegram MCP 服务器

🤖 MCP 实际应用

以下是在 Claude 中展示的 Telegram MCP 功能：

基本使用示例：

1. 示例：请求 Claude 分析聊天记录并发送回复：

1. 成功向群组发送消息：

如您所见，AI 可以无缝地与您的 Telegram 账户交互，以自然的方式检索和显示您的聊天、消息和其他数据。

这是一个功能齐全的 Telegram 集成，适用于 Claude、Cursor 和任何支持 MCP 的客户端，由 Telethon 和 模型上下文协议 (MCP) 提供支持。该项目允许您以编程方式与您的 Telegram 账户进行交互，自动化从消息传递到群组管理的所有内容。

🚀 功能与工具

此 MCP 服务器提供了一整套 Telegram 工具。每个主要的 Telegram/Telethon 功能都作为工具提供！

聊天与群组管理

• get_chats(page, page_size): 分页列出聊天
• list_chats(chat_type, limit): 列出带有元数据和过滤条件的聊天
• get_chat(chat_id): 获取关于某个聊天的详细信息
• create_group(title, user_ids): 创建新群组
• create_channel(title, about, megagroup): 创建频道或超级群组
• edit_chat_title(chat_id, title): 更改聊天/群组/频道标题
• delete_chat_photo(chat_id): 删除聊天/群组/频道照片
• leave_chat(chat_id): 退出群组或频道
• get_participants(chat_id): 列出所有参与者
• get_admins(chat_id): 列出所有管理员
• get_banned_users(chat_id): 列出所有被封禁的用户
• promote_admin(chat_id, user_id): 将用户提升为管理员
• demote_admin(chat_id, user_id): 将管理员降级为普通用户
• ban_user(chat_id, user_id): 封禁用户
• unban_user(chat_id, user_id): 解封用户
• get_invite_link(chat_id): 获取邀请链接
• export_chat_invite(chat_id): 导出邀请链接
• import_chat_invite(hash): 通过邀请哈希加入聊天
• join_chat_by_link(link): 通过邀请链接加入聊天

消息

• get_messages(chat_id, page, page_size): 分页消息
• list_messages(chat_id, limit, search_query, from_date, to_date): 过滤后的消息
• send_message(chat_id, message): 发送消息
• reply_to_message(chat_id, message_id, text): 回复消息
• edit_message(chat_id, message_id, new_text): 编辑消息
• delete_message(chat_id, message_id): 删除消息
• forward_message(from_chat_id, message_id, to_chat_id): 转发消息
• pin_message(chat_id, message_id): 置顶消息
• unpin_message(chat_id, message_id): 取消置顶消息
• mark_as_read(chat_id): 标记全部已读
• get_message_context(chat_id, message_id, context_size): 消息周围的上下文
• get_history(chat_id, limit): 完整聊天历史
• get_pinned_messages(chat_id): 列出置顶消息
• get_last_interaction(contact_id): 与联系人的最近一条消息

联系人管理

• list_contacts(): 列出所有联系人
• search_contacts(query): 搜索联系人
• add_contact(phone, first_name, last_name): 添加联系人
• delete_contact(user_id): 删除联系人
• block_user(user_id): 封锁用户
• unblock_user(user_id): 解封用户
• import_contacts(contacts): 批量导入联系人
• export_contacts(): 导出所有联系人为 JSON
• get_blocked_users(): 列出被封锁的用户- get_contact_ids(): 列出所有联系人ID
• get_direct_chat_by_contact(contact_query): 查找与某个联系人的直接聊天
• get_contact_chats(contact_id): 列出与某个联系人的所有聊天

用户与个人资料

• get_me(): 获取您的用户信息
• update_profile(first_name, last_name, about): 更新您的个人资料
• delete_profile_photo(): 删除您的个人资料照片
• get_user_photos(user_id, limit): 获取用户的个人资料照片
• get_user_status(user_id): 获取用户的在线状态

媒体

• get_media_info(chat_id, message_id): 获取消息中的媒体信息

搜索与发现

• search_public_chats(query): 搜索公共聊天/频道/机器人
• search_messages(chat_id, query, limit): 在聊天中搜索消息
• resolve_username(username): 将用户名解析为ID

表情包、GIF 和机器人

• get_sticker_sets(): 列出表情包集合
• get_bot_info(bot_username): 获取关于机器人的信息
• set_bot_commands(bot_username, commands): 设置机器人命令（仅限机器人账号）

隐私、设置及其他

• get_privacy_settings(): 获取隐私设置
• set_privacy_settings(key, allow_users, disallow_users): 设置隐私设置
• mute_chat(chat_id): 静音通知
• unmute_chat(chat_id): 取消静音通知
• archive_chat(chat_id): 归档聊天
• unarchive_chat(chat_id): 取消归档聊天
• get_recent_actions(chat_id): 获取最近的管理员操作

已移除的功能

请注意，需要直接访问服务器文件路径的工具（如 send_file、download_media、set_profile_photo、edit_chat_photo、send_voice、send_sticker、upload_file）已从 main.py 中移除。这是由于当前MCP环境在处理文件附件和本地文件系统路径方面的限制。

此外，由于Telethon库或Telegram API交互中的可靠性问题，GIF相关工具（如 get_gif_search、get_saved_gifs、send_gif）也已被移除。

📋 依赖项

• Python 3.10+
• Telethon
• MCP Python SDK
• Claude Desktop 或 Cursor（或任何MCP客户端）

🔧 安装与设置

1. Fork & 克隆

git clone https://github.com/chigwell/telegram-mcp.git

cd telegram-mcp

2. 创建虚拟环境

python3 -m venv .venv

source .venv/bin/activate  # On Windows: .venv\Scripts\activate

pip install -r requirements.txt

3. 生成会话字符串

python3 session_string_generator.py

按照提示进行身份验证并更新您的 .env 文件。

4. 配置 .env

将 .env.example 复制到 .env 并填写您的值：

TELEGRAM_API_ID=your_api_id_here

TELEGRAM_API_HASH=your_api_hash_here

TELEGRAM_SESSION_NAME=anon

TELEGRAM_SESSION_STRING=your_session_string_here

在 my.telegram.org/apps 获取您的API凭据。

🐳 使用Docker运行

如果您安装了Docker和Docker Compose，可以构建并在容器中运行服务器，简化依赖管理。

1. 构建镜像

从项目根目录构建Docker镜像：

docker build -t telegram-mcp:latest .

2. 运行容器

您有两个选项：

选项A：使用Docker Compose（推荐用于本地使用）

此方法使用 docker-compose.yml 文件，并自动从 .env 文件读取您的凭据。

1. 创建 .env 文件： 确保项目根目录下有一个包含 TELEGRAM_API_ID、TELEGRAM_API_HASH 和 TELEGRAM_SESSION_STRING（或 TELEGRAM_SESSION_NAME）的 .env 文件。使用 .env.example 作为模板。
2. 运行Compose：

   docker compose up --build
   

   • 使用 docker compose up -d 以分离模式（后台）运行。
   • 按 Ctrl+C 停止服务器。

选项B：使用 docker run

您可以直接运行容器，通过环境变量传递凭据。

docker run -it --rm \

  -e TELEGRAM_API_ID="YOUR_API_ID" \

  -e TELEGRAM_API_HASH="YOUR_API_HASH" \

  -e TELEGRAM_SESSION_STRING="YOUR_SESSION_STRING" \

  telegram-mcp:latest

• 用您的实际凭据替换占位符。
• 如果您更喜欢基于文件的会话，请使用 -e TELEGRAM_SESSION_NAME=your_session_file_name 而不是 TELEGRAM_SESSION_STRING（需要卷挂载，请参见 docker-compose.yml 中的示例）。
• -it 标志对于与服务器交互至关重要。## ⚙️ Claude 和 Cursor 的配置

MCP 配置

编辑你的 Claude 桌面配置（例如 ~/Library/Application Support/Claude/claude_desktop_config.json）或 Cursor 配置 (~/.cursor/mcp.json)：

{

  "mcpServers": {

    "telegram-mcp": {

      "command": "uv",

      "args": [

        "--directory",

        "/full/path/to/telegram-mcp",

        "run",

        "main.py"

      ]

    }

  }

}

📝 带代码和输出的工具示例

以下是最常用的工具及其实现和示例输出。

获取聊天记录

@mcp.tool()

async def get_chats(page: int = 1, page_size: int = 20) -> str:

    """

    Get a paginated list of chats.

    Args:

        page: Page number (1-indexed).

        page_size: Number of chats per page.

    """

    try:

        dialogs = await client.get_dialogs()

        start = (page - 1) * page_size

        end = start + page_size

        if start >= len(dialogs):

            return "Page out of range."

        chats = dialogs[start:end]

        lines = []

        for dialog in chats:

            entity = dialog.entity

            chat_id = entity.id

            title = getattr(entity, "title", None) or getattr(entity, "first_name", "Unknown")

            lines.append(f"Chat ID: {chat_id}, Title: {title}")

        return "\n".join(lines)

    except Exception as e:

        logger.exception(f"get_chats failed (page={page}, page_size={page_size})")

        return "An error occurred (code: GETCHATS-ERR-001). Check mcp_errors.log for details."

示例输出：

Chat ID: 123456789, Title: John Doe
Chat ID: -100987654321, Title: My Project Group
Chat ID: 111223344, Title: Jane Smith
Chat ID: -200123456789, Title: News Channel

发送消息

@mcp.tool()

async def send_message(chat_id: int, message: str) -> str:

    """

    Send a message to a specific chat.

    Args:

        chat_id: The ID of the chat.

        message: The message content to send.

    """

    try:

        entity = await client.get_entity(chat_id)

        await client.send_message(entity, message)

        return "Message sent successfully."

    except Exception as e:

        logger.exception(f"send_message failed (chat_id={chat_id})")

        return "An error occurred (code: SENDMSG-ERR-001). Check mcp_errors.log for details."

示例输出：

Message sent successfully.

获取聊天邀请链接

get_invite_link 函数特别强大，具有多种备用方法：

@mcp.tool()

async def get_invite_link(chat_id: int) -> str:

    """

    Get the invite link for a group or channel.

    """

    try:

        entity = await client.get_entity(chat_id)

        

        # Try using ExportChatInviteRequest first

        try:

            from telethon.tl import functions

            result = await client(functions.messages.ExportChatInviteRequest(

                peer=entity

            ))

            return result.link

        except AttributeError:

            # If the function doesn't exist in the current Telethon version

            logger.warning("ExportChatInviteRequest not available, using alternative method")

        except Exception as e1:

            # If that fails, log and try alternative approach

            logger.warning(f"ExportChatInviteRequest failed: {e1}")

            

        # Alternative approach using client.export_chat_invite_link

        try:

            invite_link = await client.export_chat_invite_link(entity)

            return invite_link

        except Exception as e2:

            logger.warning(f"export_chat_invite_link failed: {e2}")

            

        # Last resort: Try directly fetching chat info

        try:

            if isinstance(entity, (Chat, Channel)):

                full_chat = await client(functions.messages.GetFullChatRequest(

                    chat_id=entity.id

                ))

                if hasattr(full_chat, 'full_chat') and hasattr(full_chat.full_chat, 'invite_link'):

                    return full_chat.full_chat.invite_link or "No invite link available."

        except Exception as e3:

            logger.warning(f"GetFullChatRequest failed: {e3}")

            

        return "Could not retrieve invite link for this chat."

    except Exception as e:

        logger.exception(f"get_invite_link failed (chat_id={chat_id})")

        return f"Error getting invite link: {e}"

示例输出：

https://t.me/+AbCdEfGhIjKlMnOp

通过邀请链接加入聊天

@mcp.tool()

async def join_chat_by_link(link: str) -> str:

    """

    Join a chat by invite link.

    """

    try:

        # Extract the hash from the invite link

        if '/' in link:

            hash_part = link.split('/')[-1]

            if hash_part.startswith('+'):

                hash_part = hash_part[1:]  # Remove the '+' if present

        else:

            hash_part = link

            

        # Try checking the invite before joining

        try:

            # Try to check invite info first (will often fail if not a member)

            invite_info = await client(functions.messages.CheckChatInviteRequest(hash=hash_part))

            if hasattr(invite_info, 'chat') and invite_info.chat:

                # If we got chat info, we're already a member

                chat_title = getattr(invite_info.chat, 'title', 'Unknown Chat')

                return f"You are already a member of this chat: {chat_title}"

        except Exception:

            # This often fails if not a member - just continue

            pass

            

        # Join the chat using the hash

        result = await client(functions.messages.ImportChatInviteRequest(hash=hash_part))

        if result and hasattr(result, 'chats') and result.chats:

            chat_title = getattr(result.chats[0], 'title', 'Unknown Chat')

            return f"Successfully joined chat: {chat_title}"

        return f"Joined chat via invite hash."

    except Exception as e:

        err_str = str(e).lower()

        if "expired" in err_str:

            return "The invite hash has expired and is no longer valid."

        elif "invalid" in err_str:

            return "The invite hash is invalid or malformed."

        elif "already" in err_str and "participant" in err_str:

            return "You are already a member of this chat."

        logger.exception(f"join_chat_by_link failed (link={link})")

        return f"Error joining chat: {e}"

示例输出：

Successfully joined chat: Developer Community

搜索公共聊天

@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}"

示例输出：

[
  {
    "id": 123456789,
    "name": "TelegramBot",
    "type": "user",
    "username": "telegram_bot"
  },
  {
    "id": 987654321,
    "name": "Telegram News",
    "type": "user",
    "username": "telegram_news"
  }
]

获取与联系人的直接聊天

@mcp.tool()

async def get_direct_chat_by_contact(contact_query: str) -> str:

    """

    Find a direct chat with a specific contact by name, username, or phone.

    

    Args:

        contact_query: Name, username, or phone number to search for.

    """

    try:

        # 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:

                if isinstance(dialog.entity, User) and dialog.entity.id == contact.id:

                    chat_info = f"Chat ID: {dialog.entity.id}, Contact: {contact_name}"

                    if getattr(contact, 'username', ''):

                        chat_info += f", Username: @{contact.username}"

                    if dialog.unread_count:

                        chat_info += f", Unread: {dialog.unread_count}"

                    results.append(chat_info)

                    break

        

        if not results:

            return f"Found contacts matching '{contact_query}', but no direct chats with them."

        

        return "\n".join(results)

    except Exception as e:

        return f"Error searching for direct chat: {e}"

示例输出：

Chat ID: 123456789, Contact: John Smith, Username: @johnsmith, Unread: 3

🎮 使用示例

• "显示我的最近聊天"
• "向聊天 123456789 发送 'Hello world'"
• "添加电话 +1234567890、姓名 John Doe 的联系人"
• "创建一个名为 'Project Team' 的群组，包含用户 111, 222, 333"
• "从聊天 123456789 中的消息 42 下载媒体"
• "静音聊天 123456789 的通知"
• "在群组 123456789 中将用户 111 提升为管理员"
• "搜索关于 'news' 的公共频道"
• "通过邀请链接 https://t.me/+AbCdEfGhIjK 加入 Telegram 群组"
• "向我的保存的消息发送贴纸"
• "获取我所有的贴纸集"

你可以在 Claude、Cursor 或任何兼容 MCP 的客户端中使用自然语言来使用这些工具。

🧠 错误处理和鲁棒性

此实现包括全面的错误处理：

• 会话管理：支持基于文件和基于字符串的会话
• 错误报告：详细的错误日志记录到 mcp_errors.log
• 优雅降级：关键功能的多种备用方法
• 用户友好的消息：清晰、可操作的错误消息，而不是技术性的错误
• 账户类型检测：需要机器人账户的功能会在使用用户账户时检测并通知
• 邀请链接处理：处理各种链接格式和已经是成员的情况

该代码设计为对常见的 Telegram API 问题和限制具有鲁棒性。

🛠️ 贡献指南

1. Fork 此仓库：chigwell/telegram-mcp
2. 克隆你的 Fork：

   git clone https://github.com/<your-github-username>/telegram-mcp.git
   

3. 创建一个新的分支：

   git checkout -b my-feature
   

4. 进行更改，如果需要添加测试/文档。
5. 推送并打开 Pull Request 到 chigwell/telegram-mcp，并附上清晰的描述。
6. 在 PR 中标记 @chigwell 或 @l1v0n1 进行审查。

🔒 安全注意事项

• 永远不要提交你的 .env 或会话字符串。
• 会话字符串可以完全访问你的 Telegram 账户—请妥善保管！
• 所有处理都是本地的；除了 Telegram 的 API 外，数据不会发送到任何地方。
• 使用 .env.example 作为模板，并保持实际的 .env 文件私密。
• 测试文件在 .gitignore 中自动排除。

🛠️ 故障排除

• 检查 MCP 客户端（Claude/Cursor）和终端中的日志以查找错误。
• 详细的错误日志可以在 mcp_errors.log 中找到。
• 解释器错误？ 确保你的 .venv 已创建并被选中。
• 数据库锁定？ 使用基于字符串的身份验证，而不是基于文件的会话。
• iCloud/Dropbox 问题？ 如果看到奇怪的错误，请将项目移动到没有空格的本地路径。
• 重生成会话字符串如果你更改了 Telegram 密码或看到身份验证错误。
• 仅限机器人的功能在使用普通用户账户时会显示明确的消息。
• 测试脚本失败？ 检查 .env 中的测试配置是否包含有效的测试账户/群组。

📄 许可证

该项目根据 Apache 2.0 许可证获得许可。

🙏 致谢

• Telethon- 模型上下文协议
• Claude 和 Cursor
• chigwell/telegram-mcp (上游)

由 @chigwell 和 @l1v0n1 维护。欢迎提交 PR！

Star 历史