Telegram MCP-Server

🤖 MCP in Aktion

Hier ist eine Demonstration der Telegram MCP-Funktionen in Claude :

Grundlegendes Anwendungsbeispiel:

Beispiel: Claude bitten, den Chatverlauf zu analysieren und eine Antwort zu senden:

Nachricht erfolgreich an die Gruppe gesendet:

Wie Sie sehen, kann die KI nahtlos mit Ihrem Telegram-Konto interagieren und Ihre Chats, Nachrichten und andere Daten auf natürliche Weise abrufen und anzeigen.

Eine voll funktionsfähige Telegram-Integration für Claude, Cursor und alle MCP-kompatiblen Clients, basierend auf Telethon und dem Model Context Protocol (MCP) . Dieses Projekt ermöglicht Ihnen die programmgesteuerte Interaktion mit Ihrem Telegram-Konto und automatisiert alles vom Messaging bis zur Gruppenverwaltung.

🚀 Funktionen und Tools

Dieser MCP-Server stellt eine große Auswahl an Telegram-Tools bereit. Alle wichtigen Telegram/Telethon-Funktionen sind als Tool verfügbar!

Chat- und Gruppenverwaltung

get_chats(Seite, Seitengröße) : Paginierte Liste der Chats
list_chats(chat_type, limit) : Chats mit Metadaten und Filterung auflisten
get_chat(chat_id) : Detaillierte Informationen zu einem Chat
create_group(title, user_ids) : Eine neue Gruppe erstellen
create_channel(title, about, megagroup) : Einen Kanal oder eine Supergruppe erstellen
edit_chat_title(chat_id, title) : Chat-/Gruppen-/Kanaltitel ändern
delete_chat_photo(chat_id) : Chat-/Gruppen-/Kanalfoto entfernen
leave_chat(chat_id) : Eine Gruppe oder einen Kanal verlassen
get_participants(chat_id) : Listet alle Teilnehmer auf
get_admins(chat_id) : Listet alle Administratoren auf
get_banned_users(chat_id) : Listet alle gesperrten Benutzer auf
promote_admin(chat_id, user_id) : Benutzer zum Administrator befördern
demote_admin(chat_id, user_id) : Degradiert den Administrator zum Benutzer
ban_user(chat_id, user_id) : Benutzer sperren
unban_user(chat_id, user_id) : Benutzersperre aufheben
get_invite_link(chat_id) : Einladungslink abrufen
export_chat_invite(chat_id) : Einladungslink exportieren
import_chat_invite(hash) : Dem Chat per Einladungs-Hash beitreten
join_chat_by_link(Link) : Tritt dem Chat per Einladungslink bei

Nachrichten

get_messages(chat_id, page, page_size) : Paginierte Nachrichten
list_messages(chat_id, limit, search_query, from_date, to_date) : Gefilterte Nachrichten
send_message(chat_id, message) : Senden Sie eine Nachricht
reply_to_message(chat_id, message_id, text) : Auf eine Nachricht antworten
edit_message(chat_id, message_id, new_text) : Bearbeiten Sie Ihre Nachricht
delete_message(chat_id, message_id) : Löscht eine Nachricht
forward_message(from_chat_id, message_id, to_chat_id) : Leitet eine Nachricht weiter
pin_message(chat_id, message_id) : Eine Nachricht anheften
unpin_message(chat_id, message_id) : Eine Nachricht lösen
mark_as_read(chat_id) : Alle als gelesen markieren
get_message_context(chat_id, message_id, context_size) : Kontext rund um eine Nachricht
get_history(chat_id, limit) : Vollständiger Chatverlauf
get_pinned_messages(chat_id) : Listet angeheftete Nachrichten auf
get_last_interaction(contact_id) : Neueste Nachricht mit einem Kontakt

Kontaktverwaltung

