`strands.multiagent.swarm` ¶

Swarm Multi-Agent Pattern Implementation.

This module provides a collaborative agent orchestration system where agents work together as a team to solve complex tasks, with shared context and autonomous coordination.

Key Features: - Self-organizing agent teams with shared working memory - Tool-based coordination - Autonomous agent collaboration without central control - Dynamic task distribution based on agent capabilities - Collective intelligence through shared context - Human input via user interrupts raised in BeforeNodeCallEvent hooks and agent nodes

`AgentState = JSONSerializableDict` `module-attribute` ¶

`AttributeValue = Union[str, bool, float, int, List[str], List[bool], List[float], List[int], Sequence[str], Sequence[bool], Sequence[int], Sequence[float]]` `module-attribute` ¶

`Messages = List[Message]` `module-attribute` ¶

A list of messages representing a conversation.

`MultiAgentInput = str | list[ContentBlock] | list[InterruptResponseContent]` `module-attribute` ¶

`_DEFAULT_SWARM_ID = 'default_swarm'` `module-attribute` ¶

`logger = logging.getLogger(name)` `module-attribute` ¶

`AfterMultiAgentInvocationEvent` `dataclass` ¶

Bases: BaseHookEvent

Event triggered after orchestrator execution completes.

Attributes:

Name	Type	Description
`source`	`MultiAgentBase`	The multi-agent orchestrator instance
`invocation_state`	`dict[str, Any] \| None`	Configuration that user passes in

Source code in strands/experimental/hooks/multiagent/events.py

@dataclass
class AfterMultiAgentInvocationEvent(BaseHookEvent):
    """Event triggered after orchestrator execution completes.

    Attributes:
        source: The multi-agent orchestrator instance
        invocation_state: Configuration that user passes in
    """

    source: "MultiAgentBase"
    invocation_state: dict[str, Any] | None = None

    @property
    def should_reverse_callbacks(self) -> bool:
        """True to invoke callbacks in reverse order."""
        return True

`should_reverse_callbacks` `property` ¶

True to invoke callbacks in reverse order.

`AfterNodeCallEvent` `dataclass` ¶

Bases: BaseHookEvent

Event triggered after individual node execution completes.

Attributes:

Name	Type	Description
`source`	`MultiAgentBase`	The multi-agent orchestrator instance
`node_id`	`str`	ID of the node that just completed execution
`invocation_state`	`dict[str, Any] \| None`	Configuration that user passes in

Source code in strands/experimental/hooks/multiagent/events.py

@dataclass
class AfterNodeCallEvent(BaseHookEvent):
    """Event triggered after individual node execution completes.

    Attributes:
        source: The multi-agent orchestrator instance
        node_id: ID of the node that just completed execution
        invocation_state: Configuration that user passes in
    """

    source: "MultiAgentBase"
    node_id: str
    invocation_state: dict[str, Any] | None = None

    @property
    def should_reverse_callbacks(self) -> bool:
        """True to invoke callbacks in reverse order."""
        return True

`should_reverse_callbacks` `property` ¶

True to invoke callbacks in reverse order.

`Agent` ¶

Core Agent interface.

An agent orchestrates the following workflow:

Receives user input
Processes the input using a language model
Decides whether to use tools to gather information or perform actions
Executes those tools and receives results
Continues reasoning with the new information
Produces a final response

Source code in strands/agent/agent.py

class Agent:
    """Core Agent interface.

    An agent orchestrates the following workflow:

    1. Receives user input
    2. Processes the input using a language model
    3. Decides whether to use tools to gather information or perform actions
    4. Executes those tools and receives results
    5. Continues reasoning with the new information
    6. Produces a final response
    """

    # For backwards compatibility
    ToolCaller = _ToolCaller

    def __init__(
        self,
        model: Union[Model, str, None] = None,
        messages: Optional[Messages] = None,
        tools: Optional[list[Union[str, dict[str, str], "ToolProvider", Any]]] = None,
        system_prompt: Optional[str | list[SystemContentBlock]] = None,
        structured_output_model: Optional[Type[BaseModel]] = None,
        callback_handler: Optional[
            Union[Callable[..., Any], _DefaultCallbackHandlerSentinel]
        ] = _DEFAULT_CALLBACK_HANDLER,
        conversation_manager: Optional[ConversationManager] = None,
        record_direct_tool_call: bool = True,
        load_tools_from_directory: bool = False,
        trace_attributes: Optional[Mapping[str, AttributeValue]] = None,
        *,
        agent_id: Optional[str] = None,
        name: Optional[str] = None,
        description: Optional[str] = None,
        state: Optional[Union[AgentState, dict]] = None,
        hooks: Optional[list[HookProvider]] = None,
        session_manager: Optional[SessionManager] = None,
        tool_executor: Optional[ToolExecutor] = None,
    ):
        """Initialize the Agent with the specified configuration.

        Args:
            model: Provider for running inference or a string representing the model-id for Bedrock to use.
                Defaults to strands.models.BedrockModel if None.
            messages: List of initial messages to pre-load into the conversation.
                Defaults to an empty list if None.
            tools: List of tools to make available to the agent.
                Can be specified as:

                - String tool names (e.g., "retrieve")
                - File paths (e.g., "/path/to/tool.py")
                - Imported Python modules (e.g., from strands_tools import current_time)
                - Dictionaries with name/path keys (e.g., {"name": "tool_name", "path": "/path/to/tool.py"})
                - ToolProvider instances for managed tool collections
                - Functions decorated with `@strands.tool` decorator.

                If provided, only these tools will be available. If None, all tools will be available.
            system_prompt: System prompt to guide model behavior.
                Can be a string or a list of SystemContentBlock objects for advanced features like caching.
                If None, the model will behave according to its default settings.
            structured_output_model: Pydantic model type(s) for structured output.
                When specified, all agent calls will attempt to return structured output of this type.
                This can be overridden on the agent invocation.
                Defaults to None (no structured output).
            callback_handler: Callback for processing events as they happen during agent execution.
                If not provided (using the default), a new PrintingCallbackHandler instance is created.
                If explicitly set to None, null_callback_handler is used.
            conversation_manager: Manager for conversation history and context window.
                Defaults to strands.agent.conversation_manager.SlidingWindowConversationManager if None.
            record_direct_tool_call: Whether to record direct tool calls in message history.
                Defaults to True.
            load_tools_from_directory: Whether to load and automatically reload tools in the `./tools/` directory.
                Defaults to False.
            trace_attributes: Custom trace attributes to apply to the agent's trace span.
            agent_id: Optional ID for the agent, useful for session management and multi-agent scenarios.
                Defaults to "default".
            name: name of the Agent
                Defaults to "Strands Agents".
            description: description of what the Agent does
                Defaults to None.
            state: stateful information for the agent. Can be either an AgentState object, or a json serializable dict.
                Defaults to an empty AgentState object.
            hooks: hooks to be added to the agent hook registry
                Defaults to None.
            session_manager: Manager for handling agent sessions including conversation history and state.
                If provided, enables session-based persistence and state management.
            tool_executor: Definition of tool execution strategy (e.g., sequential, concurrent, etc.).

        Raises:
            ValueError: If agent id contains path separators.
        """
        self.model = BedrockModel() if not model else BedrockModel(model_id=model) if isinstance(model, str) else model
        self.messages = messages if messages is not None else []
        # initializing self._system_prompt for backwards compatibility
        self._system_prompt, self._system_prompt_content = self._initialize_system_prompt(system_prompt)
        self._default_structured_output_model = structured_output_model
        self.agent_id = _identifier.validate(agent_id or _DEFAULT_AGENT_ID, _identifier.Identifier.AGENT)
        self.name = name or _DEFAULT_AGENT_NAME
        self.description = description

        # If not provided, create a new PrintingCallbackHandler instance
        # If explicitly set to None, use null_callback_handler
        # Otherwise use the passed callback_handler
        self.callback_handler: Union[Callable[..., Any], PrintingCallbackHandler]
        if isinstance(callback_handler, _DefaultCallbackHandlerSentinel):
            self.callback_handler = PrintingCallbackHandler()
        elif callback_handler is None:
            self.callback_handler = null_callback_handler
        else:
            self.callback_handler = callback_handler

        self.conversation_manager = conversation_manager if conversation_manager else SlidingWindowConversationManager()

        # Process trace attributes to ensure they're of compatible types
        self.trace_attributes: dict[str, AttributeValue] = {}
        if trace_attributes:
            for k, v in trace_attributes.items():
                if isinstance(v, (str, int, float, bool)) or (
                    isinstance(v, list) and all(isinstance(x, (str, int, float, bool)) for x in v)
                ):
                    self.trace_attributes[k] = v

        self.record_direct_tool_call = record_direct_tool_call
        self.load_tools_from_directory = load_tools_from_directory

        self.tool_registry = ToolRegistry()

        # Process tool list if provided
        if tools is not None:
            self.tool_registry.process_tools(tools)

        # Initialize tools and configuration
        self.tool_registry.initialize_tools(self.load_tools_from_directory)
        if load_tools_from_directory:
            self.tool_watcher = ToolWatcher(tool_registry=self.tool_registry)

        self.event_loop_metrics = EventLoopMetrics()

        # Initialize tracer instance (no-op if not configured)
        self.tracer = get_tracer()
        self.trace_span: Optional[trace_api.Span] = None

        # Initialize agent state management
        if state is not None:
            if isinstance(state, dict):
                self.state = AgentState(state)
            elif isinstance(state, AgentState):
                self.state = state
            else:
                raise ValueError("state must be an AgentState object or a dict")
        else:
            self.state = AgentState()

        self.tool_caller = _ToolCaller(self)

        self.hooks = HookRegistry()

        self._interrupt_state = _InterruptState()

        # Initialize session management functionality
        self._session_manager = session_manager
        if self._session_manager:
            self.hooks.add_hook(self._session_manager)

        # Allow conversation_managers to subscribe to hooks
        self.hooks.add_hook(self.conversation_manager)

        self.tool_executor = tool_executor or ConcurrentToolExecutor()

        if hooks:
            for hook in hooks:
                self.hooks.add_hook(hook)
        self.hooks.invoke_callbacks(AgentInitializedEvent(agent=self))

    @property
    def system_prompt(self) -> str | None:
        """Get the system prompt as a string for backwards compatibility.

        Returns the system prompt as a concatenated string when it contains text content,
        or None if no text content is present. This maintains backwards compatibility
        with existing code that expects system_prompt to be a string.

        Returns:
            The system prompt as a string, or None if no text content exists.
        """
        return self._system_prompt

    @system_prompt.setter
    def system_prompt(self, value: str | list[SystemContentBlock] | None) -> None:
        """Set the system prompt and update internal content representation.

        Accepts either a string or list of SystemContentBlock objects.
        When set, both the backwards-compatible string representation and the internal
        content block representation are updated to maintain consistency.

        Args:
            value: System prompt as string, list of SystemContentBlock objects, or None.
                  - str: Simple text prompt (most common use case)
                  - list[SystemContentBlock]: Content blocks with features like caching
                  - None: Clear the system prompt
        """
        self._system_prompt, self._system_prompt_content = self._initialize_system_prompt(value)

    @property
    def tool(self) -> _ToolCaller:
        """Call tool as a function.

        Returns:
            Tool caller through which user can invoke tool as a function.

        Example:
            ```
            agent = Agent(tools=[calculator])
            agent.tool.calculator(...)
            ```
        """
        return self.tool_caller

    @property
    def tool_names(self) -> list[str]:
        """Get a list of all registered tool names.

        Returns:
            Names of all tools available to this agent.
        """
        all_tools = self.tool_registry.get_all_tools_config()
        return list(all_tools.keys())

    def __call__(
        self,
        prompt: AgentInput = None,
        *,
        invocation_state: dict[str, Any] | None = None,
        structured_output_model: Type[BaseModel] | None = None,
        **kwargs: Any,
    ) -> AgentResult:
        """Process a natural language prompt through the agent's event loop.

        This method implements the conversational interface with multiple input patterns:
        - String input: `agent("hello!")`
        - ContentBlock list: `agent([{"text": "hello"}, {"image": {...}}])`
        - Message list: `agent([{"role": "user", "content": [{"text": "hello"}]}])`
        - No input: `agent()` - uses existing conversation history

        Args:
            prompt: User input in various formats:
                - str: Simple text input
                - list[ContentBlock]: Multi-modal content blocks
                - list[Message]: Complete messages with roles
                - None: Use existing conversation history
            invocation_state: Additional parameters to pass through the event loop.
            structured_output_model: Pydantic model type(s) for structured output (overrides agent default).
            **kwargs: Additional parameters to pass through the event loop.[Deprecating]

        Returns:
            Result object containing:

                - stop_reason: Why the event loop stopped (e.g., "end_turn", "max_tokens")
                - message: The final message from the model
                - metrics: Performance metrics from the event loop
                - state: The final state of the event loop
                - structured_output: Parsed structured output when structured_output_model was specified
        """
        return run_async(
            lambda: self.invoke_async(
                prompt, invocation_state=invocation_state, structured_output_model=structured_output_model, **kwargs
            )
        )

    async def invoke_async(
        self,
        prompt: AgentInput = None,
        *,
        invocation_state: dict[str, Any] | None = None,
        structured_output_model: Type[BaseModel] | None = None,
        **kwargs: Any,
    ) -> AgentResult:
        """Process a natural language prompt through the agent's event loop.

        This method implements the conversational interface with multiple input patterns:
        - String input: Simple text input
        - ContentBlock list: Multi-modal content blocks
        - Message list: Complete messages with roles
        - No input: Use existing conversation history

        Args:
            prompt: User input in various formats:
                - str: Simple text input
                - list[ContentBlock]: Multi-modal content blocks
                - list[Message]: Complete messages with roles
                - None: Use existing conversation history
            invocation_state: Additional parameters to pass through the event loop.
            structured_output_model: Pydantic model type(s) for structured output (overrides agent default).
            **kwargs: Additional parameters to pass through the event loop.[Deprecating]

        Returns:
            Result: object containing:

                - stop_reason: Why the event loop stopped (e.g., "end_turn", "max_tokens")
                - message: The final message from the model
                - metrics: Performance metrics from the event loop
                - state: The final state of the event loop
        """
        events = self.stream_async(
            prompt, invocation_state=invocation_state, structured_output_model=structured_output_model, **kwargs
        )
        async for event in events:
            _ = event

        return cast(AgentResult, event["result"])

    def structured_output(self, output_model: Type[T], prompt: AgentInput = None) -> T:
        """This method allows you to get structured output from the agent.

        If you pass in a prompt, it will be used temporarily without adding it to the conversation history.
        If you don't pass in a prompt, it will use only the existing conversation history to respond.

        For smaller models, you may want to use the optional prompt to add additional instructions to explicitly
        instruct the model to output the structured data.

        Args:
            output_model: The output model (a JSON schema written as a Pydantic BaseModel)
                that the agent will use when responding.
            prompt: The prompt to use for the agent in various formats:
                - str: Simple text input
                - list[ContentBlock]: Multi-modal content blocks
                - list[Message]: Complete messages with roles
                - None: Use existing conversation history

        Raises:
            ValueError: If no conversation history or prompt is provided.
        """
        warnings.warn(
            "Agent.structured_output method is deprecated."
            " You should pass in `structured_output_model` directly into the agent invocation."
            " see: https://strandsagents.com/latest/documentation/docs/user-guide/concepts/agents/structured-output/",
            category=DeprecationWarning,
            stacklevel=2,
        )

        return run_async(lambda: self.structured_output_async(output_model, prompt))

    async def structured_output_async(self, output_model: Type[T], prompt: AgentInput = None) -> T:
        """This method allows you to get structured output from the agent.

        If you pass in a prompt, it will be used temporarily without adding it to the conversation history.
        If you don't pass in a prompt, it will use only the existing conversation history to respond.

        For smaller models, you may want to use the optional prompt to add additional instructions to explicitly
        instruct the model to output the structured data.

        Args:
            output_model: The output model (a JSON schema written as a Pydantic BaseModel)
                that the agent will use when responding.
            prompt: The prompt to use for the agent (will not be added to conversation history).

        Raises:
            ValueError: If no conversation history or prompt is provided.
        -
        """
        if self._interrupt_state.activated:
            raise RuntimeError("cannot call structured output during interrupt")

        warnings.warn(
            "Agent.structured_output_async method is deprecated."
            " You should pass in `structured_output_model` directly into the agent invocation."
            " see: https://strandsagents.com/latest/documentation/docs/user-guide/concepts/agents/structured-output/",
            category=DeprecationWarning,
            stacklevel=2,
        )
        await self.hooks.invoke_callbacks_async(BeforeInvocationEvent(agent=self))
        with self.tracer.tracer.start_as_current_span(
            "execute_structured_output", kind=trace_api.SpanKind.CLIENT
        ) as structured_output_span:
            try:
                if not self.messages and not prompt:
                    raise ValueError("No conversation history or prompt provided")

                temp_messages: Messages = self.messages + await self._convert_prompt_to_messages(prompt)

                structured_output_span.set_attributes(
                    {
                        "gen_ai.system": "strands-agents",
                        "gen_ai.agent.name": self.name,
                        "gen_ai.agent.id": self.agent_id,
                        "gen_ai.operation.name": "execute_structured_output",
                    }
                )
                if self.system_prompt:
                    structured_output_span.add_event(
                        "gen_ai.system.message",
                        attributes={"role": "system", "content": serialize([{"text": self.system_prompt}])},
                    )
                for message in temp_messages:
                    structured_output_span.add_event(
                        f"gen_ai.{message['role']}.message",
                        attributes={"role": message["role"], "content": serialize(message["content"])},
                    )
                events = self.model.structured_output(output_model, temp_messages, system_prompt=self.system_prompt)
                async for event in events:
                    if isinstance(event, TypedEvent):
                        event.prepare(invocation_state={})
                        if event.is_callback_event:
                            self.callback_handler(**event.as_dict())

                structured_output_span.add_event(
                    "gen_ai.choice", attributes={"message": serialize(event["output"].model_dump())}
                )
                return event["output"]

            finally:
                await self.hooks.invoke_callbacks_async(AfterInvocationEvent(agent=self))

    def cleanup(self) -> None:
        """Clean up resources used by the agent.

        This method cleans up all tool providers that require explicit cleanup,
        such as MCP clients. It should be called when the agent is no longer needed
        to ensure proper resource cleanup.

        Note: This method uses a "belt and braces" approach with automatic cleanup
        through finalizers as a fallback, but explicit cleanup is recommended.
        """
        self.tool_registry.cleanup()

    def __del__(self) -> None:
        """Clean up resources when agent is garbage collected."""
        # __del__ is called even when an exception is thrown in the constructor,
        # so there is no guarantee tool_registry was set..
        if hasattr(self, "tool_registry"):
            self.tool_registry.cleanup()

    async def stream_async(
        self,
        prompt: AgentInput = None,
        *,
        invocation_state: dict[str, Any] | None = None,
        structured_output_model: Type[BaseModel] | None = None,
        **kwargs: Any,
    ) -> AsyncIterator[Any]:
        """Process a natural language prompt and yield events as an async iterator.

        This method provides an asynchronous interface for streaming agent events with multiple input patterns:
        - String input: Simple text input
        - ContentBlock list: Multi-modal content blocks
        - Message list: Complete messages with roles
        - No input: Use existing conversation history

        Args:
            prompt: User input in various formats:
                - str: Simple text input
                - list[ContentBlock]: Multi-modal content blocks
                - list[Message]: Complete messages with roles
                - None: Use existing conversation history
            invocation_state: Additional parameters to pass through the event loop.
            structured_output_model: Pydantic model type(s) for structured output (overrides agent default).
            **kwargs: Additional parameters to pass to the event loop.[Deprecating]

        Yields:
            An async iterator that yields events. Each event is a dictionary containing
                information about the current state of processing, such as:

                - data: Text content being generated
                - complete: Whether this is the final chunk
                - current_tool_use: Information about tools being executed
                - And other event data provided by the callback handler

        Raises:
            Exception: Any exceptions from the agent invocation will be propagated to the caller.

        Example:
            ```python
            async for event in agent.stream_async("Analyze this data"):
                if "data" in event:
                    yield event["data"]
            ```
        """
        self._interrupt_state.resume(prompt)

        self.event_loop_metrics.reset_usage_metrics()

        merged_state = {}
        if kwargs:
            warnings.warn("`**kwargs` parameter is deprecating, use `invocation_state` instead.", stacklevel=2)
            merged_state.update(kwargs)
            if invocation_state is not None:
                merged_state["invocation_state"] = invocation_state
        else:
            if invocation_state is not None:
                merged_state = invocation_state

        callback_handler = self.callback_handler
        if kwargs:
            callback_handler = kwargs.get("callback_handler", self.callback_handler)

        # Process input and get message to add (if any)
        messages = await self._convert_prompt_to_messages(prompt)

        self.trace_span = self._start_agent_trace_span(messages)

        with trace_api.use_span(self.trace_span):
            try:
                events = self._run_loop(messages, merged_state, structured_output_model)

                async for event in events:
                    event.prepare(invocation_state=merged_state)

                    if event.is_callback_event:
                        as_dict = event.as_dict()
                        callback_handler(**as_dict)
                        yield as_dict

                result = AgentResult(*event["stop"])
                callback_handler(result=result)
                yield AgentResultEvent(result=result).as_dict()

                self._end_agent_trace_span(response=result)

            except Exception as e:
                self._end_agent_trace_span(error=e)
                raise

    async def _run_loop(
        self,
        messages: Messages,
        invocation_state: dict[str, Any],
        structured_output_model: Type[BaseModel] | None = None,
    ) -> AsyncGenerator[TypedEvent, None]:
        """Execute the agent's event loop with the given message and parameters.

        Args:
            messages: The input messages to add to the conversation.
            invocation_state: Additional parameters to pass to the event loop.
            structured_output_model: Optional Pydantic model type for structured output.

        Yields:
            Events from the event loop cycle.
        """
        await self.hooks.invoke_callbacks_async(BeforeInvocationEvent(agent=self))

        agent_result: AgentResult | None = None
        try:
            yield InitEventLoopEvent()

            await self._append_messages(*messages)

            structured_output_context = StructuredOutputContext(
                structured_output_model or self._default_structured_output_model
            )

            # Execute the event loop cycle with retry logic for context limits
            events = self._execute_event_loop_cycle(invocation_state, structured_output_context)
            async for event in events:
                # Signal from the model provider that the message sent by the user should be redacted,
                # likely due to a guardrail.
                if (
                    isinstance(event, ModelStreamChunkEvent)
                    and event.chunk
                    and event.chunk.get("redactContent")
                    and event.chunk["redactContent"].get("redactUserContentMessage")
                ):
                    self.messages[-1]["content"] = self._redact_user_content(
                        self.messages[-1]["content"], str(event.chunk["redactContent"]["redactUserContentMessage"])
                    )
                    if self._session_manager:
                        self._session_manager.redact_latest_message(self.messages[-1], self)
                yield event

            # Capture the result from the final event if available
            if isinstance(event, EventLoopStopEvent):
                agent_result = AgentResult(*event["stop"])

        finally:
            self.conversation_manager.apply_management(self)
            await self.hooks.invoke_callbacks_async(AfterInvocationEvent(agent=self, result=agent_result))

    async def _execute_event_loop_cycle(
        self, invocation_state: dict[str, Any], structured_output_context: StructuredOutputContext | None = None
    ) -> AsyncGenerator[TypedEvent, None]:
        """Execute the event loop cycle with retry logic for context window limits.

        This internal method handles the execution of the event loop cycle and implements
        retry logic for handling context window overflow exceptions by reducing the
        conversation context and retrying.

        Args:
            invocation_state: Additional parameters to pass to the event loop.
            structured_output_context: Optional structured output context for this invocation.

        Yields:
            Events of the loop cycle.
        """
        # Add `Agent` to invocation_state to keep backwards-compatibility
        invocation_state["agent"] = self

        if structured_output_context:
            structured_output_context.register_tool(self.tool_registry)

        try:
            events = event_loop_cycle(
                agent=self,
                invocation_state=invocation_state,
                structured_output_context=structured_output_context,
            )
            async for event in events:
                yield event

        except ContextWindowOverflowException as e:
            # Try reducing the context size and retrying
            self.conversation_manager.reduce_context(self, e=e)

            # Sync agent after reduce_context to keep conversation_manager_state up to date in the session
            if self._session_manager:
                self._session_manager.sync_agent(self)

            events = self._execute_event_loop_cycle(invocation_state, structured_output_context)
            async for event in events:
                yield event

        finally:
            if structured_output_context:
                structured_output_context.cleanup(self.tool_registry)

    async def _convert_prompt_to_messages(self, prompt: AgentInput) -> Messages:
        if self._interrupt_state.activated:
            return []

        messages: Messages | None = None
        if prompt is not None:
            # Check if the latest message is toolUse
            if len(self.messages) > 0 and any("toolUse" in content for content in self.messages[-1]["content"]):
                # Add toolResult message after to have a valid conversation
                logger.info(
                    "Agents latest message is toolUse, appending a toolResult message to have valid conversation."
                )
                tool_use_ids = [
                    content["toolUse"]["toolUseId"] for content in self.messages[-1]["content"] if "toolUse" in content
                ]
                await self._append_messages(
                    {
                        "role": "user",
                        "content": generate_missing_tool_result_content(tool_use_ids),
                    }
                )
            if isinstance(prompt, str):
                # String input - convert to user message
                messages = [{"role": "user", "content": [{"text": prompt}]}]
            elif isinstance(prompt, list):
                if len(prompt) == 0:
                    # Empty list
                    messages = []
                # Check if all item in input list are dictionaries
                elif all(isinstance(item, dict) for item in prompt):
                    # Check if all items are messages
                    if all(all(key in item for key in Message.__annotations__.keys()) for item in prompt):
                        # Messages input - add all messages to conversation
                        messages = cast(Messages, prompt)

                    # Check if all items are content blocks
                    elif all(any(key in ContentBlock.__annotations__.keys() for key in item) for item in prompt):
                        # Treat as List[ContentBlock] input - convert to user message
                        # This allows invalid structures to be passed through to the model
                        messages = [{"role": "user", "content": cast(list[ContentBlock], prompt)}]
        else:
            messages = []
        if messages is None:
            raise ValueError("Input prompt must be of type: `str | list[Contentblock] | Messages | None`.")
        return messages

    def _start_agent_trace_span(self, messages: Messages) -> trace_api.Span:
        """Starts a trace span for the agent.

        Args:
            messages: The input messages.
        """
        model_id = self.model.config.get("model_id") if hasattr(self.model, "config") else None
        return self.tracer.start_agent_span(
            messages=messages,
            agent_name=self.name,
            model_id=model_id,
            tools=self.tool_names,
            system_prompt=self.system_prompt,
            custom_trace_attributes=self.trace_attributes,
            tools_config=self.tool_registry.get_all_tools_config(),
        )

    def _end_agent_trace_span(
        self,
        response: Optional[AgentResult] = None,
        error: Optional[Exception] = None,
    ) -> None:
        """Ends a trace span for the agent.

        Args:
            span: The span to end.
            response: Response to record as a trace attribute.
            error: Error to record as a trace attribute.
        """
        if self.trace_span:
            trace_attributes: dict[str, Any] = {
                "span": self.trace_span,
            }

            if response:
                trace_attributes["response"] = response
            if error:
                trace_attributes["error"] = error

            self.tracer.end_agent_span(**trace_attributes)

    def _initialize_system_prompt(
        self, system_prompt: str | list[SystemContentBlock] | None
    ) -> tuple[str | None, list[SystemContentBlock] | None]:
        """Initialize system prompt fields from constructor input.

        Maintains backwards compatibility by keeping system_prompt as str when string input
        provided, avoiding breaking existing consumers.

        Maps system_prompt input to both string and content block representations:
        - If string: system_prompt=string, _system_prompt_content=[{text: string}]
        - If list with text elements: system_prompt=concatenated_text, _system_prompt_content=list
        - If list without text elements: system_prompt=None, _system_prompt_content=list
        - If None: system_prompt=None, _system_prompt_content=None
        """
        if isinstance(system_prompt, str):
            return system_prompt, [{"text": system_prompt}]
        elif isinstance(system_prompt, list):
            # Concatenate all text elements for backwards compatibility, None if no text found
            text_parts = [block["text"] for block in system_prompt if "text" in block]
            system_prompt_str = "\n".join(text_parts) if text_parts else None
            return system_prompt_str, system_prompt
        else:
            return None, None

    async def _append_messages(self, *messages: Message) -> None:
        """Appends messages to history and invoke the callbacks for the MessageAddedEvent."""
        for message in messages:
            self.messages.append(message)
            await self.hooks.invoke_callbacks_async(MessageAddedEvent(agent=self, message=message))

    def _redact_user_content(self, content: list[ContentBlock], redact_message: str) -> list[ContentBlock]:
        """Redact user content preserving toolResult blocks.

        Args:
            content: content blocks to be redacted
            redact_message: redact message to be replaced

        Returns:
            Redacted content, as follows:
            - if the message contains at least a toolResult block,
                all toolResult blocks(s) are kept, redacting only the result content;
            - otherwise, the entire content of the message is replaced
                with a single text block with the redact message.
        """
        redacted_content = []
        for block in content:
            if "toolResult" in block:
                block["toolResult"]["content"] = [{"text": redact_message}]
                redacted_content.append(block)

        if not redacted_content:
            # Text content is added only if no toolResult blocks were found
            redacted_content = [{"text": redact_message}]

        return redacted_content

