# Slack

> Give a Slack app a memory of each person who talks to it, in a channel or a direct message. What is known about them goes in front of the model before each reply, and each exchange is saved.

Give a Slack app a memory of each person who talks to it. Slack sends every mention and direct message to
your server; your server puts what is known about the person in front of Claude, replies in the thread, and
saves the exchange. The next question starts knowing this one.

## Install

```bash
pip install fastapi uvicorn httpx anthropic geniffy
```

```bash
uv add fastapi uvicorn httpx anthropic geniffy
```

Set `GENIFFY_API_KEY` from **API keys** in the Geniffy app and `ANTHROPIC_API_KEY`. From your Slack app, set
`SLACK_BOT_TOKEN` (under **OAuth & Permissions**, the `xoxb-` one) and `SLACK_SIGNING_SECRET` (under **Basic
Information**).

## Your server

```python
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}])
```

Run it with `uvicorn server:app`, somewhere Slack can reach over HTTPS.

The server answers each event at once and replies after, and it keeps the event ids it has seen in memory, which
is right for one process; with more than one, keep them where every process can see them. Each person is a
space of their own, so a busy channel keeps everyone's memory apart. The reply goes in the thread of the
message it answers.

## In Slack

1. Under **Event Subscriptions**, turn events on and set the **Request URL** to
   `https://your-server/slack/events`. Slack checks it at once.
2. Under **Subscribe to bot events**, add `app_mention` and `message.im`.
3. Under **OAuth & Permissions**, give the bot the `app_mentions:read`, `im:history` and `chat:write`
   scopes, then install the app to your workspace.

When nothing is known about the person, the memory says so in one sentence, so Claude says it doesn't know
instead of guessing. Each exchange is added as a conversation, so what the person said becomes a fact about
them, and what your app said stays the app's.

Source: https://docs.geniffy.com/integrations/slack