Telegram MCP サーバー

🤖 MCP の活用

以下は、 Claudeの Telegram MCP 機能のデモです。

基本的な使用例:

例: Claude にチャット履歴を分析して応答を送信するよう依頼する:

グループにメッセージを正常に送信しました:

ご覧のとおり、AI は Telegram アカウントとシームレスにやり取りし、チャット、メッセージ、その他のデータを自然な方法で取得して表示します。

Claude、Cursor、そしてMCP対応クライアント向けのフル機能のTelegram統合。TelethonとModel Context Protocol（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) : 禁止されたユーザーをすべて一覧表示する
prove_admin(chat_id, user_id) : ユーザーを管理者に昇格させる
demote_admin(chat_id, user_id) : 管理者をユーザーに降格する
ban_user(chat_id, user_id) : ユーザーを禁止する
unban_user(chat_id, user_id) : ユーザーのbanを解除する
get_invite_link(chat_id) : 招待リンクを取得する
export_chat_invite(chat_id) : 招待リンクをエクスポートする
import_chat_invite(ハッシュ) : 招待ハッシュでチャットに参加する
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. コンテナの実行

次の 2 つのオプションがあります。

オプション A: Docker Compose を使用する (ローカル使用に推奨)

この方法ではdocker-compose.ymlファイルを使用し、 .envファイルから資格情報を自動的に読み取ります。

**.envファイルの作成:**プロジェクトのルートに、 TELEGRAM_API_ID 、 TELEGRAM_API_HASH 、 TELEGRAM_SESSION_STRING （またはTELEGRAM_SESSION_NAME ）を含む.envファイルがあることを確認してください.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

プレースホルダーを実際の資格情報に置き換えます。
ファイルベースのセッションを希望する場合は、 TELEGRAM_SESSION_STRINGではなく-e TELEGRAM_SESSION_NAME=your_session_file_nameを使用します (ボリュームのマウントが必要です。例についてはdocker-compose.ymlを参照してください)。
-itフラグはサーバーと対話するために重要です。

⚙️ Claude と Cursor の設定

MCP構成

Claude デスクトップ構成 (例: ~/Library/Application Support/Claude/claude_desktop_config.json ) またはカーソル構成 ( ~/.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 の連絡先を追加」
「ユーザー111、222、333で「プロジェクトチーム」グループを作成する」
「チャット123456789のメッセージ42からメディアをダウンロードしてください」
「チャット 123456789 の通知をミュートする」
「ユーザー111をグループ123456789の管理者に昇格する」
「ニュース」に関する公開チャンネルを検索
「招待リンクhttps://t.me/+AbCdEfGhIjKで Telegram グループに参加してください」
「保存したメッセージにステッカーを送信する」
「すべてのステッカーセットを入手」

これらのツールは、Claude、Cursor、または MCP 互換クライアントの自然言語で使用できます。

🧠 エラー処理と堅牢性

この実装には包括的なエラー処理が含まれます。

セッション管理: ファイルベースと文字列ベースの両方のセッションで動作します
エラー報告: 詳細なエラーがmcp_errors.logに記録されます
グレースフルデグラデーション：重要な機能のための複数のフォールバックアプローチ
ユーザーフレンドリーなメッセージ: 技術的なエラーではなく、明確で実用的なエラーメッセージ
アカウント種別検出:ボットアカウントを必要とする機能は、ユーザーアカウントで使用されると検出して通知します。
招待リンク処理：様々なリンク形式と既存メンバーのケースを処理

このコードは、一般的な Telegram API の問題や制限に対して堅牢になるように設計されています。

🛠️ 貢献ガイド

このリポジトリをフォークする: chigwell/telegram-mcp
フォークをクローンします:
git clone https://github.com/<your-github-username>/telegram-mcp.git
新しいブランチを作成します。
git checkout -b my-feature
変更を加え、必要に応じてテスト/ドキュメントを追加します。
明確な説明を添えて、chigwell/telegram-mcpにプルリクエストをプッシュして開きます。
レビューのために、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.

Claude for Desktop などの MCP 互換ホストを介して 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