Telegram

pip install fastapi uvicorn httpx anthropic geniffy

uv add fastapi uvicorn httpx anthropic geniffy

import hmac
import os

import httpx
from anthropic import AsyncAnthropic
from fastapi import BackgroundTasks, FastAPI, HTTPException, Request
from geniffy import AsyncGeniffy

geniffy = AsyncGeniffy()                          # reads GENIFFY_API_KEY
claude = AsyncAnthropic()                         # reads ANTHROPIC_API_KEY
telegram = httpx.AsyncClient(base_url=f"https://api.telegram.org/bot{os.environ['TELEGRAM_BOT_TOKEN']}")
app = FastAPI()
seen: dict[int, None] = {}                        # recent update ids: Telegram sends one again when it has to wait
recent: dict[int, list] = {}                      # each person's last few turns, for "yes" and "that one"


@app.post("/telegram")
async def receive(request: Request, background: BackgroundTasks):
    given = request.headers.get("X-Telegram-Bot-Api-Secret-Token", "")
    if not hmac.compare_digest(given, os.environ["TELEGRAM_SECRET"]):
        raise HTTPException(401)
    update = await request.json()
    message = update.get("message") or {}
    text, person = message.get("text") or "", message.get("from") or {}
    if update.get("update_id") in seen or not text or text.startswith("/") or not person:
        return {}                                 # commands such as /start are yours to answer
    seen[update["update_id"]] = None
    if len(seen) > 10_000:
        seen.pop(next(iter(seen)))
    background.add_task(answer, message["chat"]["id"], person["id"], text)
    return {}                                     # at once: Telegram sends an update again when it has to wait


async def answer(chat_id: int, person_id: int, text: str) -> None:
    mem = geniffy.space(f"user_{person_id}")      # the person, by their Telegram id, whichever chat they write in
    turn = {"role": "user", "content": text}
    context = await mem.context(text)             # what is known that bears on the message
    past = recent.setdefault(person_id, [])
    response = await claude.messages.create(
        model="claude-opus-5-5",
        max_tokens=1024,
        system=f"You are a helpful assistant on Telegram. Keep replies short.\n\n<memory>\n{context}\n</memory>",
        messages=[*past[-10:], turn],
    )
    reply = "".join(block.text for block in response.content if block.type == "text")
    sent = await telegram.post("/sendMessage", json={"chat_id": chat_id, "text": reply})
    sent.raise_for_status()
    past += [turn, {"role": "assistant", "content": reply}]
    await mem.memories.add(messages=[turn, {"role": "assistant", "content": reply}])

curl "https://api.telegram.org/bot$TELEGRAM_BOT_TOKEN/setWebhook" \
  -d url=https://your-server/telegram \
  -d secret_token=$TELEGRAM_SECRET \
  -d allowed_updates='["message"]'