list_contacts() : Listet alle Kontakte auf
search_contacts(query) : Kontakte suchen
add_contact(Telefon, Vorname, Nachname) : Einen Kontakt hinzufügen
delete_contact(user_id) : Einen Kontakt löschen
block_user(user_id) : Einen Benutzer blockieren
unblock_user(user_id) : Entsperren Sie einen Benutzer
import_contacts(Kontakte) : Massenimport von Kontakten
export_contacts() : Alle Kontakte als JSON exportieren
get_blocked_users() : Listet blockierte Benutzer auf
get_contact_ids() : Listet alle Kontakt-IDs auf
get_direct_chat_by_contact(contact_query) : Direkten Chat mit einem Kontakt finden
get_contact_chats(contact_id) : Listet alle Chats mit einem Kontakt auf

Benutzer und Profil

get_me() : Holen Sie sich Ihre Benutzerinformationen
update_profile(Vorname, Nachname, Info) : Aktualisieren Sie Ihr Profil
delete_profile_photo() : Entfernen Sie Ihr Profilfoto
get_user_photos(user_id, limit) : Profilfotos eines Benutzers abrufen
get_user_status(user_id) : Ruft den Online-Status eines Benutzers ab

Medien

get_media_info(chat_id, message_id) : Informationen zu Medien in einer Nachricht abrufen

Suche und Entdeckung

search_public_chats(Abfrage) : Öffentliche Chats/Kanäle/Bots durchsuchen
search_messages(chat_id, query, limit) : Nachrichten in einem Chat suchen
resolve_username(Benutzername) : Löst einen Benutzernamen in eine ID auf

Sticker, GIFs, Bots

get_sticker_sets() : Listet Sticker-Sets auf
get_bot_info(bot_username) : Informationen über einen Bot abrufen
set_bot_commands(bot_username, commands) : Bot-Befehle festlegen (nur Bot-Konten)

Datenschutz, Einstellungen und Sonstiges

get_privacy_settings() : Datenschutzeinstellungen abrufen
set_privacy_settings(key, allow_users, disallow_users) : Datenschutzeinstellungen festlegen
mute_chat(chat_id) : Benachrichtigungen stummschalten
unmute_chat(chat_id) : Stummschaltung von Benachrichtigungen aufheben
archive_chat(chat_id) : Einen Chat archivieren
unarchive_chat(chat_id) : Einen Chat aus dem Archiv nehmen
get_recent_actions(chat_id) : Aktuelle Administratoraktionen abrufen

Entfernte Funktionalität

Bitte beachten Sie, dass Tools, die direkten Dateipfadzugriff auf dem Server benötigen ( send_file , download_media , set_profile_photo , edit_chat_photo , send_voice , send_sticker , upload_file ), aus main.py entfernt wurden. Dies liegt an Einschränkungen in der aktuellen MCP-Umgebung hinsichtlich der Verarbeitung von Dateianhängen und lokalen Dateisystempfaden.

Darüber hinaus wurden GIF-bezogene Tools ( get_gif_search , get_saved_gifs , send_gif ) aufgrund anhaltender Zuverlässigkeitsprobleme in der Telethon-Bibliothek oder bei Telegram-API-Interaktionen entfernt.

📋 Voraussetzungen

Python 3.10+
Spendenmarathon
MCP Python SDK
Claude Desktop oder Cursor (oder ein beliebiger MCP-Client)

🔧 Installation und Einrichtung

1. Fork & Clone

git clone https://github.com/chigwell/telegram-mcp.git
cd telegram-mcp

2. Erstellen Sie eine virtuelle Umgebung

python3 -m venv .venv
source .venv/bin/activate  # On Windows: .venv\Scripts\activate
pip install -r requirements.txt

3. Generieren Sie eine Sitzungszeichenfolge

python3 session_string_generator.py

Folgen Sie den Anweisungen zur Authentifizierung und Aktualisierung Ihrer .env Datei.

4. Konfigurieren Sie .env

Kopieren Sie .env.example nach .env und geben Sie Ihre Werte ein:

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

Holen Sie sich Ihre API-Anmeldeinformationen unter my.telegram.org/apps .

