Build a Telegram AI bot with Python and Claude — complete 2026 tutorial

pip install python-telegram-bot==21.6 anthropic redis

from telegram.ext import Application, MessageHandler, CommandHandler, filters
from telegram import Update
import anthropic

BOT_TOKEN = "your_bot_token_here"
client = anthropic.Anthropic()

# In-memory conversation history (upgrade to Redis for production)
conversations: dict[int, list] = {}

async def handle_message(update: Update, context) -> None:
    user_id = update.effective_user.id
    user_message = update.message.text
    
    if user_id not in conversations:
        conversations[user_id] = []
    
    conversations[user_id].append({"role": "user", "content": user_message})
    
    # Keep last 20 messages to limit context window costs
    if len(conversations[user_id]) > 20:
        conversations[user_id] = conversations[user_id][-20:]
    
    await update.message.chat.send_action("typing")
    
    response = client.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=500,
        system="You are a helpful assistant. Keep responses concise — this is a mobile chat.",
        messages=conversations[user_id],
    )
    
    reply = response.content[0].text
    conversations[user_id].append({"role": "assistant", "content": reply})
    
    await update.message.reply_text(reply)

async def start_command(update: Update, context) -> None:
    await update.message.reply_text(
        "Hi! I'm an AI assistant. Just send me a message.\n\n"
        "Commands:\n"
        "/clear — start a new conversation\n"
        "/mode — switch between chat/research/code modes"
    )

def main():
    app = Application.builder().token(BOT_TOKEN).build()
    app.add_handler(CommandHandler("start", start_command))
    app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))
    app.run_polling()

if __name__ == "__main__":
    main()

async def clear_command(update: Update, context) -> None:
    user_id = update.effective_user.id
    conversations[user_id] = []
    await update.message.reply_text("Conversation cleared. Starting fresh.")

MODES = {
    "chat": "You are a helpful, friendly assistant. Keep responses concise.",
    "research": (
        "You are a research assistant. Provide detailed, well-sourced answers. "
        "Structure your response with clear sections. Cite sources when relevant."
    ),
    "code": (
        "You are a coding assistant. When writing code, always use proper formatting. "
        "Explain what the code does. Point out edge cases and potential bugs."
    ),
    "gst": (
        "You are a GST compliance assistant for Indian businesses. "
        "Help with GST rates, return filing, input tax credit, and invoice requirements. "
        "Always mention the relevant GST section or notification when citing rules."
    ),
}

user_modes: dict[int, str] = {}

async def mode_command(update: Update, context) -> None:
    user_id = update.effective_user.id
    args = context.args
    
    if not args or args[0] not in MODES:
        current = user_modes.get(user_id, "chat")
        available = ", ".join(MODES.keys())
        await update.message.reply_text(
            f"Current mode: *{current}*\n\nAvailable: {available}\n\nUsage: /mode research",
            parse_mode="Markdown",
        )
        return
    
    new_mode = args[0]
    user_modes[user_id] = new_mode
    conversations[user_id] = []  # Clear conversation when switching modes
    await update.message.reply_text(f"Switched to *{new_mode}* mode. Conversation reset.", parse_mode="Markdown")

def get_system_prompt(user_id: int) -> str:
    mode = user_modes.get(user_id, "chat")
    return MODES[mode]

response = client.messages.create(
    model="claude-haiku-4-5-20251001",
    max_tokens=500,
    system=get_system_prompt(user_id),
    messages=conversations[user_id],
)

import time
from collections import defaultdict

# Per-user: (message_count, window_start_timestamp)
rate_limit_tracker: dict[int, tuple[int, float]] = defaultdict(lambda: (0, time.time()))

RATE_LIMIT = 10  # messages per window
RATE_WINDOW = 60  # seconds

def is_rate_limited(user_id: int) -> bool:
    count, window_start = rate_limit_tracker[user_id]
    now = time.time()
    
    if now - window_start > RATE_WINDOW:
        # New window
        rate_limit_tracker[user_id] = (1, now)
        return False
    
    if count >= RATE_LIMIT:
        return True
    
    rate_limit_tracker[user_id] = (count + 1, window_start)
    return False

# In handle_message, add at the top:
async def handle_message(update: Update, context) -> None:
    user_id = update.effective_user.id
    
    if is_rate_limited(user_id):
        await update.message.reply_text("You're sending messages too fast. Please wait a moment.")
        return
    
    # ... rest of handler

async def handle_group_message(update: Update, context) -> None:
    message = update.message
    bot_username = context.bot.username
    
    # Respond only when mentioned or when replying to bot
    is_mentioned = f"@{bot_username}" in (message.text or "")
    is_reply_to_bot = (
        message.reply_to_message and 
        message.reply_to_message.from_user.id == context.bot.id
    )
    
    if not is_mentioned and not is_reply_to_bot:
        return
    
    # Strip the mention from the message
    clean_text = (message.text or "").replace(f"@{bot_username}", "").strip()
    if not clean_text:
        return
    
    # Use group_id as conversation key (shared group context)
    chat_id = message.chat.id
    # ... process with Claude

# Register different handlers for private vs group chats
app.add_handler(MessageHandler(
    filters.TEXT & ~filters.COMMAND & filters.ChatType.PRIVATE,
    handle_message,
))
app.add_handler(MessageHandler(
    filters.TEXT & ~filters.COMMAND & (filters.ChatType.GROUP | filters.ChatType.SUPERGROUP),
    handle_group_message,
))

