返回市场
电报-MCP

电报-MCP

作者:chigwell460 星标更新:2025-11-18

项目介绍

Telegram MCP Server

MCP 徽章 许可证:Apache 2.0 Python 代码检查 Docker 构建与验证


🤖 MCP 在行动

这里展示了在 Claude 中使用 Telegram MCP 的能力:

基本用法示例:

Telegram MCP 在行动

  1. 示例:请求 Claude 分析聊天历史并发送回复:

Telegram MCP 请求

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

Telegram MCP 结果

如您所见,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): 通过邀请链接加入聊天
  • subscribe_public_channel(channel): 通过用户名或ID订阅公共频道或超级群组

消息

  • get_messages(chat_id, page, page_size): 分页的消息
  • list_messages(chat_id, limit, search_query, from_date, to_date): 过滤消息
  • list_topics(chat_id, limit, offset_topic, search_query): 列出超级群组中的论坛主题
  • 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): 最近与联系人的消息
  • create_poll(chat_id, question, options, multiple_choice, quiz_mode, public_votes, close_date): 创建投票
  • list_inline_buttons(chat_id, message_id, limit): 检查内联键盘以发现按钮文本/索引
  • press_inline_button(chat_id, message_id, button_text, button_index): 通过标签或索引触发内联键盘回调

联系人管理

  • 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): 获取最近的管理员操作

输入验证

为了提高健壮性,所有接受 chat_iduser_id 参数的函数现在包括了输入验证。您可以使用以下任意格式之一来表示这些ID:

  • 整数ID:用户、聊天或频道的直接整数ID(例如,123456789-1001234567890)。
  • 字符串ID:作为字符串提供的整数ID(例如,"123456789")。
  • 用户名:用户或频道的公开用户名(例如,"@username""username")。

服务器会自动验证输入,并将其转换为正确的格式后再向 Telegram 发送请求。如果输入无效,将返回清晰的错误消息。

移除的功能

请注意,需要直接访问服务器文件路径的工具(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. 使用 uv 安装依赖

uv sync

3. 生成会话字符串

uv run 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_IDTELEGRAM_API_HASHTELEGRAM_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:
    """
    获取分页的聊天列表。
    参数:
        page: 页面编号(从1开始)。
        page_size: 每页的聊天数量。
    """
    try:
        dialogs = await client.get_dialogs()
        start = (page - 1) * page_size
        end = start + page_size
        if start >= len(dialogs):
            return "页面超出范围。"
        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", "未知")
            lines.append(f"聊天ID: {chat_id}, 标题: {title}")
        return "\n".join(lines)
    except Exception as e:
        logger.exception(f"get_chats 失败 (page={page}, page_size={page_size})")
        return "发生错误 (代码: GETCHATS-ERR-001)。请查看 mcp_errors.log 获取详细信息。"

示例输出:

聊天ID: 123456789, 标题: John Doe
聊天ID: -100987654321, 标题: 我的项目小组
聊天ID: 111223344, 标题: Jane Smith
聊天ID: -200123456789, 标题: 新闻频道

发送消息

@mcp.tool()
async def send_message(chat_id: int, message: str) -> str:
    """
    向特定聊天发送消息。
    参数:
        chat_id: 聊天的ID。
        message: 要发送的消息内容。
    """
    try:
        entity = await client.get_entity(chat_id)
        await client.send_message(entity, message)
        return "消息发送成功。"
    except Exception as e:
        logger.exception(f"send_message 失败 (chat_id={chat_id})")
        return "发生错误 (代码: SENDMSG-ERR--001)。请查看 mcp_errors.log 获取详细信息。"

示例输出:

消息发送成功。

列出内联按钮

@mcp.tool()
async def list_inline_buttons(
    chat_id: Union[int, str],
    message_id: Optional[int] = None,
    limit: int = 20,
) -> str:
    """
    发现内联键盘布局,包括按钮索引、回调可用性和URL。
    """

示例用法:

list_inline_buttons(chat_id="@sample_tasks_bot")

这返回类似的内容:

消息42的按钮(日期 2025-01-01 12:00:00+00:00):
[0] 文本='📋 查看任务', 回调=是
[1] 文本='ℹ️ 帮助', 回调=是
[2] 文本='🌐 访问网站', 回调=否, URL=https://example.org

按下内联按钮

@mcp.tool()
async def press_inline_button(
    chat_id: Union[int, str],
    message_id: Optional[int] = None,
    button_text: Optional[str] = None,
    button_index: Optional[int] = None,
) -> str:
    """
    通过标签或零基索引按下内联键盘按钮。
    如果省略 message_id,服务器会在最近的消息中搜索最新的内联键盘。
    """

示例用法:

press_inline_button(chat_id="@sample_tasks_bot", button_text="📋 查看任务")

如果需要检查可用按钮,请先使用 list_inline_buttons —— 传递一个虚假的 button_text 快速列出选项或直接调用 list_inline_buttons。一旦知道文本或索引,press_inline_button 将发送回调,就像在原生 Telegram 客户端中点击按钮一样。

订阅公共频道

@mcp.tool()
async def subscribe_public_channel(channel: Union[int, str]) -> str:
    """
    通过用户名(例如,“@examplechannel”)或ID加入公共频道或超级群组。
    """

示例用法:

subscribe_public_channel(channel="@daily_updates_feed")

如果账户已经是参与者,则该工具报告这一点而不是失败,使其在需要幂等连接的工作流中安全运行。

获取聊天邀请链接

get_invite_link 函数特别强大,具有多种备用方法:

@mcp.tool()
async def get_invite_link(chat_id: int) -> str:
    """
    获取群组或频道的邀请链接。
    """
    try:
        entity = await client.get_entity(chat_id)
        
        # 尝试首先使用 ExportChatInviteRequest
        try:
            from telethon.tl import functions
            result = await client(functions.messages.ExportChatInviteRequest(
                peer=entity
            ))
            return result.link
        except AttributeError:
            # 如果当前 Telethon 版本中不存在该功能
            logger.warning("ExportChatInviteRequest 不可用,使用替代方法")
        except Exception as e1:
            # 如果失败,记录并尝试其他方法
            logger.warning(f"ExportChatInviteRequest 失败: {e1}")
            
        # 使用 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 失败: {e2}")
            
        # 最后一招:直接获取聊天信息
        try:
            if isinstance(entity, (Chat, Channel)):
                full_chat = await