`system_prompt` `property` `writable` ¶

Get the system prompt as a string for backwards compatibility.

Returns the system prompt as a concatenated string when it contains text content, or None if no text content is present. This maintains backwards compatibility with existing code that expects system_prompt to be a string.

Returns:

Type	Description
`str \| None`	The system prompt as a string, or None if no text content exists.

`tool` `property` ¶

Call tool as a function.

Returns:

Type	Description
`_ToolCaller`	Tool caller through which user can invoke tool as a function.

Example

agent = Agent(tools=[calculator])
agent.tool.calculator(...)

`tool_names` `property` ¶

Get a list of all registered tool names.

Returns:

Type	Description
`list[str]`	Names of all tools available to this agent.

`call(prompt=None, *, invocation_state=None, structured_output_model=None, **kwargs)` ¶

Process a natural language prompt through the agent's event loop.

This method implements the conversational interface with multiple input patterns: - String input: agent("hello!") - ContentBlock list: agent([{"text": "hello"}, {"image": {...}}]) - Message list: agent([{"role": "user", "content": [{"text": "hello"}]}]) - No input: agent() - uses existing conversation history

Parameters:

Name	Type	Description	Default
`prompt`	`AgentInput`	User input in various formats: - str: Simple text input - list[ContentBlock]: Multi-modal content blocks - list[Message]: Complete messages with roles - None: Use existing conversation history	`None`
`invocation_state`	`dict[str, Any] \| None`	Additional parameters to pass through the event loop.	`None`
`structured_output_model`	`Type[BaseModel] \| None`	Pydantic model type(s) for structured output (overrides agent default).	`None`
`**kwargs`	`Any`	Additional parameters to pass through the event loop.[Deprecating]	`{}`

Returns:

Type	Description
`AgentResult`	Result object containing: stop_reason: Why the event loop stopped (e.g., "end_turn", "max_tokens") message: The final message from the model metrics: Performance metrics from the event loop state: The final state of the event loop structured_output: Parsed structured output when structured_output_model was specified

Source code in strands/agent/agent.py

def __call__(
    self,
    prompt: AgentInput = None,
    *,
    invocation_state: dict[str, Any] | None = None,
    structured_output_model: Type[BaseModel] | None = None,
    **kwargs: Any,
) -> AgentResult:
    """Process a natural language prompt through the agent's event loop.

    This method implements the conversational interface with multiple input patterns:
    - String input: `agent("hello!")`
    - ContentBlock list: `agent([{"text": "hello"}, {"image": {...}}])`
    - Message list: `agent([{"role": "user", "content": [{"text": "hello"}]}])`
    - No input: `agent()` - uses existing conversation history

    Args:
        prompt: User input in various formats:
            - str: Simple text input
            - list[ContentBlock]: Multi-modal content blocks
            - list[Message]: Complete messages with roles
            - None: Use existing conversation history
        invocation_state: Additional parameters to pass through the event loop.
        structured_output_model: Pydantic model type(s) for structured output (overrides agent default).
        **kwargs: Additional parameters to pass through the event loop.[Deprecating]

    Returns:
        Result object containing:

            - stop_reason: Why the event loop stopped (e.g., "end_turn", "max_tokens")
            - message: The final message from the model
            - metrics: Performance metrics from the event loop
            - state: The final state of the event loop
            - structured_output: Parsed structured output when structured_output_model was specified
    """
    return run_async(
        lambda: self.invoke_async(
            prompt, invocation_state=invocation_state, structured_output_model=structured_output_model, **kwargs
        )
    )

`del()` ¶

Clean up resources when agent is garbage collected.

Source code in strands/agent/agent.py

def __del__(self) -> None:
    """Clean up resources when agent is garbage collected."""
    # __del__ is called even when an exception is thrown in the constructor,
    # so there is no guarantee tool_registry was set..
    if hasattr(self, "tool_registry"):
        self.tool_registry.cleanup()

`init(model=None, messages=None, tools=None, system_prompt=None, structured_output_model=None, callback_handler=_DEFAULT_CALLBACK_HANDLER, conversation_manager=None, record_direct_tool_call=True, load_tools_from_directory=False, trace_attributes=None, *, agent_id=None, name=None, description=None, state=None, hooks=None, session_manager=None, tool_executor=None)` ¶

Initialize the Agent with the specified configuration.

Parameters:

Name	Type	Description	Default
`model`	`Union[Model, str, None]`	Provider for running inference or a string representing the model-id for Bedrock to use. Defaults to strands.models.BedrockModel if None.	`None`
`messages`	`Optional[Messages]`	List of initial messages to pre-load into the conversation. Defaults to an empty list if None.	`None`
`tools`	`Optional[list[Union[str, dict[str, str], ToolProvider, Any]]]`	List of tools to make available to the agent. Can be specified as: String tool names (e.g., "retrieve") File paths (e.g., "/path/to/tool.py") Imported Python modules (e.g., from strands_tools import current_time) Dictionaries with name/path keys (e.g., {"name": "tool_name", "path": "/path/to/tool.py"}) ToolProvider instances for managed tool collections Functions decorated with `@strands.tool` decorator. If provided, only these tools will be available. If None, all tools will be available.	`None`
`system_prompt`	`Optional[str \| list[SystemContentBlock]]`	System prompt to guide model behavior. Can be a string or a list of SystemContentBlock objects for advanced features like caching. If None, the model will behave according to its default settings.	`None`
`structured_output_model`	`Optional[Type[BaseModel]]`	Pydantic model type(s) for structured output. When specified, all agent calls will attempt to return structured output of this type. This can be overridden on the agent invocation. Defaults to None (no structured output).	`None`
`callback_handler`	`Optional[Union[Callable[..., Any], _DefaultCallbackHandlerSentinel]]`	Callback for processing events as they happen during agent execution. If not provided (using the default), a new PrintingCallbackHandler instance is created. If explicitly set to None, null_callback_handler is used.	`_DEFAULT_CALLBACK_HANDLER`
`conversation_manager`	`Optional[ConversationManager]`	Manager for conversation history and context window. Defaults to strands.agent.conversation_manager.SlidingWindowConversationManager if None.	`None`
`record_direct_tool_call`	`bool`	Whether to record direct tool calls in message history. Defaults to True.	`True`
`load_tools_from_directory`	`bool`	Whether to load and automatically reload tools in the `./tools/` directory. Defaults to False.	`False`
`trace_attributes`	`Optional[Mapping[str, AttributeValue]]`	Custom trace attributes to apply to the agent's trace span.	`None`
`agent_id`	`Optional[str]`	Optional ID for the agent, useful for session management and multi-agent scenarios. Defaults to "default".	`None`
`name`	`Optional[str]`	name of the Agent Defaults to "Strands Agents".	`None`
`description`	`Optional[str]`	description of what the Agent does Defaults to None.	`None`
`state`	`Optional[Union[AgentState, dict]]`	stateful information for the agent. Can be either an AgentState object, or a json serializable dict. Defaults to an empty AgentState object.	`None`
`hooks`	`Optional[list[HookProvider]]`	hooks to be added to the agent hook registry Defaults to None.	`None`
`session_manager`	`Optional[SessionManager]`	Manager for handling agent sessions including conversation history and state. If provided, enables session-based persistence and state management.	`None`
`tool_executor`	`Optional[ToolExecutor]`	Definition of tool execution strategy (e.g., sequential, concurrent, etc.).	`None`

Raises:

Type	Description
`ValueError`	If agent id contains path separators.

Source code in strands/agent/agent.py

def __init__(
    self,
    model: Union[Model, str, None] = None,
    messages: Optional[Messages] = None,
    tools: Optional[list[Union[str, dict[str, str], "ToolProvider", Any]]] = None,
    system_prompt: Optional[str | list[SystemContentBlock]] = None,
    structured_output_model: Optional[Type[BaseModel]] = None,
    callback_handler: Optional[
        Union[Callable[..., Any], _DefaultCallbackHandlerSentinel]
    ] = _DEFAULT_CALLBACK_HANDLER,
    conversation_manager: Optional[ConversationManager] = None,
    record_direct_tool_call: bool = True,
    load_tools_from_directory: bool = False,
    trace_attributes: Optional[Mapping[str, AttributeValue]] = None,
    *,
    agent_id: Optional[str] = None,
    name: Optional[str] = None,
    description: Optional[str] = None,
    state: Optional[Union[AgentState, dict]] = None,
    hooks: Optional[list[HookProvider]] = None,
    session_manager: Optional[SessionManager] = None,
    tool_executor: Optional[ToolExecutor] = None,
):
    """Initialize the Agent with the specified configuration.

    Args:
        model: Provider for running inference or a string representing the model-id for Bedrock to use.
            Defaults to strands.models.BedrockModel if None.
        messages: List of initial messages to pre-load into the conversation.
            Defaults to an empty list if None.
        tools: List of tools to make available to the agent.
            Can be specified as:

            - String tool names (e.g., "retrieve")
            - File paths (e.g., "/path/to/tool.py")
            - Imported Python modules (e.g., from strands_tools import current_time)
            - Dictionaries with name/path keys (e.g., {"name": "tool_name", "path": "/path/to/tool.py"})
            - ToolProvider instances for managed tool collections
            - Functions decorated with `@strands.tool` decorator.

            If provided, only these tools will be available. If None, all tools will be available.
        system_prompt: System prompt to guide model behavior.
            Can be a string or a list of SystemContentBlock objects for advanced features like caching.
            If None, the model will behave according to its default settings.
        structured_output_model: Pydantic model type(s) for structured output.
            When specified, all agent calls will attempt to return structured output of this type.
            This can be overridden on the agent invocation.
            Defaults to None (no structured output).
        callback_handler: Callback for processing events as they happen during agent execution.
            If not provided (using the default), a new PrintingCallbackHandler instance is created.
            If explicitly set to None, null_callback_handler is used.
        conversation_manager: Manager for conversation history and context window.
            Defaults to strands.agent.conversation_manager.SlidingWindowConversationManager if None.
        record_direct_tool_call: Whether to record direct tool calls in message history.
            Defaults to True.
        load_tools_from_directory: Whether to load and automatically reload tools in the `./tools/` directory.
            Defaults to False.
        trace_attributes: Custom trace attributes to apply to the agent's trace span.
        agent_id: Optional ID for the agent, useful for session management and multi-agent scenarios.
            Defaults to "default".
        name: name of the Agent
            Defaults to "Strands Agents".
        description: description of what the Agent does
            Defaults to None.
        state: stateful information for the agent. Can be either an AgentState object, or a json serializable dict.
            Defaults to an empty AgentState object.
        hooks: hooks to be added to the agent hook registry
            Defaults to None.
        session_manager: Manager for handling agent sessions including conversation history and state.
            If provided, enables session-based persistence and state management.
        tool_executor: Definition of tool execution strategy (e.g., sequential, concurrent, etc.).

    Raises:
        ValueError: If agent id contains path separators.
    """
    self.model = BedrockModel() if not model else BedrockModel(model_id=model) if isinstance(model, str) else model
    self.messages = messages if messages is not None else []
    # initializing self._system_prompt for backwards compatibility
    self._system_prompt, self._system_prompt_content = self._initialize_system_prompt(system_prompt)
    self._default_structured_output_model = structured_output_model
    self.agent_id = _identifier.validate(agent_id or _DEFAULT_AGENT_ID, _identifier.Identifier.AGENT)
    self.name = name or _DEFAULT_AGENT_NAME
    self.description = description

    # If not provided, create a new PrintingCallbackHandler instance
    # If explicitly set to None, use null_callback_handler
    # Otherwise use the passed callback_handler
    self.callback_handler: Union[Callable[..., Any], PrintingCallbackHandler]
    if isinstance(callback_handler, _DefaultCallbackHandlerSentinel):
        self.callback_handler = PrintingCallbackHandler()
    elif callback_handler is None:
        self.callback_handler = null_callback_handler
    else:
        self.callback_handler = callback_handler

    self.conversation_manager = conversation_manager if conversation_manager else SlidingWindowConversationManager()

    # Process trace attributes to ensure they're of compatible types
    self.trace_attributes: dict[str, AttributeValue] = {}
    if trace_attributes:
        for k, v in trace_attributes.items():
            if isinstance(v, (str, int, float, bool)) or (
                isinstance(v, list) and all(isinstance(x, (str, int, float, bool)) for x in v)
            ):
                self.trace_attributes[k] = v

    self.record_direct_tool_call = record_direct_tool_call
    self.load_tools_from_directory = load_tools_from_directory

    self.tool_registry = ToolRegistry()

    # Process tool list if provided
    if tools is not None:
        self.tool_registry.process_tools(tools)

    # Initialize tools and configuration
    self.tool_registry.initialize_tools(self.load_tools_from_directory)
    if load_tools_from_directory:
        self.tool_watcher = ToolWatcher(tool_registry=self.tool_registry)

    self.event_loop_metrics = EventLoopMetrics()

    # Initialize tracer instance (no-op if not configured)
    self.tracer = get_tracer()
    self.trace_span: Optional[trace_api.Span] = None

    # Initialize agent state management
    if state is not None:
        if isinstance(state, dict):
            self.state = AgentState(state)
        elif isinstance(state, AgentState):
            self.state = state
        else:
            raise ValueError("state must be an AgentState object or a dict")
    else:
        self.state = AgentState()

    self.tool_caller = _ToolCaller(self)

    self.hooks = HookRegistry()

    self._interrupt_state = _InterruptState()

    # Initialize session management functionality
    self._session_manager = session_manager
    if self._session_manager:
        self.hooks.add_hook(self._session_manager)

    # Allow conversation_managers to subscribe to hooks
    self.hooks.add_hook(self.conversation_manager)

    self.tool_executor = tool_executor or ConcurrentToolExecutor()

    if hooks:
        for hook in hooks:
            self.hooks.add_hook(hook)
    self.hooks.invoke_callbacks(AgentInitializedEvent(agent=self))

`cleanup()` ¶

Clean up resources used by the agent.

This method cleans up all tool providers that require explicit cleanup, such as MCP clients. It should be called when the agent is no longer needed to ensure proper resource cleanup.

Note: This method uses a "belt and braces" approach with automatic cleanup through finalizers as a fallback, but explicit cleanup is recommended.

Source code in strands/agent/agent.py

def cleanup(self) -> None:
    """Clean up resources used by the agent.

    This method cleans up all tool providers that require explicit cleanup,
    such as MCP clients. It should be called when the agent is no longer needed
    to ensure proper resource cleanup.

    Note: This method uses a "belt and braces" approach with automatic cleanup
    through finalizers as a fallback, but explicit cleanup is recommended.
    """
    self.tool_registry.cleanup()

`invoke_async(prompt=None, *, invocation_state=None, structured_output_model=None, **kwargs)` `async` ¶

Process a natural language prompt through the agent's event loop.

This method implements the conversational interface with multiple input patterns: - String input: Simple text input - ContentBlock list: Multi-modal content blocks - Message list: Complete messages with roles - No input: Use existing conversation history

Parameters:

Name	Type	Description	Default
`prompt`	`AgentInput`	User input in various formats: - str: Simple text input - list[ContentBlock]: Multi-modal content blocks - list[Message]: Complete messages with roles - None: Use existing conversation history	`None`
`invocation_state`	`dict[str, Any] \| None`	Additional parameters to pass through the event loop.	`None`
`structured_output_model`	`Type[BaseModel] \| None`	Pydantic model type(s) for structured output (overrides agent default).	`None`
`**kwargs`	`Any`	Additional parameters to pass through the event loop.[Deprecating]	`{}`

Returns:

Name	Type	Description
`Result`	`AgentResult`	object containing: stop_reason: Why the event loop stopped (e.g., "end_turn", "max_tokens") message: The final message from the model metrics: Performance metrics from the event loop state: The final state of the event loop

Source code in strands/agent/agent.py

async def invoke_async(
    self,
    prompt: AgentInput = None,
    *,
    invocation_state: dict[str, Any] | None = None,
    structured_output_model: Type[BaseModel] | None = None,
    **kwargs: Any,
) -> AgentResult:
    """Process a natural language prompt through the agent's event loop.

    This method implements the conversational interface with multiple input patterns:
    - String input: Simple text input
    - ContentBlock list: Multi-modal content blocks
    - Message list: Complete messages with roles
    - No input: Use existing conversation history

    Args:
        prompt: User input in various formats:
            - str: Simple text input
            - list[ContentBlock]: Multi-modal content blocks
            - list[Message]: Complete messages with roles
            - None: Use existing conversation history
        invocation_state: Additional parameters to pass through the event loop.
        structured_output_model: Pydantic model type(s) for structured output (overrides agent default).
        **kwargs: Additional parameters to pass through the event loop.[Deprecating]

    Returns:
        Result: object containing:

            - stop_reason: Why the event loop stopped (e.g., "end_turn", "max_tokens")
            - message: The final message from the model
            - metrics: Performance metrics from the event loop
            - state: The final state of the event loop
    """
    events = self.stream_async(
        prompt, invocation_state=invocation_state, structured_output_model=structured_output_model, **kwargs
    )
    async for event in events:
        _ = event

    return cast(AgentResult, event["result"])

`stream_async(prompt=None, *, invocation_state=None, structured_output_model=None, **kwargs)` `async` ¶

Process a natural language prompt and yield events as an async iterator.

This method provides an asynchronous interface for streaming agent events with multiple input patterns: - String input: Simple text input - ContentBlock list: Multi-modal content blocks - Message list: Complete messages with roles - No input: Use existing conversation history

Parameters:

Name	Type	Description	Default
`prompt`	`AgentInput`	User input in various formats: - str: Simple text input - list[ContentBlock]: Multi-modal content blocks - list[Message]: Complete messages with roles - None: Use existing conversation history	`None`
`invocation_state`	`dict[str, Any] \| None`	Additional parameters to pass through the event loop.	`None`
`structured_output_model`	`Type[BaseModel] \| None`	Pydantic model type(s) for structured output (overrides agent default).	`None`
`**kwargs`	`Any`	Additional parameters to pass to the event loop.[Deprecating]	`{}`

Yields:

Type	Description
`AsyncIterator[Any]`	An async iterator that yields events. Each event is a dictionary containing information about the current state of processing, such as: data: Text content being generated complete: Whether this is the final chunk current_tool_use: Information about tools being executed And other event data provided by the callback handler

