Telegram MCP Server
by sparfenyuk
- src
- mcp_telegram
# ruff: noqa: T201
from __future__ import annotations
from functools import cache
from getpass import getpass
from pydantic_settings import BaseSettings
from telethon import TelegramClient # type: ignore[import-untyped]
from telethon.errors.rpcerrorlist import SessionPasswordNeededError # type: ignore[import-untyped]
from telethon.tl.types import User # type: ignore[import-untyped]
from xdg_base_dirs import xdg_state_home # type: ignore[import-error]
class TelegramSettings(BaseSettings):
api_id: str
api_hash: str
class Config:
env_prefix = "TELEGRAM_"
env_file = ".env"
async def connect_to_telegram(api_id: str, api_hash: str, phone_number: str) -> None:
user_session = create_client(api_id=api_id, api_hash=api_hash)
await user_session.connect()
result = await user_session.send_code_request(phone_number)
code = input("Enter login code: ")
try:
await user_session.sign_in(
phone=phone_number,
code=code,
phone_code_hash=result.phone_code_hash,
)
except SessionPasswordNeededError:
password = getpass("Enter 2FA password: ")
await user_session.sign_in(password=password)
user = await user_session.get_me()
if isinstance(user, User):
print(f"Hey {user.username}! You are connected!")
else:
print("Connected!")
print("You can now use the mcp-telegram server.")
async def logout_from_telegram() -> None:
user_session = create_client()
await user_session.connect()
await user_session.log_out()
print("You are now logged out from Telegram.")
@cache
def create_client(
api_id: str | None = None,
api_hash: str | None = None,
session_name: str = "mcp_telegram_session",
) -> TelegramClient:
if api_id is not None and api_hash is not None:
config = TelegramSettings(api_id=api_id, api_hash=api_hash)
else:
config = TelegramSettings()
state_home = xdg_state_home() / "mcp-telegram"
state_home.mkdir(parents=True, exist_ok=True)
return TelegramClient(state_home / session_name, config.api_id, config.api_hash, base_logger="telethon")