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) : チャット/グループ/チャンネルのタイトルを変更する
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) : 禁止されたユーザーをすべて一覧表示する
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) : ピン留めされたメッセージを一覧表示する

連絡先管理

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) : 最近の管理者のアクションを取得する

📋 要件

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 資格情報を取得します。

⚙️ 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-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

🎮 使用例

「最近のチャットを表示」
「チャット 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 を歓迎します!

Telegram MCP Server

Integrations

Telegram MCP サーバー

🤖 MCP の活用

🚀 機能とツール

チャットとグループ管理

メッセージング

連絡先管理

ユーザーとプロフィール

メディア

検索と発見

ステッカー、GIF、ボット

プライバシー、設定、その他

📋 要件

🔧 インストールとセットアップ

1. フォーク＆クローン

2. 仮想環境を作成する

3. セッション文字列を生成する

4. .env を設定する

⚙️ Claude と Cursor の設定

MCP構成

📝 コードと出力を含むツールの例

チャットの取得

メッセージの送信

チャット招待リンクの取得

招待リンク経由でチャットに参加する

公開チャットの検索

連絡先と直接チャットする

🎮 使用例

🧠 エラー処理と堅牢性

🛠️ 貢献ガイド

🔒 セキュリティに関する考慮事項

🛠️ トラブルシューティング

📄 ライセンス

🙏 謝辞

スターの歴史

Appeared in Searches