feat(agent): default to message 'ts' when no 'thread_ts' is avaialble for 'agent.chat_stream(...)'

slack_bolt/agent/agent.py

@@ -1,6 +1,7 @@
-from typing import Optional
+from typing import Dict, List, Optional, Sequence, Union
 
 from slack_sdk import WebClient
+from slack_sdk.web import SlackResponse
 from slack_sdk.web.chat_stream import ChatStream
 
 
@@ -71,3 +72,69 @@ def chat_stream(
             recipient_user_id=recipient_user_id or self._user_id,
             **kwargs,
         )
+
+    def set_status(
+        self,
+        *,
+        status: str,
+        loading_messages: Optional[List[str]] = None,
+        channel: Optional[str] = None,
+        thread_ts: Optional[str] = None,
+        **kwargs,
+    ) -> SlackResponse:
+        """Sets the status of an assistant thread.
+
+        Args:
+            status: The status text to display.
+            loading_messages: Optional list of loading messages to cycle through.
+            channel: Channel ID. Defaults to the channel from the event context.
+            thread_ts: Thread timestamp. Defaults to the thread_ts from the event context.
+            **kwargs: Additional arguments passed to ``WebClient.assistant_threads_setStatus()``.
+
+        Returns:
+            ``SlackResponse`` from the API call.
+        """
+        return self._client.assistant_threads_setStatus(
+            channel_id=channel or self._channel_id,  # type: ignore[arg-type]
+            thread_ts=thread_ts or self._thread_ts,  # type: ignore[arg-type]
+            status=status,
+            loading_messages=loading_messages,
+            **kwargs,
+        )
+
+    def set_suggested_prompts(
+        self,
+        *,
+        prompts: Sequence[Union[str, Dict[str, str]]],
+        title: Optional[str] = None,
+        channel: Optional[str] = None,
+        thread_ts: Optional[str] = None,
+        **kwargs,
+    ) -> SlackResponse:
+        """Sets suggested prompts for an assistant thread.
+
+        Args:
+            prompts: A sequence of prompts. Each prompt can be either a string
+                (used as both title and message) or a dict with 'title' and 'message' keys.
+            title: Optional title for the suggested prompts section.
+            channel: Channel ID. Defaults to the channel from the event context.
+            thread_ts: Thread timestamp. Defaults to the thread_ts from the event context.
+            **kwargs: Additional arguments passed to ``WebClient.assistant_threads_setSuggestedPrompts()``.
+
+        Returns:
+            ``SlackResponse`` from the API call.
+        """
+        prompts_arg: List[Dict[str, str]] = []
+        for prompt in prompts:
+            if isinstance(prompt, str):
+                prompts_arg.append({"title": prompt, "message": prompt})
+            else:
+                prompts_arg.append(prompt)
+
+        return self._client.assistant_threads_setSuggestedPrompts(
+            channel_id=channel or self._channel_id,  # type: ignore[arg-type]
+            thread_ts=thread_ts or self._thread_ts,  # type: ignore[arg-type]
+            prompts=prompts_arg,
+            title=title,
+            **kwargs,
+        )

slack_bolt/agent/async_agent.py

@@ -1,6 +1,6 @@
-from typing import Optional
+from typing import Dict, List, Optional, Sequence, Union
 
-from slack_sdk.web.async_client import AsyncWebClient
+from slack_sdk.web.async_client import AsyncSlackResponse, AsyncWebClient
 from slack_sdk.web.async_chat_stream import AsyncChatStream
 
 
@@ -68,3 +68,69 @@ async def chat_stream(
             recipient_user_id=recipient_user_id or self._user_id,
             **kwargs,
         )
