Telegram MCP 服务器
🤖 MCP 实际应用
以下是Claude中 Telegram MCP 功能的演示：
基本用法示例：
示例：要求 Claude 分析聊天记录并发送回复：
成功发送消息到群组：
如您所见，AI 可以与您的 Telegram 帐户无缝交互，以自然的方式检索和显示您的聊天、消息和其他数据。
一个功能齐全的 Telegram 集成项目，适用于 Claude、Cursor 以及任何兼容 MCP 的客户端，由Telethon和模型上下文协议 (MCP)提供支持。该项目允许您以编程方式与您的 Telegram 帐户进行交互，从而实现从消息传递到群组管理的所有自动化操作。
Related MCP server: Telegram MCP Server
🚀 功能和工具
此 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) ：获取用户的在线状态
媒体
搜索与发现
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 ）已被删除。
📋 要求
🔧 安装和设置
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
选项 B：使用
您可以直接运行容器，并将凭据作为环境变量传递。
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
⚙️ 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以供审核。
🔒 安全注意事项
永远不要提交你的
会话字符串可完全访问您的 Telegram 帐户 - 请确保其安全！
所有处理都是本地的；除了 Telegram 的 API 之外，没有数据被发送到任何地方。
使用.env.example作为模板并将实际的.env文件保密。
测试文件会自动排除在.gitignore中。
🛠️ 故障排除
检查 MCP 客户端（Claude/Cursor）和终端中的日志是否有错误。
详细的错误日志可以在mcp_errors.log中找到。
**解释器错误？**请确保已创建并选择.venv文件。
**数据库锁定？**请使用会话字符串身份验证，而不是基于文件的会话。
**iCloud/Dropbox 出了问题？**如果遇到奇怪的错误，请将项目移动到没有空格的本地路径。
如果您更改了 Telegram 密码或看到身份验证错误，请重新生成会话字符串。
当与常规用户帐户一起使用时，仅限机器人的功能将显示清晰的消息。
**测试脚本失败？**请检查.env中的测试配置，确认测试账户/测试组有效。
📄 许可证
该项目采用Apache 2.0 许可证进行授权。
🙏 致谢
由
星史
