auth.py

#!/usr/bin/env python3
"""
One-time authorization script for Telegram User MCP
Run this once to create a session file
"""

import asyncio
import os
from pathlib import Path
from dotenv import load_dotenv
from telethon import TelegramClient
from telethon.errors import SessionPasswordNeededError

load_dotenv()

API_ID = os.getenv("TELEGRAM_API_ID")
API_HASH = os.getenv("TELEGRAM_API_HASH")
PHONE = os.getenv("TELEGRAM_PHONE")

SESSION_PATH = Path(__file__).parent / "telegram_session"


async def main():
    if not API_ID or not API_HASH:
        print("Error: Set TELEGRAM_API_ID and TELEGRAM_API_HASH in .env file")
        print("Get them from https://my.telegram.org")
        return
    
    if not PHONE:
        print("Error: Set TELEGRAM_PHONE in .env file")
        return

    print(f"Connecting with phone: {PHONE}")
    
    client = TelegramClient(str(SESSION_PATH), int(API_ID), API_HASH)
    
    await client.connect()
    
    if not await client.is_user_authorized():
        print(f"Sending code to {PHONE}...")
        await client.send_code_request(PHONE)
        
        code = input("Enter the code you received: ")
        
        try:
            await client.sign_in(PHONE, code)
        except SessionPasswordNeededError:
            password = input("Two-factor authentication enabled. Enter password: ")
            await client.sign_in(password=password)
    
    me = await client.get_me()
    print(f"\n✅ Successfully authorized as: {me.first_name} {me.last_name or ''}")
    print(f"   Username: @{me.username}")
    print(f"   Phone: {me.phone}")
    print(f"\nSession saved to: {SESSION_PATH}.session")
    print("\nYou can now use telegram-user-mcp!")
    
    await client.disconnect()


if __name__ == "__main__":
    asyncio.run(main())