pip install redis

import redis
import json

r = redis.Redis(host="localhost", port=6379, decode_responses=True)

def get_conversation(user_id: int) -> list:
    data = r.get(f"conv:{user_id}")
    return json.loads(data) if data else []

def save_conversation(user_id: int, messages: list) -> None:
    r.setex(
        f"conv:{user_id}",
        86400 * 7,  # 7-day TTL — conversations expire after a week of inactivity
        json.dumps(messages[-20:]),  # Keep last 20 messages
    )

def clear_conversation(user_id: int) -> None:
    r.delete(f"conv:{user_id}")

# Replace the in-memory dict usage:
async def handle_message(update: Update, context) -> None:
    user_id = update.effective_user.id
    user_message = update.message.text
    
    if is_rate_limited(user_id):
        await update.message.reply_text("Slow down — wait a moment before sending again.")
        return
    
    history = get_conversation(user_id)
    history.append({"role": "user", "content": user_message})
    
    await update.message.chat.send_action("typing")
    
    response = client.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=500,
        system=get_system_prompt(user_id),
        messages=history,
    )
    
    reply = response.content[0].text
    history.append({"role": "assistant", "content": reply})
    save_conversation(user_id, history)
    
    await update.message.reply_text(reply)

from telegram.ext import Application, MessageHandler, CommandHandler, filters
from telegram import Update, BotCommand
import anthropic, redis, json, time
from collections import defaultdict
import os

BOT_TOKEN = os.environ["TELEGRAM_BOT_TOKEN"]
client = anthropic.Anthropic()
r = redis.Redis(host="localhost", port=6379, decode_responses=True)

MODES = {
    "chat": "You are a helpful assistant. Be concise — this is a mobile chat.",
    "research": "You are a research assistant. Provide detailed, structured answers.",
    "code": "You are a coding assistant. Format code properly, explain what it does, note edge cases.",
}

rate_tracker: dict[int, tuple[int, float]] = defaultdict(lambda: (0, 0.0))

def is_rate_limited(user_id: int) -> bool:
    count, start = rate_tracker[user_id]
    now = time.time()
    if now - start > 60:
        rate_tracker[user_id] = (1, now)
        return False
    if count >= 10:
        return True
    rate_tracker[user_id] = (count + 1, start)
    return False

def get_conv(uid: int) -> list:
    d = r.get(f"conv:{uid}")
    return json.loads(d) if d else []

def save_conv(uid: int, msgs: list) -> None:
    r.setex(f"conv:{uid}", 86400 * 7, json.dumps(msgs[-20:]))

def get_mode(uid: int) -> str:
    return r.get(f"mode:{uid}") or "chat"

def set_mode(uid: int, mode: str) -> None:
    r.setex(f"mode:{uid}", 86400 * 30, mode)

async def start(update: Update, context) -> None:
    await update.message.reply_text(
        "*AI Assistant*\n\nJust send a message to start. Commands:\n"
        "/clear — new conversation\n/mode [chat|research|code] — switch mode",
        parse_mode="Markdown",
    )

async def clear(update: Update, context) -> None:
    r.delete(f"conv:{update.effective_user.id}")
    await update.message.reply_text("Conversation cleared.")

async def mode(update: Update, context) -> None:
    uid = update.effective_user.id
    args = context.args
    if not args or args[0] not in MODES:
        await update.message.reply_text(f"Current: {get_mode(uid)}\nOptions: {', '.join(MODES)}\nUsage: /mode research")
        return
    set_mode(uid, args[0])
    r.delete(f"conv:{uid}")
    await update.message.reply_text(f"Mode: {args[0]}")

async def message(update: Update, context) -> None:
    uid = update.effective_user.id
    if is_rate_limited(uid):
        await update.message.reply_text("Too fast — wait a moment.")
        return
    
    history = get_conv(uid)
    history.append({"role": "user", "content": update.message.text})
    await update.message.chat.send_action("typing")
    
    resp = client.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=500,
        system=MODES[get_mode(uid)],
        messages=history,
    )
    
    reply = resp.content[0].text
    history.append({"role": "assistant", "content": reply})
    save_conv(uid, history)
    await update.message.reply_text(reply)

def main():
    app = Application.builder().token(BOT_TOKEN).build()
    app.add_handler(CommandHandler("start", start))
    app.add_handler(CommandHandler("clear", clear))
    app.add_handler(CommandHandler("mode", mode))
    app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND & filters.ChatType.PRIVATE, message))
    app.run_polling()

if __name__ == "__main__":
    main()

# On the VPS
sudo apt update && sudo apt install python3-pip redis-server -y
sudo systemctl enable redis

# Upload your bot file (scp or git pull)
pip3 install python-telegram-bot anthropic redis

# Create a systemd service
sudo nano /etc/systemd/system/telegram-bot.service

[Unit]
Description=Telegram AI Bot
After=network.target redis.service

[Service]
User=ubuntu
WorkingDirectory=/home/ubuntu/telegram-bot
ExecStart=/usr/bin/python3 bot.py
Restart=always
RestartSec=10
Environment="TELEGRAM_BOT_TOKEN=your_token"
Environment="ANTHROPIC_API_KEY=your_key"

[Install]
WantedBy=multi-user.target

sudo systemctl daemon-reload
sudo systemctl enable telegram-bot
sudo systemctl start telegram-bot
sudo journalctl -u telegram-bot -f  # Watch logs