🐳 Ausführen mit Docker

Wenn Sie Docker und Docker Compose installiert haben, können Sie den Server in einem Container erstellen und ausführen, was die Abhängigkeitsverwaltung vereinfacht.

1. Erstellen Sie das Image

Erstellen Sie das Docker-Image aus dem Stammverzeichnis des Projekts:

docker build -t telegram-mcp:latest .

2. Ausführen des Containers

Sie haben zwei Möglichkeiten:

Option A: Verwenden von Docker Compose (empfohlen für die lokale Verwendung)

Diese Methode verwendet die Datei docker-compose.yml und liest Ihre Anmeldeinformationen automatisch aus einer .env Datei.

.env Datei erstellen: Stellen Sie sicher, dass Sie im Projektstamm eine .env Datei mit Ihren TELEGRAM_API_ID , TELEGRAM_API_HASH und TELEGRAM_SESSION_STRING (oder TELEGRAM_SESSION_NAME ) haben. Verwenden Sie .env.example als Vorlage.
Compose ausführen:
docker compose up --build
- Verwenden Sie docker compose up -d um im getrennten Modus (Hintergrund) auszuführen.
- Drücken Sie Ctrl+C , um den Server zu stoppen.

Option B: Verwenden von docker run

Sie können den Container direkt ausführen und Anmeldeinformationen als Umgebungsvariablen übergeben.

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

Ersetzen Sie Platzhalter durch Ihre tatsächlichen Anmeldeinformationen.
Verwenden Sie -e TELEGRAM_SESSION_NAME=your_session_file_name anstelle von TELEGRAM_SESSION_STRING , wenn Sie dateibasierte Sitzungen bevorzugen (erfordert Volume-Mounting, siehe docker-compose.yml für ein Beispiel).
Die -it -Flags sind für die Interaktion mit dem Server von entscheidender Bedeutung.

⚙️ Konfiguration für Claude & Cursor

MCP-Konfiguration

Bearbeiten Sie Ihre Claude-Desktopkonfiguration (z. ~/Library/Application Support/Claude/claude_desktop_config.json ) oder Cursorkonfiguration ( ~/.cursor/mcp.json ):

{
  "mcpServers": {
    "telegram-mcp": {
      "command": "uv",
      "args": [
        "--directory",
        "/full/path/to/telegram-mcp",
        "run",
        "main.py"
      ]
    }
  }
}

📝 Tool-Beispiele mit Code und Ausgabe

Nachfolgend finden Sie Beispiele der am häufigsten verwendeten Tools mit ihrer Implementierung und Beispielausgabe.

Erhalten Sie Ihre Chats

@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."

Beispielausgabe:

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

Senden von Nachrichten

@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."

Beispielausgabe:

Message sent successfully.

Chat-Einladungslinks erhalten

Die Funktion get_invite_link ist mit mehreren Fallback-Methoden besonders robust:

@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}"

Beispielausgabe:

https://t.me/+AbCdEfGhIjKlMnOp

Chats über Einladungslinks beitreten

@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}"

Beispielausgabe:

Successfully joined chat: Developer Community

Öffentliche Chats durchsuchen

@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}"

Beispielausgabe:

[
  {
    "id": 123456789,
    "name": "TelegramBot",
    "type": "user",
    "username": "telegram_bot"
  },
  {
    "id": 987654321,
    "name": "Telegram News",
    "type": "user",
    "username": "telegram_news"
  }
]

Direkte Chats mit Kontakten

@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}"

Beispielausgabe:

Chat ID: 123456789, Contact: John Smith, Username: @johnsmith, Unread: 3

🎮 Anwendungsbeispiele