+
+    async def set_status(
+        self,
+        *,
+        status: str,
+        loading_messages: Optional[List[str]] = None,
+        channel: Optional[str] = None,
+        thread_ts: Optional[str] = None,
+        **kwargs,
+    ) -> AsyncSlackResponse:
+        """Sets the status of an assistant thread.
+
+        Args:
+            status: The status text to display.
+            loading_messages: Optional list of loading messages to cycle through.
+            channel: Channel ID. Defaults to the channel from the event context.
+            thread_ts: Thread timestamp. Defaults to the thread_ts from the event context.
+            **kwargs: Additional arguments passed to ``AsyncWebClient.assistant_threads_setStatus()``.
+
+        Returns:
+            ``AsyncSlackResponse`` from the API call.
+        """
+        return await self._client.assistant_threads_setStatus(
+            channel_id=channel or self._channel_id,  # type: ignore[arg-type]
+            thread_ts=thread_ts or self._thread_ts,  # type: ignore[arg-type]
+            status=status,
+            loading_messages=loading_messages,
+            **kwargs,
+        )
+
+    async def set_suggested_prompts(
+        self,
+        *,
+        prompts: Sequence[Union[str, Dict[str, str]]],
+        title: Optional[str] = None,
+        channel: Optional[str] = None,
+        thread_ts: Optional[str] = None,
+        **kwargs,
+    ) -> AsyncSlackResponse:
+        """Sets suggested prompts for an assistant thread.
+
+        Args:
+            prompts: A sequence of prompts. Each prompt can be either a string
+                (used as both title and message) or a dict with 'title' and 'message' keys.
+            title: Optional title for the suggested prompts section.
+            channel: Channel ID. Defaults to the channel from the event context.
+            thread_ts: Thread timestamp. Defaults to the thread_ts from the event context.
+            **kwargs: Additional arguments passed to ``AsyncWebClient.assistant_threads_setSuggestedPrompts()``.
+
+        Returns:
+            ``AsyncSlackResponse`` from the API call.
+        """
+        prompts_arg: List[Dict[str, str]] = []
+        for prompt in prompts:
+            if isinstance(prompt, str):
+                prompts_arg.append({"title": prompt, "message": prompt})
+            else:
+                prompts_arg.append(prompt)
+
+        return await self._client.assistant_threads_setSuggestedPrompts(
+            channel_id=channel or self._channel_id,  # type: ignore[arg-type]
+            thread_ts=thread_ts or self._thread_ts,  # type: ignore[arg-type]
+            prompts=prompts_arg,
+            title=title,
+            **kwargs,
+        )

slack_bolt/kwargs_injection/async_utils.py