Raises:

Type	Description
`Exception`	Any exceptions from the agent invocation will be propagated to the caller.

Example

async for event in agent.stream_async("Analyze this data"):
    if "data" in event:
        yield event["data"]

Source code in strands/agent/agent.py

async def stream_async(
    self,
    prompt: AgentInput = None,
    *,
    invocation_state: dict[str, Any] | None = None,
    structured_output_model: Type[BaseModel] | None = None,
    **kwargs: Any,
) -> AsyncIterator[Any]:
    """Process a natural language prompt and yield events as an async iterator.

    This method provides an asynchronous interface for streaming agent events with multiple input patterns:
    - String input: Simple text input
    - ContentBlock list: Multi-modal content blocks
    - Message list: Complete messages with roles
    - No input: Use existing conversation history

    Args:
        prompt: User input in various formats:
            - str: Simple text input
            - list[ContentBlock]: Multi-modal content blocks
            - list[Message]: Complete messages with roles
            - None: Use existing conversation history
        invocation_state: Additional parameters to pass through the event loop.
        structured_output_model: Pydantic model type(s) for structured output (overrides agent default).
        **kwargs: Additional parameters to pass to the event loop.[Deprecating]

    Yields:
        An async iterator that yields events. Each event is a dictionary containing
            information about the current state of processing, such as:

            - data: Text content being generated
            - complete: Whether this is the final chunk
            - current_tool_use: Information about tools being executed
            - And other event data provided by the callback handler

    Raises:
        Exception: Any exceptions from the agent invocation will be propagated to the caller.

    Example:
        ```python
        async for event in agent.stream_async("Analyze this data"):
            if "data" in event:
                yield event["data"]
        ```
    """
    self._interrupt_state.resume(prompt)

    self.event_loop_metrics.reset_usage_metrics()

    merged_state = {}
    if kwargs:
        warnings.warn("`**kwargs` parameter is deprecating, use `invocation_state` instead.", stacklevel=2)
        merged_state.update(kwargs)
        if invocation_state is not None:
            merged_state["invocation_state"] = invocation_state
    else:
        if invocation_state is not None:
            merged_state = invocation_state

    callback_handler = self.callback_handler
    if kwargs:
        callback_handler = kwargs.get("callback_handler", self.callback_handler)

    # Process input and get message to add (if any)
    messages = await self._convert_prompt_to_messages(prompt)

    self.trace_span = self._start_agent_trace_span(messages)

    with trace_api.use_span(self.trace_span):
        try:
            events = self._run_loop(messages, merged_state, structured_output_model)

            async for event in events:
                event.prepare(invocation_state=merged_state)

                if event.is_callback_event:
                    as_dict = event.as_dict()
                    callback_handler(**as_dict)
                    yield as_dict

            result = AgentResult(*event["stop"])
            callback_handler(result=result)
            yield AgentResultEvent(result=result).as_dict()

            self._end_agent_trace_span(response=result)

        except Exception as e:
            self._end_agent_trace_span(error=e)
            raise

`structured_output(output_model, prompt=None)` ¶

This method allows you to get structured output from the agent.

If you pass in a prompt, it will be used temporarily without adding it to the conversation history. If you don't pass in a prompt, it will use only the existing conversation history to respond.

For smaller models, you may want to use the optional prompt to add additional instructions to explicitly instruct the model to output the structured data.

Parameters:

Name	Type	Description	Default
`output_model`	`Type[T]`	The output model (a JSON schema written as a Pydantic BaseModel) that the agent will use when responding.	required
`prompt`	`AgentInput`	The prompt to use for the agent in various formats: - str: Simple text input - list[ContentBlock]: Multi-modal content blocks - list[Message]: Complete messages with roles - None: Use existing conversation history	`None`

Raises:

Type	Description
`ValueError`	If no conversation history or prompt is provided.

Source code in strands/agent/agent.py

def structured_output(self, output_model: Type[T], prompt: AgentInput = None) -> T:
    """This method allows you to get structured output from the agent.

    If you pass in a prompt, it will be used temporarily without adding it to the conversation history.
    If you don't pass in a prompt, it will use only the existing conversation history to respond.

    For smaller models, you may want to use the optional prompt to add additional instructions to explicitly
    instruct the model to output the structured data.

    Args:
        output_model: The output model (a JSON schema written as a Pydantic BaseModel)
            that the agent will use when responding.
        prompt: The prompt to use for the agent in various formats:
            - str: Simple text input
            - list[ContentBlock]: Multi-modal content blocks
            - list[Message]: Complete messages with roles
            - None: Use existing conversation history

    Raises:
        ValueError: If no conversation history or prompt is provided.
    """
    warnings.warn(
        "Agent.structured_output method is deprecated."
        " You should pass in `structured_output_model` directly into the agent invocation."
        " see: https://strandsagents.com/latest/documentation/docs/user-guide/concepts/agents/structured-output/",
        category=DeprecationWarning,
        stacklevel=2,
    )

    return run_async(lambda: self.structured_output_async(output_model, prompt))

`structured_output_async(output_model, prompt=None)` `async` ¶

This method allows you to get structured output from the agent.

If you pass in a prompt, it will be used temporarily without adding it to the conversation history. If you don't pass in a prompt, it will use only the existing conversation history to respond.

For smaller models, you may want to use the optional prompt to add additional instructions to explicitly instruct the model to output the structured data.

Parameters:

Name	Type	Description	Default
`output_model`	`Type[T]`	The output model (a JSON schema written as a Pydantic BaseModel) that the agent will use when responding.	required
`prompt`	`AgentInput`	The prompt to use for the agent (will not be added to conversation history).	`None`

Raises:

Type	Description
`ValueError`	If no conversation history or prompt is provided.

-

Source code in strands/agent/agent.py

async def structured_output_async(self, output_model: Type[T], prompt: AgentInput = None) -> T:
    """This method allows you to get structured output from the agent.

    If you pass in a prompt, it will be used temporarily without adding it to the conversation history.
    If you don't pass in a prompt, it will use only the existing conversation history to respond.

    For smaller models, you may want to use the optional prompt to add additional instructions to explicitly
    instruct the model to output the structured data.

    Args:
        output_model: The output model (a JSON schema written as a Pydantic BaseModel)
            that the agent will use when responding.
        prompt: The prompt to use for the agent (will not be added to conversation history).

    Raises:
        ValueError: If no conversation history or prompt is provided.
    -
    """
    if self._interrupt_state.activated:
        raise RuntimeError("cannot call structured output during interrupt")

    warnings.warn(
        "Agent.structured_output_async method is deprecated."
        " You should pass in `structured_output_model` directly into the agent invocation."
        " see: https://strandsagents.com/latest/documentation/docs/user-guide/concepts/agents/structured-output/",
        category=DeprecationWarning,
        stacklevel=2,
    )
    await self.hooks.invoke_callbacks_async(BeforeInvocationEvent(agent=self))
    with self.tracer.tracer.start_as_current_span(
        "execute_structured_output", kind=trace_api.SpanKind.CLIENT
    ) as structured_output_span:
        try:
            if not self.messages and not prompt:
                raise ValueError("No conversation history or prompt provided")

            temp_messages: Messages = self.messages + await self._convert_prompt_to_messages(prompt)

            structured_output_span.set_attributes(
                {
                    "gen_ai.system": "strands-agents",
                    "gen_ai.agent.name": self.name,
                    "gen_ai.agent.id": self.agent_id,
                    "gen_ai.operation.name": "execute_structured_output",
                }
            )
            if self.system_prompt:
                structured_output_span.add_event(
                    "gen_ai.system.message",
                    attributes={"role": "system", "content": serialize([{"text": self.system_prompt}])},
                )
            for message in temp_messages:
                structured_output_span.add_event(
                    f"gen_ai.{message['role']}.message",
                    attributes={"role": message["role"], "content": serialize(message["content"])},
                )
            events = self.model.structured_output(output_model, temp_messages, system_prompt=self.system_prompt)
            async for event in events:
                if isinstance(event, TypedEvent):
                    event.prepare(invocation_state={})
                    if event.is_callback_event:
                        self.callback_handler(**event.as_dict())

            structured_output_span.add_event(
                "gen_ai.choice", attributes={"message": serialize(event["output"].model_dump())}
            )
            return event["output"]

        finally:
            await self.hooks.invoke_callbacks_async(AfterInvocationEvent(agent=self))

`BeforeMultiAgentInvocationEvent` `dataclass` ¶

Bases: BaseHookEvent

Event triggered before orchestrator execution starts.

Attributes:

Name	Type	Description
`source`	`MultiAgentBase`	The multi-agent orchestrator instance
`invocation_state`	`dict[str, Any] \| None`	Configuration that user passes in

Source code in strands/experimental/hooks/multiagent/events.py

@dataclass
class BeforeMultiAgentInvocationEvent(BaseHookEvent):
    """Event triggered before orchestrator execution starts.

    Attributes:
        source: The multi-agent orchestrator instance
        invocation_state: Configuration that user passes in
    """

    source: "MultiAgentBase"
    invocation_state: dict[str, Any] | None = None

`BeforeNodeCallEvent` `dataclass` ¶

Bases: BaseHookEvent, _Interruptible

Event triggered before individual node execution starts.

Attributes:

Name	Type	Description
`source`	`MultiAgentBase`	The multi-agent orchestrator instance
`node_id`	`str`	ID of the node about to execute
`invocation_state`	`dict[str, Any] \| None`	Configuration that user passes in
`cancel_node`	`bool \| str`	A user defined message that when set, will cancel the node execution with status FAILED. The message will be emitted under a MultiAgentNodeCancel event. If set to `True`, Strands will cancel the node using a default cancel message.

Source code in strands/experimental/hooks/multiagent/events.py

@dataclass
class BeforeNodeCallEvent(BaseHookEvent, _Interruptible):
    """Event triggered before individual node execution starts.

    Attributes:
        source: The multi-agent orchestrator instance
        node_id: ID of the node about to execute
        invocation_state: Configuration that user passes in
        cancel_node: A user defined message that when set, will cancel the node execution with status FAILED.
            The message will be emitted under a MultiAgentNodeCancel event. If set to `True`, Strands will cancel the
            node using a default cancel message.
    """

    source: "MultiAgentBase"
    node_id: str
    invocation_state: dict[str, Any] | None = None
    cancel_node: bool | str = False

    def _can_write(self, name: str) -> bool:
        return name in ["cancel_node"]

    @override
    def _interrupt_id(self, name: str) -> str:
        """Unique id for the interrupt.

        Args:
            name: User defined name for the interrupt.

        Returns:
            Interrupt id.
        """
        node_id = uuid.uuid5(uuid.NAMESPACE_OID, self.node_id)
        call_id = uuid.uuid5(uuid.NAMESPACE_OID, name)
        return f"v1:before_node_call:{node_id}:{call_id}"

`ContentBlock` ¶

Bases: TypedDict

A block of content for a message that you pass to, or receive from, a model.

Attributes:

Name	Type	Description
`cachePoint`	`CachePoint`	A cache point configuration to optimize conversation history.
`document`	`DocumentContent`	A document to include in the message.
`guardContent`	`GuardContent`	Contains the content to assess with the guardrail.
`image`	`ImageContent`	Image to include in the message.
`reasoningContent`	`ReasoningContentBlock`	Contains content regarding the reasoning that is carried out by the model.
`text`	`str`	Text to include in the message.
`toolResult`	`ToolResult`	The result for a tool request that a model makes.
`toolUse`	`ToolUse`	Information about a tool use request from a model.
`video`	`VideoContent`	Video to include in the message.
`citationsContent`	`CitationsContentBlock`	Contains the citations for a document.

Source code in strands/types/content.py

class ContentBlock(TypedDict, total=False):
    """A block of content for a message that you pass to, or receive from, a model.

    Attributes:
        cachePoint: A cache point configuration to optimize conversation history.
        document: A document to include in the message.
        guardContent: Contains the content to assess with the guardrail.
        image: Image to include in the message.
        reasoningContent: Contains content regarding the reasoning that is carried out by the model.
        text: Text to include in the message.
        toolResult: The result for a tool request that a model makes.
        toolUse: Information about a tool use request from a model.
        video: Video to include in the message.
        citationsContent: Contains the citations for a document.
    """

    cachePoint: CachePoint
    document: DocumentContent
    guardContent: GuardContent
    image: ImageContent
    reasoningContent: ReasoningContentBlock
    text: str
    toolResult: ToolResult
    toolUse: ToolUse
    video: VideoContent
    citationsContent: CitationsContentBlock

`HookProvider` ¶

Bases: Protocol

Protocol for objects that provide hook callbacks to an agent.

Hook providers offer a composable way to extend agent functionality by subscribing to various events in the agent lifecycle. This protocol enables building reusable components that can hook into agent events.

Example

class MyHookProvider(HookProvider):
    def register_hooks(self, registry: HookRegistry) -> None:
        registry.add_callback(StartRequestEvent, self.on_request_start)
        registry.add_callback(EndRequestEvent, self.on_request_end)

agent = Agent(hooks=[MyHookProvider()])

Source code in strands/hooks/registry.py

@runtime_checkable
class HookProvider(Protocol):
    """Protocol for objects that provide hook callbacks to an agent.

    Hook providers offer a composable way to extend agent functionality by
    subscribing to various events in the agent lifecycle. This protocol enables
    building reusable components that can hook into agent events.

    Example:
        ```python
        class MyHookProvider(HookProvider):
            def register_hooks(self, registry: HookRegistry) -> None:
                registry.add_callback(StartRequestEvent, self.on_request_start)
                registry.add_callback(EndRequestEvent, self.on_request_end)

        agent = Agent(hooks=[MyHookProvider()])
        ```
    """

    def register_hooks(self, registry: "HookRegistry", **kwargs: Any) -> None:
        """Register callback functions for specific event types.

        Args:
            registry: The hook registry to register callbacks with.
            **kwargs: Additional keyword arguments for future extensibility.
        """
        ...

`register_hooks(registry, **kwargs)` ¶

Register callback functions for specific event types.

Parameters:

Name	Type	Description	Default
`registry`	`HookRegistry`	The hook registry to register callbacks with.	required
`**kwargs`	`Any`	Additional keyword arguments for future extensibility.	`{}`

Source code in strands/hooks/registry.py

def register_hooks(self, registry: "HookRegistry", **kwargs: Any) -> None:
    """Register callback functions for specific event types.

    Args:
        registry: The hook registry to register callbacks with.
        **kwargs: Additional keyword arguments for future extensibility.
    """
    ...

`HookRegistry` ¶

Registry for managing hook callbacks associated with event types.

The HookRegistry maintains a mapping of event types to callback functions and provides methods for registering callbacks and invoking them when events occur.

The registry handles callback ordering, including reverse ordering for cleanup events, and provides type-safe event dispatching.

Source code in strands/hooks/registry.py

class HookRegistry:
    """Registry for managing hook callbacks associated with event types.

    The HookRegistry maintains a mapping of event types to callback functions
    and provides methods for registering callbacks and invoking them when
    events occur.

    The registry handles callback ordering, including reverse ordering for
    cleanup events, and provides type-safe event dispatching.
    """

    def __init__(self) -> None:
        """Initialize an empty hook registry."""
        self._registered_callbacks: dict[Type, list[HookCallback]] = {}

    def add_callback(self, event_type: Type[TEvent], callback: HookCallback[TEvent]) -> None:
        """Register a callback function for a specific event type.

        Args:
            event_type: The class type of events this callback should handle.
            callback: The callback function to invoke when events of this type occur.

        Example:
            ```python
            def my_handler(event: StartRequestEvent):
                print("Request started")

            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)

    def add_hook(self, hook: HookProvider) -> None:
        """Register all callbacks from a hook provider.

        This method allows bulk registration of callbacks by delegating to
        the hook provider's register_hooks method. This is the preferred
        way to register multiple related callbacks.

        Args:
            hook: The hook provider containing callbacks to register.

        Example:
            ```python
            class MyHooks(HookProvider):
                def register_hooks(self, registry: HookRegistry):
                    registry.add_callback(StartRequestEvent, self.on_start)
                    registry.add_callback(EndRequestEvent, self.on_end)

            registry.add_hook(MyHooks())
            ```
        """
        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.

        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:
            RuntimeError: If at least one callback is async.
            ValueError: If interrupt name is used more than once.

        Example:
            ```python
            event = StartRequestEvent(agent=my_agent)
            registry.invoke_callbacks(event)
            ```
        """
        callbacks = list(self.get_callbacks_for(event))
        interrupts: dict[str, Interrupt] = {}

        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:
                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 has_callbacks(self) -> bool:
        """Check if the registry has any registered callbacks.

        Returns:
            True if there are any registered callbacks, False otherwise.

        Example:
            ```python
            if registry.has_callbacks():
                print("Registry has callbacks registered")
            ```
        """
        return bool(self._registered_callbacks)

    def get_callbacks_for(self, event: TEvent) -> Generator[HookCallback[TEvent], None, None]:
        """Get callbacks registered for the given event in the appropriate order.

        This method returns callbacks in registration order for normal events,
        or reverse registration order for events that have should_reverse_callbacks=True.
        This enables proper cleanup ordering for teardown events.

        Args:
            event: The event to get callbacks for.

        Yields:
            Callback functions registered for this event type, in the appropriate order.

        Example:
            ```python
            event = EndRequestEvent(agent=my_agent)
            for callback in registry.get_callbacks_for(event):
                callback(event)
            ```
        """
        event_type = type(event)

        callbacks = self._registered_callbacks.get(event_type, [])
        if event.should_reverse_callbacks:
            yield from reversed(callbacks)
        else:
            yield from callbacks

`init()` ¶

Initialize an empty hook registry.

Source code in strands/hooks/registry.py

def __init__(self) -> None:
    """Initialize an empty hook registry."""
    self._registered_callbacks: dict[Type, list[HookCallback]] = {}

`add_callback(event_type, callback)` ¶

Register a callback function for a specific event type.

Parameters:

Name	Type	Description	Default
`event_type`	`Type[TEvent]`	The class type of events this callback should handle.	required
`callback`	`HookCallback[TEvent]`	The callback function to invoke when events of this type occur.	required

Example

def my_handler(event: StartRequestEvent):
    print("Request started")

registry.add_callback(StartRequestEvent, my_handler)

Source code in strands/hooks/registry.py

def add_callback(self, event_type: Type[TEvent], callback: HookCallback[TEvent]) -> None:
    """Register a callback function for a specific event type.

    Args:
        event_type: The class type of events this callback should handle.
        callback: The callback function to invoke when events of this type occur.

    Example:
        ```python
        def my_handler(event: StartRequestEvent):
            print("Request started")

        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)

`add_hook(hook)` ¶

Register all callbacks from a hook provider.

This method allows bulk registration of callbacks by delegating to the hook provider's register_hooks method. This is the preferred way to register multiple related callbacks.

Parameters:

Name	Type	Description	Default
`hook`	`HookProvider`	The hook provider containing callbacks to register.	required

Example

class MyHooks(HookProvider):
    def register_hooks(self, registry: HookRegistry):
        registry.add_callback(StartRequestEvent, self.on_start)
        registry.add_callback(EndRequestEvent, self.on_end)

registry.add_hook(MyHooks())

Source code in strands/hooks/registry.py

def add_hook(self, hook: HookProvider) -> None:
    """Register all callbacks from a hook provider.

    This method allows bulk registration of callbacks by delegating to
    the hook provider's register_hooks method. This is the preferred
    way to register multiple related callbacks.

    Args:
        hook: The hook provider containing callbacks to register.

    Example:
        ```python
        class MyHooks(HookProvider):
            def register_hooks(self, registry: HookRegistry):
                registry.add_callback(StartRequestEvent, self.on_start)
                registry.add_callback(EndRequestEvent, self.on_end)

        registry.add_hook(MyHooks())
        ```
    """
    hook.register_hooks(self)

`get_callbacks_for(event)` ¶

Get callbacks registered for the given event in the appropriate order.

This method returns callbacks in registration order for normal events, or reverse registration order for events that have should_reverse_callbacks=True. This enables proper cleanup ordering for teardown events.

Parameters:

Name	Type	Description	Default
`event`	`TEvent`	The event to get callbacks for.	required

Yields:

Type	Description
`HookCallback[TEvent]`	Callback functions registered for this event type, in the appropriate order.

Example

event = EndRequestEvent(agent=my_agent)
for callback in registry.get_callbacks_for(event):
    callback(event)

Source code in strands/hooks/registry.py

def get_callbacks_for(self, event: TEvent) -> Generator[HookCallback[TEvent], None, None]:
    """Get callbacks registered for the given event in the appropriate order.

    This method returns callbacks in registration order for normal events,
    or reverse registration order for events that have should_reverse_callbacks=True.
    This enables proper cleanup ordering for teardown events.

    Args:
        event: The event to get callbacks for.

    Yields:
        Callback functions registered for this event type, in the appropriate order.

    Example:
        ```python
        event = EndRequestEvent(agent=my_agent)
        for callback in registry.get_callbacks_for(event):
            callback(event)
        ```
    """
    event_type = type(event)

    callbacks = self._registered_callbacks.get(event_type, [])
    if event.should_reverse_callbacks:
        yield from reversed(callbacks)
    else:
        yield from callbacks

`has_callbacks()` ¶

Check if the registry has any registered callbacks.

Returns:

Type	Description
`bool`	True if there are any registered callbacks, False otherwise.

Example

if registry.has_callbacks():
    print("Registry has callbacks registered")

Source code in strands/hooks/registry.py

def has_callbacks(self) -> bool:
    """Check if the registry has any registered callbacks.

    Returns:
        True if there are any registered callbacks, False otherwise.

    Example:
        ```python
        if registry.has_callbacks():
            print("Registry has callbacks registered")
        ```
    """
    return bool(self._registered_callbacks)

`invoke_callbacks(event)` ¶

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.

Parameters:

Name	Type	Description	Default
`event`	`TInvokeEvent`	The event to dispatch to registered callbacks.	required

Returns:

Type	Description
`tuple[TInvokeEvent, list[Interrupt]]`	The event dispatched to registered callbacks and any interrupts raised by the user.

Raises:

Type	Description
`RuntimeError`	If at least one callback is async.
`ValueError`	If interrupt name is used more than once.

Example

event = StartRequestEvent(agent=my_agent)
registry.invoke_callbacks(event)

Source code in strands/hooks/registry.py

def invoke_callbacks(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:
        RuntimeError: If at least one callback is async.
        ValueError: If interrupt name is used more than once.

    Example:
        ```python
        event = StartRequestEvent(agent=my_agent)
        registry.invoke_callbacks(event)
        ```
    """
    callbacks = list(self.get_callbacks_for(event))
    interrupts: dict[str, Interrupt] = {}

    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:
            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())

`invoke_callbacks_async(event)` `async` ¶

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.

Parameters:

Name	Type	Description	Default
`event`	`TInvokeEvent`	The event to dispatch to registered callbacks.	required

Returns:

Type	Description
`tuple[TInvokeEvent, list[Interrupt]]`	The event dispatched to registered callbacks and any interrupts raised by the user.

Raises:

Type	Description
`ValueError`	If interrupt name is used more than once.

Example

event = StartRequestEvent(agent=my_agent)
await registry.invoke_callbacks_async(event)

Source code in strands/hooks/registry.py

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())

`Interrupt` `dataclass` ¶

Represents an interrupt that can pause agent execution for human-in-the-loop workflows.

Attributes:

Name	Type	Description
`id`	`str`	Unique identifier.
`name`	`str`	User defined name.
`reason`	`Any`	User provided reason for raising the interrupt.
`response`	`Any`	Human response provided when resuming the agent after an interrupt.

Source code in strands/interrupt.py

