Building a Slack Data Analyst Bot with Claude Managed Agents

The anthropic/claude-cookbooks repository ships a production-ready pattern in sre_bot_slack.py that combines Slack Bolt, the Claude Agent SDK, and a subprocess-based MCP server to deploy an interactive data analysis assistant directly into Slack channels.

This implementation lets you transform Claude into an on-demand data analyst capable of querying warehouses, fetching spreadsheets, and returning formatted insights within Slack threads. By adapting the Site Reliability Engineering (SRE) bot architecture, you replace infrastructure-focused tools with analytics-oriented ones while retaining the same robust event handling and streaming response system.

Architecture Overview

The solution comprises four integrated components that handle authentication, AI orchestration, tool execution, and message formatting:

Slack Bolt Async App (sre_bot_slack.py)
Receives app_mention and direct-message events via Socket Mode, authenticating with SLACK_BOT_TOKEN and SLACK_APP_TOKEN. The async handler runs behind NAT and firewalls without exposing public endpoints. Critical initialization occurs at lines 49-55, where AsyncApp and AsyncSocketModeHandler are instantiated.

Claude Agent SDK
Drives the Managed Agents through the query() async generator. In sre_bot_slack.py (lines 86-100), ClaudeAgentOptions configures the model (claude-opus-4-6), whitelists allowed tools, and sets permission_mode="acceptEdits" to enable direct tool invocation without human-in-the-loop interruptions.

MCP Subprocess Server (sre_mcp_server.py)
Implements the tool backend using JSON-RPC over stdio. Running in a separate process prevents SDK race conditions while exposing a TOOLS list (lines 57-120) that defines available functions, descriptions, and JSON schemas. Each tool maps to an async coroutine returning structured content blocks.

Utility Helpers
The convert_markdown_to_slack() function (lines 387-398 in sre_bot_slack.py) sanitizes Claude's markdown output for Slack compatibility, converting **bold** to *bold* and stripping headers before posting via the Web API.

Step-by-Step Message Flow

1. Event Ingestion: A user mentions the bot (@analyst show weekly revenue) or sends a DM. The handle_mention() or handle_message() handler strips the bot ID and extracts raw text.

2. Background Processing: The handler immediately launches process_investigation() (or your renamed equivalent) as an asyncio.create_task() to prevent Slack's 3-second timeout window from triggering.

3. Agent Initialization: The background task instantiates ClaudeAgentOptions, pointing mcp_servers to the Python subprocess running sre_mcp_server.py.

4. Streaming Responses: The SDK yields AssistantMessage, ToolUseBlock, and ResultMessage objects. Text blocks pass through convert_markdown_to_slack() before posting to the thread. Tool invocations trigger a "🔧 Checking {tool_name}…" status message.

5. Tool Execution: When Claude calls a tool, the MCP server executes the corresponding async function (e.g., run_sql_query) and returns JSON results, which render as formatted code blocks in Slack.

6. Completion: The loop terminates when Claude finishes reasoning, closing the background task while the Slack handler remains available for subsequent queries.

Converting SRE Tools to Analytics Capabilities

The repository's default implementation focuses on site reliability tasks. To pivot toward data analysis, modify the TOOLS list in sre_mcp_server.py to include analytics functions:

• Replace query_metrics (Prometheus) with run_sql_query (BigQuery/ClickHouse)
• Replace get_logs (Kubernetes) with fetch_audit_logs (Snowflake)
• Replace run_shell_command (kubectl) with execute_python_analysis (Pandas scripts)
• Add fetch_google_sheet for ad-hoc spreadsheet analysis

The Slack integration layer remains unchanged; only the tool implementations and the allowed_tools whitelist in ClaudeAgentOptions require updates.

Implementation Guide

Minimal Bot Entry Point

Create a new file (e.g., analyst_bot_slack.py) adjacent to the example. This implementation reuses the architecture from sre_bot_slack.py but targets analytics workflows:

#!/usr/bin/env python3
"""
Slack Data Analyst Bot – Claude Managed Agents implementation.
"""

