Slack

pip install fastapi uvicorn httpx anthropic geniffy

uv add fastapi uvicorn httpx anthropic geniffy

import hashlib
import hmac
import json
import os
import re
import time

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
slack = httpx.AsyncClient(base_url="https://slack.com/api",
                          headers={"Authorization": f"Bearer {os.environ['SLACK_BOT_TOKEN']}"})
app = FastAPI()
seen: dict[str, None] = {}                        # recent event ids: Slack sends an event again when it has to wait


def space_for(team_id: str, user_id: str) -> str:
    """The person, by their workspace and Slack id. If you know them as customers, use your customer id."""
    return f"slack_{team_id}_{user_id}"


def check(request: Request, body: bytes) -> None:
    """Slack signs every request with your signing secret; one older than five minutes is a replay."""
    stamp = request.headers.get("X-Slack-Request-Timestamp", "0")
    if not stamp.isdigit() or abs(time.time() - int(stamp)) > 300:
        raise HTTPException(401)
    base = f"v0:{stamp}:".encode() + body
    signed = "v0=" + hmac.new(os.environ["SLACK_SIGNING_SECRET"].encode(), base, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(request.headers.get("X-Slack-Signature", ""), signed):
        raise HTTPException(401)


@app.post("/slack/events")
async def events(request: Request, background: BackgroundTasks):
    body = await request.body()
    check(request, body)
    payload = json.loads(body)
    if payload.get("type") == "url_verification":       # Slack checks the address once, when you add it
        return {"challenge": payload["challenge"]}
    event = payload.get("event") or {}
    if payload.get("event_id") in seen or event.get("bot_id") or event.get("subtype") or not event.get("user"):
        return {}                                 # a retry, a bot (this one included), or an edit: nothing to answer
    if event.get("type") == "app_mention" or (event.get("type") == "message" and event.get("channel_type") == "im"):
        seen[payload["event_id"]] = None
        if len(seen) > 10_000:
            seen.pop(next(iter(seen)))
        background.add_task(answer, payload["team_id"], event)
    return {}                                     # at once: Slack sends an event again after three seconds


async def answer(team_id: str, event: dict) -> None:
    text = re.sub(r"<@[A-Z0-9]+>\s*", "", event.get("text") or "").strip()   # without the app's own mention
    if not text:
        return
    mem = geniffy.space(space_for(team_id, event["user"]))
    turn = {"role": "user", "content": text}
    context = await mem.context(text)             # what is known that bears on the message
    response = await claude.messages.create(
        model="claude-opus-5-5",
        max_tokens=1024,
        system=f"You are a helpful assistant in Slack. Keep replies short.\n\n<memory>\n{context}\n</memory>",
        messages=[turn],
    )
    reply = "".join(block.text for block in response.content if block.type == "text")
    sent = await slack.post("/chat.postMessage", json={"channel": event["channel"], "text": reply,
                                                       "thread_ts": event.get("thread_ts") or event["ts"]})
    sent.raise_for_status()
    await mem.memories.add(messages=[turn, {"role": "assistant", "content": reply}])