@dataclass
class Interrupt:
    """Represents an interrupt that can pause agent execution for human-in-the-loop workflows.

    Attributes:
        id: Unique identifier.
        name: User defined name.
        reason: User provided reason for raising the interrupt.
        response: Human response provided when resuming the agent after an interrupt.
    """

    id: str
    name: str
    reason: Any = None
    response: Any = None

    def to_dict(self) -> dict[str, Any]:
        """Serialize to dict for session management."""
        return asdict(self)

`to_dict()` ¶

Serialize to dict for session management.

Source code in strands/interrupt.py

def to_dict(self) -> dict[str, Any]:
    """Serialize to dict for session management."""
    return asdict(self)

`Metrics` ¶

Bases: TypedDict

Performance metrics for model interactions.

Attributes:

Name	Type	Description
`latencyMs`	`int`	Latency of the model request in milliseconds.
`timeToFirstByteMs`	`int`	Latency from sending model request to first content chunk (contentBlockDelta or contentBlockStart) from the model in milliseconds.

Source code in strands/types/event_loop.py

class Metrics(TypedDict, total=False):
    """Performance metrics for model interactions.

    Attributes:
        latencyMs (int): Latency of the model request in milliseconds.
        timeToFirstByteMs (int): Latency from sending model request to first
            content chunk (contentBlockDelta or contentBlockStart) from the model in milliseconds.
    """

    latencyMs: Required[int]
    timeToFirstByteMs: int

`MultiAgentBase` ¶

Bases: ABC

Base class for multi-agent helpers.

This class integrates with existing Strands Agent instances and provides multi-agent orchestration capabilities.

Attributes:

Name	Type	Description
`id`	`str`	Unique MultiAgent id for session management,etc.

Source code in strands/multiagent/base.py

class MultiAgentBase(ABC):
    """Base class for multi-agent helpers.

    This class integrates with existing Strands Agent instances and provides
    multi-agent orchestration capabilities.

    Attributes:
        id: Unique MultiAgent id for session management,etc.
    """

    id: str

    @abstractmethod
    async def invoke_async(
        self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
    ) -> MultiAgentResult:
        """Invoke asynchronously.

        Args:
            task: The task to execute
            invocation_state: Additional state/context passed to underlying agents.
                Defaults to None to avoid mutable default argument issues.
            **kwargs: Additional keyword arguments passed to underlying agents.
        """
        raise NotImplementedError("invoke_async not implemented")

    async def stream_async(
        self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
    ) -> AsyncIterator[dict[str, Any]]:
        """Stream events during multi-agent execution.

        Default implementation executes invoke_async and yields the result as a single event.
        Subclasses can override this method to provide true streaming capabilities.

        Args:
            task: The task to execute
            invocation_state: Additional state/context passed to underlying agents.
                Defaults to None to avoid mutable default argument issues.
            **kwargs: Additional keyword arguments passed to underlying agents.

        Yields:
            Dictionary events containing multi-agent execution information including:
            - Multi-agent coordination events (node start/complete, handoffs)
            - Forwarded single-agent events with node context
            - Final result event
        """
        # Default implementation for backward compatibility
        # Execute invoke_async and yield the result as a single event
        result = await self.invoke_async(task, invocation_state, **kwargs)
        yield {"result": result}

    def __call__(
        self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
    ) -> MultiAgentResult:
        """Invoke synchronously.

        Args:
            task: The task to execute
            invocation_state: Additional state/context passed to underlying agents.
                Defaults to None to avoid mutable default argument issues.
            **kwargs: Additional keyword arguments passed to underlying agents.
        """
        if invocation_state is None:
            invocation_state = {}

        if kwargs:
            invocation_state.update(kwargs)
            warnings.warn("`**kwargs` parameter is deprecating, use `invocation_state` instead.", stacklevel=2)

        return run_async(lambda: self.invoke_async(task, invocation_state))

    def serialize_state(self) -> dict[str, Any]:
        """Return a JSON-serializable snapshot of the orchestrator state."""
        raise NotImplementedError

    def deserialize_state(self, payload: dict[str, Any]) -> None:
        """Restore orchestrator state from a session dict."""
        raise NotImplementedError

    def _parse_trace_attributes(
        self, attributes: Mapping[str, AttributeValue] | None = None
    ) -> dict[str, AttributeValue]:
        trace_attributes: dict[str, AttributeValue] = {}
        if attributes:
            for k, v in attributes.items():
                if isinstance(v, (str, int, float, bool)) or (
                    isinstance(v, list) and all(isinstance(x, (str, int, float, bool)) for x in v)
                ):
                    trace_attributes[k] = v
        return trace_attributes

`call(task, invocation_state=None, **kwargs)` ¶

Invoke synchronously.

Parameters:

Name	Type	Description	Default
`task`	`MultiAgentInput`	The task to execute	required
`invocation_state`	`dict[str, Any] \| None`	Additional state/context passed to underlying agents. Defaults to None to avoid mutable default argument issues.	`None`
`**kwargs`	`Any`	Additional keyword arguments passed to underlying agents.	`{}`

Source code in strands/multiagent/base.py

def __call__(
    self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
) -> MultiAgentResult:
    """Invoke synchronously.

    Args:
        task: The task to execute
        invocation_state: Additional state/context passed to underlying agents.
            Defaults to None to avoid mutable default argument issues.
        **kwargs: Additional keyword arguments passed to underlying agents.
    """
    if invocation_state is None:
        invocation_state = {}

    if kwargs:
        invocation_state.update(kwargs)
        warnings.warn("`**kwargs` parameter is deprecating, use `invocation_state` instead.", stacklevel=2)

    return run_async(lambda: self.invoke_async(task, invocation_state))

`deserialize_state(payload)` ¶

Restore orchestrator state from a session dict.

Source code in strands/multiagent/base.py

def deserialize_state(self, payload: dict[str, Any]) -> None:
    """Restore orchestrator state from a session dict."""
    raise NotImplementedError

`invoke_async(task, invocation_state=None, **kwargs)` `abstractmethod` `async` ¶

Invoke asynchronously.

Parameters:

Name	Type	Description	Default
`task`	`MultiAgentInput`	The task to execute	required
`invocation_state`	`dict[str, Any] \| None`	Additional state/context passed to underlying agents. Defaults to None to avoid mutable default argument issues.	`None`
`**kwargs`	`Any`	Additional keyword arguments passed to underlying agents.	`{}`

Source code in strands/multiagent/base.py

@abstractmethod
async def invoke_async(
    self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
) -> MultiAgentResult:
    """Invoke asynchronously.

    Args:
        task: The task to execute
        invocation_state: Additional state/context passed to underlying agents.
            Defaults to None to avoid mutable default argument issues.
        **kwargs: Additional keyword arguments passed to underlying agents.
    """
    raise NotImplementedError("invoke_async not implemented")

`serialize_state()` ¶

Return a JSON-serializable snapshot of the orchestrator state.

Source code in strands/multiagent/base.py

def serialize_state(self) -> dict[str, Any]:
    """Return a JSON-serializable snapshot of the orchestrator state."""
    raise NotImplementedError

`stream_async(task, invocation_state=None, **kwargs)` `async` ¶

Stream events during multi-agent execution.

Default implementation executes invoke_async and yields the result as a single event. Subclasses can override this method to provide true streaming capabilities.

Parameters:

Name	Type	Description	Default
`task`	`MultiAgentInput`	The task to execute	required
`invocation_state`	`dict[str, Any] \| None`	Additional state/context passed to underlying agents. Defaults to None to avoid mutable default argument issues.	`None`
`**kwargs`	`Any`	Additional keyword arguments passed to underlying agents.	`{}`

Yields:

Type	Description
`AsyncIterator[dict[str, Any]]`	Dictionary events containing multi-agent execution information including:
`AsyncIterator[dict[str, Any]]`	Multi-agent coordination events (node start/complete, handoffs)
`AsyncIterator[dict[str, Any]]`	Forwarded single-agent events with node context
`AsyncIterator[dict[str, Any]]`	Final result event

Source code in strands/multiagent/base.py

async def stream_async(
    self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
) -> AsyncIterator[dict[str, Any]]:
    """Stream events during multi-agent execution.

    Default implementation executes invoke_async and yields the result as a single event.
    Subclasses can override this method to provide true streaming capabilities.

    Args:
        task: The task to execute
        invocation_state: Additional state/context passed to underlying agents.
            Defaults to None to avoid mutable default argument issues.
        **kwargs: Additional keyword arguments passed to underlying agents.

    Yields:
        Dictionary events containing multi-agent execution information including:
        - Multi-agent coordination events (node start/complete, handoffs)
        - Forwarded single-agent events with node context
        - Final result event
    """
    # Default implementation for backward compatibility
    # Execute invoke_async and yield the result as a single event
    result = await self.invoke_async(task, invocation_state, **kwargs)
    yield {"result": result}

`MultiAgentHandoffEvent` ¶

Bases: TypedEvent

Event emitted during node transitions in multi-agent systems.

Supports both single handoffs (Swarm) and batch transitions (Graph). For Swarm: Single node-to-node handoffs with a message. For Graph: Batch transitions where multiple nodes complete and multiple nodes begin.

Source code in strands/types/_events.py

class MultiAgentHandoffEvent(TypedEvent):
    """Event emitted during node transitions in multi-agent systems.

    Supports both single handoffs (Swarm) and batch transitions (Graph).
    For Swarm: Single node-to-node handoffs with a message.
    For Graph: Batch transitions where multiple nodes complete and multiple nodes begin.
    """

    def __init__(
        self,
        from_node_ids: list[str],
        to_node_ids: list[str],
        message: str | None = None,
    ) -> None:
        """Initialize with handoff information.

        Args:
            from_node_ids: List of node ID(s) completing execution.
                - Swarm: Single-element list ["agent_a"]
                - Graph: Multi-element list ["node1", "node2"]
            to_node_ids: List of node ID(s) beginning execution.
                - Swarm: Single-element list ["agent_b"]
                - Graph: Multi-element list ["node3", "node4"]
            message: Optional message explaining the transition (typically used in Swarm)

        Examples:
            Swarm handoff: MultiAgentHandoffEvent(["researcher"], ["analyst"], "Need calculations")
            Graph batch: MultiAgentHandoffEvent(["node1", "node2"], ["node3", "node4"])
        """
        event_data = {
            "type": "multiagent_handoff",
            "from_node_ids": from_node_ids,
            "to_node_ids": to_node_ids,
        }

        if message is not None:
            event_data["message"] = message

        super().__init__(event_data)

`init(from_node_ids, to_node_ids, message=None)` ¶

Initialize with handoff information.

Parameters:

Name	Type	Description	Default
`from_node_ids`	`list[str]`	List of node ID(s) completing execution. - Swarm: Single-element list ["agent_a"] - Graph: Multi-element list ["node1", "node2"]	required
`to_node_ids`	`list[str]`	List of node ID(s) beginning execution. - Swarm: Single-element list ["agent_b"] - Graph: Multi-element list ["node3", "node4"]	required
`message`	`str \| None`	Optional message explaining the transition (typically used in Swarm)	`None`

Examples:

Swarm handoff: MultiAgentHandoffEvent(["researcher"], ["analyst"], "Need calculations") Graph batch: MultiAgentHandoffEvent(["node1", "node2"], ["node3", "node4"])

Source code in strands/types/_events.py

def __init__(
    self,
    from_node_ids: list[str],
    to_node_ids: list[str],
    message: str | None = None,
) -> None:
    """Initialize with handoff information.

    Args:
        from_node_ids: List of node ID(s) completing execution.
            - Swarm: Single-element list ["agent_a"]
            - Graph: Multi-element list ["node1", "node2"]
        to_node_ids: List of node ID(s) beginning execution.
            - Swarm: Single-element list ["agent_b"]
            - Graph: Multi-element list ["node3", "node4"]
        message: Optional message explaining the transition (typically used in Swarm)

    Examples:
        Swarm handoff: MultiAgentHandoffEvent(["researcher"], ["analyst"], "Need calculations")
        Graph batch: MultiAgentHandoffEvent(["node1", "node2"], ["node3", "node4"])
    """
    event_data = {
        "type": "multiagent_handoff",
        "from_node_ids": from_node_ids,
        "to_node_ids": to_node_ids,
    }

    if message is not None:
        event_data["message"] = message

    super().__init__(event_data)

`MultiAgentInitializedEvent` `dataclass` ¶

Bases: BaseHookEvent

Event triggered when multi-agent orchestrator initialized.

Attributes:

Name	Type	Description
`source`	`MultiAgentBase`	The multi-agent orchestrator instance
`invocation_state`	`dict[str, Any] \| None`	Configuration that user passes in

Source code in strands/experimental/hooks/multiagent/events.py

@dataclass
class MultiAgentInitializedEvent(BaseHookEvent):
    """Event triggered when multi-agent orchestrator initialized.

    Attributes:
        source: The multi-agent orchestrator instance
        invocation_state: Configuration that user passes in
    """

    source: "MultiAgentBase"
    invocation_state: dict[str, Any] | None = None

`MultiAgentNodeCancelEvent` ¶

Bases: TypedEvent

Event emitted when a user cancels node execution from their BeforeNodeCallEvent hook.

Source code in strands/types/_events.py

class MultiAgentNodeCancelEvent(TypedEvent):
    """Event emitted when a user cancels node execution from their BeforeNodeCallEvent hook."""

    def __init__(self, node_id: str, message: str) -> None:
        """Initialize with cancel message.

        Args:
            node_id: Unique identifier for the node.
            message: The node cancellation message.
        """
        super().__init__(
            {
                "type": "multiagent_node_cancel",
                "node_id": node_id,
                "message": message,
            }
        )

`init(node_id, message)` ¶

Initialize with cancel message.

Parameters:

Name	Type	Description	Default
`node_id`	`str`	Unique identifier for the node.	required
`message`	`str`	The node cancellation message.	required

Source code in strands/types/_events.py

def __init__(self, node_id: str, message: str) -> None:
    """Initialize with cancel message.

    Args:
        node_id: Unique identifier for the node.
        message: The node cancellation message.
    """
    super().__init__(
        {
            "type": "multiagent_node_cancel",
            "node_id": node_id,
            "message": message,
        }
    )

`MultiAgentNodeInterruptEvent` ¶

Bases: TypedEvent

Event emitted when a node is interrupted.

Source code in strands/types/_events.py

class MultiAgentNodeInterruptEvent(TypedEvent):
    """Event emitted when a node is interrupted."""

    def __init__(self, node_id: str, interrupts: list[Interrupt]) -> None:
        """Set interrupt in the event payload.

        Args:
            node_id: Unique identifier for the node generating the event.
            interrupts: Interrupts raised by user.
        """
        super().__init__(
            {
                "type": "multiagent_node_interrupt",
                "node_id": node_id,
                "interrupts": interrupts,
            }
        )

    @property
    def interrupts(self) -> list[Interrupt]:
        """The interrupt instances."""
        return cast(list[Interrupt], self["interrupts"])

`interrupts` `property` ¶

The interrupt instances.

`init(node_id, interrupts)` ¶

Set interrupt in the event payload.

Parameters:

Name	Type	Description	Default
`node_id`	`str`	Unique identifier for the node generating the event.	required
`interrupts`	`list[Interrupt]`	Interrupts raised by user.	required

Source code in strands/types/_events.py

def __init__(self, node_id: str, interrupts: list[Interrupt]) -> None:
    """Set interrupt in the event payload.

    Args:
        node_id: Unique identifier for the node generating the event.
        interrupts: Interrupts raised by user.
    """
    super().__init__(
        {
            "type": "multiagent_node_interrupt",
            "node_id": node_id,
            "interrupts": interrupts,
        }
    )

`MultiAgentNodeStartEvent` ¶

Bases: TypedEvent

Event emitted when a node begins execution in multi-agent context.

Source code in strands/types/_events.py

class MultiAgentNodeStartEvent(TypedEvent):
    """Event emitted when a node begins execution in multi-agent context."""

    def __init__(self, node_id: str, node_type: str) -> None:
        """Initialize with node information.

        Args:
            node_id: Unique identifier for the node
            node_type: Type of node ("agent", "swarm", "graph")
        """
        super().__init__({"type": "multiagent_node_start", "node_id": node_id, "node_type": node_type})

`init(node_id, node_type)` ¶

Initialize with node information.

Parameters:

Name	Type	Description	Default
`node_id`	`str`	Unique identifier for the node	required
`node_type`	`str`	Type of node ("agent", "swarm", "graph")	required

Source code in strands/types/_events.py

def __init__(self, node_id: str, node_type: str) -> None:
    """Initialize with node information.

    Args:
        node_id: Unique identifier for the node
        node_type: Type of node ("agent", "swarm", "graph")
    """
    super().__init__({"type": "multiagent_node_start", "node_id": node_id, "node_type": node_type})

`MultiAgentNodeStopEvent` ¶

Bases: TypedEvent

Event emitted when a node stops execution.

Similar to EventLoopStopEvent but for individual nodes in multi-agent orchestration. Provides the complete NodeResult which contains execution details, metrics, and status.

Source code in strands/types/_events.py

class MultiAgentNodeStopEvent(TypedEvent):
    """Event emitted when a node stops execution.

    Similar to EventLoopStopEvent but for individual nodes in multi-agent orchestration.
    Provides the complete NodeResult which contains execution details, metrics, and status.
    """

    def __init__(
        self,
        node_id: str,
        node_result: "NodeResult",
    ) -> None:
        """Initialize with stop information.

        Args:
            node_id: Unique identifier for the node
            node_result: Complete result from the node execution containing result,
                execution_time, status, accumulated_usage, accumulated_metrics, and execution_count
        """
        super().__init__(
            {
                "type": "multiagent_node_stop",
                "node_id": node_id,
                "node_result": node_result,
            }
        )

`init(node_id, node_result)` ¶

Initialize with stop information.

Parameters:

Name	Type	Description	Default
`node_id`	`str`	Unique identifier for the node	required
`node_result`	`NodeResult`	Complete result from the node execution containing result, execution_time, status, accumulated_usage, accumulated_metrics, and execution_count	required

Source code in strands/types/_events.py

def __init__(
    self,
    node_id: str,
    node_result: "NodeResult",
) -> None:
    """Initialize with stop information.

    Args:
        node_id: Unique identifier for the node
        node_result: Complete result from the node execution containing result,
            execution_time, status, accumulated_usage, accumulated_metrics, and execution_count
    """
    super().__init__(
        {
            "type": "multiagent_node_stop",
            "node_id": node_id,
            "node_result": node_result,
        }
    )

`MultiAgentNodeStreamEvent` ¶

Bases: TypedEvent

Event emitted during node execution - forwards agent events with node context.

Source code in strands/types/_events.py

class MultiAgentNodeStreamEvent(TypedEvent):
    """Event emitted during node execution - forwards agent events with node context."""

    def __init__(self, node_id: str, agent_event: dict[str, Any]) -> None:
        """Initialize with node context and agent event.

        Args:
            node_id: Unique identifier for the node generating the event
            agent_event: The original agent event data
        """
        super().__init__(
            {
                "type": "multiagent_node_stream",
                "node_id": node_id,
                "event": agent_event,  # Nest agent event to avoid field conflicts
            }
        )

`init(node_id, agent_event)` ¶

Initialize with node context and agent event.

Parameters:

Name	Type	Description	Default
`node_id`	`str`	Unique identifier for the node generating the event	required
`agent_event`	`dict[str, Any]`	The original agent event data	required

Source code in strands/types/_events.py

def __init__(self, node_id: str, agent_event: dict[str, Any]) -> None:
    """Initialize with node context and agent event.

    Args:
        node_id: Unique identifier for the node generating the event
        agent_event: The original agent event data
    """
    super().__init__(
        {
            "type": "multiagent_node_stream",
            "node_id": node_id,
            "event": agent_event,  # Nest agent event to avoid field conflicts
        }
    )

`MultiAgentResult` `dataclass` ¶

Result from multi-agent execution with accumulated metrics.

Source code in strands/multiagent/base.py

@dataclass
class MultiAgentResult:
    """Result from multi-agent execution with accumulated metrics."""

    status: Status = Status.PENDING
    results: dict[str, NodeResult] = field(default_factory=lambda: {})
    accumulated_usage: Usage = field(default_factory=lambda: Usage(inputTokens=0, outputTokens=0, totalTokens=0))
    accumulated_metrics: Metrics = field(default_factory=lambda: Metrics(latencyMs=0))
    execution_count: int = 0
    execution_time: int = 0
    interrupts: list[Interrupt] = field(default_factory=list)

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "MultiAgentResult":
        """Rehydrate a MultiAgentResult from persisted JSON."""
        if data.get("type") != "multiagent_result":
            raise TypeError(f"MultiAgentResult.from_dict: unexpected type {data.get('type')!r}")

        results = {k: NodeResult.from_dict(v) for k, v in data.get("results", {}).items()}
        usage = _parse_usage(data.get("accumulated_usage", {}))
        metrics = _parse_metrics(data.get("accumulated_metrics", {}))

        interrupts = []
        for interrupt_data in data.get("interrupts", []):
            interrupts.append(Interrupt(**interrupt_data))

        multiagent_result = cls(
            status=Status(data["status"]),
            results=results,
            accumulated_usage=usage,
            accumulated_metrics=metrics,
            execution_count=int(data.get("execution_count", 0)),
            execution_time=int(data.get("execution_time", 0)),
            interrupts=interrupts,
        )
        return multiagent_result

    def to_dict(self) -> dict[str, Any]:
        """Convert MultiAgentResult to JSON-serializable dict."""
        return {
            "type": "multiagent_result",
            "status": self.status.value,
            "results": {k: v.to_dict() for k, v in self.results.items()},
            "accumulated_usage": self.accumulated_usage,
            "accumulated_metrics": self.accumulated_metrics,
            "execution_count": self.execution_count,
            "execution_time": self.execution_time,
            "interrupts": [interrupt.to_dict() for interrupt in self.interrupts],
        }

`from_dict(data)` `classmethod` ¶

Rehydrate a MultiAgentResult from persisted JSON.

Source code in strands/multiagent/base.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "MultiAgentResult":
    """Rehydrate a MultiAgentResult from persisted JSON."""
    if data.get("type") != "multiagent_result":
        raise TypeError(f"MultiAgentResult.from_dict: unexpected type {data.get('type')!r}")

    results = {k: NodeResult.from_dict(v) for k, v in data.get("results", {}).items()}
    usage = _parse_usage(data.get("accumulated_usage", {}))
    metrics = _parse_metrics(data.get("accumulated_metrics", {}))

    interrupts = []
    for interrupt_data in data.get("interrupts", []):
        interrupts.append(Interrupt(**interrupt_data))

    multiagent_result = cls(
        status=Status(data["status"]),
        results=results,
        accumulated_usage=usage,
        accumulated_metrics=metrics,
        execution_count=int(data.get("execution_count", 0)),
        execution_time=int(data.get("execution_time", 0)),
        interrupts=interrupts,
    )
    return multiagent_result

`to_dict()` ¶

Convert MultiAgentResult to JSON-serializable dict.

Source code in strands/multiagent/base.py

def to_dict(self) -> dict[str, Any]:
    """Convert MultiAgentResult to JSON-serializable dict."""
    return {
        "type": "multiagent_result",
        "status": self.status.value,
        "results": {k: v.to_dict() for k, v in self.results.items()},
        "accumulated_usage": self.accumulated_usage,
        "accumulated_metrics": self.accumulated_metrics,
        "execution_count": self.execution_count,
        "execution_time": self.execution_time,
        "interrupts": [interrupt.to_dict() for interrupt in self.interrupts],
    }

`MultiAgentResultEvent` ¶

Bases: TypedEvent

Event emitted when multi-agent execution completes with final result.

Source code in strands/types/_events.py

class MultiAgentResultEvent(TypedEvent):
    """Event emitted when multi-agent execution completes with final result."""

    def __init__(self, result: "MultiAgentResult") -> None:
        """Initialize with multi-agent result.

        Args:
            result: The final result from multi-agent execution (SwarmResult, GraphResult, etc.)
        """
        super().__init__({"type": "multiagent_result", "result": result})

