Add 'agent: BoltAgent' listener argument

.gitignore

@@ -17,6 +17,9 @@ venv/
 .venv*
 .env/
 
+# claude
+.claude/*.local.json
+
 # codecov / coverage
 .coverage
 cov_*

docs/english/_sidebar.json

@@ -85,6 +85,11 @@
 			"tools/bolt-python/concepts/token-rotation"
 		]
 	},
+	{
+		"type": "category",
+		"label": "Experiments",
+		"items": ["tools/bolt-python/experiments"]
+	},
 	{
 		"type": "category",
 		"label": "Legacy",

docs/english/experiments.md

@@ -0,0 +1,34 @@
+# Experiments
+
+Bolt for Python includes experimental features still under active development. These features may be fleeting, may not be perfectly polished, and should be thought of as available for use "at your own risk." 
+
+Experimental features are categorized as `semver:patch` until the experimental status is removed.
+
+We love feedback from our community, so we encourage you to explore and interact with the [GitHub repo](https://github.com/slackapi/bolt-python). Contributions, bug reports, and any feedback are all helpful; let us nurture the Slack CLI together to help make building Slack apps more pleasant for everyone.
+
+## Available experiments
+* [Agent listener argument](#agent)
+
+## Agent listener argument {#agent}
+
+The `agent: BoltAgent` listener argument provides access to AI agent-related features.
+
+The `BoltAgent` and `AsyncBoltAgent` classes offer a `chat_stream()` method that comes pre-configured with event context defaults: `channel_id`, `thread_ts`, `team_id`, and `user_id` fields. 
+
+The listener argument is wired into the Bolt `kwargs` injection system, so listeners can declare it as a parameter or access it via the `context.agent` property.
+
+### Example
+
+```python
+from slack_bolt import BoltAgent
+
+@app.event("app_mention")
+def handle_mention(agent: BoltAgent):
+    stream = agent.chat_stream()
+    stream.append(markdown_text="Hello!")
+    stream.stop()
+```
+
+### Limitations
+
+The `chat_stream()` method currently only works when the `thread_ts` field is available in the event context (DMs and threaded replies). Top-level channel messages do not have a `thread_ts` field, and the `ts` field is not yet provided to `BoltAgent`.

slack_bolt/__init__.py

@@ -21,6 +21,7 @@
 from .response import BoltResponse
 
 # AI Agents & Assistants
+from .agent import BoltAgent
 from .middleware.assistant.assistant import (
     Assistant,
 )
@@ -46,6 +47,7 @@
     "CustomListenerMatcher",
     "BoltRequest",
     "BoltResponse",
+    "BoltAgent",
     "Assistant",
     "AssistantThreadContext",
     "AssistantThreadContextStore",

slack_bolt/adapter/__init__.py

@@ -1,2 +1 @@
-"""Adapter modules for running Bolt apps along with Web frameworks or Socket Mode.
-"""
+"""Adapter modules for running Bolt apps along with Web frameworks or Socket Mode."""

slack_bolt/agent/__init__.py

@@ -0,0 +1,5 @@
+from .agent import BoltAgent
+
+__all__ = [
+    "BoltAgent",
+]

slack_bolt/agent/agent.py

@@ -0,0 +1,73 @@
+from typing import Optional
+
+from slack_sdk import WebClient
+from slack_sdk.web.chat_stream import ChatStream
+
+
+class BoltAgent:
+    """Agent listener argument for building AI-powered Slack agents.
+
+    Experimental:
+        This API is experimental and may change in future releases.
+
+        FIXME: chat_stream() only works when thread_ts is available (DMs and threaded replies).
+        It does not work on channel messages because ts is not provided to BoltAgent yet.
+
+        @app.event("app_mention")
+        def handle_mention(agent):
+            stream = agent.chat_stream()
+            stream.append(markdown_text="Hello!")
+            stream.stop()
+    """
+
+    def __init__(
+        self,
+        *,
+        client: WebClient,
+        channel_id: Optional[str] = None,
+        thread_ts: Optional[str] = None,
+        team_id: Optional[str] = None,
+        user_id: Optional[str] = None,
+    ):
+        self._client = client
+        self._channel_id = channel_id
+        self._thread_ts = thread_ts
+        self._team_id = team_id
+        self._user_id = user_id
+
+    def chat_stream(
+        self,
+        *,
+        channel: Optional[str] = None,
+        thread_ts: Optional[str] = None,
+        recipient_team_id: Optional[str] = None,
+        recipient_user_id: Optional[str] = None,
+        **kwargs,
+    ) -> ChatStream:
+        """Creates a ChatStream with defaults from event context.
+
+        Each call creates a new instance. Create multiple for parallel streams.
+
+        Args:
+            channel: Channel ID. Defaults to the channel from the event context.
+            thread_ts: Thread timestamp. Defaults to the thread_ts from the event context.
+            recipient_team_id: Team ID of the recipient. Defaults to the team from the event context.
+            recipient_user_id: User ID of the recipient. Defaults to the user from the event context.
+            **kwargs: Additional arguments passed to ``WebClient.chat_stream()``.
+
+        Returns:
+            A new ``ChatStream`` instance.
+        """
+        provided = [arg for arg in (channel, thread_ts, recipient_team_id, recipient_user_id) if arg is not None]
+        if provided and len(provided) < 4:
+            raise ValueError(
+                "Either provide all of channel, thread_ts, recipient_team_id, and recipient_user_id, or none of them"
+            )
+        # Argument validation is delegated to chat_stream() and the API
+        return self._client.chat_stream(
+            channel=channel or self._channel_id,  # type: ignore[arg-type]
+            thread_ts=thread_ts or self._thread_ts,  # type: ignore[arg-type]
+            recipient_team_id=recipient_team_id or self._team_id,
+            recipient_user_id=recipient_user_id or self._user_id,
+            **kwargs,
+        )

slack_bolt/agent/async_agent.py

@@ -0,0 +1,70 @@
+from typing import Optional
+
+from slack_sdk.web.async_client import AsyncWebClient
+from slack_sdk.web.async_chat_stream import AsyncChatStream
+
+
+class AsyncBoltAgent:
+    """Async agent listener argument for building AI-powered Slack agents.
+
+    Experimental:
+        This API is experimental and may change in future releases.
+
+        @app.event("app_mention")
+        async def handle_mention(agent):
+            stream = await agent.chat_stream()
+            await stream.append(markdown_text="Hello!")
+            await stream.stop()
+    """
+
+    def __init__(
+        self,
+        *,
+        client: AsyncWebClient,
+        channel_id: Optional[str] = None,
+        thread_ts: Optional[str] = None,
+        team_id: Optional[str] = None,
+        user_id: Optional[str] = None,
+    ):
+        self._client = client
+        self._channel_id = channel_id
+        self._thread_ts = thread_ts
+        self._team_id = team_id
+        self._user_id = user_id
+
+    async def chat_stream(
+        self,
+        *,
+        channel: Optional[str] = None,
+        thread_ts: Optional[str] = None,
+        recipient_team_id: Optional[str] = None,
+        recipient_user_id: Optional[str] = None,
+        **kwargs,
+    ) -> AsyncChatStream:
+        """Creates an AsyncChatStream with defaults from event context.
+
+        Each call creates a new instance. Create multiple for parallel streams.
+
+        Args:
+            channel: Channel ID. Defaults to the channel from the event context.
+            thread_ts: Thread timestamp. Defaults to the thread_ts from the event context.
+            recipient_team_id: Team ID of the recipient. Defaults to the team from the event context.
+            recipient_user_id: User ID of the recipient. Defaults to the user from the event context.
+            **kwargs: Additional arguments passed to ``AsyncWebClient.chat_stream()``.
+
+        Returns:
+            A new ``AsyncChatStream`` instance.
+        """
+        provided = [arg for arg in (channel, thread_ts, recipient_team_id, recipient_user_id) if arg is not None]
+        if provided and len(provided) < 4:
+            raise ValueError(
+                "Either provide all of channel, thread_ts, recipient_team_id, and recipient_user_id, or none of them"
+            )
+        # Argument validation is delegated to chat_stream() and the API
+        return await self._client.chat_stream(
+            channel=channel or self._channel_id,  # type: ignore[arg-type]
+            thread_ts=thread_ts or self._thread_ts,  # type: ignore[arg-type]
+            recipient_team_id=recipient_team_id or self._team_id,
+            recipient_user_id=recipient_user_id or self._user_id,
+            **kwargs,
+        )

slack_bolt/kwargs_injection/args.py

@@ -8,6 +8,7 @@
 from slack_bolt.context.fail import Fail
 from slack_bolt.context.get_thread_context.get_thread_context import GetThreadContext
 from slack_bolt.context.respond import Respond
+from slack_bolt.agent.agent import BoltAgent
 from slack_bolt.context.save_thread_context import SaveThreadContext
 from slack_bolt.context.say import Say
 from slack_bolt.context.set_status import SetStatus
@@ -102,6 +103,8 @@ def handle_buttons(args):
     """`get_thread_context()` utility function for AI Agents & Assistants"""
     save_thread_context: Optional[SaveThreadContext]
     """`save_thread_context()` utility function for AI Agents & Assistants"""
+    agent: Optional[BoltAgent]
+    """`agent` listener argument for AI Agents & Assistants"""
     # middleware
     next: Callable[[], None]
     """`next()` utility function, which tells the middleware chain that it can continue with the next one"""
@@ -135,6 +138,7 @@ def __init__(
         set_suggested_prompts: Optional[SetSuggestedPrompts] = None,
         get_thread_context: Optional[GetThreadContext] = None,
         save_thread_context: Optional[SaveThreadContext] = None,
+        agent: Optional[BoltAgent] = None,
         # As this method is not supposed to be invoked by bolt-python users,
         # the naming conflict with the built-in one affects
         # only the internals of this method
@@ -168,6 +172,7 @@ def __init__(
         self.set_suggested_prompts = set_suggested_prompts
         self.get_thread_context = get_thread_context
         self.save_thread_context = save_thread_context
+        self.agent = agent
 
         self.next: Callable[[], None] = next
         self.next_: Callable[[], None] = next

slack_bolt/kwargs_injection/async_args.py

@@ -1,6 +1,7 @@
 from logging import Logger
 from typing import Callable, Awaitable, Dict, Any, Optional
 
+from slack_bolt.agent.async_agent import AsyncBoltAgent
 from slack_bolt.context.ack.async_ack import AsyncAck
 from slack_bolt.context.async_context import AsyncBoltContext
 from slack_bolt.context.complete.async_complete import AsyncComplete
@@ -101,6 +102,8 @@ async def handle_buttons(args):
     """`get_thread_context()` utility function for AI Agents & Assistants"""
     save_thread_context: Optional[AsyncSaveThreadContext]
     """`save_thread_context()` utility function for AI Agents & Assistants"""
+    agent: Optional[AsyncBoltAgent]
+    """`agent` listener argument for AI Agents & Assistants"""
     # middleware
     next: Callable[[], Awaitable[None]]
     """`next()` utility function, which tells the middleware chain that it can continue with the next one"""
@@ -134,6 +137,7 @@ def __init__(
         set_suggested_prompts: Optional[AsyncSetSuggestedPrompts] = None,
         get_thread_context: Optional[AsyncGetThreadContext] = None,
         save_thread_context: Optional[AsyncSaveThreadContext] = None,
+        agent: Optional[AsyncBoltAgent] = None,
         next: Callable[[], Awaitable[None]],
         **kwargs,  # noqa
     ):
@@ -164,6 +168,7 @@ def __init__(
         self.set_suggested_prompts = set_suggested_prompts
         self.get_thread_context = get_thread_context
         self.save_thread_context = save_thread_context
+        self.agent = agent
 
         self.next: Callable[[], Awaitable[None]] = next
         self.next_: Callable[[], Awaitable[None]] = next

slack_bolt/kwargs_injection/async_utils.py

@@ -1,9 +1,11 @@
 import inspect
 import logging
+import warnings
 from typing import Callable, Dict, MutableSequence, Optional, Any
 
 from slack_bolt.request.async_request import AsyncBoltRequest
 from slack_bolt.response import BoltResponse
+from slack_bolt.warning import ExperimentalWarning
 from .async_args import AsyncArgs
 from slack_bolt.request.payload_utils import (
     to_options,
@@ -29,7 +31,7 @@ def build_async_required_kwargs(
     error: Optional[Exception] = None,  # for error handlers
     next_keys_required: bool = True,  # False for listeners / middleware / error handlers
 ) -> Dict[str, Any]:
-    all_available_args = {
+    all_available_args: Dict[str, Any] = {
         "logger": logger,
         "client": request.context.client,
         "req": request,
@@ -83,6 +85,23 @@ def build_async_required_kwargs(
         if k not in all_available_args:
             all_available_args[k] = v
 
+    # Defer agent creation to avoid constructing AsyncBoltAgent on every request
+    if "agent" in required_arg_names:
+        from slack_bolt.agent.async_agent import AsyncBoltAgent
+
+        all_available_args["agent"] = AsyncBoltAgent(
+            client=request.context.client,
+            channel_id=request.context.channel_id,
+            thread_ts=request.context.thread_ts,
+            team_id=request.context.team_id,
+            user_id=request.context.user_id,
+        )
+        warnings.warn(
+            "The agent listener argument is experimental and may change in future versions.",
+            category=ExperimentalWarning,
+            stacklevel=2,  # Point to the caller, not this internal helper
+        )
+
     if len(required_arg_names) > 0:
         # To support instance/class methods in a class for listeners/middleware,
         # check if the first argument is either self or cls
@@ -102,7 +121,7 @@ def build_async_required_kwargs(
     for name in required_arg_names:
         if name == "args":
             if isinstance(request, AsyncBoltRequest):
-                kwargs[name] = AsyncArgs(**all_available_args)  # type: ignore[arg-type]
+                kwargs[name] = AsyncArgs(**all_available_args)
             else:
                 logger.warning(f"Unknown Request object type detected ({type(request)})")