@@ -89,10 +89,14 @@ def build_async_required_kwargs(
     if "agent" in required_arg_names:
         from slack_bolt.agent.async_agent import AsyncBoltAgent
 
+        # Resolve thread_ts: assistant events set context.thread_ts, otherwise read from event
+        event = request.body.get("event", {})
+        thread_ts = request.context.thread_ts or event.get("thread_ts") or event.get("ts")
+
         all_available_args["agent"] = AsyncBoltAgent(
             client=request.context.client,
             channel_id=request.context.channel_id,
-            thread_ts=request.context.thread_ts,
+            thread_ts=thread_ts,
             team_id=request.context.team_id,
             user_id=request.context.user_id,
         )

slack_bolt/kwargs_injection/utils.py

@@ -88,10 +88,14 @@ def build_required_kwargs(
     if "agent" in required_arg_names:
         from slack_bolt.agent.agent import BoltAgent
 
+        # Resolve thread_ts: assistant events set context.thread_ts, otherwise read from event
+        event = request.body.get("event", {})
+        thread_ts = request.context.thread_ts or event.get("thread_ts") or event.get("ts")
+
         all_available_args["agent"] = BoltAgent(
             client=request.context.client,
             channel_id=request.context.channel_id,
-            thread_ts=request.context.thread_ts,
+            thread_ts=thread_ts,
             team_id=request.context.team_id,
             user_id=request.context.user_id,
         )

slack_bolt/request/internals.py

@@ -218,6 +218,9 @@ def extract_thread_ts(payload: Dict[str, Any]) -> Optional[str]:
     # This utility initially supports only the use cases for AI assistants, but it may be fine to add more patterns.
     # That said, note that thread_ts is always required for assistant threads, but it's not for channels.
     # Thus, blindly setting this thread_ts to say utility can break existing apps' behaviors.
+    #
+    # The BoltAgent class handles non-assistant thread_ts separately by reading from the event directly,
+    # allowing it to work correctly without affecting say() behavior.
     if is_assistant_event(payload):
         event = payload["event"]
         if (

tests/scenario_tests/test_events_agent.py

@@ -80,6 +80,54 @@ def handle_action(ack, agent: BoltAgent):
         assert response.status == 200
         assert_target_called()
 
+    def test_agent_thread_ts_from_event_in_thread(self):
+        """Agent gets thread_ts from event when in a thread."""
+        app = App(client=self.web_client)
+
+        state = {"thread_ts": None}
+
+        def assert_target_called():
+            count = 0
+            while state["thread_ts"] is None and count < 20:
+                sleep(0.1)
+                count += 1
+            assert state["thread_ts"] is not None
+
+        @app.event("app_mention")
+        def handle_mention(agent: BoltAgent):
+            state["thread_ts"] = agent._thread_ts
+
+        request = BoltRequest(body=app_mention_in_thread_body, mode="socket_mode")
+        response = app.dispatch(request)
+        assert response.status == 200
+        assert_target_called()
+        # Should use event.thread_ts (the thread root), not event.ts
+        assert state["thread_ts"] == "1111111111.111111"
+
+    def test_agent_thread_ts_falls_back_to_ts(self):
+        """Agent falls back to event.ts when not in a thread."""
+        app = App(client=self.web_client)
+
+        state = {"thread_ts": None}
+
+        def assert_target_called():
+            count = 0
+            while state["thread_ts"] is None and count < 20:
+                sleep(0.1)
+                count += 1
+            assert state["thread_ts"] is not None
+
+        @app.event("app_mention")
+        def handle_mention(agent: BoltAgent):
+            state["thread_ts"] = agent._thread_ts
+
+        request = BoltRequest(body=app_mention_event_body, mode="socket_mode")
+        response = app.dispatch(request)
+        assert response.status == 200
+        assert_target_called()
+        # Should fall back to event.ts since no thread_ts
+        assert state["thread_ts"] == "1234567890.123456"
+
     def test_agent_kwarg_emits_experimental_warning(self):
         app = App(client=self.web_client)
 
@@ -140,6 +188,18 @@ def build_payload(event: dict) -> dict:
     }
 )
 
+app_mention_in_thread_body = build_payload(
+    {
+        "type": "app_mention",
+        "user": "W222",
+        "text": "<@W111> hello in thread",
+        "ts": "2222222222.222222",
+        "thread_ts": "1111111111.111111",  # Thread root timestamp
+        "channel": "C111",
+        "event_ts": "2222222222.222222",
+    }
+)
+
 action_event_body = {
     "type": "block_actions",
     "user": {"id": "W222", "username": "test_user", "name": "test_user", "team_id": "T111"},

tests/scenario_tests_async/test_events_agent.py

@@ -86,6 +86,56 @@ async def handle_action(ack, agent: AsyncBoltAgent):
         assert response.status == 200
         await assert_target_called()
 
+    @pytest.mark.asyncio
+    async def test_agent_thread_ts_from_event_in_thread(self):
+        """Agent gets thread_ts from event when in a thread."""
+        app = AsyncApp(client=self.web_client)
+
+        state = {"thread_ts": None}
+
+        async def assert_target_called():
+            count = 0
+            while state["thread_ts"] is None and count < 20:
+                await asyncio.sleep(0.1)
+                count += 1
+            assert state["thread_ts"] is not None
+
+        @app.event("app_mention")
+        async def handle_mention(agent: AsyncBoltAgent):
+            state["thread_ts"] = agent._thread_ts
+
+        request = AsyncBoltRequest(body=app_mention_in_thread_body, mode="socket_mode")
+        response = await app.async_dispatch(request)
+        assert response.status == 200
+        await assert_target_called()
+        # Should use event.thread_ts (the thread root), not event.ts
+        assert state["thread_ts"] == "1111111111.111111"
+
+    @pytest.mark.asyncio
+    async def test_agent_thread_ts_falls_back_to_ts(self):
+        """Agent falls back to event.ts when not in a thread."""
+        app = AsyncApp(client=self.web_client)
+
+        state = {"thread_ts": None}
+
+        async def assert_target_called():
+            count = 0
+            while state["thread_ts"] is None and count < 20:
+                await asyncio.sleep(0.1)
+                count += 1
+            assert state["thread_ts"] is not None
+
+        @app.event("app_mention")
+        async def handle_mention(agent: AsyncBoltAgent):
+            state["thread_ts"] = agent._thread_ts
+
+        request = AsyncBoltRequest(body=app_mention_event_body, mode="socket_mode")
+        response = await app.async_dispatch(request)
+        assert response.status == 200
+        await assert_target_called()
+        # Should fall back to event.ts since no thread_ts
+        assert state["thread_ts"] == "1234567890.123456"
+
     @pytest.mark.asyncio
     async def test_agent_kwarg_emits_experimental_warning(self):
         app = AsyncApp(client=self.web_client)
@@ -147,6 +197,18 @@ def build_payload(event: dict) -> dict:
     }
 )
 
+app_mention_in_thread_body = build_payload(
+    {
+        "type": "app_mention",
+        "user": "W222",
+        "text": "<@W111> hello in thread",
+        "ts": "2222222222.222222",
+        "thread_ts": "1111111111.111111",  # Thread root timestamp
+        "channel": "C111",
+        "event_ts": "2222222222.222222",
+    }
+)
+
 action_event_body = {
     "type": "block_actions",
     "user": {"id": "W222", "username": "test_user", "name": "test_user", "team_id": "T111"},