`init(result)` ¶

Initialize with multi-agent result.

Parameters:

Name	Type	Description	Default
`result`	`MultiAgentResult`	The final result from multi-agent execution (SwarmResult, GraphResult, etc.)	required

Source code in strands/types/_events.py

def __init__(self, result: "MultiAgentResult") -> None:
    """Initialize with multi-agent result.

    Args:
        result: The final result from multi-agent execution (SwarmResult, GraphResult, etc.)
    """
    super().__init__({"type": "multiagent_result", "result": result})

`NodeResult` `dataclass` ¶

Unified result from node execution - handles both Agent and nested MultiAgentBase results.

Source code in strands/multiagent/base.py

@dataclass
class NodeResult:
    """Unified result from node execution - handles both Agent and nested MultiAgentBase results."""

    # Core result data - single AgentResult, nested MultiAgentResult, or Exception
    result: Union[AgentResult, "MultiAgentResult", Exception]

    # Execution metadata
    execution_time: int = 0
    status: Status = Status.PENDING

    # Accumulated metrics from this node and all children
    accumulated_usage: Usage = field(default_factory=lambda: Usage(inputTokens=0, outputTokens=0, totalTokens=0))
    accumulated_metrics: Metrics = field(default_factory=lambda: Metrics(latencyMs=0))
    execution_count: int = 0
    interrupts: list[Interrupt] = field(default_factory=list)

    def get_agent_results(self) -> list[AgentResult]:
        """Get all AgentResult objects from this node, flattened if nested."""
        if isinstance(self.result, Exception):
            return []  # No agent results for exceptions
        elif isinstance(self.result, AgentResult):
            return [self.result]
        else:
            # Flatten nested results from MultiAgentResult
            flattened = []
            for nested_node_result in self.result.results.values():
                flattened.extend(nested_node_result.get_agent_results())
            return flattened

    def to_dict(self) -> dict[str, Any]:
        """Convert NodeResult to JSON-serializable dict, ignoring state field."""
        if isinstance(self.result, Exception):
            result_data: dict[str, Any] = {"type": "exception", "message": str(self.result)}
        elif isinstance(self.result, AgentResult):
            result_data = self.result.to_dict()
        else:
            # MultiAgentResult case
            result_data = self.result.to_dict()

        return {
            "result": result_data,
            "execution_time": self.execution_time,
            "status": self.status.value,
            "accumulated_usage": self.accumulated_usage,
            "accumulated_metrics": self.accumulated_metrics,
            "execution_count": self.execution_count,
            "interrupts": [interrupt.to_dict() for interrupt in self.interrupts],
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "NodeResult":
        """Rehydrate a NodeResult from persisted JSON."""
        if "result" not in data:
            raise TypeError("NodeResult.from_dict: missing 'result'")
        raw = data["result"]

        result: Union[AgentResult, "MultiAgentResult", Exception]
        if isinstance(raw, dict) and raw.get("type") == "agent_result":
            result = AgentResult.from_dict(raw)
        elif isinstance(raw, dict) and raw.get("type") == "exception":
            result = Exception(str(raw.get("message", "node failed")))
        elif isinstance(raw, dict) and raw.get("type") == "multiagent_result":
            result = MultiAgentResult.from_dict(raw)
        else:
            raise TypeError(f"NodeResult.from_dict: unsupported result payload: {raw!r}")

        usage = _parse_usage(data.get("accumulated_usage", {}))
        metrics = _parse_metrics(data.get("accumulated_metrics", {}))

        interrupts = []
        for interrupt_data in data.get("interrupts", []):
            interrupts.append(Interrupt(**interrupt_data))

        return cls(
            result=result,
            execution_time=int(data.get("execution_time", 0)),
            status=Status(data.get("status", "pending")),
            accumulated_usage=usage,
            accumulated_metrics=metrics,
            execution_count=int(data.get("execution_count", 0)),
            interrupts=interrupts,
        )

`from_dict(data)` `classmethod` ¶

Rehydrate a NodeResult from persisted JSON.

Source code in strands/multiagent/base.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "NodeResult":
    """Rehydrate a NodeResult from persisted JSON."""
    if "result" not in data:
        raise TypeError("NodeResult.from_dict: missing 'result'")
    raw = data["result"]

    result: Union[AgentResult, "MultiAgentResult", Exception]
    if isinstance(raw, dict) and raw.get("type") == "agent_result":
        result = AgentResult.from_dict(raw)
    elif isinstance(raw, dict) and raw.get("type") == "exception":
        result = Exception(str(raw.get("message", "node failed")))
    elif isinstance(raw, dict) and raw.get("type") == "multiagent_result":
        result = MultiAgentResult.from_dict(raw)
    else:
        raise TypeError(f"NodeResult.from_dict: unsupported result payload: {raw!r}")

    usage = _parse_usage(data.get("accumulated_usage", {}))
    metrics = _parse_metrics(data.get("accumulated_metrics", {}))

    interrupts = []
    for interrupt_data in data.get("interrupts", []):
        interrupts.append(Interrupt(**interrupt_data))

    return cls(
        result=result,
        execution_time=int(data.get("execution_time", 0)),
        status=Status(data.get("status", "pending")),
        accumulated_usage=usage,
        accumulated_metrics=metrics,
        execution_count=int(data.get("execution_count", 0)),
        interrupts=interrupts,
    )

`get_agent_results()` ¶

Get all AgentResult objects from this node, flattened if nested.

Source code in strands/multiagent/base.py

def get_agent_results(self) -> list[AgentResult]:
    """Get all AgentResult objects from this node, flattened if nested."""
    if isinstance(self.result, Exception):
        return []  # No agent results for exceptions
    elif isinstance(self.result, AgentResult):
        return [self.result]
    else:
        # Flatten nested results from MultiAgentResult
        flattened = []
        for nested_node_result in self.result.results.values():
            flattened.extend(nested_node_result.get_agent_results())
        return flattened

`to_dict()` ¶

Convert NodeResult to JSON-serializable dict, ignoring state field.

Source code in strands/multiagent/base.py

def to_dict(self) -> dict[str, Any]:
    """Convert NodeResult to JSON-serializable dict, ignoring state field."""
    if isinstance(self.result, Exception):
        result_data: dict[str, Any] = {"type": "exception", "message": str(self.result)}
    elif isinstance(self.result, AgentResult):
        result_data = self.result.to_dict()
    else:
        # MultiAgentResult case
        result_data = self.result.to_dict()

    return {
        "result": result_data,
        "execution_time": self.execution_time,
        "status": self.status.value,
        "accumulated_usage": self.accumulated_usage,
        "accumulated_metrics": self.accumulated_metrics,
        "execution_count": self.execution_count,
        "interrupts": [interrupt.to_dict() for interrupt in self.interrupts],
    }

`SessionManager` ¶

Bases: HookProvider, ABC

Abstract interface for managing sessions.

A session manager is in charge of persisting the conversation and state of an agent across its interaction. Changes made to the agents conversation, state, or other attributes should be persisted immediately after they are changed. The different methods introduced in this class are called at important lifecycle events for an agent, and should be persisted in the session.

Source code in strands/session/session_manager.py

class SessionManager(HookProvider, ABC):
    """Abstract interface for managing sessions.

    A session manager is in charge of persisting the conversation and state of an agent across its interaction.
    Changes made to the agents conversation, state, or other attributes should be persisted immediately after
    they are changed. The different methods introduced in this class are called at important lifecycle events
    for an agent, and should be persisted in the session.
    """

    def register_hooks(self, registry: HookRegistry, **kwargs: Any) -> None:
        """Register hooks for persisting the agent to the session."""
        # After the normal Agent initialization behavior, call the session initialize function to restore the agent
        registry.add_callback(AgentInitializedEvent, lambda event: self.initialize(event.agent))

        # For each message appended to the Agents messages, store that message in the session
        registry.add_callback(MessageAddedEvent, lambda event: self.append_message(event.message, event.agent))

        # Sync the agent into the session for each message in case the agent state was updated
        registry.add_callback(MessageAddedEvent, lambda event: self.sync_agent(event.agent))

        # After an agent was invoked, sync it with the session to capture any conversation manager state updates
        registry.add_callback(AfterInvocationEvent, lambda event: self.sync_agent(event.agent))

        registry.add_callback(MultiAgentInitializedEvent, lambda event: self.initialize_multi_agent(event.source))
        registry.add_callback(AfterNodeCallEvent, lambda event: self.sync_multi_agent(event.source))
        registry.add_callback(AfterMultiAgentInvocationEvent, lambda event: self.sync_multi_agent(event.source))

        # Register BidiAgent hooks
        registry.add_callback(BidiAgentInitializedEvent, lambda event: self.initialize_bidi_agent(event.agent))
        registry.add_callback(BidiMessageAddedEvent, lambda event: self.append_bidi_message(event.message, event.agent))
        registry.add_callback(BidiMessageAddedEvent, lambda event: self.sync_bidi_agent(event.agent))
        registry.add_callback(BidiAfterInvocationEvent, lambda event: self.sync_bidi_agent(event.agent))

    @abstractmethod
    def redact_latest_message(self, redact_message: Message, agent: "Agent", **kwargs: Any) -> None:
        """Redact the message most recently appended to the agent in the session.

        Args:
            redact_message: New message to use that contains the redact content
            agent: Agent to apply the message redaction to
            **kwargs: Additional keyword arguments for future extensibility.
        """

    @abstractmethod
    def append_message(self, message: Message, agent: "Agent", **kwargs: Any) -> None:
        """Append a message to the agent's session.

        Args:
            message: Message to add to the agent in the session
            agent: Agent to append the message to
            **kwargs: Additional keyword arguments for future extensibility.
        """

    @abstractmethod
    def sync_agent(self, agent: "Agent", **kwargs: Any) -> None:
        """Serialize and sync the agent with the session storage.

        Args:
            agent: Agent who should be synchronized with the session storage
            **kwargs: Additional keyword arguments for future extensibility.
        """

    @abstractmethod
    def initialize(self, agent: "Agent", **kwargs: Any) -> None:
        """Initialize an agent with a session.

        Args:
            agent: Agent to initialize
            **kwargs: Additional keyword arguments for future extensibility.
        """

    def sync_multi_agent(self, source: "MultiAgentBase", **kwargs: Any) -> None:
        """Serialize and sync multi-agent with the session storage.

        Args:
            source: Multi-agent source object to persist
            **kwargs: Additional keyword arguments for future extensibility.
        """
        raise NotImplementedError(
            f"{self.__class__.__name__} does not support multi-agent persistence "
            "(sync_multi_agent). Provide an implementation or use a "
            "SessionManager with session_type=SessionType.MULTI_AGENT."
        )

    def initialize_multi_agent(self, source: "MultiAgentBase", **kwargs: Any) -> None:
        """Read multi-agent state from persistent storage.

        Args:
            **kwargs: Additional keyword arguments for future extensibility.
            source: Multi-agent state to initialize.

        Returns:
            Multi-agent state dictionary or empty dict if not found.

        """
        raise NotImplementedError(
            f"{self.__class__.__name__} does not support multi-agent persistence "
            "(initialize_multi_agent). Provide an implementation or use a "
            "SessionManager with session_type=SessionType.MULTI_AGENT."
        )

    def initialize_bidi_agent(self, agent: "BidiAgent", **kwargs: Any) -> None:
        """Initialize a bidirectional agent with a session.

        Args:
            agent: BidiAgent to initialize
            **kwargs: Additional keyword arguments for future extensibility.
        """
        raise NotImplementedError(
            f"{self.__class__.__name__} does not support bidirectional agent persistence "
            "(initialize_bidi_agent). Provide an implementation or use a "
            "SessionManager with bidirectional agent support."
        )

    def append_bidi_message(self, message: Message, agent: "BidiAgent", **kwargs: Any) -> None:
        """Append a message to the bidirectional agent's session.

        Args:
            message: Message to add to the agent in the session
            agent: BidiAgent to append the message to
            **kwargs: Additional keyword arguments for future extensibility.
        """
        raise NotImplementedError(
            f"{self.__class__.__name__} does not support bidirectional agent persistence "
            "(append_bidi_message). Provide an implementation or use a "
            "SessionManager with bidirectional agent support."
        )

    def sync_bidi_agent(self, agent: "BidiAgent", **kwargs: Any) -> None:
        """Serialize and sync the bidirectional agent with the session storage.

        Args:
            agent: BidiAgent who should be synchronized with the session storage
            **kwargs: Additional keyword arguments for future extensibility.
        """
        raise NotImplementedError(
            f"{self.__class__.__name__} does not support bidirectional agent persistence "
            "(sync_bidi_agent). Provide an implementation or use a "
            "SessionManager with bidirectional agent support."
        )

`append_bidi_message(message, agent, **kwargs)` ¶

Append a message to the bidirectional agent's session.

Parameters:

Name	Type	Description	Default
`message`	`Message`	Message to add to the agent in the session	required
`agent`	`BidiAgent`	BidiAgent to append the message to	required
`**kwargs`	`Any`	Additional keyword arguments for future extensibility.	`{}`

Source code in strands/session/session_manager.py

def append_bidi_message(self, message: Message, agent: "BidiAgent", **kwargs: Any) -> None:
    """Append a message to the bidirectional agent's session.

    Args:
        message: Message to add to the agent in the session
        agent: BidiAgent to append the message to
        **kwargs: Additional keyword arguments for future extensibility.
    """
    raise NotImplementedError(
        f"{self.__class__.__name__} does not support bidirectional agent persistence "
        "(append_bidi_message). Provide an implementation or use a "
        "SessionManager with bidirectional agent support."
    )

`append_message(message, agent, **kwargs)` `abstractmethod` ¶

Append a message to the agent's session.

Parameters:

Name	Type	Description	Default
`message`	`Message`	Message to add to the agent in the session	required
`agent`	`Agent`	Agent to append the message to	required
`**kwargs`	`Any`	Additional keyword arguments for future extensibility.	`{}`

Source code in strands/session/session_manager.py

@abstractmethod
def append_message(self, message: Message, agent: "Agent", **kwargs: Any) -> None:
    """Append a message to the agent's session.

    Args:
        message: Message to add to the agent in the session
        agent: Agent to append the message to
        **kwargs: Additional keyword arguments for future extensibility.
    """

`initialize(agent, **kwargs)` `abstractmethod` ¶

Initialize an agent with a session.

Parameters:

Name	Type	Description	Default
`agent`	`Agent`	Agent to initialize	required
`**kwargs`	`Any`	Additional keyword arguments for future extensibility.	`{}`

Source code in strands/session/session_manager.py

@abstractmethod
def initialize(self, agent: "Agent", **kwargs: Any) -> None:
    """Initialize an agent with a session.

    Args:
        agent: Agent to initialize
        **kwargs: Additional keyword arguments for future extensibility.
    """

`initialize_bidi_agent(agent, **kwargs)` ¶

Initialize a bidirectional agent with a session.

Parameters:

Name	Type	Description	Default
`agent`	`BidiAgent`	BidiAgent to initialize	required
`**kwargs`	`Any`	Additional keyword arguments for future extensibility.	`{}`

Source code in strands/session/session_manager.py

def initialize_bidi_agent(self, agent: "BidiAgent", **kwargs: Any) -> None:
    """Initialize a bidirectional agent with a session.

    Args:
        agent: BidiAgent to initialize
        **kwargs: Additional keyword arguments for future extensibility.
    """
    raise NotImplementedError(
        f"{self.__class__.__name__} does not support bidirectional agent persistence "
        "(initialize_bidi_agent). Provide an implementation or use a "
        "SessionManager with bidirectional agent support."
    )

`initialize_multi_agent(source, **kwargs)` ¶

Read multi-agent state from persistent storage.

Parameters:

Name	Type	Description	Default
`**kwargs`	`Any`	Additional keyword arguments for future extensibility.	`{}`
`source`	`MultiAgentBase`	Multi-agent state to initialize.	required

Returns:

Type	Description
`None`	Multi-agent state dictionary or empty dict if not found.

Source code in strands/session/session_manager.py

def initialize_multi_agent(self, source: "MultiAgentBase", **kwargs: Any) -> None:
    """Read multi-agent state from persistent storage.

    Args:
        **kwargs: Additional keyword arguments for future extensibility.
        source: Multi-agent state to initialize.

    Returns:
        Multi-agent state dictionary or empty dict if not found.

    """
    raise NotImplementedError(
        f"{self.__class__.__name__} does not support multi-agent persistence "
        "(initialize_multi_agent). Provide an implementation or use a "
        "SessionManager with session_type=SessionType.MULTI_AGENT."
    )

`redact_latest_message(redact_message, agent, **kwargs)` `abstractmethod` ¶

Redact the message most recently appended to the agent in the session.

Parameters:

Name	Type	Description	Default
`redact_message`	`Message`	New message to use that contains the redact content	required
`agent`	`Agent`	Agent to apply the message redaction to	required
`**kwargs`	`Any`	Additional keyword arguments for future extensibility.	`{}`

Source code in strands/session/session_manager.py

@abstractmethod
def redact_latest_message(self, redact_message: Message, agent: "Agent", **kwargs: Any) -> None:
    """Redact the message most recently appended to the agent in the session.

    Args:
        redact_message: New message to use that contains the redact content
        agent: Agent to apply the message redaction to
        **kwargs: Additional keyword arguments for future extensibility.
    """

`register_hooks(registry, **kwargs)` ¶

Register hooks for persisting the agent to the session.

Source code in strands/session/session_manager.py

def register_hooks(self, registry: HookRegistry, **kwargs: Any) -> None:
    """Register hooks for persisting the agent to the session."""
    # After the normal Agent initialization behavior, call the session initialize function to restore the agent
    registry.add_callback(AgentInitializedEvent, lambda event: self.initialize(event.agent))

    # For each message appended to the Agents messages, store that message in the session
    registry.add_callback(MessageAddedEvent, lambda event: self.append_message(event.message, event.agent))

    # Sync the agent into the session for each message in case the agent state was updated
    registry.add_callback(MessageAddedEvent, lambda event: self.sync_agent(event.agent))

    # After an agent was invoked, sync it with the session to capture any conversation manager state updates
    registry.add_callback(AfterInvocationEvent, lambda event: self.sync_agent(event.agent))

    registry.add_callback(MultiAgentInitializedEvent, lambda event: self.initialize_multi_agent(event.source))
    registry.add_callback(AfterNodeCallEvent, lambda event: self.sync_multi_agent(event.source))
    registry.add_callback(AfterMultiAgentInvocationEvent, lambda event: self.sync_multi_agent(event.source))

    # Register BidiAgent hooks
    registry.add_callback(BidiAgentInitializedEvent, lambda event: self.initialize_bidi_agent(event.agent))
    registry.add_callback(BidiMessageAddedEvent, lambda event: self.append_bidi_message(event.message, event.agent))
    registry.add_callback(BidiMessageAddedEvent, lambda event: self.sync_bidi_agent(event.agent))
    registry.add_callback(BidiAfterInvocationEvent, lambda event: self.sync_bidi_agent(event.agent))

`sync_agent(agent, **kwargs)` `abstractmethod` ¶

Serialize and sync the agent with the session storage.

Parameters:

Name	Type	Description	Default
`agent`	`Agent`	Agent who should be synchronized with the session storage	required
`**kwargs`	`Any`	Additional keyword arguments for future extensibility.	`{}`

Source code in strands/session/session_manager.py

@abstractmethod
def sync_agent(self, agent: "Agent", **kwargs: Any) -> None:
    """Serialize and sync the agent with the session storage.

    Args:
        agent: Agent who should be synchronized with the session storage
        **kwargs: Additional keyword arguments for future extensibility.
    """

`sync_bidi_agent(agent, **kwargs)` ¶

Serialize and sync the bidirectional agent with the session storage.

Parameters:

Name	Type	Description	Default
`agent`	`BidiAgent`	BidiAgent who should be synchronized with the session storage	required
`**kwargs`	`Any`	Additional keyword arguments for future extensibility.	`{}`

Source code in strands/session/session_manager.py

def sync_bidi_agent(self, agent: "BidiAgent", **kwargs: Any) -> None:
    """Serialize and sync the bidirectional agent with the session storage.

    Args:
        agent: BidiAgent who should be synchronized with the session storage
        **kwargs: Additional keyword arguments for future extensibility.
    """
    raise NotImplementedError(
        f"{self.__class__.__name__} does not support bidirectional agent persistence "
        "(sync_bidi_agent). Provide an implementation or use a "
        "SessionManager with bidirectional agent support."
    )

`sync_multi_agent(source, **kwargs)` ¶

Serialize and sync multi-agent with the session storage.

Parameters:

Name	Type	Description	Default
`source`	`MultiAgentBase`	Multi-agent source object to persist	required
`**kwargs`	`Any`	Additional keyword arguments for future extensibility.	`{}`

Source code in strands/session/session_manager.py

def sync_multi_agent(self, source: "MultiAgentBase", **kwargs: Any) -> None:
    """Serialize and sync multi-agent with the session storage.

    Args:
        source: Multi-agent source object to persist
        **kwargs: Additional keyword arguments for future extensibility.
    """
    raise NotImplementedError(
        f"{self.__class__.__name__} does not support multi-agent persistence "
        "(sync_multi_agent). Provide an implementation or use a "
        "SessionManager with session_type=SessionType.MULTI_AGENT."
    )

`SharedContext` `dataclass` ¶

Shared context between swarm nodes.

Source code in strands/multiagent/swarm.py

@dataclass
class SharedContext:
    """Shared context between swarm nodes."""

    context: dict[str, dict[str, Any]] = field(default_factory=dict)

    def add_context(self, node: SwarmNode, key: str, value: Any) -> None:
        """Add context."""
        self._validate_key(key)
        self._validate_json_serializable(value)

        if node.node_id not in self.context:
            self.context[node.node_id] = {}
        self.context[node.node_id][key] = value

    def _validate_key(self, key: str) -> None:
        """Validate that a key is valid.

        Args:
            key: The key to validate

        Raises:
            ValueError: If key is invalid
        """
        if key is None:
            raise ValueError("Key cannot be None")
        if not isinstance(key, str):
            raise ValueError("Key must be a string")
        if not key.strip():
            raise ValueError("Key cannot be empty")

    def _validate_json_serializable(self, value: Any) -> None:
        """Validate that a value is JSON serializable.

        Args:
            value: The value to validate

        Raises:
            ValueError: If value is not JSON serializable
        """
        try:
            json.dumps(value)
        except (TypeError, ValueError) as e:
            raise ValueError(
                f"Value is not JSON serializable: {type(value).__name__}. "
                f"Only JSON-compatible types (str, int, float, bool, list, dict, None) are allowed."
            ) from e

`add_context(node, key, value)` ¶

Add context.

Source code in strands/multiagent/swarm.py

def add_context(self, node: SwarmNode, key: str, value: Any) -> None:
    """Add context."""
    self._validate_key(key)
    self._validate_json_serializable(value)

    if node.node_id not in self.context:
        self.context[node.node_id] = {}
    self.context[node.node_id][key] = value

`Status` ¶

Bases: Enum

Execution status for both graphs and nodes.

Attributes:

Name	Type	Description
`PENDING`		Task has not started execution yet.
`EXECUTING`		Task is currently running.
`COMPLETED`		Task finished successfully.
`FAILED`		Task encountered an error and could not complete.
`INTERRUPTED`		Task was interrupted by user.

Source code in strands/multiagent/base.py

class Status(Enum):
    """Execution status for both graphs and nodes.

    Attributes:
        PENDING: Task has not started execution yet.
        EXECUTING: Task is currently running.
        COMPLETED: Task finished successfully.
        FAILED: Task encountered an error and could not complete.
        INTERRUPTED: Task was interrupted by user.
    """

    PENDING = "pending"
    EXECUTING = "executing"
    COMPLETED = "completed"
    FAILED = "failed"
    INTERRUPTED = "interrupted"

`Swarm` ¶

Bases: MultiAgentBase

Self-organizing collaborative agent teams with shared working memory.

Source code in strands/multiagent/swarm.py

class Swarm(MultiAgentBase):
    """Self-organizing collaborative agent teams with shared working memory."""

    def __init__(
        self,
        nodes: list[Agent],
        *,
        entry_point: Agent | None = None,
        max_handoffs: int = 20,
        max_iterations: int = 20,
        execution_timeout: float = 900.0,
        node_timeout: float = 300.0,
        repetitive_handoff_detection_window: int = 0,
        repetitive_handoff_min_unique_agents: int = 0,
        session_manager: Optional[SessionManager] = None,
        hooks: Optional[list[HookProvider]] = None,
        id: str = _DEFAULT_SWARM_ID,
        trace_attributes: Optional[Mapping[str, AttributeValue]] = None,
    ) -> None:
        """Initialize Swarm with agents and configuration.

        Args:
            id: Unique swarm id (default: "default_swarm")
            nodes: List of nodes (e.g. Agent) to include in the swarm
            entry_point: Agent to start with. If None, uses the first agent (default: None)
            max_handoffs: Maximum handoffs to agents and users (default: 20)
            max_iterations: Maximum node executions within the swarm (default: 20)
            execution_timeout: Total execution timeout in seconds (default: 900.0)
            node_timeout: Individual node timeout in seconds (default: 300.0)
            repetitive_handoff_detection_window: Number of recent nodes to check for repetitive handoffs
                Disabled by default (default: 0)
            repetitive_handoff_min_unique_agents: Minimum unique agents required in recent sequence
                Disabled by default (default: 0)
            session_manager: Session manager for persisting graph state and execution history (default: None)
            hooks: List of hook providers for monitoring and extending graph execution behavior (default: None)
            trace_attributes: Custom trace attributes to apply to the agent's trace span (default: None)
        """
        super().__init__()
        self.id = id
        self.entry_point = entry_point
        self.max_handoffs = max_handoffs
        self.max_iterations = max_iterations
        self.execution_timeout = execution_timeout
        self.node_timeout = node_timeout
        self.repetitive_handoff_detection_window = repetitive_handoff_detection_window
        self.repetitive_handoff_min_unique_agents = repetitive_handoff_min_unique_agents

        self.shared_context = SharedContext()
        self.nodes: dict[str, SwarmNode] = {}

        self.state = SwarmState(
            current_node=None,  # Placeholder, will be set properly
            task="",
            completion_status=Status.PENDING,
        )
        self._interrupt_state = _InterruptState()

        self.tracer = get_tracer()
        self.trace_attributes: dict[str, AttributeValue] = self._parse_trace_attributes(trace_attributes)

        self.session_manager = session_manager
        self.hooks = HookRegistry()
        if hooks:
            for hook in hooks:
                self.hooks.add_hook(hook)
        if self.session_manager:
            self.hooks.add_hook(self.session_manager)

        self._resume_from_session = False

        self._setup_swarm(nodes)
        self._inject_swarm_tools()
        run_async(lambda: self.hooks.invoke_callbacks_async(MultiAgentInitializedEvent(self)))

    def __call__(
        self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
    ) -> SwarmResult:
        """Invoke the swarm synchronously.

        Args:
            task: The task to execute
            invocation_state: Additional state/context passed to underlying agents.
                Defaults to None to avoid mutable default argument issues.
            **kwargs: Keyword arguments allowing backward compatible future changes.
        """
        if invocation_state is None:
            invocation_state = {}
        return run_async(lambda: self.invoke_async(task, invocation_state))

    async def invoke_async(
        self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
    ) -> SwarmResult:
        """Invoke the swarm asynchronously.

        This method uses stream_async internally and consumes all events until completion,
        following the same pattern as the Agent class.

        Args:
            task: The task to execute
            invocation_state: Additional state/context passed to underlying agents.
                Defaults to None to avoid mutable default argument issues.
            **kwargs: Keyword arguments allowing backward compatible future changes.
        """
        events = self.stream_async(task, invocation_state, **kwargs)
        final_event = None
        async for event in events:
            final_event = event

        if final_event is None or "result" not in final_event:
            raise ValueError("Swarm streaming completed without producing a result event")

        return cast(SwarmResult, final_event["result"])

    async def stream_async(
        self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
    ) -> AsyncIterator[dict[str, Any]]:
        """Stream events during swarm execution.

        Args:
            task: The task to execute
            invocation_state: Additional state/context passed to underlying agents.
                Defaults to None to avoid mutable default argument issues.
            **kwargs: Keyword arguments allowing backward compatible future changes.

        Yields:
            Dictionary events during swarm execution, such as:
            - multi_agent_node_start: When a node begins execution
            - multi_agent_node_stream: Forwarded agent events with node context
            - multi_agent_handoff: When control is handed off between agents
            - multi_agent_node_stop: When a node stops execution
            - result: Final swarm result
        """
        self._interrupt_state.resume(task)

        if invocation_state is None:
            invocation_state = {}

        await self.hooks.invoke_callbacks_async(BeforeMultiAgentInvocationEvent(self, invocation_state))

        logger.debug("starting swarm execution")

        if self._resume_from_session or self._interrupt_state.activated:
            self.state.completion_status = Status.EXECUTING
            self.state.start_time = time.time()
        else:
            # Initialize swarm state with configuration
            initial_node = self._initial_node()

            self.state = SwarmState(
                current_node=initial_node,
                task=task,
                completion_status=Status.EXECUTING,
                shared_context=self.shared_context,
            )

        span = self.tracer.start_multiagent_span(task, "swarm", custom_trace_attributes=self.trace_attributes)
        with trace_api.use_span(span, end_on_exit=True):
            interrupts = []

            try:
                current_node = cast(SwarmNode, self.state.current_node)
                logger.debug("current_node=<%s> | starting swarm execution with node", current_node.node_id)
                logger.debug(
                    "max_handoffs=<%d>, max_iterations=<%d>, timeout=<%s>s | swarm execution config",
                    self.max_handoffs,
                    self.max_iterations,
                    self.execution_timeout,
                )

                async for event in self._execute_swarm(invocation_state):
                    if isinstance(event, MultiAgentNodeInterruptEvent):
                        interrupts = event.interrupts

                    yield event.as_dict()

            except Exception:
                logger.exception("swarm execution failed")
                self.state.completion_status = Status.FAILED
                raise
            finally:
                self.state.execution_time = round((time.time() - self.state.start_time) * 1000)
                await self.hooks.invoke_callbacks_async(AfterMultiAgentInvocationEvent(self, invocation_state))
                self._resume_from_session = False

            # Yield final result after execution_time is set
            result = self._build_result(interrupts)
            yield MultiAgentResultEvent(result=result).as_dict()

    async def _stream_with_timeout(
        self, async_generator: AsyncIterator[Any], timeout: float | None, timeout_message: str
    ) -> AsyncIterator[Any]:
        """Wrap an async generator with timeout for total execution time.

        Tracks elapsed time from start and enforces timeout across all events.
        Each event wait uses remaining time from the total timeout budget.

        Args:
            async_generator: The generator to wrap
            timeout: Total timeout in seconds for entire stream, or None for no timeout
            timeout_message: Message to include in timeout exception

        Yields:
            Events from the wrapped generator as they arrive

        Raises:
            Exception: If total execution time exceeds timeout
        """
        if timeout is None:
            # No timeout - just pass through
            async for event in async_generator:
                yield event
        else:
            # Track start time for total timeout
            start_time = asyncio.get_event_loop().time()

            while True:
                # Calculate remaining time from total timeout budget
                elapsed = asyncio.get_event_loop().time() - start_time
                remaining = timeout - elapsed

                if remaining <= 0:
                    raise Exception(timeout_message)

                try:
                    event = await asyncio.wait_for(async_generator.__anext__(), timeout=remaining)
                    yield event
                except StopAsyncIteration:
                    break
                except asyncio.TimeoutError as err:
                    raise Exception(timeout_message) from err

    def _setup_swarm(self, nodes: list[Agent]) -> None:
        """Initialize swarm configuration."""
        # Validate nodes before setup
        self._validate_swarm(nodes)

        # Validate agents have names and create SwarmNode objects
        for i, node in enumerate(nodes):
            if not node.name:
                node_id = f"node_{i}"
                node.name = node_id
                logger.debug("node_id=<%s> | agent has no name, dynamically generating one", node_id)

            node_id = str(node.name)

            # Ensure node IDs are unique
            if node_id in self.nodes:
                raise ValueError(f"Node ID '{node_id}' is not unique. Each agent must have a unique name.")

            self.nodes[node_id] = SwarmNode(node_id, node, swarm=self)

        # Validate entry point if specified
        if self.entry_point is not None:
            entry_point_node_id = str(self.entry_point.name)
            if (
                entry_point_node_id not in self.nodes
                or self.nodes[entry_point_node_id].executor is not self.entry_point
            ):
                available_agents = [
                    f"{node_id} ({type(node.executor).__name__})" for node_id, node in self.nodes.items()
                ]
                raise ValueError(f"Entry point agent not found in swarm nodes. Available agents: {available_agents}")

        swarm_nodes = list(self.nodes.values())
        logger.debug("nodes=<%s> | initialized swarm with nodes", [node.node_id for node in swarm_nodes])

        if self.entry_point:
            entry_point_name = getattr(self.entry_point, "name", "unnamed_agent")
            logger.debug("entry_point=<%s> | configured entry point", entry_point_name)
        else:
            first_node = next(iter(self.nodes.keys()))
            logger.debug("entry_point=<%s> | using first node as entry point", first_node)

    def _validate_swarm(self, nodes: list[Agent]) -> None:
        """Validate swarm structure and nodes."""
        # Check for duplicate object instances
        seen_instances = set()
        for node in nodes:
            if id(node) in seen_instances:
                raise ValueError("Duplicate node instance detected. Each node must have a unique object instance.")
            seen_instances.add(id(node))

            # Check for session persistence
            if node._session_manager is not None:
                raise ValueError("Session persistence is not supported for Swarm agents yet.")

    def _inject_swarm_tools(self) -> None:
        """Add swarm coordination tools to each agent."""
        # Create tool functions with proper closures
        swarm_tools = [
            self._create_handoff_tool(),
        ]

        for node in self.nodes.values():
            # Check for existing tools with conflicting names
            existing_tools = node.executor.tool_registry.registry
            conflicting_tools = []

            if "handoff_to_agent" in existing_tools:
                conflicting_tools.append("handoff_to_agent")

            if conflicting_tools:
                raise ValueError(
                    f"Agent '{node.node_id}' already has tools with names that conflict with swarm coordination tools: "
                    f"{', '.join(conflicting_tools)}. Please rename these tools to avoid conflicts."
                )

            # Use the agent's tool registry to process and register the tools
            node.executor.tool_registry.process_tools(swarm_tools)

        logger.debug(
            "tool_count=<%d>, node_count=<%d> | injected coordination tools into agents",
            len(swarm_tools),
            len(self.nodes),
        )

    def _create_handoff_tool(self) -> Callable[..., Any]:
        """Create handoff tool for agent coordination."""
        swarm_ref = self  # Capture swarm reference

        @tool
        def handoff_to_agent(agent_name: str, message: str, context: dict[str, Any] | None = None) -> dict[str, Any]:
            """Transfer control to another agent in the swarm for specialized help.

            Args:
                agent_name: Name of the agent to hand off to
                message: Message explaining what needs to be done and why you're handing off
                context: Additional context to share with the next agent

            Returns:
                Confirmation of handoff initiation
            """
            try:
                context = context or {}

                # Validate target agent exists
                target_node = swarm_ref.nodes.get(agent_name)
                if not target_node:
                    return {"status": "error", "content": [{"text": f"Error: Agent '{agent_name}' not found in swarm"}]}

                # Execute handoff
                swarm_ref._handle_handoff(target_node, message, context)

                return {"status": "success", "content": [{"text": f"Handing off to {agent_name}: {message}"}]}
            except Exception as e:
                return {"status": "error", "content": [{"text": f"Error in handoff: {str(e)}"}]}

        return handoff_to_agent

    def _handle_handoff(self, target_node: SwarmNode, message: str, context: dict[str, Any]) -> None:
        """Handle handoff to another agent."""
        # If task is already completed, don't allow further handoffs
        if self.state.completion_status != Status.EXECUTING:
            logger.debug(
                "task_status=<%s> | ignoring handoff request - task already completed",
                self.state.completion_status,
            )
            return

        current_node = cast(SwarmNode, self.state.current_node)

        self.state.handoff_node = target_node
        self.state.handoff_message = message

        # Store handoff context as shared context
        if context:
            for key, value in context.items():
                self.shared_context.add_context(current_node, key, value)

        logger.debug(
            "from_node=<%s>, to_node=<%s> | handing off from agent to agent",
            current_node.node_id,
            target_node.node_id,
        )

    def _build_node_input(self, target_node: SwarmNode) -> str:
        """Build input text for a node based on shared context and handoffs.

        Example formatted output:
        ```
        Handoff Message: The user needs help with Python debugging - I've identified the issue but need someone with more expertise to fix it.

        User Request: My Python script is throwing a KeyError when processing JSON data from an API

        Previous agents who worked on this: data_analyst → code_reviewer

        Shared knowledge from previous agents:
        • data_analyst: {"issue_location": "line 42", "error_type": "missing key validation", "suggested_fix": "add key existence check"}
        • code_reviewer: {"code_quality": "good overall structure", "security_notes": "API key should be in environment variable"}

        Other agents available for collaboration:
        Agent name: data_analyst. Agent description: Analyzes data and provides deeper insights
        Agent name: code_reviewer.
        Agent name: security_specialist. Agent description: Focuses on secure coding practices and vulnerability assessment

        You have access to swarm coordination tools if you need help from other agents. If you don't hand off to another agent, the swarm will consider the task complete.
        ```
        """  # noqa: E501
        context_info: dict[str, Any] = {
            "task": self.state.task,
            "node_history": [node.node_id for node in self.state.node_history],
            "shared_context": {k: v for k, v in self.shared_context.context.items()},
        }
        context_text = ""

        # Include handoff message prominently at the top if present
        if self.state.handoff_message:
            context_text += f"Handoff Message: {self.state.handoff_message}\n\n"

        # Include task information if available
        if "task" in context_info:
            task = context_info.get("task")
            if isinstance(task, str):
                context_text += f"User Request: {task}\n\n"
            elif isinstance(task, list):
                context_text += "User Request: Multi-modal task\n\n"

        # Include detailed node history
        if context_info.get("node_history"):
            context_text += f"Previous agents who worked on this: {' → '.join(context_info['node_history'])}\n\n"

        # Include actual shared context, not just a mention
        shared_context = context_info.get("shared_context", {})
        if shared_context:
            context_text += "Shared knowledge from previous agents:\n"
            for node_name, context in shared_context.items():
                if context:  # Only include if node has contributed context
                    context_text += f"• {node_name}: {context}\n"
            context_text += "\n"

        # Include available nodes with descriptions if available
        other_nodes = [node_id for node_id in self.nodes.keys() if node_id != target_node.node_id]
        if other_nodes:
            context_text += "Other agents available for collaboration:\n"
            for node_id in other_nodes:
                node = self.nodes.get(node_id)
                context_text += f"Agent name: {node_id}."
                if node and hasattr(node.executor, "description") and node.executor.description:
                    context_text += f" Agent description: {node.executor.description}"
                context_text += "\n"
            context_text += "\n"

        context_text += (
            "You have access to swarm coordination tools if you need help from other agents. "
            "If you don't hand off to another agent, the swarm will consider the task complete."
        )

        return context_text

    def _activate_interrupt(self, node: SwarmNode, interrupts: list[Interrupt]) -> MultiAgentNodeInterruptEvent:
        """Activate the interrupt state.

        Note, a Swarm may be interrupted either from a BeforeNodeCallEvent hook or from within an agent node. In either
        case, we must manage the interrupt state of both the Swarm and the individual agent nodes.

        Args:
            node: The interrupted node.
            interrupts: The interrupts raised by the user.

        Returns:
            MultiAgentNodeInterruptEvent
        """
        logger.debug("node=<%s> | node interrupted", node.node_id)
        self.state.completion_status = Status.INTERRUPTED

        self._interrupt_state.context[node.node_id] = {
            "activated": node.executor._interrupt_state.activated,
            "interrupt_state": node.executor._interrupt_state.to_dict(),
            "state": node.executor.state.get(),
            "messages": node.executor.messages,
        }

        self._interrupt_state.interrupts.update({interrupt.id: interrupt for interrupt in interrupts})
        self._interrupt_state.activate()

        return MultiAgentNodeInterruptEvent(node.node_id, interrupts)

    async def _execute_swarm(self, invocation_state: dict[str, Any]) -> AsyncIterator[Any]:
        """Execute swarm and yield TypedEvent objects."""
        try:
            # Main execution loop
            while True:
                if self.state.completion_status != Status.EXECUTING:
                    reason = f"Completion status is: {self.state.completion_status}"
                    logger.debug("reason=<%s> | stopping streaming execution", reason)
                    break

                should_continue, reason = self.state.should_continue(
                    max_handoffs=self.max_handoffs,
                    max_iterations=self.max_iterations,
                    execution_timeout=self.execution_timeout,
                    repetitive_handoff_detection_window=self.repetitive_handoff_detection_window,
                    repetitive_handoff_min_unique_agents=self.repetitive_handoff_min_unique_agents,
                )
                if not should_continue:
                    self.state.completion_status = Status.FAILED
                    logger.debug("reason=<%s> | stopping execution", reason)
                    break

                current_node = self.state.current_node
                if not current_node or current_node.node_id not in self.nodes:
                    logger.error("node=<%s> | node not found", current_node.node_id if current_node else "None")
                    self.state.completion_status = Status.FAILED
                    break

                logger.debug(
                    "current_node=<%s>, iteration=<%d> | executing node",
                    current_node.node_id,
                    len(self.state.node_history) + 1,
                )

                before_event, interrupts = await self.hooks.invoke_callbacks_async(
                    BeforeNodeCallEvent(self, current_node.node_id, invocation_state)
                )

                # TODO: Implement cancellation token to stop _execute_node from continuing
                try:
                    if interrupts:
                        yield self._activate_interrupt(current_node, interrupts)
                        break

                    if before_event.cancel_node:
                        cancel_message = (
                            before_event.cancel_node
                            if isinstance(before_event.cancel_node, str)
                            else "node cancelled by user"
                        )
                        logger.debug("reason=<%s> | cancelling execution", cancel_message)
                        yield MultiAgentNodeCancelEvent(current_node.node_id, cancel_message)
                        self.state.completion_status = Status.FAILED
                        break

                    node_stream = self._stream_with_timeout(
                        self._execute_node(current_node, self.state.task, invocation_state),
                        self.node_timeout,
                        f"Node '{current_node.node_id}' execution timed out after {self.node_timeout}s",
                    )
                    async for event in node_stream:
                        yield event

                    stop_event = cast(MultiAgentNodeStopEvent, event)
                    node_result = stop_event["node_result"]
                    if node_result.status == Status.INTERRUPTED:
                        yield self._activate_interrupt(current_node, node_result.interrupts)
                        break

                    self._interrupt_state.deactivate()

                    self.state.node_history.append(current_node)

                except Exception:
                    logger.exception("node=<%s> | node execution failed", current_node.node_id)
                    self.state.completion_status = Status.FAILED
                    break

                finally:
                    await self.hooks.invoke_callbacks_async(
                        AfterNodeCallEvent(self, current_node.node_id, invocation_state)
                    )

                logger.debug("node=<%s> | node execution completed", current_node.node_id)

                # Check if handoff requested during execution
                if self.state.handoff_node:
                    previous_node = current_node
                    current_node = self.state.handoff_node

                    self.state.handoff_node = None
                    self.state.current_node = current_node

                    handoff_event = MultiAgentHandoffEvent(
                        from_node_ids=[previous_node.node_id],
                        to_node_ids=[current_node.node_id],
                        message=self.state.handoff_message or "Agent handoff occurred",
                    )
                    yield handoff_event
                    logger.debug(
                        "from_node=<%s>, to_node=<%s> | handoff detected",
                        previous_node.node_id,
                        current_node.node_id,
                    )

                else:
                    logger.debug("node=<%s> | no handoff occurred, marking swarm as complete", current_node.node_id)
                    self.state.completion_status = Status.COMPLETED
                    break

        except Exception:
            logger.exception("swarm execution failed")
            self.state.completion_status = Status.FAILED
        finally:
            elapsed_time = time.time() - self.state.start_time
            logger.debug("status=<%s> | swarm execution completed", self.state.completion_status)
            logger.debug(
                "node_history_length=<%d>, time=<%s>s | metrics",
                len(self.state.node_history),
                f"{elapsed_time:.2f}",
            )

    async def _execute_node(
        self, node: SwarmNode, task: MultiAgentInput, invocation_state: dict[str, Any]
    ) -> AsyncIterator[Any]:
        """Execute swarm node and yield TypedEvent objects."""
        start_time = time.time()
        node_name = node.node_id

        # Emit node start event
        start_event = MultiAgentNodeStartEvent(node_id=node_name, node_type="agent")
        yield start_event

        try:
            if self._interrupt_state.activated and self._interrupt_state.context[node_name]["activated"]:
                node_input = self._interrupt_state.context["responses"]

            else:
                # Prepare context for node
                context_text = self._build_node_input(node)
                node_input = [ContentBlock(text=f"Context:\n{context_text}\n\n")]

                # Clear handoff message after it's been included in context
                self.state.handoff_message = None

                if not isinstance(task, str):
                    # Include additional ContentBlocks in node input
                    node_input = node_input + cast(list[ContentBlock], task)

            # Execute node with streaming
            node.reset_executor_state()

            # Stream agent events with node context and capture final result
            result = None
            async for event in node.executor.stream_async(node_input, invocation_state=invocation_state):
                # Forward agent events with node context
                wrapped_event = MultiAgentNodeStreamEvent(node_name, event)
                yield wrapped_event
                # Capture the final result event
                if "result" in event:
                    result = event["result"]

            if result is None:
                raise ValueError(f"Node '{node_name}' did not produce a result event")

            execution_time = round((time.time() - start_time) * 1000)
            status = Status.INTERRUPTED if result.stop_reason == "interrupt" else Status.COMPLETED

            # Create NodeResult with extracted metrics
            result_metrics = getattr(result, "metrics", None)
            usage = getattr(result_metrics, "accumulated_usage", Usage(inputTokens=0, outputTokens=0, totalTokens=0))
            metrics = getattr(result_metrics, "accumulated_metrics", Metrics(latencyMs=execution_time))

            node_result = NodeResult(
                result=result,
                execution_time=execution_time,
                status=status,
                accumulated_usage=usage,
                accumulated_metrics=metrics,
                execution_count=1,
                interrupts=result.interrupts or [],
            )

            # Store result in state
            self.state.results[node_name] = node_result

            # Accumulate metrics
            self._accumulate_metrics(node_result)

            # Emit node stop event with full NodeResult
            complete_event = MultiAgentNodeStopEvent(
                node_id=node_name,
                node_result=node_result,
            )
            yield complete_event

        except Exception as e:
            execution_time = round((time.time() - start_time) * 1000)
            logger.exception("node=<%s> | node execution failed", node_name)

            # Create a NodeResult for the failed node
            node_result = NodeResult(
                result=e,
                execution_time=execution_time,
                status=Status.FAILED,
                accumulated_usage=Usage(inputTokens=0, outputTokens=0, totalTokens=0),
                accumulated_metrics=Metrics(latencyMs=execution_time),
                execution_count=1,
            )

            # Store result in state
            self.state.results[node_name] = node_result

            # Emit node stop event even for failures
            complete_event = MultiAgentNodeStopEvent(
                node_id=node_name,
                node_result=node_result,
            )
            yield complete_event

            raise

    def _accumulate_metrics(self, node_result: NodeResult) -> None:
        """Accumulate metrics from a node result."""
        self.state.accumulated_usage["inputTokens"] += node_result.accumulated_usage.get("inputTokens", 0)
        self.state.accumulated_usage["outputTokens"] += node_result.accumulated_usage.get("outputTokens", 0)
        self.state.accumulated_usage["totalTokens"] += node_result.accumulated_usage.get("totalTokens", 0)
        self.state.accumulated_metrics["latencyMs"] += node_result.accumulated_metrics.get("latencyMs", 0)

    def _build_result(self, interrupts: list[Interrupt]) -> SwarmResult:
        """Build swarm result from current state."""
        return SwarmResult(
            status=self.state.completion_status,
            results=self.state.results,
            accumulated_usage=self.state.accumulated_usage,
            accumulated_metrics=self.state.accumulated_metrics,
            execution_count=len(self.state.node_history),
            execution_time=self.state.execution_time,
            node_history=self.state.node_history,
            interrupts=interrupts,
        )

    def serialize_state(self) -> dict[str, Any]:
        """Serialize the current swarm state to a dictionary."""
        status_str = self.state.completion_status.value
        if self.state.completion_status == Status.EXECUTING and self.state.current_node:
            next_nodes = [self.state.current_node.node_id]
        elif self.state.completion_status == Status.INTERRUPTED and self.state.current_node:
            next_nodes = [self.state.current_node.node_id]
        elif self.state.handoff_node:
            next_nodes = [self.state.handoff_node.node_id]
        else:
            next_nodes = []

        return {
            "type": "swarm",
            "id": self.id,
            "status": status_str,
            "node_history": [n.node_id for n in self.state.node_history],
            "node_results": {k: v.to_dict() for k, v in self.state.results.items()},
            "next_nodes_to_execute": next_nodes,
            "current_task": self.state.task,
            "context": {
                "shared_context": getattr(self.state.shared_context, "context", {}) or {},
                "handoff_node": self.state.handoff_node.node_id if self.state.handoff_node else None,
                "handoff_message": self.state.handoff_message,
            },
            "_internal_state": {
                "interrupt_state": self._interrupt_state.to_dict(),
            },
        }

    def deserialize_state(self, payload: dict[str, Any]) -> None:
        """Restore swarm state from a session dict and prepare for execution.

        This method handles two scenarios:
        1. If the persisted status is COMPLETED, FAILED resets all nodes and graph state
           to allow re-execution from the beginning.
        2. Otherwise, restores the persisted state and prepares to resume execution
           from the next ready nodes.

        Args:
            payload: Dictionary containing persisted state data including status,
                    completed nodes, results, and next nodes to execute.
        """
        if "_internal_state" in payload:
            internal_state = payload["_internal_state"]
            self._interrupt_state = _InterruptState.from_dict(internal_state["interrupt_state"])

        self._resume_from_session = "next_nodes_to_execute" in payload
        if self._resume_from_session:
            self._from_dict(payload)
            return

        for node in self.nodes.values():
            node.reset_executor_state()

        self.state = SwarmState(
            current_node=SwarmNode("", Agent(), swarm=self),
            task="",
            completion_status=Status.PENDING,
        )

    def _from_dict(self, payload: dict[str, Any]) -> None:
        self.state.completion_status = Status(payload["status"])
        # Hydrate completed nodes & results
        context = payload["context"] or {}
        self.shared_context.context = context.get("shared_context") or {}
        self.state.handoff_message = context.get("handoff_message")
        self.state.handoff_node = self.nodes[context["handoff_node"]] if context.get("handoff_node") else None

        self.state.node_history = [self.nodes[nid] for nid in (payload.get("node_history") or []) if nid in self.nodes]

        raw_results = payload.get("node_results") or {}
        results: dict[str, NodeResult] = {}
        for node_id, entry in raw_results.items():
            if node_id not in self.nodes:
                continue
            try:
                results[node_id] = NodeResult.from_dict(entry)
            except Exception:
                logger.exception("Failed to hydrate NodeResult for node_id=%s; skipping.", node_id)
                raise
        self.state.results = results
        self.state.task = payload.get("current_task", self.state.task)

        next_node_ids = payload.get("next_nodes_to_execute") or []
        if next_node_ids:
            self.state.current_node = self.nodes[next_node_ids[0]] if next_node_ids[0] else self._initial_node()

    def _initial_node(self) -> SwarmNode:
        if self.entry_point:
            return self.nodes[str(self.entry_point.name)]
        return next(iter(self.nodes.values()))  # First SwarmNode

`call(task, invocation_state=None, **kwargs)` ¶

Invoke the swarm synchronously.

Parameters:

Name	Type	Description	Default
`task`	`MultiAgentInput`	The task to execute	required
`invocation_state`	`dict[str, Any] \| None`	Additional state/context passed to underlying agents. Defaults to None to avoid mutable default argument issues.	`None`
`**kwargs`	`Any`	Keyword arguments allowing backward compatible future changes.	`{}`

Source code in strands/multiagent/swarm.py

def __call__(
    self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
) -> SwarmResult:
    """Invoke the swarm synchronously.

    Args:
        task: The task to execute
        invocation_state: Additional state/context passed to underlying agents.
            Defaults to None to avoid mutable default argument issues.
        **kwargs: Keyword arguments allowing backward compatible future changes.
    """
    if invocation_state is None:
        invocation_state = {}
    return run_async(lambda: self.invoke_async(task, invocation_state))

`init(nodes, *, entry_point=None, max_handoffs=20, max_iterations=20, execution_timeout=900.0, node_timeout=300.0, repetitive_handoff_detection_window=0, repetitive_handoff_min_unique_agents=0, session_manager=None, hooks=None, id=_DEFAULT_SWARM_ID, trace_attributes=None)` ¶

Initialize Swarm with agents and configuration.

Parameters:

Name	Type	Description	Default
`id`	`str`	Unique swarm id (default: "default_swarm")	`_DEFAULT_SWARM_ID`
`nodes`	`list[Agent]`	List of nodes (e.g. Agent) to include in the swarm	required
`entry_point`	`Agent \| None`	Agent to start with. If None, uses the first agent (default: None)	`None`
`max_handoffs`	`int`	Maximum handoffs to agents and users (default: 20)	`20`
`max_iterations`	`int`	Maximum node executions within the swarm (default: 20)	`20`
`execution_timeout`	`float`	Total execution timeout in seconds (default: 900.0)	`900.0`
`node_timeout`	`float`	Individual node timeout in seconds (default: 300.0)	`300.0`
`repetitive_handoff_detection_window`	`int`	Number of recent nodes to check for repetitive handoffs Disabled by default (default: 0)	`0`
`repetitive_handoff_min_unique_agents`	`int`	Minimum unique agents required in recent sequence Disabled by default (default: 0)	`0`
`session_manager`	`Optional[SessionManager]`	Session manager for persisting graph state and execution history (default: None)	`None`
`hooks`	`Optional[list[HookProvider]]`	List of hook providers for monitoring and extending graph execution behavior (default: None)	`None`
`trace_attributes`	`Optional[Mapping[str, AttributeValue]]`	Custom trace attributes to apply to the agent's trace span (default: None)	`None`

Source code in strands/multiagent/swarm.py

def __init__(
    self,
    nodes: list[Agent],
    *,
    entry_point: Agent | None = None,
    max_handoffs: int = 20,
    max_iterations: int = 20,
    execution_timeout: float = 900.0,
    node_timeout: float = 300.0,
    repetitive_handoff_detection_window: int = 0,
    repetitive_handoff_min_unique_agents: int = 0,
    session_manager: Optional[SessionManager] = None,
    hooks: Optional[list[HookProvider]] = None,
    id: str = _DEFAULT_SWARM_ID,
    trace_attributes: Optional[Mapping[str, AttributeValue]] = None,
) -> None:
    """Initialize Swarm with agents and configuration.

    Args:
        id: Unique swarm id (default: "default_swarm")
        nodes: List of nodes (e.g. Agent) to include in the swarm
        entry_point: Agent to start with. If None, uses the first agent (default: None)
        max_handoffs: Maximum handoffs to agents and users (default: 20)
        max_iterations: Maximum node executions within the swarm (default: 20)
        execution_timeout: Total execution timeout in seconds (default: 900.0)
        node_timeout: Individual node timeout in seconds (default: 300.0)
        repetitive_handoff_detection_window: Number of recent nodes to check for repetitive handoffs
            Disabled by default (default: 0)
        repetitive_handoff_min_unique_agents: Minimum unique agents required in recent sequence
            Disabled by default (default: 0)
        session_manager: Session manager for persisting graph state and execution history (default: None)
        hooks: List of hook providers for monitoring and extending graph execution behavior (default: None)
        trace_attributes: Custom trace attributes to apply to the agent's trace span (default: None)
    """
    super().__init__()
    self.id = id
    self.entry_point = entry_point
    self.max_handoffs = max_handoffs
    self.max_iterations = max_iterations
    self.execution_timeout = execution_timeout
    self.node_timeout = node_timeout
    self.repetitive_handoff_detection_window = repetitive_handoff_detection_window
    self.repetitive_handoff_min_unique_agents = repetitive_handoff_min_unique_agents

    self.shared_context = SharedContext()
    self.nodes: dict[str, SwarmNode] = {}

    self.state = SwarmState(
        current_node=None,  # Placeholder, will be set properly
        task="",
        completion_status=Status.PENDING,
    )
    self._interrupt_state = _InterruptState()

    self.tracer = get_tracer()
    self.trace_attributes: dict[str, AttributeValue] = self._parse_trace_attributes(trace_attributes)

    self.session_manager = session_manager
    self.hooks = HookRegistry()
    if hooks:
        for hook in hooks:
            self.hooks.add_hook(hook)
    if self.session_manager:
        self.hooks.add_hook(self.session_manager)

    self._resume_from_session = False

    self._setup_swarm(nodes)
    self._inject_swarm_tools()
    run_async(lambda: self.hooks.invoke_callbacks_async(MultiAgentInitializedEvent(self)))

`deserialize_state(payload)` ¶

Restore swarm state from a session dict and prepare for execution.

This method handles two scenarios: 1. If the persisted status is COMPLETED, FAILED resets all nodes and graph state to allow re-execution from the beginning. 2. Otherwise, restores the persisted state and prepares to resume execution from the next ready nodes.

Parameters:

Name	Type	Description	Default
`payload`	`dict[str, Any]`	Dictionary containing persisted state data including status, completed nodes, results, and next nodes to execute.	required

Source code in strands/multiagent/swarm.py

def deserialize_state(self, payload: dict[str, Any]) -> None:
    """Restore swarm state from a session dict and prepare for execution.

    This method handles two scenarios:
    1. If the persisted status is COMPLETED, FAILED resets all nodes and graph state
       to allow re-execution from the beginning.
    2. Otherwise, restores the persisted state and prepares to resume execution
       from the next ready nodes.

    Args:
        payload: Dictionary containing persisted state data including status,
                completed nodes, results, and next nodes to execute.
    """
    if "_internal_state" in payload:
        internal_state = payload["_internal_state"]
        self._interrupt_state = _InterruptState.from_dict(internal_state["interrupt_state"])

    self._resume_from_session = "next_nodes_to_execute" in payload
    if self._resume_from_session:
        self._from_dict(payload)
        return

    for node in self.nodes.values():
        node.reset_executor_state()

    self.state = SwarmState(
        current_node=SwarmNode("", Agent(), swarm=self),
        task="",
        completion_status=Status.PENDING,
    )

`invoke_async(task, invocation_state=None, **kwargs)` `async` ¶

Invoke the swarm asynchronously.

This method uses stream_async internally and consumes all events until completion, following the same pattern as the Agent class.

Parameters:

Name	Type	Description	Default
`task`	`MultiAgentInput`	The task to execute	required
`invocation_state`	`dict[str, Any] \| None`	Additional state/context passed to underlying agents. Defaults to None to avoid mutable default argument issues.	`None`
`**kwargs`	`Any`	Keyword arguments allowing backward compatible future changes.	`{}`

Source code in strands/multiagent/swarm.py

async def invoke_async(
    self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
) -> SwarmResult:
    """Invoke the swarm asynchronously.

    This method uses stream_async internally and consumes all events until completion,
    following the same pattern as the Agent class.

    Args:
        task: The task to execute
        invocation_state: Additional state/context passed to underlying agents.
            Defaults to None to avoid mutable default argument issues.
        **kwargs: Keyword arguments allowing backward compatible future changes.
    """
    events = self.stream_async(task, invocation_state, **kwargs)
    final_event = None
    async for event in events:
        final_event = event

    if final_event is None or "result" not in final_event:
        raise ValueError("Swarm streaming completed without producing a result event")

    return cast(SwarmResult, final_event["result"])

`serialize_state()` ¶

Serialize the current swarm state to a dictionary.

Source code in strands/multiagent/swarm.py

def serialize_state(self) -> dict[str, Any]:
    """Serialize the current swarm state to a dictionary."""
    status_str = self.state.completion_status.value
    if self.state.completion_status == Status.EXECUTING and self.state.current_node:
        next_nodes = [self.state.current_node.node_id]
    elif self.state.completion_status == Status.INTERRUPTED and self.state.current_node:
        next_nodes = [self.state.current_node.node_id]
    elif self.state.handoff_node:
        next_nodes = [self.state.handoff_node.node_id]
    else:
        next_nodes = []

    return {
        "type": "swarm",
        "id": self.id,
        "status": status_str,
        "node_history": [n.node_id for n in self.state.node_history],
        "node_results": {k: v.to_dict() for k, v in self.state.results.items()},
        "next_nodes_to_execute": next_nodes,
        "current_task": self.state.task,
        "context": {
            "shared_context": getattr(self.state.shared_context, "context", {}) or {},
            "handoff_node": self.state.handoff_node.node_id if self.state.handoff_node else None,
            "handoff_message": self.state.handoff_message,
        },
        "_internal_state": {
            "interrupt_state": self._interrupt_state.to_dict(),
        },
    }

`stream_async(task, invocation_state=None, **kwargs)` `async` ¶

Stream events during swarm execution.

Parameters:

Name	Type	Description	Default
`task`	`MultiAgentInput`	The task to execute	required
`invocation_state`	`dict[str, Any] \| None`	Additional state/context passed to underlying agents. Defaults to None to avoid mutable default argument issues.	`None`
`**kwargs`	`Any`	Keyword arguments allowing backward compatible future changes.	`{}`

Yields:

Type	Description
`AsyncIterator[dict[str, Any]]`	Dictionary events during swarm execution, such as:
`AsyncIterator[dict[str, Any]]`	multi_agent_node_start: When a node begins execution
`AsyncIterator[dict[str, Any]]`	multi_agent_node_stream: Forwarded agent events with node context
`AsyncIterator[dict[str, Any]]`	multi_agent_handoff: When control is handed off between agents
`AsyncIterator[dict[str, Any]]`	multi_agent_node_stop: When a node stops execution
`AsyncIterator[dict[str, Any]]`	result: Final swarm result

Source code in strands/multiagent/swarm.py

async def stream_async(
    self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
) -> AsyncIterator[dict[str, Any]]:
    """Stream events during swarm execution.

    Args:
        task: The task to execute
        invocation_state: Additional state/context passed to underlying agents.
            Defaults to None to avoid mutable default argument issues.
        **kwargs: Keyword arguments allowing backward compatible future changes.

    Yields:
        Dictionary events during swarm execution, such as:
        - multi_agent_node_start: When a node begins execution
        - multi_agent_node_stream: Forwarded agent events with node context
        - multi_agent_handoff: When control is handed off between agents
        - multi_agent_node_stop: When a node stops execution
        - result: Final swarm result
    """
    self._interrupt_state.resume(task)

    if invocation_state is None:
        invocation_state = {}

    await self.hooks.invoke_callbacks_async(BeforeMultiAgentInvocationEvent(self, invocation_state))

    logger.debug("starting swarm execution")

    if self._resume_from_session or self._interrupt_state.activated:
        self.state.completion_status = Status.EXECUTING
        self.state.start_time = time.time()
    else:
        # Initialize swarm state with configuration
        initial_node = self._initial_node()

        self.state = SwarmState(
            current_node=initial_node,
            task=task,
            completion_status=Status.EXECUTING,
            shared_context=self.shared_context,
        )

    span = self.tracer.start_multiagent_span(task, "swarm", custom_trace_attributes=self.trace_attributes)
    with trace_api.use_span(span, end_on_exit=True):
        interrupts = []

        try:
            current_node = cast(SwarmNode, self.state.current_node)
            logger.debug("current_node=<%s> | starting swarm execution with node", current_node.node_id)
            logger.debug(
                "max_handoffs=<%d>, max_iterations=<%d>, timeout=<%s>s | swarm execution config",
                self.max_handoffs,
                self.max_iterations,
                self.execution_timeout,
            )

            async for event in self._execute_swarm(invocation_state):
                if isinstance(event, MultiAgentNodeInterruptEvent):
                    interrupts = event.interrupts

                yield event.as_dict()

        except Exception:
            logger.exception("swarm execution failed")
            self.state.completion_status = Status.FAILED
            raise
        finally:
            self.state.execution_time = round((time.time() - self.state.start_time) * 1000)
            await self.hooks.invoke_callbacks_async(AfterMultiAgentInvocationEvent(self, invocation_state))
            self._resume_from_session = False

        # Yield final result after execution_time is set
        result = self._build_result(interrupts)
        yield MultiAgentResultEvent(result=result).as_dict()

`SwarmNode` `dataclass` ¶

Represents a node (e.g. Agent) in the swarm.

Source code in strands/multiagent/swarm.py

@dataclass
class SwarmNode:
    """Represents a node (e.g. Agent) in the swarm."""

    node_id: str
    executor: Agent
    swarm: Optional["Swarm"] = None
    _initial_messages: Messages = field(default_factory=list, init=False)
    _initial_state: AgentState = field(default_factory=AgentState, init=False)

    def __post_init__(self) -> None:
        """Capture initial executor state after initialization."""
        # Deep copy the initial messages and state to preserve them
        self._initial_messages = copy.deepcopy(self.executor.messages)
        self._initial_state = AgentState(self.executor.state.get())

    def __hash__(self) -> int:
        """Return hash for SwarmNode based on node_id."""
        return hash(self.node_id)

    def __eq__(self, other: Any) -> bool:
        """Return equality for SwarmNode based on node_id."""
        if not isinstance(other, SwarmNode):
            return False
        return self.node_id == other.node_id

    def __str__(self) -> str:
        """Return string representation of SwarmNode."""
        return self.node_id

    def __repr__(self) -> str:
        """Return detailed representation of SwarmNode."""
        return f"SwarmNode(node_id='{self.node_id}')"

    def reset_executor_state(self) -> None:
        """Reset SwarmNode executor state to initial state when swarm was created.

        If Swarm is resuming from an interrupt, we reset the executor state from the interrupt context.
        """
        if self.swarm and self.swarm._interrupt_state.activated:
            context = self.swarm._interrupt_state.context[self.node_id]
            self.executor.messages = context["messages"]
            self.executor.state = AgentState(context["state"])
            self.executor._interrupt_state = _InterruptState.from_dict(context["interrupt_state"])
            return

        self.executor.messages = copy.deepcopy(self._initial_messages)
        self.executor.state = AgentState(self._initial_state.get())

`eq(other)` ¶

Return equality for SwarmNode based on node_id.

Source code in strands/multiagent/swarm.py

def __eq__(self, other: Any) -> bool:
    """Return equality for SwarmNode based on node_id."""
    if not isinstance(other, SwarmNode):
        return False
    return self.node_id == other.node_id

`hash()` ¶

Return hash for SwarmNode based on node_id.

Source code in strands/multiagent/swarm.py

def __hash__(self) -> int:
    """Return hash for SwarmNode based on node_id."""
    return hash(self.node_id)

`__post_init__()` ¶

Capture initial executor state after initialization.

Source code in strands/multiagent/swarm.py

def __post_init__(self) -> None:
    """Capture initial executor state after initialization."""
    # Deep copy the initial messages and state to preserve them
    self._initial_messages = copy.deepcopy(self.executor.messages)
    self._initial_state = AgentState(self.executor.state.get())

`repr()` ¶

Return detailed representation of SwarmNode.

Source code in strands/multiagent/swarm.py

def __repr__(self) -> str:
    """Return detailed representation of SwarmNode."""
    return f"SwarmNode(node_id='{self.node_id}')"

`str()` ¶

Return string representation of SwarmNode.

Source code in strands/multiagent/swarm.py

def __str__(self) -> str:
    """Return string representation of SwarmNode."""
    return self.node_id

`reset_executor_state()` ¶

Reset SwarmNode executor state to initial state when swarm was created.

If Swarm is resuming from an interrupt, we reset the executor state from the interrupt context.

Source code in strands/multiagent/swarm.py

def reset_executor_state(self) -> None:
    """Reset SwarmNode executor state to initial state when swarm was created.

    If Swarm is resuming from an interrupt, we reset the executor state from the interrupt context.
    """
    if self.swarm and self.swarm._interrupt_state.activated:
        context = self.swarm._interrupt_state.context[self.node_id]
        self.executor.messages = context["messages"]
        self.executor.state = AgentState(context["state"])
        self.executor._interrupt_state = _InterruptState.from_dict(context["interrupt_state"])
        return

    self.executor.messages = copy.deepcopy(self._initial_messages)
    self.executor.state = AgentState(self._initial_state.get())

`SwarmResult` `dataclass` ¶

Bases: MultiAgentResult

Result from swarm execution - extends MultiAgentResult with swarm-specific details.

Source code in strands/multiagent/swarm.py

@dataclass
class SwarmResult(MultiAgentResult):
    """Result from swarm execution - extends MultiAgentResult with swarm-specific details."""

    node_history: list[SwarmNode] = field(default_factory=list)

`SwarmState` `dataclass` ¶

Current state of swarm execution.

Source code in strands/multiagent/swarm.py

@dataclass
class SwarmState:
    """Current state of swarm execution."""

    current_node: SwarmNode | None  # The agent currently executing
    task: MultiAgentInput  # The original task from the user that is being executed
    completion_status: Status = Status.PENDING  # Current swarm execution status
    shared_context: SharedContext = field(default_factory=SharedContext)  # Context shared between agents
    node_history: list[SwarmNode] = field(default_factory=list)  # Complete history of agents that have executed
    start_time: float = field(default_factory=time.time)  # When swarm execution began
    results: dict[str, NodeResult] = field(default_factory=dict)  # Results from each agent execution
    # Total token usage across all agents
    accumulated_usage: Usage = field(default_factory=lambda: Usage(inputTokens=0, outputTokens=0, totalTokens=0))
    # Total metrics across all agents
    accumulated_metrics: Metrics = field(default_factory=lambda: Metrics(latencyMs=0))
    execution_time: int = 0  # Total execution time in milliseconds
    handoff_node: SwarmNode | None = None  # The agent to execute next
    handoff_message: str | None = None  # Message passed during agent handoff

    def should_continue(
        self,
        *,
        max_handoffs: int,
        max_iterations: int,
        execution_timeout: float,
        repetitive_handoff_detection_window: int,
        repetitive_handoff_min_unique_agents: int,
    ) -> Tuple[bool, str]:
        """Check if the swarm should continue.

        Returns: (should_continue, reason)
        """
        # Check handoff limit
        if len(self.node_history) >= max_handoffs:
            return False, f"Max handoffs reached: {max_handoffs}"

        # Check iteration limit
        if len(self.node_history) >= max_iterations:
            return False, f"Max iterations reached: {max_iterations}"

        # Check timeout
        elapsed = time.time() - self.start_time
        if elapsed > execution_timeout:
            return False, f"Execution timed out: {execution_timeout}s"

        # Check for repetitive handoffs (agents passing back and forth)
        if repetitive_handoff_detection_window > 0 and len(self.node_history) >= repetitive_handoff_detection_window:
            recent = self.node_history[-repetitive_handoff_detection_window:]
            unique_nodes = len(set(recent))
            if unique_nodes < repetitive_handoff_min_unique_agents:
                return (
                    False,
                    (
                        f"Repetitive handoff: {unique_nodes} unique nodes "
                        f"out of {repetitive_handoff_detection_window} recent iterations"
                    ),
                )

        return True, "Continuing"

`should_continue(*, max_handoffs, max_iterations, execution_timeout, repetitive_handoff_detection_window, repetitive_handoff_min_unique_agents)` ¶

Check if the swarm should continue.

Returns: (should_continue, reason)

Source code in strands/multiagent/swarm.py

def should_continue(
    self,
    *,
    max_handoffs: int,
    max_iterations: int,
    execution_timeout: float,
    repetitive_handoff_detection_window: int,
    repetitive_handoff_min_unique_agents: int,
) -> Tuple[bool, str]:
    """Check if the swarm should continue.

    Returns: (should_continue, reason)
    """
    # Check handoff limit
    if len(self.node_history) >= max_handoffs:
        return False, f"Max handoffs reached: {max_handoffs}"

    # Check iteration limit
    if len(self.node_history) >= max_iterations:
        return False, f"Max iterations reached: {max_iterations}"

    # Check timeout
    elapsed = time.time() - self.start_time
    if elapsed > execution_timeout:
        return False, f"Execution timed out: {execution_timeout}s"

    # Check for repetitive handoffs (agents passing back and forth)
    if repetitive_handoff_detection_window > 0 and len(self.node_history) >= repetitive_handoff_detection_window:
        recent = self.node_history[-repetitive_handoff_detection_window:]
        unique_nodes = len(set(recent))
        if unique_nodes < repetitive_handoff_min_unique_agents:
            return (
                False,
                (
                    f"Repetitive handoff: {unique_nodes} unique nodes "
                    f"out of {repetitive_handoff_detection_window} recent iterations"
                ),
            )

    return True, "Continuing"

`Usage` ¶

Bases: TypedDict

Token usage information for model interactions.

Attributes:

Name	Type	Description
`inputTokens`	`Required[int]`	Number of tokens sent in the request to the model.
`outputTokens`	`Required[int]`	Number of tokens that the model generated for the request.
`totalTokens`	`Required[int]`	Total number of tokens (input + output).
`cacheReadInputTokens`	`int`	Number of tokens read from cache (optional).
`cacheWriteInputTokens`	`int`	Number of tokens written to cache (optional).

Source code in strands/types/event_loop.py

class Usage(TypedDict, total=False):
    """Token usage information for model interactions.

    Attributes:
        inputTokens: Number of tokens sent in the request to the model.
        outputTokens: Number of tokens that the model generated for the request.
        totalTokens: Total number of tokens (input + output).
        cacheReadInputTokens: Number of tokens read from cache (optional).
        cacheWriteInputTokens: Number of tokens written to cache (optional).
    """

    inputTokens: Required[int]
    outputTokens: Required[int]
    totalTokens: Required[int]
    cacheReadInputTokens: int
    cacheWriteInputTokens: int

`_InterruptState` `dataclass` ¶

Track the state of interrupt events raised by the user.

Note, interrupt state is cleared after resuming.

Attributes:

Name	Type	Description
`interrupts`	`dict[str, Interrupt]`	Interrupts raised by the user.
`context`	`dict[str, Any]`	Additional context associated with an interrupt event.
`activated`	`bool`	True if agent is in an interrupt state, False otherwise.

Source code in strands/interrupt.py

@dataclass
class _InterruptState:
    """Track the state of interrupt events raised by the user.

    Note, interrupt state is cleared after resuming.

    Attributes:
        interrupts: Interrupts raised by the user.
        context: Additional context associated with an interrupt event.
        activated: True if agent is in an interrupt state, False otherwise.
    """

    interrupts: dict[str, Interrupt] = field(default_factory=dict)
    context: dict[str, Any] = field(default_factory=dict)
    activated: bool = False

    def activate(self) -> None:
        """Activate the interrupt state."""
        self.activated = True

    def deactivate(self) -> None:
        """Deacitvate the interrupt state.

        Interrupts and context are cleared.
        """
        self.interrupts = {}
        self.context = {}
        self.activated = False

    def resume(self, prompt: "AgentInput") -> None:
        """Configure the interrupt state if resuming from an interrupt event.

        Args:
            prompt: User responses if resuming from interrupt.

        Raises:
            TypeError: If in interrupt state but user did not provide responses.
        """
        if not self.activated:
            return

        if not isinstance(prompt, list):
            raise TypeError(f"prompt_type={type(prompt)} | must resume from interrupt with list of interruptResponse's")

        invalid_types = [
            content_type for content in prompt for content_type in content if content_type != "interruptResponse"
        ]
        if invalid_types:
            raise TypeError(
                f"content_types=<{invalid_types}> | must resume from interrupt with list of interruptResponse's"
            )

        contents = cast(list["InterruptResponseContent"], prompt)
        for content in contents:
            interrupt_id = content["interruptResponse"]["interruptId"]
            interrupt_response = content["interruptResponse"]["response"]

            if interrupt_id not in self.interrupts:
                raise KeyError(f"interrupt_id=<{interrupt_id}> | no interrupt found")

            self.interrupts[interrupt_id].response = interrupt_response

        self.context["responses"] = contents

    def to_dict(self) -> dict[str, Any]:
        """Serialize to dict for session management."""
        return asdict(self)

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "_InterruptState":
        """Initiailize interrupt state from serialized interrupt state.

        Interrupt state can be serialized with the `to_dict` method.
        """
        return cls(
            interrupts={
                interrupt_id: Interrupt(**interrupt_data) for interrupt_id, interrupt_data in data["interrupts"].items()
            },
            context=data["context"],
            activated=data["activated"],
        )

`activate()` ¶

Activate the interrupt state.

Source code in strands/interrupt.py

def activate(self) -> None:
    """Activate the interrupt state."""
    self.activated = True

`deactivate()` ¶

Deacitvate the interrupt state.

Interrupts and context are cleared.

Source code in strands/interrupt.py

def deactivate(self) -> None:
    """Deacitvate the interrupt state.

    Interrupts and context are cleared.
    """
    self.interrupts = {}
    self.context = {}
    self.activated = False

`from_dict(data)` `classmethod` ¶

Initiailize interrupt state from serialized interrupt state.

Interrupt state can be serialized with the to_dict method.

Source code in strands/interrupt.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "_InterruptState":
    """Initiailize interrupt state from serialized interrupt state.

    Interrupt state can be serialized with the `to_dict` method.
    """
    return cls(
        interrupts={
            interrupt_id: Interrupt(**interrupt_data) for interrupt_id, interrupt_data in data["interrupts"].items()
        },
        context=data["context"],
        activated=data["activated"],
    )

`resume(prompt)` ¶

Configure the interrupt state if resuming from an interrupt event.

Parameters:

Name	Type	Description	Default
`prompt`	`AgentInput`	User responses if resuming from interrupt.	required

Raises:

Type	Description
`TypeError`	If in interrupt state but user did not provide responses.

Source code in strands/interrupt.py

def resume(self, prompt: "AgentInput") -> None:
    """Configure the interrupt state if resuming from an interrupt event.

    Args:
        prompt: User responses if resuming from interrupt.

    Raises:
        TypeError: If in interrupt state but user did not provide responses.
    """
    if not self.activated:
        return

    if not isinstance(prompt, list):
        raise TypeError(f"prompt_type={type(prompt)} | must resume from interrupt with list of interruptResponse's")

    invalid_types = [
        content_type for content in prompt for content_type in content if content_type != "interruptResponse"
    ]
    if invalid_types:
        raise TypeError(
            f"content_types=<{invalid_types}> | must resume from interrupt with list of interruptResponse's"
        )

    contents = cast(list["InterruptResponseContent"], prompt)
    for content in contents:
        interrupt_id = content["interruptResponse"]["interruptId"]
        interrupt_response = content["interruptResponse"]["response"]

        if interrupt_id not in self.interrupts:
            raise KeyError(f"interrupt_id=<{interrupt_id}> | no interrupt found")

        self.interrupts[interrupt_id].response = interrupt_response

    self.context["responses"] = contents

`to_dict()` ¶

Serialize to dict for session management.

Source code in strands/interrupt.py

def to_dict(self) -> dict[str, Any]:
    """Serialize to dict for session management."""
    return asdict(self)

`get_tracer()` ¶

Get or create the global tracer.

Returns:

Type	Description
`Tracer`	The global tracer instance.

Source code in strands/telemetry/tracer.py

def get_tracer() -> Tracer:
    """Get or create the global tracer.

    Returns:
        The global tracer instance.
    """
    global _tracer_instance

    if not _tracer_instance:
        _tracer_instance = Tracer()

    return _tracer_instance

`run_async(async_func)` ¶

Run an async function in a separate thread to avoid event loop conflicts.

This utility handles the common pattern of running async code from sync contexts by using ThreadPoolExecutor to isolate the async execution.

Parameters:

Name	Type	Description	Default
`async_func`	`Callable[[], Awaitable[T]]`	A callable that returns an awaitable	required

Returns:

Type	Description
`T`	The result of the async function

Source code in strands/_async.py

def run_async(async_func: Callable[[], Awaitable[T]]) -> T:
    """Run an async function in a separate thread to avoid event loop conflicts.

    This utility handles the common pattern of running async code from sync contexts
    by using ThreadPoolExecutor to isolate the async execution.

    Args:
        async_func: A callable that returns an awaitable

    Returns:
        The result of the async function
    """

    async def execute_async() -> T:
        return await async_func()

    def execute() -> T:
        return asyncio.run(execute_async())

    with ThreadPoolExecutor() as executor:
        context = contextvars.copy_context()
        future = executor.submit(context.run, execute)
        return future.result()

`tool(func=None, description=None, inputSchema=None, name=None, context=False)` ¶

tool(__func: Callable[P, R]) -> DecoratedFunctionTool[P, R]

tool(
    description: Optional[str] = None,
    inputSchema: Optional[JSONSchema] = None,
    name: Optional[str] = None,
    context: bool | str = False,
) -> Callable[
    [Callable[P, R]], DecoratedFunctionTool[P, R]
]

Decorator that transforms a Python function into a Strands tool.

This decorator seamlessly enables a function to be called both as a regular Python function and as a Strands tool. It extracts metadata from the function's signature, docstring, and type hints to generate an OpenAPI-compatible tool specification.

When decorated, a function:

Still works as a normal function when called directly with arguments
Processes tool use API calls when provided with a tool use dictionary
Validates inputs against the function's type hints and parameter spec
Formats return values according to the expected Strands tool result format
Provides automatic error handling and reporting

The decorator can be used in two ways: - As a simple decorator: @tool - With parameters: @tool(name="custom_name", description="Custom description")

Parameters:

Name	Type	Description	Default
`func`	`Optional[Callable[P, R]]`	The function to decorate. When used as a simple decorator, this is the function being decorated. When used with parameters, this will be None.	`None`
`description`	`Optional[str]`	Optional custom description to override the function's docstring.	`None`
`inputSchema`	`Optional[JSONSchema]`	Optional custom JSON schema to override the automatically generated schema.	`None`
`name`	`Optional[str]`	Optional custom name to override the function's name.	`None`
`context`	`bool \| str`	When provided, places an object in the designated parameter. If True, the param name defaults to 'tool_context', or if an override is needed, set context equal to a string to designate the param name.	`False`

Returns:

Type	Description
`Union[DecoratedFunctionTool[P, R], Callable[[Callable[P, R]], DecoratedFunctionTool[P, R]]]`	An AgentTool that also mimics the original function when invoked

Example

@tool
def my_tool(name: str, count: int = 1) -> str:
    # Does something useful with the provided parameters.
    #
    # Parameters:
    #   name: The name to process
    #   count: Number of times to process (default: 1)
    #
    # Returns:
    #   A message with the result
    return f"Processed {name} {count} times"

agent = Agent(tools=[my_tool])
agent.my_tool(name="example", count=3)
# Returns: {
#   "toolUseId": "123",
#   "status": "success",
#   "content": [{"text": "Processed example 3 times"}]
# }

Example with parameters

@tool(name="custom_tool", description="A tool with a custom name and description", context=True)
def my_tool(name: str, count: int = 1, tool_context: ToolContext) -> str:
    tool_id = tool_context["tool_use"]["toolUseId"]
    return f"Processed {name} {count} times with tool ID {tool_id}"

Source code in strands/tools/decorator.py

def tool(  # type: ignore
    func: Optional[Callable[P, R]] = None,
    description: Optional[str] = None,
    inputSchema: Optional[JSONSchema] = None,
    name: Optional[str] = None,
    context: bool | str = False,
) -> Union[DecoratedFunctionTool[P, R], Callable[[Callable[P, R]], DecoratedFunctionTool[P, R]]]:
    """Decorator that transforms a Python function into a Strands tool.

    This decorator seamlessly enables a function to be called both as a regular Python function and as a Strands tool.
    It extracts metadata from the function's signature, docstring, and type hints to generate an OpenAPI-compatible tool
    specification.

    When decorated, a function:

    1. Still works as a normal function when called directly with arguments
    2. Processes tool use API calls when provided with a tool use dictionary
    3. Validates inputs against the function's type hints and parameter spec
    4. Formats return values according to the expected Strands tool result format
    5. Provides automatic error handling and reporting

    The decorator can be used in two ways:
    - As a simple decorator: `@tool`
    - With parameters: `@tool(name="custom_name", description="Custom description")`

    Args:
        func: The function to decorate. When used as a simple decorator, this is the function being decorated.
            When used with parameters, this will be None.
        description: Optional custom description to override the function's docstring.
        inputSchema: Optional custom JSON schema to override the automatically generated schema.
        name: Optional custom name to override the function's name.
        context: When provided, places an object in the designated parameter. If True, the param name
            defaults to 'tool_context', or if an override is needed, set context equal to a string to designate
            the param name.

    Returns:
        An AgentTool that also mimics the original function when invoked

    Example:
        ```python
        @tool
        def my_tool(name: str, count: int = 1) -> str:
            # Does something useful with the provided parameters.
            #
            # Parameters:
            #   name: The name to process
            #   count: Number of times to process (default: 1)
            #
            # Returns:
            #   A message with the result
            return f"Processed {name} {count} times"

        agent = Agent(tools=[my_tool])
        agent.my_tool(name="example", count=3)
        # Returns: {
        #   "toolUseId": "123",
        #   "status": "success",
        #   "content": [{"text": "Processed example 3 times"}]
        # }
        ```

    Example with parameters:
        ```python
        @tool(name="custom_tool", description="A tool with a custom name and description", context=True)
        def my_tool(name: str, count: int = 1, tool_context: ToolContext) -> str:
            tool_id = tool_context["tool_use"]["toolUseId"]
            return f"Processed {name} {count} times with tool ID {tool_id}"
        ```
    """

    def decorator(f: T) -> "DecoratedFunctionTool[P, R]":
        # Resolve context parameter name
        if isinstance(context, bool):
            context_param = "tool_context" if context else None
        else:
            context_param = context.strip()
            if not context_param:
                raise ValueError("Context parameter name cannot be empty")

        # Create function tool metadata
        tool_meta = FunctionToolMetadata(f, context_param)
        tool_spec = tool_meta.extract_metadata()
        if name is not None:
            tool_spec["name"] = name
        if description is not None:
            tool_spec["description"] = description
        if inputSchema is not None:
            tool_spec["inputSchema"] = inputSchema

        tool_name = tool_spec.get("name", f.__name__)

        if not isinstance(tool_name, str):
            raise ValueError(f"Tool name must be a string, got {type(tool_name)}")

        return DecoratedFunctionTool(tool_name, tool_spec, f, tool_meta)

    # Handle both @tool and @tool() syntax
    if func is None:
        # Need to ignore type-checking here since it's hard to represent the support
        # for both flows using the type system
        return decorator

    return decorator(func)

strands.multiagent.swarm ¶

AgentState = JSONSerializableDict module-attribute ¶

AttributeValue = Union[str, bool, float, int, List[str], List[bool], List[float], List[int], Sequence[str], Sequence[bool], Sequence[int], Sequence[float]] module-attribute ¶

Messages = List[Message] module-attribute ¶

MultiAgentInput = str | list[ContentBlock] | list[InterruptResponseContent] module-attribute ¶

_DEFAULT_SWARM_ID = 'default_swarm' module-attribute ¶

logger = logging.getLogger(__name__) module-attribute ¶

AfterMultiAgentInvocationEvent dataclass ¶

should_reverse_callbacks property ¶

AfterNodeCallEvent dataclass ¶

should_reverse_callbacks property ¶

Agent ¶

system_prompt property writable ¶

tool property ¶

tool_names property ¶

__call__(prompt=None, *, invocation_state=None, structured_output_model=None, **kwargs) ¶

__del__() ¶

cleanup() ¶

invoke_async(prompt=None, *, invocation_state=None, structured_output_model=None, **kwargs) async ¶

stream_async(prompt=None, *, invocation_state=None, structured_output_model=None, **kwargs) async ¶

structured_output(output_model, prompt=None) ¶

structured_output_async(output_model, prompt=None) async ¶

BeforeMultiAgentInvocationEvent dataclass ¶

BeforeNodeCallEvent dataclass ¶

ContentBlock ¶

HookProvider ¶

register_hooks(registry, **kwargs) ¶

HookRegistry ¶

__init__() ¶

add_callback(event_type, callback) ¶

add_hook(hook) ¶

get_callbacks_for(event) ¶

has_callbacks() ¶

invoke_callbacks(event) ¶

invoke_callbacks_async(event) async ¶

Interrupt dataclass ¶

to_dict() ¶

Metrics ¶

MultiAgentBase ¶

__call__(task, invocation_state=None, **kwargs) ¶

deserialize_state(payload) ¶

invoke_async(task, invocation_state=None, **kwargs) abstractmethod async ¶

serialize_state() ¶

stream_async(task, invocation_state=None, **kwargs) async ¶

MultiAgentHandoffEvent ¶

__init__(from_node_ids, to_node_ids, message=None) ¶

MultiAgentInitializedEvent dataclass ¶

MultiAgentNodeCancelEvent ¶

__init__(node_id, message) ¶

MultiAgentNodeInterruptEvent ¶

interrupts property ¶

__init__(node_id, interrupts) ¶

MultiAgentNodeStartEvent ¶

__init__(node_id, node_type) ¶

MultiAgentNodeStopEvent ¶

__init__(node_id, node_result) ¶

MultiAgentNodeStreamEvent ¶

__init__(node_id, agent_event) ¶

MultiAgentResult dataclass ¶

from_dict(data) classmethod ¶

to_dict() ¶

MultiAgentResultEvent ¶

__init__(result) ¶

NodeResult dataclass ¶

from_dict(data) classmethod ¶

get_agent_results() ¶

to_dict() ¶

SessionManager ¶

append_bidi_message(message, agent, **kwargs) ¶

append_message(message, agent, **kwargs) abstractmethod ¶

initialize(agent, **kwargs) abstractmethod ¶

initialize_bidi_agent(agent, **kwargs) ¶

initialize_multi_agent(source, **kwargs) ¶

redact_latest_message(redact_message, agent, **kwargs) abstractmethod ¶

register_hooks(registry, **kwargs) ¶

sync_agent(agent, **kwargs) abstractmethod ¶

sync_bidi_agent(agent, **kwargs) ¶

sync_multi_agent(source, **kwargs) ¶

SharedContext dataclass ¶

add_context(node, key, value) ¶

`strands.multiagent.swarm` ¶

`AgentState = JSONSerializableDict` `module-attribute` ¶

`AttributeValue = Union[str, bool, float, int, List[str], List[bool], List[float], List[int], Sequence[str], Sequence[bool], Sequence[int], Sequence[float]]` `module-attribute` ¶

`Messages = List[Message]` `module-attribute` ¶

`MultiAgentInput = str | list[ContentBlock] | list[InterruptResponseContent]` `module-attribute` ¶

`_DEFAULT_SWARM_ID = 'default_swarm'` `module-attribute` ¶

`logger = logging.getLogger(name)` `module-attribute` ¶

`AfterMultiAgentInvocationEvent` `dataclass` ¶

`should_reverse_callbacks` `property` ¶

`AfterNodeCallEvent` `dataclass` ¶

`should_reverse_callbacks` `property` ¶

`Agent` ¶

`system_prompt` `property` `writable` ¶

`tool` `property` ¶

`tool_names` `property` ¶

`call(prompt=None, *, invocation_state=None, structured_output_model=None, **kwargs)` ¶

`del()` ¶

`cleanup()` ¶

`invoke_async(prompt=None, *, invocation_state=None, structured_output_model=None, **kwargs)` `async` ¶

`stream_async(prompt=None, *, invocation_state=None, structured_output_model=None, **kwargs)` `async` ¶

`structured_output(output_model, prompt=None)` ¶

`structured_output_async(output_model, prompt=None)` `async` ¶

`BeforeMultiAgentInvocationEvent` `dataclass` ¶

`BeforeNodeCallEvent` `dataclass` ¶

`ContentBlock` ¶

`HookProvider` ¶

`register_hooks(registry, **kwargs)` ¶

`HookRegistry` ¶

`init()` ¶

`add_callback(event_type, callback)` ¶

`add_hook(hook)` ¶

`get_callbacks_for(event)` ¶

`has_callbacks()` ¶

`invoke_callbacks(event)` ¶

`invoke_callbacks_async(event)` `async` ¶

`Interrupt` `dataclass` ¶

`to_dict()` ¶

`Metrics` ¶

`MultiAgentBase` ¶

`call(task, invocation_state=None, **kwargs)` ¶

`deserialize_state(payload)` ¶

`invoke_async(task, invocation_state=None, **kwargs)` `abstractmethod` `async` ¶

`serialize_state()` ¶

`stream_async(task, invocation_state=None, **kwargs)` `async` ¶

`MultiAgentHandoffEvent` ¶

`init(from_node_ids, to_node_ids, message=None)` ¶

`MultiAgentInitializedEvent` `dataclass` ¶

`MultiAgentNodeCancelEvent` ¶

`init(node_id, message)` ¶

`MultiAgentNodeInterruptEvent` ¶

`interrupts` `property` ¶

`init(node_id, interrupts)` ¶

`MultiAgentNodeStartEvent` ¶

`init(node_id, node_type)` ¶

`MultiAgentNodeStopEvent` ¶

`init(node_id, node_result)` ¶

`MultiAgentNodeStreamEvent` ¶

`init(node_id, agent_event)` ¶

`MultiAgentResult` `dataclass` ¶

`from_dict(data)` `classmethod` ¶

`to_dict()` ¶

`MultiAgentResultEvent` ¶

`init(result)` ¶

`NodeResult` `dataclass` ¶

`from_dict(data)` `classmethod` ¶

`get_agent_results()` ¶

`to_dict()` ¶

`SessionManager` ¶

`append_bidi_message(message, agent, **kwargs)` ¶

`append_message(message, agent, **kwargs)` `abstractmethod` ¶

`initialize(agent, **kwargs)` `abstractmethod` ¶

`initialize_bidi_agent(agent, **kwargs)` ¶

`initialize_multi_agent(source, **kwargs)` ¶

`redact_latest_message(redact_message, agent, **kwargs)` `abstractmethod` ¶

`register_hooks(registry, **kwargs)` ¶

`sync_agent(agent, **kwargs)` `abstractmethod` ¶

`sync_bidi_agent(agent, **kwargs)` ¶

`sync_multi_agent(source, **kwargs)` ¶

`SharedContext` `dataclass` ¶

`add_context(node, key, value)` ¶