import asyncio
import os
import re
import sys
from pathlib import Path
from dotenv import load_dotenv
from slack_bolt.async_app import AsyncApp
from slack_bolt.adapter.socket_mode.async_handler import AsyncSocketModeHandler
from claude_agent_sdk import (
    query,
    ClaudeAgentOptions,
    AssistantMessage,
    TextBlock,
    ToolUseBlock,
    ResultMessage,
)

# Load environment variables first

load_dotenv()

app = AsyncApp(token=os.getenv("SLACK_BOT_TOKEN"))

def convert_markdown_to_slack(text: str) -> str:
    """Sanitize markdown for Slack mrkdwn compatibility."""
    text = re.sub(r"^###?\s*(.+)$", r"*\1*", text, flags=re.MULTILINE)
    text = re.sub(r"\*\*([^*]+)\*\*", r"*\1*", text)
    return text

async def process_query(user_prompt: str, channel: str, thread_ts: str, say):
    """Stream Claude responses and tool results to Slack."""
    options = ClaudeAgentOptions(
        model="claude-opus-4-6",
        allowed_tools=[
            "Task", "Read", "Write", "Edit", "Bash",
            "run_sql_query", "fetch_google_sheet",
        ],
        permission_mode="acceptEdits",
        mcp_servers={
            "analytics": {
                "command": sys.executable,
                "args": [str(Path(__file__).parent / "analytics_mcp_server.py")],
            }
        },
        system_prompt=(
            "You are a data analyst in Slack. Answer questions using available "
            "tools. Return concise, markdown-formatted insights."
        ),
    )

    async for msg in query(prompt=user_prompt, options=options):
        if isinstance(msg, AssistantMessage):
            for block in msg.content:
                if isinstance(block, TextBlock) and block.text.strip():
                    await say(
                        text=convert_markdown_to_slack(block.text.strip()),
                        thread_ts=thread_ts
                    )
                elif isinstance(block, ToolUseBlock):
                    tool_name = block.name.replace("mcp__analytics__", "")
                    await say(text=f"🔧 *Checking {tool_name}…*", thread_ts=thread_ts)
        elif isinstance(msg, ResultMessage) and msg.is_error:
            await say(text=f"❌ Error: {msg.result}", thread_ts=thread_ts)

@app.event("app_mention")
async def handle_mention(event, say):
    channel = event["channel"]
    thread_ts = event.get("thread_ts", event["ts"])
    # Remove bot mention from text

    text = re.sub(r"<@[A-Z0-9]+>\s*", "", event["text"]).strip()
    
    if not text:
        await say(text="👋 Mention me with a query.", thread_ts=thread_ts)
        return
    
    # Prevent Slack timeout with background task

    asyncio.create_task(process_query(text, channel, thread_ts, say))

async def main():
    handler = AsyncSocketModeHandler(app, os.getenv("SLACK_APP_TOKEN"))
    await handler.start_async()

if __name__ == "__main__":
    asyncio.run(main())

Adding a SQL Query Tool

Extend your MCP server (adapted from sre_mcp_server.py) by adding a new entry to the TOOLS list and implementing the handler:

# analytics_mcp_server.py

import asyncpg
import tabulate
from typing import Any

TOOLS = [
    {
        "name": "run_sql_query",
        "description": "Execute read-only SQL against the analytics warehouse. Returns markdown table.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "sql": {
                    "type": "string",
                    "description": "SELECT statement (no DML allowed)"
                }
            },
            "required": ["sql"]
        },
    },
    # ... other tools

]

async def run_sql_query(sql: str) -> dict[str, Any]:
    """Execute query and return formatted results."""
    conn = await asyncpg.connect(os.getenv("WAREHOUSE_DSN"))
    try:
        rows = await conn.fetch(sql)
        if not rows:
            return {"content": [{"type": "text", "text": "No results returned."}]}
        
        headers = list(dict(rows[0]).keys())
        data = [list(dict(r).values()) for r in rows[:20]]  # Limit 20 rows

        
        table = tabulate.tabulate(data, headers=headers, tablefmt="github")
        return {
            "content": [{"type": "text", "text": f"```\n{table}\n```"}]
        }
    finally:
        await conn.close()

