Add 'agent: BoltAgent' listener argument

.gitignore

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

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/agent/__init__.py

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

slack_bolt/agent/agent.py

@@ -0,0 +1,77 @@
+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.
+        """
+        resolved_channel = channel or self._channel_id
+        resolved_thread_ts = thread_ts or self._thread_ts
+        if resolved_channel is None:
+            raise ValueError(
+                "channel is required: provide it as an argument or ensure channel_id is set in the event context"
+            )
+        if resolved_thread_ts is None:
+            raise ValueError(
+                "thread_ts is required: provide it as an argument or ensure thread_ts is set in the event context"
+            )
+        return self._client.chat_stream(
+            channel=resolved_channel,
+            thread_ts=resolved_thread_ts,
+            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,74 @@
+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.
+        """
+        resolved_channel = channel or self._channel_id
+        resolved_thread_ts = thread_ts or self._thread_ts
+        if resolved_channel is None:
+            raise ValueError(
+                "channel is required: provide it as an argument or ensure channel_id is set in the event context"
+            )
+        if resolved_thread_ts is None:
+            raise ValueError(
+                "thread_ts is required: provide it as an argument or ensure thread_ts is set in the event context"
+            )
+        return await self._client.chat_stream(
+            channel=resolved_channel,
+            thread_ts=resolved_thread_ts,
+            recipient_team_id=recipient_team_id or self._team_id,
+            recipient_user_id=recipient_user_id or self._user_id,
+            **kwargs,
+        )

slack_bolt/context/async_context.py

@@ -1,4 +1,4 @@
-from typing import Optional
+from typing import TYPE_CHECKING, Optional
 
 from slack_sdk.web.async_client import AsyncWebClient
 
@@ -15,6 +15,9 @@
 from slack_bolt.context.set_title.async_set_title import AsyncSetTitle
 from slack_bolt.util.utils import create_copy
 
+if TYPE_CHECKING:
+    from slack_bolt.agent.async_agent import AsyncBoltAgent
+
 
 class AsyncBoltContext(BaseContext):
     """Context object associated with a request from Slack."""
@@ -187,6 +190,34 @@ async def handle_button_clicks(context):
             self["fail"] = AsyncFail(client=self.client, function_execution_id=self.function_execution_id)
         return self["fail"]
 
+    @property
+    def agent(self) -> "AsyncBoltAgent":
+        """`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()
+
+        Returns:
+            `AsyncBoltAgent` instance
+        """
+        if "agent" not in self:
+            from slack_bolt.agent.async_agent import AsyncBoltAgent
+
+            self["agent"] = AsyncBoltAgent(
+                client=self.client,
+                channel_id=self.channel_id,
+                thread_ts=self.thread_ts,
+                team_id=self.team_id,
+                user_id=self.user_id,
+            )
+        return self["agent"]
+
     @property
     def set_title(self) -> Optional[AsyncSetTitle]:
         return self.get("set_title")

slack_bolt/context/base_context.py

@@ -38,6 +38,7 @@ class BaseContext(dict):
         "set_status",
         "set_title",
         "set_suggested_prompts",
+        "agent",
     ]
     # Note that these items are not copyable, so when you add new items to this list,
     # you must modify ThreadListenerRunner/AsyncioListenerRunner's _build_lazy_request method to pass the values.

slack_bolt/context/context.py

@@ -1,4 +1,4 @@
-from typing import Optional
+from typing import TYPE_CHECKING, Optional
 
 from slack_sdk import WebClient
 
@@ -15,6 +15,9 @@
 from slack_bolt.context.set_title import SetTitle
 from slack_bolt.util.utils import create_copy
 
+if TYPE_CHECKING:
+    from slack_bolt.agent.agent import BoltAgent
+
 
 class BoltContext(BaseContext):
     """Context object associated with a request from Slack."""
@@ -188,6 +191,34 @@ def handle_button_clicks(context):
             self["fail"] = Fail(client=self.client, function_execution_id=self.function_execution_id)
         return self["fail"]
 
+    @property
+    def agent(self) -> "BoltAgent":
+        """`agent` listener argument for building AI-powered Slack agents.
+
+        Experimental:
+            This API is experimental and may change in future releases.
+
+            @app.event("app_mention")
+            def handle_mention(agent):
+                stream = agent.chat_stream()
+                stream.append(markdown_text="Hello!")
+                stream.stop()
+
+        Returns:
+            `BoltAgent` instance
+        """
+        if "agent" not in self:
+            from slack_bolt.agent.agent import BoltAgent
+
+            self["agent"] = BoltAgent(
+                client=self.client,
+                channel_id=self.channel_id,
+                thread_ts=self.thread_ts,
+                team_id=self.team_id,
+                user_id=self.user_id,
+            )
+        return self["agent"]
+
     @property
     def set_title(self) -> Optional[SetTitle]:
         return self.get("set_title")

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

@@ -29,7 +29,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 +83,10 @@ 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 or "args" in required_arg_names:
+        all_available_args["agent"] = request.context.agent
+
     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 +106,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)})")