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) ：更改聊天/群组/频道标题
- edit_chat_photo(chat_id, file_path) ：设置聊天/群组/频道照片
- 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) ：列出已置顶的消息
联系人管理
- 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_last_interaction(contact_id) ：与联系人的最新消息
用户和个人资料
- get_me() ：获取您的用户信息
- update_profile(first_name, last_name, about) ：更新您的个人资料
- set_profile_photo(file_path) ：设置您的个人资料照片
- delete_profile_photo() ：删除您的个人资料照片
- get_user_photos(user_id，limit) ：获取用户的个人资料照片
- get_user_status(user_id) ：获取用户的在线状态
媒体
- send_file(chat_id, file_path, caption) ：发送文件
- send_voice(chat_id, file_path) ：发送语音消息
- download_media（chat_id、message_id、file_path） ：下载媒体
- upload_file(file_path) ：将文件上传到 Telegram 服务器
- 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() ：列出贴纸集
- send_sticker(chat_id, file_path) ：发送贴纸
- get_gif_search(query, limit) ：搜索 GIF
- send_gif(chat_id，gif_id) ：发送GIF
- 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) ：获取最近的管理员操作
📋 要求
🔧 安装和设置
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 凭证。
⚙️ 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-server",
"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！
星史