Merge branch 'strands-agents:main' into feat-agent-interface

Author: awsarron

src/strands/_async.py

@@ -1,6 +1,7 @@
 """Private async execution utilities."""
 
 import asyncio
+import contextvars
 from concurrent.futures import ThreadPoolExecutor
 from typing import Awaitable, Callable, TypeVar
 
@@ -27,5 +28,6 @@ def execute() -> T:
         return asyncio.run(execute_async())
 
     with ThreadPoolExecutor() as executor:
-        future = executor.submit(execute)
+        context = contextvars.copy_context()
+        future = executor.submit(context.run, execute)
         return future.result()

src/strands/agent/agent.py

@@ -227,7 +227,7 @@ async def event_loop_cycle(
             )
         structured_output_context.set_forced_mode()
         logger.debug("Forcing structured output tool")
-        agent._append_message(
+        await agent._append_message(
             {"role": "user", "content": [{"text": "You must format the previous response as structured output."}]}
         )
 
@@ -322,7 +322,7 @@ async def _handle_model_execution(
             model_id=model_id,
         )
         with trace_api.use_span(model_invoke_span):
-            agent.hooks.invoke_callbacks(
+            await agent.hooks.invoke_callbacks_async(
                 BeforeModelCallEvent(
                     agent=agent,
                 )
@@ -335,14 +335,19 @@ async def _handle_model_execution(
                 tool_specs = agent.tool_registry.get_all_tool_specs()
             try:
                 async for event in stream_messages(
-                    agent.model, agent.system_prompt, agent.messages, tool_specs, structured_output_context.tool_choice
+                    agent.model,
+                    agent.system_prompt,
+                    agent.messages,
+                    tool_specs,
+                    system_prompt_content=agent._system_prompt_content,
+                    tool_choice=structured_output_context.tool_choice,
                 ):
                     yield event
 
                 stop_reason, message, usage, metrics = event["stop"]
                 invocation_state.setdefault("request_state", {})
 
-                agent.hooks.invoke_callbacks(
+                await agent.hooks.invoke_callbacks_async(
                     AfterModelCallEvent(
                         agent=agent,
                         stop_response=AfterModelCallEvent.ModelStopResponse(
@@ -363,7 +368,7 @@ async def _handle_model_execution(
                 if model_invoke_span:
                     tracer.end_span_with_error(model_invoke_span, str(e), e)
 
-                agent.hooks.invoke_callbacks(
+                await agent.hooks.invoke_callbacks_async(
                     AfterModelCallEvent(
                         agent=agent,
                         exception=e,
@@ -397,7 +402,7 @@ async def _handle_model_execution(
 
         # Add the response message to the conversation
         agent.messages.append(message)
-        agent.hooks.invoke_callbacks(MessageAddedEvent(agent=agent, message=message))
+        await agent.hooks.invoke_callbacks_async(MessageAddedEvent(agent=agent, message=message))
 
         # Update metrics
         agent.event_loop_metrics.update_usage(usage)
@@ -502,7 +507,7 @@ async def _handle_tool_execution(
     }
 
     agent.messages.append(tool_result_message)
-    agent.hooks.invoke_callbacks(MessageAddedEvent(agent=agent, message=tool_result_message))
+    await agent.hooks.invoke_callbacks_async(MessageAddedEvent(agent=agent, message=tool_result_message))
 
     yield ToolResultMessageEvent(message=tool_result_message)
 

src/strands/agent/interrupt.py

@@ -22,7 +22,7 @@
     TypedEvent,
 )
 from ..types.citations import CitationsContentBlock
-from ..types.content import ContentBlock, Message, Messages
+from ..types.content import ContentBlock, Message, Messages, SystemContentBlock
 from ..types.streaming import (
     ContentBlockDeltaEvent,
     ContentBlockStart,
@@ -418,16 +418,22 @@ async def stream_messages(
     system_prompt: Optional[str],
     messages: Messages,
     tool_specs: list[ToolSpec],
+    *,
     tool_choice: Optional[Any] = None,
+    system_prompt_content: Optional[list[SystemContentBlock]] = None,
+    **kwargs: Any,
 ) -> AsyncGenerator[TypedEvent, None]:
     """Streams messages to the model and processes the response.
 
     Args:
         model: Model provider.
-        system_prompt: The system prompt to send.
+        system_prompt: The system prompt string, used for backwards compatibility with models that expect it.
         messages: List of messages to send.
         tool_specs: The list of tool specs.
         tool_choice: Optional tool choice constraint for forcing specific tool usage.
+        system_prompt_content: The authoritative system prompt content blocks that always contains the
+            system prompt data.
+        **kwargs: Additional keyword arguments for future extensibility.
 
     Yields:
         The reason for stopping, the final message, and the usage metrics
@@ -436,7 +442,14 @@ async def stream_messages(
 
     messages = _normalize_messages(messages)
     start_time = time.time()
-    chunks = model.stream(messages, tool_specs if tool_specs else None, system_prompt, tool_choice=tool_choice)
+
+    chunks = model.stream(
+        messages,
+        tool_specs if tool_specs else None,
+        system_prompt,
+        tool_choice=tool_choice,
+        system_prompt_content=system_prompt_content,
+    )
 
     async for event in process_stream(chunks, start_time):
         yield event

src/strands/event_loop/event_loop.py

@@ -7,9 +7,10 @@
 via hook provider objects.
 """
 
+import inspect
 import logging
 from dataclasses import dataclass
-from typing import TYPE_CHECKING, Any, Generator, Generic, Protocol, Type, TypeVar
+from typing import TYPE_CHECKING, Any, Awaitable, Generator, Generic, Protocol, Type, TypeVar
 
 from ..interrupt import Interrupt, InterruptException
 
@@ -122,10 +123,15 @@ class HookCallback(Protocol, Generic[TEvent]):
         ```python
         def my_callback(event: StartRequestEvent) -> None:
             print(f"Request started for agent: {event.agent.name}")
+
+        # Or
+
+        async def my_callback(event: StartRequestEvent) -> None:
+            # await an async operation
         ```
     """
 
-    def __call__(self, event: TEvent) -> None:
+    def __call__(self, event: TEvent) -> None | Awaitable[None]:
         """Handle a hook event.
 
         Args:
@@ -164,6 +170,10 @@ def my_handler(event: StartRequestEvent):
             registry.add_callback(StartRequestEvent, my_handler)
             ```
         """
+        # Related issue: https://github.com/strands-agents/sdk-python/issues/330
+        if event_type.__name__ == "AgentInitializedEvent" and inspect.iscoroutinefunction(callback):
+            raise ValueError("AgentInitializedEvent can only be registered with a synchronous callback")
+
         callbacks = self._registered_callbacks.setdefault(event_type, [])
         callbacks.append(callback)
 
@@ -189,6 +199,52 @@ def register_hooks(self, registry: HookRegistry):
         """
         hook.register_hooks(self)
 
+    async def invoke_callbacks_async(self, event: TInvokeEvent) -> tuple[TInvokeEvent, list[Interrupt]]:
+        """Invoke all registered callbacks for the given event.
+
+        This method finds all callbacks registered for the event's type and
+        invokes them in the appropriate order. For events with should_reverse_callbacks=True,
+        callbacks are invoked in reverse registration order. Any exceptions raised by callback
+        functions will propagate to the caller.
+
+        Additionally, this method aggregates interrupts raised by the user to instantiate human-in-the-loop workflows.
+
+        Args:
+            event: The event to dispatch to registered callbacks.
+
+        Returns:
+            The event dispatched to registered callbacks and any interrupts raised by the user.
+
+        Raises:
+            ValueError: If interrupt name is used more than once.
+
+        Example:
+            ```python
+            event = StartRequestEvent(agent=my_agent)
+            await registry.invoke_callbacks_async(event)
+            ```
+        """
+        interrupts: dict[str, Interrupt] = {}
+
+        for callback in self.get_callbacks_for(event):
+            try:
+                if inspect.iscoroutinefunction(callback):
+                    await callback(event)
+                else:
+                    callback(event)
+
+            except InterruptException as exception:
+                interrupt = exception.interrupt
+                if interrupt.name in interrupts:
+                    message = f"interrupt_name=<{interrupt.name}> | interrupt name used more than once"
+                    logger.error(message)
+                    raise ValueError(message) from exception
+
+                # Each callback is allowed to raise their own interrupt.
+                interrupts[interrupt.name] = interrupt
+
+        return event, list(interrupts.values())
+
     def invoke_callbacks(self, event: TInvokeEvent) -> tuple[TInvokeEvent, list[Interrupt]]:
         """Invoke all registered callbacks for the given event.
 
@@ -206,6 +262,7 @@ def invoke_callbacks(self, event: TInvokeEvent) -> tuple[TInvokeEvent, list[Inte
             The event dispatched to registered callbacks and any interrupts raised by the user.
 
         Raises:
+            RuntimeError: If at least one callback is async.
             ValueError: If interrupt name is used more than once.
 
         Example:
@@ -214,9 +271,13 @@ def invoke_callbacks(self, event: TInvokeEvent) -> tuple[TInvokeEvent, list[Inte
             registry.invoke_callbacks(event)
             ```
         """
+        callbacks = list(self.get_callbacks_for(event))
         interrupts: dict[str, Interrupt] = {}
 
-        for callback in self.get_callbacks_for(event):
+        if any(inspect.iscoroutinefunction(callback) for callback in callbacks):
+            raise RuntimeError(f"event=<{event}> | use invoke_callbacks_async to invoke async callback")
+
+        for callback in callbacks:
             try:
                 callback(event)
             except InterruptException as exception: