Telegram MCP 服务器

🤖 MCP 实际应用

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

基本用法示例：

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

成功发送消息到群组：

如您所见，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+
电视募捐节目
MCP Python SDK
Claude Desktop或Cursor （或任何 MCP 客户端）

🔧 安装和设置

1. 分叉与克隆

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文件中读取您的凭据。

**创建.env文件：**确保项目根目录中有一个.env文件，其中包含TELEGRAM_API_ID 、 TELEGRAM_API_HASH和TELEGRAM_SESSION_STRING （或TELEGRAM_SESSION_NAME ）。使用.env.example作为模板。
运行 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

🎮 使用示例

“显示我最近的聊天”
“发送‘Hello world’到聊天 123456789”
“添加联系人，电话 +1234567890，名字为 John Doe”
“创建一个组‘项目团队’，用户编号为 111、222、333”
“从聊天 123456789 中的消息 42 下载媒体”
“关闭聊天 123456789 的通知”
“将用户 111 提升为组 123456789 中的管理员”
“搜索有关‘新闻’的公共频道”
通过邀请链接https://t.me/+AbCdEfGhIjK加入 Telegram 群组
“向我的已保存消息发送贴纸”
“获取我的所有贴纸套装”

您可以通过 Claude、Cursor 或任何与 MCP 兼容的客户端中的自然语言使用这些工具。

🧠 错误处理和稳健性

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

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

该代码旨在抵御常见的 Telegram API 问题和限制。

🛠️ 贡献指南

分支此 repo： chigwell/telegram-mcp
克隆你的 fork：
git clone https://github.com/<your-github-username>/telegram-mcp.git
创建新分支：
git checkout -b my-feature
进行更改，如有必要，添加测试/文档。
向chigwell/telegram-mcp**推送并打开一个 Pull 请求，**并附上清晰的描述。
在您的 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 许可证进行授权。

🙏 致谢

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

星史

This server cannot be installed

security - not tested

license - permissive license

quality - not tested

How are these scores calculated?

remote-capable server

The server can be hosted and run remotely because it primarily relies on remote services or has no dependency on the local environment.

一种服务器，可以通过与 MCP 兼容的主机（如 Claude for Desktop）直接与 Telegram 聊天进行交互，提供检索聊天、获取消息和发送消息的工具。

Related MCP Servers

MCP-Communicator-Telegram
qpd-v
-
security
F
license
-
quality
An MCP server that enables communication with users through Telegram. This server provides a tool to ask questions to users and receive their responses via a Telegram bot.
Last updated -
1
16
JavaScript
ClaudeKeep
sdairs
A
security
A
license
A
quality
An MCP server that enables saving and sharing Claude Desktop conversations, allowing users to store chats privately or make them public through a web interface.
Last updated -
2
6
TypeScript
MIT License
Discord MCP Server
RossH121
-
security
F
license
-
quality
An MCP server that allows Claude to interact with Discord by providing tools for sending/reading messages and managing server resources through Discord's API.
Last updated -
JavaScript
Telegram MCP Server
alexandertsai
-
security
A
license
-
quality
A simple MCP server that allows Claude to access your Telegram account to read and send messages on your behalf.
Last updated -
4
Python
Apache 2.0

View all related MCP servers