Key Source Files

Your development workflow centers on these files from the anthropics/claude-cookbooks repository:

• claude_agent_sdk/site_reliability_agent/examples/sre_bot_slack.py: Reference implementation containing Slack initialization (lines 49-55), ClaudeAgentOptions configuration (lines 86-100), and the convert_markdown_to_slack() helper (lines 387-398).

• claude_agent_sdk/site_reliability_agent/sre_mcp_server.py: Template for your analytics MCP server, showing the TOOLS definition structure (lines 57-120) and async tool implementation patterns.

• claude_agent_sdk/utils/agent_visualizer.py: Optional debugging utility for visualizing tool invocation chains during development.

• managed_agents/chief_of_staff_agent/agent.py: Demonstrates sub-agent delegation via the Task tool (lines 61-66), useful if you want to create specialized sub-agents (e.g., "financial_analyst", "marketing_analyst").

Local Development Setup

1. Install dependencies using the repository's uv configuration:

uv sync --all-extras

2. Configure environment variables in .env:

cp .env.example .env

# Edit .env to include:

# ANTHROPIC_API_KEY=sk-...

# SLACK_BOT_TOKEN=xoxb-...

# SLACK_APP_TOKEN=xapp-...

# WAREHOUSE_DSN=postgresql://...

3. Launch the bot:

python analyst_bot_slack.py

You should see console output indicating the Socket Mode handler is running. Invite the bot to a test channel and query:

@Analyst SELECT region, SUM(revenue) FROM sales GROUP BY region LIMIT 5;

The bot will stream status updates and return a formatted markdown table.

Summary

• Reuse the SRE pattern: The sre_bot_slack.py architecture provides Socket Mode handling, background task management, and markdown conversion suitable for any data-driven Slack bot.

• Implement analytics tools: Extend sre_mcp_server.py with SQL connectors, API clients, or Python analysis functions, exposing them through the MCP TOOLS schema.

• Configure ClaudeAgentOptions: Set permission_mode="acceptEdits" and whitelist your specific tools (run_sql_query, fetch_google_sheet) to authorize autonomous data retrieval.

• Prevent timeouts: Always wrap agent processing in asyncio.create_task() to return control to Slack within the 3-second acknowledgement window.

• Format for Slack: Pass all Claude output through convert_markdown_to_slack() to ensure mrkdwn compatibility.

Frequently Asked Questions

What is the Model Context Protocol (MCP) in this context?

MCP is the JSON-RPC protocol used between the Claude Agent SDK and your tool server. In sre_mcp_server.py, the MCP server runs as a subprocess that receives tool invocation requests over stdio and returns structured results, enabling Claude to execute Python code, SQL queries, or API calls securely without blocking the main async event loop.

How do I prevent Slack timeouts when queries take too long?

The sre_bot_slack.py implementation uses asyncio.create_task() to spawn process_investigation() (or equivalent) as a background task immediately upon receiving an event. This sends an immediate HTTP 200 response to Slack's event delivery system, preventing the 3-second timeout, while the agent continues processing and posts results asynchronously to the thread.

Can I use multiple specialized agents instead of a single data analyst?

Yes. Following the pattern in managed_agents/chief_of_staff_agent/agent.py (lines 61-66), define sub-agents (e.g., financial_analyst.json, marketing_analyst.json) in the .claude/agents/ directory. Reference them in ClaudeAgentOptions under allowed_tools using the Task tool, allowing Claude to delegate specific analysis types to purpose-built agents.

What security considerations apply to the MCP server?

Run the MCP server in a subprocess with restricted environment variables, as shown in the mcp_servers configuration dict. Ensure database credentials (like WAREHOUSE_DSN) are injected via environment variables or secret managers, never committed to source control. The permission_mode="acceptEdits" setting allows autonomous tool use, so validate all SQL inputs within your tool implementations to prevent injection attacks.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}