Skip to main content

Adding agent features with Bolt for Python

Check out the Support Agent sample app

The code snippets throughout this guide are from our Support Agent sample app, Casey, which supports integration with Pydantic, Anthropic, and OpenAI.

View our agent quickstart to get up and running with Casey. Otherwise, read on for exploration and explanation of agent-focused Bolt features found within Casey.

Your agent can utilize features applicable to messages throughout Slack, like chat streaming and feedback buttons. They can also utilize the Assistant class for a side-panel view designed with AI in mind.

If you're unfamiliar with using these feature within Slack, you may want to read the API docs on the subject. Then come back here to implement them with Bolt!

Slack MCP Server​

Casey can harness the Slack MCP Server when deployed via an HTTP Server with OAuth.

To enable the Slack MCP Server:

  1. Install ngrok and start a tunnel:
ngrok http 3000

  1. Copy the https://*.ngrok-free.app URL from the ngrok output.

  2. Update manifest.json for HTTP mode:

    • Set socket_mode_enabled to false
    • Replace ngrok-free.app with your ngrok domain (e.g. YOUR_NGROK_SUBDOMAIN.ngrok-free.app)

  3. Create a new local dev app:

slack install -E local

  1. Enable MCP for your app:

    • Run slack app settings to open your app's settings
    • Navigate to Agents & AI Apps in the left-side navigation
    • Toggle Model Context Protocol on

  2. Update your .env OAuth environment variables:

    • Run slack app settings to open App Settings
    • Copy Client ID, Client Secret, and Signing Secret
    • Update SLACK_REDIRECT_URI in .env with your ngrok domain
SLACK_CLIENT_ID=YOUR_CLIENT_ID
SLACK_CLIENT_SECRET=YOUR_CLIENT_SECRET
SLACK_REDIRECT_URI=https://YOUR_NGROK_SUBDOMAIN.ngrok-free.app/slack/oauth_redirect
SLACK_SIGNING_SECRET=YOUR_SIGNING_SECRET
  1. Start the app:
slack run app_oauth.py
  1. Click the install URL printed in the terminal to install the app to your workspace via OAuth.

Your agent can now access the Slack MCP server!

Listening for user invocation​

Agents can be invoked throughout Slack, such as via @mentions in channels, messaging the agent, and using the assistant side panel.

import re
from logging import Logger

from agents import Runner
from slack_bolt import BoltContext, Say, SayStream, SetStatus
from slack_sdk import WebClient

from agent import CaseyDeps, casey_agent
from thread_context import conversation_store
from listeners.views.feedback_builder import build_feedback_blocks


def handle_app_mentioned(
    client: WebClient,
    context: BoltContext,
    event: dict,
    logger: Logger,
    say: Say,
    say_stream: SayStream,
    set_status: SetStatus,
):
    """Handle @Casey mentions in channels."""
    try:
        channel_id = context.channel_id
        text = event.get("text", "")
        thread_ts = event.get("thread_ts") or event["ts"]
        user_id = context.user_id

        # Strip the bot mention from the text
        cleaned_text = re.sub(r"<@[A-Z0-9]+>", "", text).strip()

        if not cleaned_text:
            say(
                text="Hey there! How can I help you? Describe your IT issue and I'll do my best to assist.",
                thread_ts=thread_ts,
            )
            return

        # Add eyes reaction only to the first message (not threaded replies)
        if not event.get("thread_ts"):
            client.reactions_add(
                channel=channel_id,
                timestamp=event["ts"],
                name="eyes",
            )
        ...

Setting status​

Your app can show its users action is happening behind the scenes by setting its thread status.

def handle_app_mentioned(
    set_status: SetStatus,
    ...
):
    set_status(
        status="Thinking...",
        loading_messages=[
            "Teaching the hamsters to type faster…",
            "Untangling the internet cables…",
            "Consulting the office goldfish…",
            "Polishing up the response just for you…",
            "Convincing the AI to stop overthinking…",
        ],
    )

Streaming messages​

You can have your app's messages stream in to replicate conventional agent behavior. Bolt for Python provides a say_stream utility as a listener argument available for app.event and app.message listeners.

The say_stream utility streamlines calling the Python Slack SDK's WebClient.chat_stream helper utility by sourcing parameter values from the relevant event payload.

ParameterValue
channel_idSourced from the event payload.
thread_tsSourced from the event payload. Falls back to the ts value if available.
recipient_team_idSourced from the event team_id (enterprise_id if the app is installed on an org).
recipient_user_idSourced from the user_id of the event.

If neither a channel_id or thread_ts can be sourced, then the utility will be None.

app.message('*', async ({ sayStream }) => {
  const stream = sayStream();
  await stream.append({ markdown_text: "Here's my response..." });
  await stream.append({ markdown_text: "And here's more..." });
  await stream.stop();
});

Adding and handling feedback​

You can use the feedback buttons block element to allow users to immediately provide feedback regarding the app's responses. Here's what the feedback buttons look like from the Support Agent sample app:

.../listeners/views/feedback_builder.py
from slack_sdk.models.blocks import (
    Block,
    ContextActionsBlock,
    FeedbackButtonObject,
    FeedbackButtonsElement,
)


def build_feedback_blocks() -> list[Block]:
    """Build feedback blocks with thumbs up/down buttons."""
    return [
        ContextActionsBlock(
            elements=[
                FeedbackButtonsElement(
                    action_id="feedback",
                    positive_button=FeedbackButtonObject(
                        text="Good Response",
                        accessibility_label="Submit positive feedback on this response",
                        value="good-feedback",
                    ),
                    negative_button=FeedbackButtonObject(
                        text="Bad Response",
                        accessibility_label="Submit negative feedback on this response",
                        value="bad-feedback",
                    ),
                )
            ]
        )
    ]

