Build a Slack data analyst bot with Claude Managed Agents Apr 2026 • Agent Patterns Integrations Mention the bot with a CSV to get an analysis report in-thread, with multi-turn follow-ups on the same session.

import io
import os
import threading
from getpass import getpass
import requests
from anthropic import Anthropic
from dotenv import load_dotenv, set_key
from markdown_to_mrkdwn import SlackMarkdownConverter
from slack_bolt import App
from slack_bolt.adapter.socket_mode import SocketModeHandler

load_dotenv(override=True)

# Prompt for Slack tokens on first run and save them to .env
for key in ("SLACK_BOT_TOKEN", "SLACK_APP_TOKEN"):
    if not os.environ.get(key):
        os.environ[key] = getpass(f"{key}: ")
        set_key(".env", key, os.environ[key])

client = Anthropic()
app = App(token=os.environ["SLACK_BOT_TOKEN"])

for key in ("ANALYST_ENV_ID", "ANALYST_AGENT_ID", "ANALYST_AGENT_VERSION"):
    if not os.environ.get(key):
        raise RuntimeError(f"{key} not set. Run data_analyst_agent.ipynb first.")

ANALYST_AGENT = {
    "id": os.environ["ANALYST_AGENT_ID"],
    "version": int(os.environ["ANALYST_AGENT_VERSION"]),
}
ANALYST_ENV_ID = os.environ["ANALYST_ENV_ID"]

# thread_ts -> session_id mapping for follow-ups
thread_sessions: dict[str, str] = {}
mrkdwn = SlackMarkdownConverter()

@app.event("app_mention")
def on_mention(event, say, ack):
    ack()
    channel = event["channel"]
    thread_ts = event.get("thread_ts") or event["ts"]
    # Mention text arrives as "<@BOTID> question"; drop the mention prefix
    question = event["text"].split(">", 1)[-1].strip()
    slack_file = (event.get("files") or [None])[0]
    
    say(text="On it. Analyzing now.", thread_ts=thread_ts)
    
    # Run slow work in background thread
    threading.Thread(
        target=start_analysis,
        args=(channel, thread_ts, question, slack_file)
    ).start()

def start_analysis(channel: str, thread_ts: str, question: str, slack_file: dict | None) -> None:
    try:
        resources = []
        if slack_file:
            # Download file from Slack
            resp = requests.get(
                slack_file["url_private"],
                headers={"Authorization": f"Bearer {app.client.token}"},
                timeout=30,
            )
            resp.raise_for_status()
            mime = slack_file.get("mimetype", "text/csv")
            
            # Upload to Anthropic Files API
            uploaded = client.beta.files.upload(
                file=(slack_file["name"], io.BytesIO(resp.content), mime)
            )
            mount = "/mnt/session/uploads/data.csv"
            resources.append({
                "type": "file",
                "file_id": uploaded.id,
                "mount_path": mount
            })
            question += f"\n\nThe data is mounted at {mount}."
        
        # Create session per Slack thread
        session = client.beta.sessions.create(
            environment_id=ANALYST_ENV_ID,
            agent={"type": "agent", **ANALYST_AGENT},
            resources=resources,
            title="".join(c for c in question if c.isprintable())[:80],
            metadata={
                "slack_channel": channel,
                "slack_thread_ts": thread_ts
            },
        )
        thread_sessions[thread_ts] = session.id
        
        # Send question to agent
        client.beta.sessions.events.send(
            session.id,
            events=[{
                "type": "user.message",
                "content": [{"type": "text", "text": question}]
            }],
        )
        relay_stream(session.id, channel, thread_ts)
    except Exception as e:
        app.client.chat_postMessage(
            channel=channel,
            thread_ts=thread_ts,
            text=f"Analysis failed: {type(e).__name__}: {e}"
        )

def relay_stream(session_id: str, channel: str, thread_ts: str) -> None:
    summary = ""
    posted_progress = False
    
    for ev in client.beta.sessions.events.stream(session_id):
        t = ev.type
        if t == "agent.message":
            # Keep the latest text block; it becomes the final summary
            for b in ev.content:
                if b.type == "text" and b.text.strip():
                    summary = b.text
        elif t == "agent.tool_use" and not posted_progress:
            # Post one-time progress update when agent starts running commands
            app.client.chat_postMessage(
                channel=channel,
                thread_ts=thread_ts,
                text="Running analysis..."
            )
            posted_progress = True
        elif t == "session.status_idle":
            break
        elif t == "session.status_terminated":
            trace = f"https://platform.claude.com/sessions/{session_id}"
            app.client.chat_postMessage(
                channel=channel,
                thread_ts=thread_ts,
                text=f"Session terminated. [View trace]({trace})"
            )
            return
    
    # Post final summary
    if summary:
        app.client.chat_postMessage(
            channel=channel,
            thread_ts=thread_ts,
            text=mrkdwn.convert(summary)
        )
    
    # Upload agent-generated files
    for f in client.beta.files.list(scope_id=session_id):
        if f.downloadable:
            content = client.beta.files.download(f.id).content
            app.client.files_upload_v2(
                channel=channel,
                thread_ts=thread_ts,
                file=content,
                filename=f.filename
            )

@app.event("message")
def on_message(event, say, ack):
    ack()
    
    # Only handle replies in threads where bot has a session
    thread_ts = event.get("thread_ts")
    if not thread_ts or thread_ts not in thread_sessions:
        return
    
    # Skip bot's own messages
    if event.get("bot_id"):
        return
    
    session_id = thread_sessions[thread_ts]
    channel = event["channel"]
    text = event.get("text", "")
    
    say(text="Processing follow-up...", thread_ts=thread_ts)
    
    # Send follow-up to same session
    try:
        client.beta.sessions.events.send(
            session_id,
            events=[{
                "type": "user.message",
                "content": [{"type": "text", "text": text}]
            }],
        )
        relay_stream(session_id, channel, thread_ts)
    except Exception as e:
        app.client.chat_postMessage(
            channel=channel,
            thread_ts=thread_ts,
            text=f"Follow-up failed: {type(e).__name__}: {e}"
        )

# Main entry point - run the bot with Socket Mode
if __name__ == "__main__":
    handler = SocketModeHandler(app, os.environ["SLACK_APP_TOKEN"])
    handler.start()

anthropic>=0.91.0
python-dotenv
slack-bolt
requests
markdown-to-mrkdwn

SLACK_BOT_TOKEN=xoxb-your-token-here
SLACK_APP_TOKEN=xapp-your-token-here
ANALYST_ENV_ID=your-env-id
ANALYST_AGENT_ID=your-agent-id
ANALYST_AGENT_VERSION=1