refactor: scope set_title to assistant threads via AssistantUtilities

Move set_title construction into AssistantUtilities as a property and attach it from the assistant-event branch of AttachingConversationKwargs (sync and async). Previously set_title was attached to any IM message event that had a thread_ts; it is now assistant-thread-only, matching say/get_thread_context/save_thread_context. set_suggested_prompts remains available for any DM to the app. AssistantUtilities now derives channel_id/thread_ts directly from the payload: from the assistant_thread property for assistant_* events (via the new has_channel_id_and_thread_ts helper in internals.py) and from channel/thread_ts for message events, instead of reading them off the BoltContext. Also simplify the assistant-thread message predicates in payload_utils, since is_im_message_event / is_any_im_message_event already imply is_event. Co-Authored-By: Claude <svc-devxp-claude@slack-corp.com>
WilliamBergamin · Jul 14, 2026 · Jul 2, 2026 · Jul 3, 2026 · Jul 6, 2026 · Jul 7, 2026
commit 19fced4144f6f07a440abe26542980b2dfe4240b
@@ -7,8 +7,10 @@
 
 from slack_bolt.context.context import BoltContext
 from slack_bolt.context.say import Say
+from .internals import has_channel_id_and_thread_ts
 from ..get_thread_context.get_thread_context import GetThreadContext
 from ..save_thread_context import SaveThreadContext
+from ..set_title import SetTitle
 
 
 class AssistantUtilities:
@@ -29,9 +31,15 @@ def __init__(
         self.client = context.client
         self.thread_context_store = thread_context_store or DefaultAssistantThreadContextStore(context)
 
-        if context.channel_id is not None and context.thread_ts is not None:
-            self.channel_id = context.channel_id
-            self.thread_ts = context.thread_ts
+        if has_channel_id_and_thread_ts(self.payload):
+            # assistant_thread_started
+            thread = self.payload["assistant_thread"]
+            self.channel_id = thread["channel_id"]
+            self.thread_ts = thread["thread_ts"]
+        elif self.payload.get("channel") is not None and self.payload.get("thread_ts") is not None:
+            # message event
+            self.channel_id = self.payload["channel"]
+            self.thread_ts = self.payload["thread_ts"]
         else:
             # When moving this code to Bolt internals, no need to raise an exception for this pattern
             raise ValueError(f"Cannot instantiate Assistant for this event pattern ({self.payload})")
@@ -51,6 +59,10 @@ def build_metadata() -> Optional[dict]:
             build_metadata=build_metadata,
         )
 
+    @property
+    def set_title(self) -> SetTitle:
+        return SetTitle(self.client, self.channel_id, self.thread_ts)
+
     @property
     def get_thread_context(self) -> GetThreadContext:
         return GetThreadContext(self.thread_context_store, self.channel_id, self.thread_ts, self.payload)

@@ -10,8 +10,10 @@
 
 from slack_bolt.context.async_context import AsyncBoltContext
 from slack_bolt.context.say.async_say import AsyncSay
+from .internals import has_channel_id_and_thread_ts
 from ..get_thread_context.async_get_thread_context import AsyncGetThreadContext
 from ..save_thread_context.async_save_thread_context import AsyncSaveThreadContext
+from ..set_title.async_set_title import AsyncSetTitle
 
 
 class AsyncAssistantUtilities:
@@ -32,9 +34,15 @@ def __init__(
         self.client = context.client
         self.thread_context_store = thread_context_store or DefaultAsyncAssistantThreadContextStore(context)
 
-        if context.channel_id is not None and context.thread_ts is not None:
-            self.channel_id = context.channel_id
-            self.thread_ts = context.thread_ts
+        if has_channel_id_and_thread_ts(self.payload):
+            # assistant_thread_started
+            thread = self.payload["assistant_thread"]
+            self.channel_id = thread["channel_id"]
+            self.thread_ts = thread["thread_ts"]
+        elif self.payload.get("channel") is not None and self.payload.get("thread_ts") is not None:
+            # message event
+            self.channel_id = self.payload["channel"]
+            self.thread_ts = self.payload["thread_ts"]
         else:
             # When moving this code to Bolt internals, no need to raise an exception for this pattern
             raise ValueError(f"Cannot instantiate Assistant for this event pattern ({self.payload})")
@@ -48,6 +56,10 @@ def say(self) -> AsyncSay:
             build_metadata=self._build_message_metadata,
         )
 
