Slack bot

pip install memsy slack-bolt openai

export SLACK_BOT_TOKEN="xoxb-..."
export SLACK_APP_TOKEN="xapp-..."
export MEMSY_BASE_URL="https://api.memsy.io/v1"
export MEMSY_API_KEY="msy_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
export OPENAI_API_KEY="sk-..."

# identity.py
from typing import Any

from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError

# Slack user id -> email. One lookup per person, not per mention.
_EMAILS: dict[str, str] = {}


def actor_id(client: WebClient, event: dict[str, Any]) -> str:
    """
    Who is speaking, as their email address, so that a memory can be traced
    back to a person without a second lookup.

    Falls back to the Slack user id, which is NOT belt-and-braces: a bot, a
    guest account, or a missing `users:read.email` scope all return a
    profile with no address, and without the fallback every one of those
    turns would be filed under the same empty actor.
    """
    user = event["user"]
    if user not in _EMAILS:
        try:
            profile = client.users_info(user=user)["user"]["profile"]
            _EMAILS[user] = profile.get("email") or f"slack_{user}"
        except SlackApiError:
            _EMAILS[user] = f"slack_{user}"
    return _EMAILS[user]


def session_id(event: dict[str, Any]) -> str:
    """
    Which conversation. `thread_ts` is present on every message inside a
    thread; a top-level message has only `ts`, and becomes the root of the
    thread the bot's reply creates. Falling back to `ts` therefore gives the
    thread and its root message the same session id — which is correct,
    because they are the same conversation.
    """
    return f"slack_{event['channel']}_{event.get('thread_ts', event['ts'])}"

# app.py
import json
import logging
import os
import re

from memsy import EventPayload, MemsyClient
from memsy.exceptions import MemsyError
from openai import OpenAI
from slack_bolt import App
from slack_bolt.adapter.socket_mode import SocketModeHandler

from identity import actor_id, session_id

logging.basicConfig(level=logging.INFO)
log = logging.getLogger(__name__)

memsy = MemsyClient(
    base_url=os.environ["MEMSY_BASE_URL"],
    api_key=os.environ["MEMSY_API_KEY"],
)
openai = OpenAI()
app = App(token=os.environ["SLACK_BOT_TOKEN"])

# "<@U123ABC> what did we decide?" -> "what did we decide?"
MENTION = re.compile(r"<@[A-Z0-9]+>")


@app.event("app_mention")
def handle_mention(event, say, client):
    question = MENTION.sub("", event["text"]).strip()
    if not question:
        say(text="Ask me something.", thread_ts=event.get("thread_ts", event["ts"]))
        return

    actor = actor_id(client, event)
    session = session_id(event)

    # 1. Recall. Scoped to this person — see the note below on team-wide recall.
    try:
        recalled = memsy.search(question, actor_id=actor, limit=5)
        memories = [r.content for r in recalled.results]
    except MemsyError:
        log.exception("memsy recall failed")
        memories = []

    context = "\n".join(f"- {m}" for m in memories) or "(no prior context)"

    # 2. Ground.
    completion = openai.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": (
                    "You are a helpful teammate in Slack. Be brief — two or "
                    "three sentences. Use the prior context below when it is "
                    "relevant.\n\n"
                    f"Prior context:\n{context}"
                ),
            },
            {"role": "user", "content": question},
        ],
    )
    answer = completion.choices[0].message.content or ""

    # Reply in-thread so the conversation stays one session.
    say(text=answer, thread_ts=event.get("thread_ts", event["ts"]))

    # 3. Record.
    try:
        memsy.ingest(
            [
                EventPayload(
                    actor_id=actor,
                    session_id=session,
                    kind="user_message",
                    content=question,
                    metadata=json.dumps({"slack_user_id": event["user"]}),
                ),
                EventPayload(
                    actor_id=actor,
                    session_id=session,
                    kind="assistant_message",
                    content=answer,
                    metadata=json.dumps({"slack_user_id": event["user"]}),
                ),
            ]
        )
    except MemsyError:
        # The reply is already in the channel; a failed write must not
        # produce a second, confusing error message in the thread.
        log.exception("memsy ingest failed")


if __name__ == "__main__":
    SocketModeHandler(app, os.environ["SLACK_APP_TOKEN"]).start()

python app.py

EventPayload(
    actor_id=actor,
    session_id=session,
    kind="user_message",
    content=question,
    team_id=f"slack_{event['channel']}",   # the channel is the team
)

recalled = memsy.search(
    question,
    actor_id=actor,
    team_ids=[f"slack_{event['channel']}"],
    limit=5,
)

# serve.py -- an HTTP endpoint in place of SocketModeHandler
from fastapi import FastAPI, Request
from slack_bolt.adapter.fastapi import SlackRequestHandler

from app import app  # the same App, the same @app.event("app_mention")

api = FastAPI()
handler = SlackRequestHandler(app)


@api.post("/slack/events")
async def slack_events(request: Request):
    return await handler.handle(request)