That feedback block is then rendered at the bottom of your app's message via the say_stream utility.

...
        # Stream response in thread with feedback buttons
        streamer = say_stream()
        streamer.append(markdown_text=result.output)
        feedback_blocks = build_feedback_blocks()
        streamer.stop(blocks=feedback_blocks)
...

You can also add a response for when the user provides feedback.

...listeners/actions/feedback_button.py
from logging import Logger

from slack_bolt import Ack, BoltContext
from slack_sdk import WebClient


def handle_feedback_button(
    ack: Ack, body: dict, client: WebClient, context: BoltContext, logger: Logger
):
    """Handle thumbs up/down feedback on Casey's responses."""
    ack()

    try:
        channel_id = context.channel_id
        user_id = context.user_id
        message_ts = body["message"]["ts"]
        feedback_value = body["actions"][0]["value"]

        if feedback_value == "good-feedback":
            client.chat_postEphemeral(
                channel=channel_id,
                user=user_id,
                thread_ts=message_ts,
                text="Glad that was helpful! :tada:",
            )
        else:
            client.chat_postEphemeral(
                channel=channel_id,
                user=user_id,
                thread_ts=message_ts,
                text="Sorry that wasn't helpful. :slightly_frowning_face: Try rephrasing your question or I can create a support ticket for you.",
            )

        logger.debug(
            f"Feedback received: value={feedback_value}, message_ts={message_ts}"
        )
    except Exception as e:
        logger.exception(f"Failed to handle feedback: {e}")

Full example​

Putting all those concepts together results in a dynamic agent ready to helpfully respond.

Full example
app_mentioned.py
import re
from logging import Logger

from slack_bolt import BoltContext, Say, SayStream, SetStatus
from slack_sdk import WebClient

from agent import CaseyDeps, casey_agent, get_model
from thread_context import conversation_store
from listeners.views.feedback_builder import build_feedback_blocks


def handle_app_mentioned(
    client: WebClient,
    context: BoltContext,
    event: dict,
    logger: Logger,
    say: Say,
    say_stream: SayStream,
    set_status: SetStatus,
):
    """Handle @Casey mentions in channels."""
    try:
        channel_id = context.channel_id
        text = event.get("text", "")
        thread_ts = event.get("thread_ts") or event["ts"]
        user_id = context.user_id

        # Strip the bot mention from the text
        cleaned_text = re.sub(r"<@[A-Z0-9]+>", "", text).strip()

        if not cleaned_text:
            say(
                text="Hey there! How can I help you? Describe your IT issue and I'll do my best to assist.",
                thread_ts=thread_ts,
            )
            return

        # Add eyes reaction only to the first message (not threaded replies)
        if not event.get("thread_ts"):
            client.reactions_add(
                channel=channel_id,
                timestamp=event["ts"],
                name="eyes",
            )

        # Set assistant thread status with loading messages
        set_status(
            status="Thinking...",
            loading_messages=[
                "Teaching the hamsters to type faster…",
                "Untangling the internet cables…",
                "Consulting the office goldfish…",
                "Polishing up the response just for you…",
                "Convincing the AI to stop overthinking…",
            ],
        )

        # Get conversation history
        history = conversation_store.get_history(channel_id, thread_ts)

        # Run the agent
        deps = CaseyDeps(
            client=client,
            user_id=user_id,
            channel_id=channel_id,
            thread_ts=thread_ts,
            message_ts=event["ts"],
        )
        result = casey_agent.run_sync(
            cleaned_text,
            model=get_model(),
            deps=deps,
            message_history=history,
        )

        # Stream response in thread with feedback buttons
        streamer = say_stream()
        streamer.append(markdown_text=result.output)
        feedback_blocks = build_feedback_blocks()
        streamer.stop(blocks=feedback_blocks)

        # Store conversation history
        conversation_store.set_history(channel_id, thread_ts, result.all_messages())

    except Exception as e:
        logger.exception(f"Failed to handle app mention: {e}")
        say(
            text=f":warning: Something went wrong! ({e})",
            thread_ts=event.get("thread_ts") or event["ts"],
        )

Onward: adding custom tools​

Casey comes with test tools and simulated systems. You can extend it with custom tools to make it a fully functioning Slack agent.

In this example, we'll add a tool that makes live calls to check the GitHub status.

  1. Create agent/tools/{tool-name}.py and define the tool with the @tool decorator:
agent/tools/check_github_status.py
from claude_agent_sdk import tool
import httpx

@tool(
    name="check_github_status",
    description="Check GitHub's current operational status",
    input_schema={},
)
async def check_github_status_tool(args):
    """Check if GitHub is operational."""
    async with httpx.AsyncClient() as client:
        response = await client.get("https://www.githubstatus.com/api/v2/status.json")
        data = response.json()
        status = data["status"]["indicator"]
        description = data["status"]["description"]
    
    return {
        "content": [
            {
                "type": "text",
                "text": f"**GitHub Status** — {status}\n{description}",
            }
        ]
    }
  1. Import the tool in agent/casey.py:
agent/casey.py
from agent.tools import check_github_status_tool
  1. Register in casey_tools_server:
agent/casey.py
casey_tools_server = create_sdk_mcp_server(
    name="casey-tools",
    version="1.0.0",
    tools=[
        check_github_status_tool,  # Add here
        # ... other tools
    ],
)
  1. Add to CASEY_TOOLS:
agent/casey.py
CASEY_TOOLS = [
    "check_github_status",  # Add here
    # ... other tools
]

Use this example as a jumping off point for building out an agent with the capabilities you need!