+    @property
+    def set_title(self) -> AsyncSetTitle:
+        return AsyncSetTitle(self.client, self.channel_id, self.thread_ts)
+
     async def _build_message_metadata(self) -> dict:
         return {
             "event_type": "assistant_thread_context",

diff --git a/slack_bolt/context/assistant/internals.py b/slack_bolt/context/assistant/internals.py
@@ -0,0 +1,9 @@
+def has_channel_id_and_thread_ts(payload: dict) -> bool:
+    """Verifies if the given payload has both channel_id and thread_ts under assistant_thread property.
+    This data pattern is available for assistant_* events.
+    """
+    return (
+        payload.get("assistant_thread") is not None
+        and payload["assistant_thread"].get("channel_id") is not None
+        and payload["assistant_thread"].get("thread_ts") is not None
+    )
@@ -5,15 +5,14 @@
 from slack_bolt.context.say_stream.async_say_stream import AsyncSayStream
 from slack_bolt.context.set_status.async_set_status import AsyncSetStatus
 from slack_bolt.context.set_suggested_prompts.async_set_suggested_prompts import AsyncSetSuggestedPrompts
-from slack_bolt.context.set_title.async_set_title import AsyncSetTitle
 from slack_bolt.middleware.async_middleware import AsyncMiddleware
 from slack_bolt.request.async_request import AsyncBoltRequest
 from slack_bolt.request.payload_utils import (
     is_assistant_event,
     is_assistant_thread_context_changed_event,
     is_assistant_thread_started_event,
-    to_event,
     is_im_message_event,
+    to_event,
 )
 from slack_bolt.response import BoltResponse
 
@@ -46,6 +45,7 @@ async def async_process(
                 thread_context_store=self.thread_context_store,
             )
             req.context["say"] = assistant.say
+            req.context["set_title"] = assistant.set_title
             req.context["get_thread_context"] = assistant.get_thread_context
             req.context["save_thread_context"] = assistant.save_thread_context
 
@@ -59,8 +59,6 @@ async def async_process(
                 channel_id=req.context.channel_id,
                 thread_ts=req.context.thread_ts,
             )
-            if req.context.thread_ts:
-                req.context["set_title"] = AsyncSetTitle(req.context.client, req.context.channel_id, req.context.thread_ts)
 
         # TODO: in the future we might want to introduce a "proper" extract_ts utility
         thread_ts_or_ts = req.context.thread_ts or event.get("ts")

@@ -4,7 +4,6 @@
 from slack_bolt.context.say_stream.say_stream import SayStream
 from slack_bolt.context.set_status.set_status import SetStatus
 from slack_bolt.context.set_suggested_prompts.set_suggested_prompts import SetSuggestedPrompts
-from slack_bolt.context.set_title import SetTitle
 from slack_bolt.middleware import Middleware
 from slack_bolt.context.assistant.assistant_utilities import AssistantUtilities
 from slack_bolt.request.payload_utils import (
@@ -40,6 +39,7 @@ def process(self, *, req: BoltRequest, resp: BoltResponse, next: Callable[[], Bo
                 thread_context_store=self.thread_context_store,
             )
             req.context["say"] = assistant.say
+            req.context["set_title"] = assistant.set_title
             req.context["get_thread_context"] = assistant.get_thread_context
             req.context["save_thread_context"] = assistant.save_thread_context
 
@@ -53,8 +53,6 @@ def process(self, *, req: BoltRequest, resp: BoltResponse, next: Callable[[], Bo
                 channel_id=req.context.channel_id,
                 thread_ts=req.context.thread_ts,
             )
-            if req.context.thread_ts:
-                req.context["set_title"] = SetTitle(req.context.client, req.context.channel_id, req.context.thread_ts)
 
         # TODO: in the future we might want to introduce a "proper" extract_ts utility
         thread_ts_or_ts = req.context.thread_ts or event.get("ts")

@@ -72,18 +72,15 @@ def is_assistant_thread_context_changed_event(body: Dict[str, Any]) -> bool:
 
 
 def is_user_message_event_in_assistant_thread(body: Dict[str, Any]) -> bool:
-    if is_event(body):
-        return (
-            is_im_message_event(body) and body["event"].get("thread_ts") is not None and body["event"].get("bot_id") is None
-        )
+    if is_im_message_event(body):
+        return body["event"].get("thread_ts") is not None and body["event"].get("bot_id") is None
     return False
 
 
 def is_bot_message_event_in_assistant_thread(body: Dict[str, Any]) -> bool:
-    if is_event(body):
+    if is_any_im_message_event(body):
         return (
-            is_any_im_message_event(body)
-            and body["event"].get("subtype") is None
+            body["event"].get("subtype") is None
             and body["event"].get("thread_ts") is not None
             and body["event"].get("bot_id") is not None
         )
@@ -92,14 +89,10 @@ def is_bot_message_event_in_assistant_thread(body: Dict[str, Any]) -> bool:
 
 def is_other_message_sub_event_in_assistant_thread(body: Dict[str, Any]) -> bool:
     # message_changed, message_deleted etc.
-    if is_event(body):
-        return (
-            is_any_im_message_event(body)
-            and not is_user_message_event_in_assistant_thread(body)
-            and (
-                _is_other_message_sub_event(body["event"].get("message"))
-                or _is_other_message_sub_event(body["event"].get("previous_message"))
-            )
+    if is_any_im_message_event(body):
+        return not is_user_message_event_in_assistant_thread(body) and (
+            _is_other_message_sub_event(body["event"].get("message"))
+            or _is_other_message_sub_event(body["event"].get("previous_message"))
         )
     return False
 

@@ -4,6 +4,7 @@
 from slack_bolt.request import BoltRequest
 from slack_bolt.response import BoltResponse
 from tests.scenario_tests.test_events_assistant import (
+    build_payload,
     thread_started_event_body,
     user_message_event_body,
     channel_user_message_event_body,
@@ -16,6 +17,19 @@ def next():
 
 ASSISTANT_KWARGS = ("say", "set_title", "set_suggested_prompts", "get_thread_context", "save_thread_context")
 
+# A top-level DM (not in a thread) is not an assistant thread, but set_suggested_prompts is still attached.
+top_level_im_message_event_body = build_payload(
+    {
+        "user": "W222",
+        "type": "message",
+        "ts": "1726133700.887259",
+        "text": "A top-level DM, not in a thread",
+        "channel": "D111",
+        "event_ts": "1726133700.887259",
+        "channel_type": "im",
+    }
+)
+
 
 class TestAttachingConversationKwargs:
     def test_assistant_event_attaches_kwargs(self):
@@ -46,6 +60,26 @@ def test_user_message_event_attaches_kwargs(self):
         assert "say_stream" in req.context
         assert "set_status" in req.context
 
+    def test_top_level_dm_attaches_suggested_prompts_but_not_set_title(self):
+        middleware = AttachingConversationKwargs()
+        req = BoltRequest(body=top_level_im_message_event_body, mode="socket_mode")
+        req.context["client"] = WebClient(token="xoxb-test")
+
+        resp = middleware.process(req=req, resp=BoltResponse(status=404), next=next)
+
+        assert resp.status == 200
+        # set_suggested_prompts is available for any DM to the app
+        assert "set_suggested_prompts" in req.context
+        # set_title is assistant-thread-only; a top-level DM is not an assistant thread
+        assert "set_title" not in req.context
+        # say/get_thread_context/save_thread_context remain assistant-only
+        assert "say" not in req.context
+        assert "get_thread_context" not in req.context
+        assert "save_thread_context" not in req.context
+        # set_status / say_stream are attached whenever a ts is resolvable
+        assert "say_stream" in req.context
+        assert "set_status" in req.context
+
     def test_non_assistant_event_does_not_attach_kwargs(self):
         middleware = AttachingConversationKwargs()
         req = BoltRequest(body=channel_user_message_event_body, mode="socket_mode")

@@ -7,6 +7,7 @@
 from slack_bolt.request.async_request import AsyncBoltRequest
 from slack_bolt.response import BoltResponse
 from tests.scenario_tests_async.test_events_assistant import (
+    build_payload,
     thread_started_event_body,
     user_message_event_body,
     channel_user_message_event_body,
@@ -19,6 +20,19 @@ async def next():
 
 ASSISTANT_KWARGS = ("say", "set_title", "set_suggested_prompts", "get_thread_context", "save_thread_context")
 
+# A top-level DM (not in a thread) is not an assistant thread, but set_suggested_prompts is still attached.
+top_level_im_message_event_body = build_payload(
+    {
+        "user": "W222",
+        "type": "message",
+        "ts": "1726133700.887259",
+        "text": "A top-level DM, not in a thread",
+        "channel": "D111",
+        "event_ts": "1726133700.887259",
+        "channel_type": "im",
+    }
+)
+
 
 class TestAsyncAttachingConversationKwargs:
     @pytest.mark.asyncio
@@ -51,6 +65,27 @@ async def test_user_message_event_attaches_kwargs(self):
         assert "say_stream" in req.context
         assert "set_status" in req.context
 
+    @pytest.mark.asyncio
+    async def test_top_level_dm_attaches_suggested_prompts_but_not_set_title(self):
+        middleware = AsyncAttachingConversationKwargs()
+        req = AsyncBoltRequest(body=top_level_im_message_event_body, mode="socket_mode")
+        req.context["client"] = AsyncWebClient(token="xoxb-test")
+
+        resp = await middleware.async_process(req=req, resp=BoltResponse(status=404), next=next)
+
+        assert resp.status == 200
+        # set_suggested_prompts is available for any DM to the app
+        assert "set_suggested_prompts" in req.context
+        # set_title is assistant-thread-only; a top-level DM is not an assistant thread
+        assert "set_title" not in req.context
+        # say/get_thread_context/save_thread_context remain assistant-only
+        assert "say" not in req.context
+        assert "get_thread_context" not in req.context
+        assert "save_thread_context" not in req.context
+        # set_status / say_stream are attached whenever a ts is resolvable
+        assert "say_stream" in req.context
+        assert "set_status" in req.context
+
     @pytest.mark.asyncio
     async def test_non_assistant_event_does_not_attach_kwargs(self):
         middleware = AsyncAttachingConversationKwargs()