„Meine letzten Chats anzeigen“
„Senden Sie ‚Hallo Welt‘ an Chat 123456789“
„Kontakt mit Telefonnummer +1234567890, Name John Doe hinzufügen“
„Erstellen Sie eine Gruppe ‚Projektteam‘ mit den Benutzern 111, 222, 333“
„Laden Sie die Medien aus Nachricht 42 im Chat 123456789 herunter.“
„Benachrichtigungen für Chat 123456789 stummschalten“
„Benutzer 111 zum Administrator in der Gruppe 123456789 befördern“
„Suche nach öffentlichen Kanälen zum Thema ‚Nachrichten‘“
„Treten Sie der Telegram-Gruppe mit dem Einladungslink https://t.me/+AbCdEfGhIjK bei“
„Sende einen Sticker an meine gespeicherten Nachrichten“
„Holen Sie sich alle meine Sticker-Sets“

Sie können diese Tools in natürlicher Sprache in Claude, Cursor oder jedem MCP-kompatiblen Client verwenden.

🧠 Fehlerbehandlung und Robustheit

Diese Implementierung beinhaltet eine umfassende Fehlerbehandlung:

Sitzungsverwaltung : Funktioniert sowohl mit dateibasierten als auch mit stringbasierten Sitzungen
Fehlerberichterstattung : Detaillierte Fehler werden in mcp_errors.log protokolliert
Graceful Degradation : Mehrere Fallback-Ansätze für kritische Funktionen
Benutzerfreundliche Meldungen : Klare, umsetzbare Fehlermeldungen anstelle von technischen Fehlern
Kontotyperkennung : Funktionen, die Bot-Konten erfordern, erkennen und benachrichtigen, wenn sie mit Benutzerkonten verwendet werden
Verarbeitung von Einladungslinks : Behandelt verschiedene Linkformate und Fälle, in denen bereits Mitglieder sind

Der Code ist so konzipiert, dass er gegen häufige Probleme und Einschränkungen der Telegram-API robust ist.

🛠️ Beitragsleitfaden

Fork dieses Repo: chigwell/telegram-mcp
Klonen Sie Ihre Gabel:
git clone https://github.com/<your-github-username>/telegram-mcp.git
Erstellen Sie einen neuen Zweig:
git checkout -b my-feature
Nehmen Sie Ihre Änderungen vor und fügen Sie bei Bedarf Tests/Dokumente hinzu.
Pushen und öffnen Sie einen Pull Request an chigwell/telegram-mcp mit einer klaren Beschreibung.
Markieren Sie @chigwell oder @l1v0n1 in Ihrem PR zur Überprüfung.

🔒 Sicherheitsüberlegungen

Übergeben Sie niemals Ihre .env oder Sitzungszeichenfolge.
Die Sitzungszeichenfolge gewährt vollen Zugriff auf Ihr Telegram-Konto – bewahren Sie es sicher auf!
Die gesamte Verarbeitung erfolgt lokal; es werden keine Daten an andere Orte als die API von Telegram gesendet.
Verwenden Sie .env.example als Vorlage und halten Sie Ihre eigentliche .env Datei privat.
Testdateien werden in .gitignore automatisch ausgeschlossen.

🛠️ Fehlerbehebung

Überprüfen Sie die Protokolle in Ihrem MCP-Client (Claude/Cursor) und im Terminal auf Fehler.
Detaillierte Fehlerprotokolle finden Sie in mcp_errors.log .
Interpreterfehler? Stellen Sie sicher, dass Ihre .venv erstellt und ausgewählt ist.
Datenbanksperre? Verwenden Sie die Sitzungszeichenfolgen-Authentifizierung, keine dateibasierten Sitzungen.
Probleme mit iCloud/Dropbox? Verschieben Sie Ihr Projekt in einen lokalen Pfad ohne Leerzeichen, wenn merkwürdige Fehler auftreten.
Generieren Sie die Sitzungszeichenfolge neu , wenn Sie Ihr Telegrammkennwort ändern oder Authentifizierungsfehler sehen.
Nur-Bot-Funktionen zeigen bei Verwendung mit regulären Benutzerkonten klare Meldungen an.
Fehler beim Testskript? Überprüfen Sie die Testkonfiguration in .env auf gültige Testkonten/-gruppen.

Integrations