strands.multiagent.graph

Directed Graph Multi-Agent Pattern Implementation.

This module provides a deterministic graph-based agent orchestration system where agents or MultiAgentBase instances (like Swarm or Graph) are nodes in a graph, executed according to edge dependencies, with output from one node passed as input to connected nodes.

Key Features: - Agents and MultiAgentBase instances (Swarm, Graph, etc.) as graph nodes - Deterministic execution based on dependency resolution - Output propagation along edges - Support for cyclic graphs (feedback loops) - Clear dependency management - Supports nested graphs (Graph as a node in another Graph)

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_GRAPH_ID = 'default_graph' 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:

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

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

Graph

Bases: MultiAgentBase

Directed Graph multi-agent orchestration with configurable revisit behavior.

Source code in strands/multiagent/graph.py

class Graph(MultiAgentBase):
    """Directed Graph multi-agent orchestration with configurable revisit behavior."""

    def __init__(
        self,
        nodes: dict[str, GraphNode],
        edges: set[GraphEdge],
        entry_points: set[GraphNode],
        max_node_executions: Optional[int] = None,
        execution_timeout: Optional[float] = None,
        node_timeout: Optional[float] = None,
        reset_on_revisit: bool = False,
        session_manager: Optional[SessionManager] = None,
        hooks: Optional[list[HookProvider]] = None,
        id: str = _DEFAULT_GRAPH_ID,
        trace_attributes: Optional[Mapping[str, AttributeValue]] = None,
    ) -> None:
        """Initialize Graph with execution limits and reset behavior.

        Args:
            nodes: Dictionary of node_id to GraphNode
            edges: Set of GraphEdge objects
            entry_points: Set of GraphNode objects that are entry points
            max_node_executions: Maximum total node executions (default: None - no limit)
            execution_timeout: Total execution timeout in seconds (default: None - no limit)
            node_timeout: Individual node timeout in seconds (default: None - no limit)
            reset_on_revisit: Whether to reset node state when revisited (default: False)
            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)
            id: Unique graph id (default: None)
            trace_attributes: Custom trace attributes to apply to the agent's trace span (default: None)
        """
        super().__init__()

        # Validate nodes for duplicate instances
        self._validate_graph(nodes)

        self.nodes = nodes
        self.edges = edges
        self.entry_points = entry_points
        self.max_node_executions = max_node_executions
        self.execution_timeout = execution_timeout
        self.node_timeout = node_timeout
        self.reset_on_revisit = reset_on_revisit
        self.state = GraphState()
        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 self.session_manager:
            self.hooks.add_hook(self.session_manager)
        if hooks:
            for hook in hooks:
                self.hooks.add_hook(hook)

        self._resume_next_nodes: list[GraphNode] = []
        self._resume_from_session = False
        self.id = id

        run_async(lambda: self.hooks.invoke_callbacks_async(MultiAgentInitializedEvent(self)))

    def __call__(
        self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
    ) -> GraphResult:
        """Invoke the graph 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
    ) -> GraphResult:
        """Invoke the graph 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("Graph streaming completed without producing a result event")

        return cast(GraphResult, 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 graph 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 graph execution, such as:
            - multi_agent_node_start: When a node begins execution
            - multi_agent_node_stream: Forwarded agent/multi-agent events with node context
            - multi_agent_node_stop: When a node stops execution
            - result: Final graph result
        """
        if invocation_state is None:
            invocation_state = {}

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

        logger.debug("task=<%s> | starting graph execution", task)

        # Initialize state
        start_time = time.time()
        if not self._resume_from_session:
            # Initialize state
            self.state = GraphState(
                status=Status.EXECUTING,
                task=task,
                total_nodes=len(self.nodes),
                edges=[(edge.from_node, edge.to_node) for edge in self.edges],
                entry_points=list(self.entry_points),
                start_time=start_time,
            )
        else:
            self.state.status = Status.EXECUTING
            self.state.start_time = start_time

        span = self.tracer.start_multiagent_span(task, "graph", custom_trace_attributes=self.trace_attributes)
        with trace_api.use_span(span, end_on_exit=True):
            try:
                logger.debug(
                    "max_node_executions=<%s>, execution_timeout=<%s>s, node_timeout=<%s>s | graph execution config",
                    self.max_node_executions or "None",
                    self.execution_timeout or "None",
                    self.node_timeout or "None",
                )

                async for event in self._execute_graph(invocation_state):
                    yield event.as_dict()

                # Set final status based on execution results
                if self.state.failed_nodes:
                    self.state.status = Status.FAILED
                elif self.state.status == Status.EXECUTING:
                    self.state.status = Status.COMPLETED

                logger.debug("status=<%s> | graph execution completed", self.state.status)

                # Yield final result (consistent with Agent's AgentResultEvent format)
                result = self._build_result()

                # Use the same event format as Agent for consistency
                yield MultiAgentResultEvent(result=result).as_dict()

            except Exception:
                logger.exception("graph execution failed")
                self.state.status = Status.FAILED
                raise
            finally:
                self.state.execution_time = round((time.time() - start_time) * 1000)
                await self.hooks.invoke_callbacks_async(AfterMultiAgentInvocationEvent(self))
                self._resume_from_session = False
                self._resume_next_nodes.clear()

    def _validate_graph(self, nodes: dict[str, GraphNode]) -> None:
        """Validate graph nodes for duplicate instances."""
        # Check for duplicate node instances
        seen_instances = set()
        for node in nodes.values():
            if id(node.executor) in seen_instances:
                raise ValueError("Duplicate node instance detected. Each node must have a unique object instance.")
            seen_instances.add(id(node.executor))

            # Validate Agent-specific constraints for each node
            _validate_node_executor(node.executor)

    async def _execute_graph(self, invocation_state: dict[str, Any]) -> AsyncIterator[Any]:
        """Execute graph and yield TypedEvent objects."""
        ready_nodes = self._resume_next_nodes if self._resume_from_session else list(self.entry_points)

        while ready_nodes:
            # Check execution limits before continuing
            should_continue, reason = self.state.should_continue(
                max_node_executions=self.max_node_executions,
                execution_timeout=self.execution_timeout,
            )
            if not should_continue:
                self.state.status = Status.FAILED
                logger.debug("reason=<%s> | stopping execution", reason)
                return  # Let the top-level exception handler deal with it

            current_batch = ready_nodes.copy()
            ready_nodes.clear()

            # Execute current batch
            async for event in self._execute_nodes_parallel(current_batch, invocation_state):
                yield event

            # Find newly ready nodes after batch execution
            # We add all nodes in current batch as completed batch,
            # because a failure would throw exception and code would not make it here
            newly_ready = self._find_newly_ready_nodes(current_batch)

            # Emit handoff event for batch transition if there are nodes to transition to
            if newly_ready:
                handoff_event = MultiAgentHandoffEvent(
                    from_node_ids=[node.node_id for node in current_batch],
                    to_node_ids=[node.node_id for node in newly_ready],
                )
                yield handoff_event
                logger.debug(
                    "from_node_ids=<%s>, to_node_ids=<%s> | batch transition",
                    [node.node_id for node in current_batch],
                    [node.node_id for node in newly_ready],
                )

            ready_nodes.extend(newly_ready)

    async def _execute_nodes_parallel(
        self, nodes: list["GraphNode"], invocation_state: dict[str, Any]
    ) -> AsyncIterator[Any]:
        """Execute multiple nodes in parallel and merge their event streams in real-time.

        Uses a shared queue where each node's stream runs independently and pushes events
        as they occur, enabling true real-time event propagation without round-robin delays.
        """
        event_queue: asyncio.Queue[Any | None | Exception] = asyncio.Queue()

        # Start all node streams as independent tasks
        tasks = [asyncio.create_task(self._stream_node_to_queue(node, event_queue, invocation_state)) for node in nodes]

        try:
            # Consume events from the queue as they arrive
            # Continue until all tasks are done
            while any(not task.done() for task in tasks):
                try:
                    # Use timeout to avoid race condition: if all tasks complete between
                    # checking task.done() and calling queue.get(), we'd hang forever.
                    # The 0.1s timeout allows us to periodically re-check task completion
                    # while still being responsive to incoming events.
                    event = await asyncio.wait_for(event_queue.get(), timeout=0.1)
                except asyncio.TimeoutError:
                    # No event available, continue checking tasks
                    continue

                # Check if it's an exception - fail fast
                if isinstance(event, Exception):
                    # Cancel all other tasks immediately
                    for task in tasks:
                        if not task.done():
                            task.cancel()
                    raise event

                if event is not None:
                    yield event

            # Process any remaining events in the queue after all tasks complete
            while not event_queue.empty():
                event = await event_queue.get()
                if isinstance(event, Exception):
                    raise event
                if event is not None:
                    yield event
        finally:
            # Cancel any remaining tasks
            remaining_tasks = [task for task in tasks if not task.done()]
            if remaining_tasks:
                logger.warning(
                    "remaining_task_count=<%d> | cancelling remaining tasks in finally block",
                    len(remaining_tasks),
                )
                for task in remaining_tasks:
                    task.cancel()
            await asyncio.gather(*tasks, return_exceptions=True)

    async def _stream_node_to_queue(
        self,
        node: GraphNode,
        event_queue: asyncio.Queue[Any | None | Exception],
        invocation_state: dict[str, Any],
    ) -> None:
        """Stream events from a node to the shared queue with optional timeout."""
        try:
            # Apply timeout to the entire streaming process if configured
            if self.node_timeout is not None:

                async def stream_node() -> None:
                    async for event in self._execute_node(node, invocation_state):
                        await event_queue.put(event)

                try:
                    await asyncio.wait_for(stream_node(), timeout=self.node_timeout)
                except asyncio.TimeoutError:
                    # Handle timeout and send exception through queue
                    timeout_exc = await self._handle_node_timeout(node, event_queue)
                    await event_queue.put(timeout_exc)
            else:
                # No timeout - stream normally
                async for event in self._execute_node(node, invocation_state):
                    await event_queue.put(event)
        except Exception as e:
            # Send exception through queue for fail-fast behavior
            await event_queue.put(e)
        finally:
            await event_queue.put(None)

    async def _handle_node_timeout(self, node: GraphNode, event_queue: asyncio.Queue[Any | None]) -> Exception:
        """Handle a node timeout by creating a failed result and emitting events.

        Returns:
            The timeout exception to be re-raised for fail-fast behavior
        """
        assert self.node_timeout is not None
        timeout_exception = Exception(f"Node '{node.node_id}' execution timed out after {self.node_timeout}s")

        node_result = NodeResult(
            result=timeout_exception,
            execution_time=round(self.node_timeout * 1000),
            status=Status.FAILED,
            accumulated_usage=Usage(inputTokens=0, outputTokens=0, totalTokens=0),
            accumulated_metrics=Metrics(latencyMs=round(self.node_timeout * 1000)),
            execution_count=1,
        )

        node.execution_status = Status.FAILED
        node.result = node_result
        node.execution_time = node_result.execution_time
        self.state.failed_nodes.add(node)
        self.state.results[node.node_id] = node_result

        complete_event = MultiAgentNodeStopEvent(
            node_id=node.node_id,
            node_result=node_result,
        )
        await event_queue.put(complete_event)

        return timeout_exception

    def _find_newly_ready_nodes(self, completed_batch: list["GraphNode"]) -> list["GraphNode"]:
        """Find nodes that became ready after the last execution."""
        newly_ready = []
        for _node_id, node in self.nodes.items():
            if self._is_node_ready_with_conditions(node, completed_batch):
                newly_ready.append(node)
        return newly_ready

    def _is_node_ready_with_conditions(self, node: GraphNode, completed_batch: list["GraphNode"]) -> bool:
        """Check if a node is ready considering conditional edges."""
        # Get incoming edges to this node
        incoming_edges = [edge for edge in self.edges if edge.to_node == node]

        # Check if at least one incoming edge condition is satisfied
        for edge in incoming_edges:
            if edge.from_node in completed_batch:
                if edge.should_traverse(self.state):
                    logger.debug(
                        "from=<%s>, to=<%s> | edge ready via satisfied condition", edge.from_node.node_id, node.node_id
                    )
                    return True
                else:
                    logger.debug(
                        "from=<%s>, to=<%s> | edge condition not satisfied", edge.from_node.node_id, node.node_id
                    )
        return False

    async def _execute_node(self, node: GraphNode, invocation_state: dict[str, Any]) -> AsyncIterator[Any]:
        """Execute a single node and yield TypedEvent objects."""
        # Reset the node's state if reset_on_revisit is enabled, and it's being revisited
        if self.reset_on_revisit and node in self.state.completed_nodes:
            logger.debug("node_id=<%s> | resetting node state for revisit", node.node_id)
            node.reset_executor_state()
            self.state.completed_nodes.remove(node)

        node.execution_status = Status.EXECUTING
        logger.debug("node_id=<%s> | executing node", node.node_id)

        # Emit node start event
        start_event = MultiAgentNodeStartEvent(
            node_id=node.node_id, node_type="agent" if isinstance(node.executor, Agent) else "multiagent"
        )
        yield start_event

        before_event, _ = await self.hooks.invoke_callbacks_async(
            BeforeNodeCallEvent(self, node.node_id, invocation_state)
        )

        start_time = time.time()
        try:
            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(node.node_id, cancel_message)
                raise RuntimeError(cancel_message)

            # Build node input from satisfied dependencies
            node_input = self._build_node_input(node)

            # Execute and stream events (timeout handled at task level)
            if isinstance(node.executor, MultiAgentBase):
                # For nested multi-agent systems, stream their events and collect result
                multi_agent_result = None
                async for event in node.executor.stream_async(node_input, invocation_state):
                    # Forward nested multi-agent events with node context
                    wrapped_event = MultiAgentNodeStreamEvent(node.node_id, event)
                    yield wrapped_event
                    # Capture the final result event
                    if "result" in event:
                        multi_agent_result = event["result"]

                # Use the captured result from streaming (no double execution)
                if multi_agent_result is None:
                    raise ValueError(f"Node '{node.node_id}' did not produce a result event")

                node_result = NodeResult(
                    result=multi_agent_result,
                    execution_time=multi_agent_result.execution_time,
                    status=Status.COMPLETED,
                    accumulated_usage=multi_agent_result.accumulated_usage,
                    accumulated_metrics=multi_agent_result.accumulated_metrics,
                    execution_count=multi_agent_result.execution_count,
                )

            elif isinstance(node.executor, Agent):
                # For agents, stream their events and collect result
                agent_response = 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.node_id, event)
                    yield wrapped_event
                    # Capture the final result event
                    if "result" in event:
                        agent_response = event["result"]

                # Use the captured result from streaming (no double execution)
                if agent_response is None:
                    raise ValueError(f"Node '{node.node_id}' did not produce a result event")

                # Check for interrupt (from main branch)
                if agent_response.stop_reason == "interrupt":
                    node.executor.messages.pop()  # remove interrupted tool use message
                    node.executor._interrupt_state.deactivate()

                    raise RuntimeError("user raised interrupt from agent | interrupts are not yet supported in graphs")

                # Extract metrics with defaults
                response_metrics = getattr(agent_response, "metrics", None)
                usage = getattr(
                    response_metrics, "accumulated_usage", Usage(inputTokens=0, outputTokens=0, totalTokens=0)
                )
                metrics = getattr(response_metrics, "accumulated_metrics", Metrics(latencyMs=0))

                node_result = NodeResult(
                    result=agent_response,
                    execution_time=round((time.time() - start_time) * 1000),
                    status=Status.COMPLETED,
                    accumulated_usage=usage,
                    accumulated_metrics=metrics,
                    execution_count=1,
                )
            else:
                raise ValueError(f"Node '{node.node_id}' of type '{type(node.executor)}' is not supported")

            # Mark as completed
            node.execution_status = Status.COMPLETED
            node.result = node_result
            node.execution_time = node_result.execution_time
            self.state.completed_nodes.add(node)
            self.state.results[node.node_id] = node_result
            self.state.execution_order.append(node)

            # Accumulate metrics
            self._accumulate_metrics(node_result)

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

            logger.debug(
                "node_id=<%s>, execution_time=<%dms> | node completed successfully",
                node.node_id,
                node.execution_time,
            )

        except Exception as e:
            # All failures (programming errors and execution failures) stop graph execution
            # This matches the old fail-fast behavior
            logger.error("node_id=<%s>, error=<%s> | node failed", node.node_id, e)
            execution_time = round((time.time() - start_time) * 1000)

            # 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,
            )

            node.execution_status = Status.FAILED
            node.result = node_result
            node.execution_time = execution_time
            self.state.failed_nodes.add(node)
            self.state.results[node.node_id] = node_result

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

            # Re-raise to stop graph execution (fail-fast behavior)
            raise

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

    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)
        self.state.execution_count += node_result.execution_count

    def _build_node_input(self, node: GraphNode) -> list[ContentBlock]:
        """Build input text for a node based on dependency outputs.

        Example formatted output:
        ```
        Original Task: Analyze the quarterly sales data and create a summary report

        Inputs from previous nodes:

        From data_processor:
          - Agent: Sales data processed successfully. Found 1,247 transactions totaling $89,432.
          - Agent: Key trends: 15% increase in Q3, top product category is Electronics.

        From validator:
          - Agent: Data validation complete. All records verified, no anomalies detected.
        ```
        """
        # Get satisfied dependencies
        dependency_results = {}
        for edge in self.edges:
            if (
                edge.to_node == node
                and edge.from_node in self.state.completed_nodes
                and edge.from_node.node_id in self.state.results
            ):
                if edge.should_traverse(self.state):
                    dependency_results[edge.from_node.node_id] = self.state.results[edge.from_node.node_id]

        if not dependency_results:
            # No dependencies - return task as ContentBlocks
            if isinstance(self.state.task, str):
                return [ContentBlock(text=self.state.task)]
            else:
                return cast(list[ContentBlock], self.state.task)

        # Combine task with dependency outputs
        node_input = []

        # Add original task
        if isinstance(self.state.task, str):
            node_input.append(ContentBlock(text=f"Original Task: {self.state.task}"))
        else:
            # Add task content blocks with a prefix
            node_input.append(ContentBlock(text="Original Task:"))
            node_input.extend(cast(list[ContentBlock], self.state.task))

        # Add dependency outputs
        node_input.append(ContentBlock(text="\nInputs from previous nodes:"))

        for dep_id, node_result in dependency_results.items():
            node_input.append(ContentBlock(text=f"\nFrom {dep_id}:"))
            # Get all agent results from this node (flattened if nested)
            agent_results = node_result.get_agent_results()
            for result in agent_results:
                agent_name = getattr(result, "agent_name", "Agent")
                result_text = str(result)
                node_input.append(ContentBlock(text=f"  - {agent_name}: {result_text}"))

        return node_input

    def _build_result(self) -> GraphResult:
        """Build graph result from current state."""
        return GraphResult(
            status=self.state.status,
            results=self.state.results,
            accumulated_usage=self.state.accumulated_usage,
            accumulated_metrics=self.state.accumulated_metrics,
            execution_count=self.state.execution_count,
            execution_time=self.state.execution_time,
            total_nodes=self.state.total_nodes,
            completed_nodes=len(self.state.completed_nodes),
            failed_nodes=len(self.state.failed_nodes),
            execution_order=self.state.execution_order,
            edges=self.state.edges,
            entry_points=self.state.entry_points,
        )

    def serialize_state(self) -> dict[str, Any]:
        """Serialize the current graph state to a dictionary."""
        compute_nodes = self._compute_ready_nodes_for_resume()
        next_nodes = [n.node_id for n in compute_nodes] if compute_nodes else []
        return {
            "type": "graph",
            "id": self.id,
            "status": self.state.status.value,
            "completed_nodes": [n.node_id for n in self.state.completed_nodes],
            "failed_nodes": [n.node_id for n in self.state.failed_nodes],
            "node_results": {k: v.to_dict() for k, v in (self.state.results or {}).items()},
            "next_nodes_to_execute": next_nodes,
            "current_task": self.state.task,
            "execution_order": [n.node_id for n in self.state.execution_order],
        }

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

        This method handles two scenarios:
        1. If the graph execution ended (no next_nodes_to_execute, eg: Completed, or Failed with dead end nodes),
        resets all nodes and graph state to allow re-execution from the beginning.
        2. If the graph execution was interrupted mid-execution (has next_nodes_to_execute),
           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 not payload.get("next_nodes_to_execute"):
            # Reset all nodes
            for node in self.nodes.values():
                node.reset_executor_state()
            # Reset graph state
            self.state = GraphState()
            self._resume_from_session = False
            return
        else:
            self._from_dict(payload)
            self._resume_from_session = True

    def _compute_ready_nodes_for_resume(self) -> list[GraphNode]:
        if self.state.status == Status.PENDING:
            return []
        ready_nodes: list[GraphNode] = []
        completed_nodes = set(self.state.completed_nodes)
        for node in self.nodes.values():
            if node in completed_nodes:
                continue
            incoming = [e for e in self.edges if e.to_node is node]
            if not incoming:
                ready_nodes.append(node)
            elif all(e.from_node in completed_nodes and e.should_traverse(self.state) for e in incoming):
                ready_nodes.append(node)

        return ready_nodes

    def _from_dict(self, payload: dict[str, Any]) -> None:
        self.state.status = Status(payload["status"])
        # Hydrate completed nodes & results
        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.failed_nodes = set(
            self.nodes[node_id] for node_id in (payload.get("failed_nodes") or []) if node_id in self.nodes
        )

        # Restore completed nodes from persisted data
        completed_node_ids = payload.get("completed_nodes") or []
        self.state.completed_nodes = {self.nodes[node_id] for node_id in completed_node_ids if node_id in self.nodes}

        # Execution order (only nodes that still exist)
        order_node_ids = payload.get("execution_order") or []
        self.state.execution_order = [self.nodes[node_id] for node_id in order_node_ids if node_id in self.nodes]

        # Task
        self.state.task = payload.get("current_task", self.state.task)

        # next nodes to execute
        next_nodes = [self.nodes[nid] for nid in (payload.get("next_nodes_to_execute") or []) if nid in self.nodes]
        self._resume_next_nodes = next_nodes

__call__(task, invocation_state=None, **kwargs)

Invoke the graph 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/graph.py

def __call__(
    self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
) -> GraphResult:
    """Invoke the graph 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, edges, entry_points, max_node_executions=None, execution_timeout=None, node_timeout=None, reset_on_revisit=False, session_manager=None, hooks=None, id=_DEFAULT_GRAPH_ID, trace_attributes=None)

Initialize Graph with execution limits and reset behavior.

Parameters:

Name	Type	Description	Default
nodes	dict[str, GraphNode]	Dictionary of node_id to GraphNode	required
edges	set[GraphEdge]	Set of GraphEdge objects	required
entry_points	set[GraphNode]	Set of GraphNode objects that are entry points	required
max_node_executions	Optional[int]	Maximum total node executions (default: None - no limit)	None
execution_timeout	Optional[float]	Total execution timeout in seconds (default: None - no limit)	None
node_timeout	Optional[float]	Individual node timeout in seconds (default: None - no limit)	None
reset_on_revisit	bool	Whether to reset node state when revisited (default: False)	False
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
id	str	Unique graph id (default: None)	_DEFAULT_GRAPH_ID
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/graph.py

def __init__(
    self,
    nodes: dict[str, GraphNode],
    edges: set[GraphEdge],
    entry_points: set[GraphNode],
    max_node_executions: Optional[int] = None,
    execution_timeout: Optional[float] = None,
    node_timeout: Optional[float] = None,
    reset_on_revisit: bool = False,
    session_manager: Optional[SessionManager] = None,
    hooks: Optional[list[HookProvider]] = None,
    id: str = _DEFAULT_GRAPH_ID,
    trace_attributes: Optional[Mapping[str, AttributeValue]] = None,
) -> None:
    """Initialize Graph with execution limits and reset behavior.

    Args:
        nodes: Dictionary of node_id to GraphNode
        edges: Set of GraphEdge objects
        entry_points: Set of GraphNode objects that are entry points
        max_node_executions: Maximum total node executions (default: None - no limit)
        execution_timeout: Total execution timeout in seconds (default: None - no limit)
        node_timeout: Individual node timeout in seconds (default: None - no limit)
        reset_on_revisit: Whether to reset node state when revisited (default: False)
        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)
        id: Unique graph id (default: None)
        trace_attributes: Custom trace attributes to apply to the agent's trace span (default: None)
    """
    super().__init__()

    # Validate nodes for duplicate instances
    self._validate_graph(nodes)

    self.nodes = nodes
    self.edges = edges
    self.entry_points = entry_points
    self.max_node_executions = max_node_executions
    self.execution_timeout = execution_timeout
    self.node_timeout = node_timeout
    self.reset_on_revisit = reset_on_revisit
    self.state = GraphState()
    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 self.session_manager:
        self.hooks.add_hook(self.session_manager)
    if hooks:
        for hook in hooks:
            self.hooks.add_hook(hook)

    self._resume_next_nodes: list[GraphNode] = []
    self._resume_from_session = False
    self.id = id

    run_async(lambda: self.hooks.invoke_callbacks_async(MultiAgentInitializedEvent(self)))

deserialize_state(payload)

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

This method handles two scenarios: 1. If the graph execution ended (no next_nodes_to_execute, eg: Completed, or Failed with dead end nodes), resets all nodes and graph state to allow re-execution from the beginning. 2. If the graph execution was interrupted mid-execution (has next_nodes_to_execute), 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/graph.py

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

    This method handles two scenarios:
    1. If the graph execution ended (no next_nodes_to_execute, eg: Completed, or Failed with dead end nodes),
    resets all nodes and graph state to allow re-execution from the beginning.
    2. If the graph execution was interrupted mid-execution (has next_nodes_to_execute),
       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 not payload.get("next_nodes_to_execute"):
        # Reset all nodes
        for node in self.nodes.values():
            node.reset_executor_state()
        # Reset graph state
        self.state = GraphState()
        self._resume_from_session = False
        return
    else:
        self._from_dict(payload)
        self._resume_from_session = True

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

Invoke the graph 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/graph.py

async def invoke_async(
    self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
) -> GraphResult:
    """Invoke the graph 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("Graph streaming completed without producing a result event")

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

serialize_state()

Serialize the current graph state to a dictionary.

Source code in strands/multiagent/graph.py

def serialize_state(self) -> dict[str, Any]:
    """Serialize the current graph state to a dictionary."""
    compute_nodes = self._compute_ready_nodes_for_resume()
    next_nodes = [n.node_id for n in compute_nodes] if compute_nodes else []
    return {
        "type": "graph",
        "id": self.id,
        "status": self.state.status.value,
        "completed_nodes": [n.node_id for n in self.state.completed_nodes],
        "failed_nodes": [n.node_id for n in self.state.failed_nodes],
        "node_results": {k: v.to_dict() for k, v in (self.state.results or {}).items()},
        "next_nodes_to_execute": next_nodes,
        "current_task": self.state.task,
        "execution_order": [n.node_id for n in self.state.execution_order],
    }

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

Stream events during graph 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 graph 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/multi-agent events with node context
AsyncIterator[dict[str, Any]]	multi_agent_node_stop: When a node stops execution
AsyncIterator[dict[str, Any]]	result: Final graph result

Source code in strands/multiagent/graph.py

async def stream_async(
    self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
) -> AsyncIterator[dict[str, Any]]:
    """Stream events during graph 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 graph execution, such as:
        - multi_agent_node_start: When a node begins execution
        - multi_agent_node_stream: Forwarded agent/multi-agent events with node context
        - multi_agent_node_stop: When a node stops execution
        - result: Final graph result
    """
    if invocation_state is None:
        invocation_state = {}

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

    logger.debug("task=<%s> | starting graph execution", task)

    # Initialize state
    start_time = time.time()
    if not self._resume_from_session:
        # Initialize state
        self.state = GraphState(
            status=Status.EXECUTING,
            task=task,
            total_nodes=len(self.nodes),
            edges=[(edge.from_node, edge.to_node) for edge in self.edges],
            entry_points=list(self.entry_points),
            start_time=start_time,
        )
    else:
        self.state.status = Status.EXECUTING
        self.state.start_time = start_time

    span = self.tracer.start_multiagent_span(task, "graph", custom_trace_attributes=self.trace_attributes)
    with trace_api.use_span(span, end_on_exit=True):
        try:
            logger.debug(
                "max_node_executions=<%s>, execution_timeout=<%s>s, node_timeout=<%s>s | graph execution config",
                self.max_node_executions or "None",
                self.execution_timeout or "None",
                self.node_timeout or "None",
            )

            async for event in self._execute_graph(invocation_state):
                yield event.as_dict()

            # Set final status based on execution results
            if self.state.failed_nodes:
                self.state.status = Status.FAILED
            elif self.state.status == Status.EXECUTING:
                self.state.status = Status.COMPLETED

            logger.debug("status=<%s> | graph execution completed", self.state.status)

            # Yield final result (consistent with Agent's AgentResultEvent format)
            result = self._build_result()

            # Use the same event format as Agent for consistency
            yield MultiAgentResultEvent(result=result).as_dict()

        except Exception:
            logger.exception("graph execution failed")
            self.state.status = Status.FAILED
            raise
        finally:
            self.state.execution_time = round((time.time() - start_time) * 1000)
            await self.hooks.invoke_callbacks_async(AfterMultiAgentInvocationEvent(self))
            self._resume_from_session = False
            self._resume_next_nodes.clear()

GraphBuilder

Builder pattern for constructing graphs.

Source code in strands/multiagent/graph.py

class GraphBuilder:
    """Builder pattern for constructing graphs."""

    def __init__(self) -> None:
        """Initialize GraphBuilder with empty collections."""
        self.nodes: dict[str, GraphNode] = {}
        self.edges: set[GraphEdge] = set()
        self.entry_points: set[GraphNode] = set()

        # Configuration options
        self._max_node_executions: Optional[int] = None
        self._execution_timeout: Optional[float] = None
        self._node_timeout: Optional[float] = None
        self._reset_on_revisit: bool = False
        self._id: str = _DEFAULT_GRAPH_ID
        self._session_manager: Optional[SessionManager] = None
        self._hooks: Optional[list[HookProvider]] = None

    def add_node(self, executor: Agent | MultiAgentBase, node_id: str | None = None) -> GraphNode:
        """Add an Agent or MultiAgentBase instance as a node to the graph."""
        _validate_node_executor(executor, self.nodes)

        # Auto-generate node_id if not provided
        if node_id is None:
            node_id = getattr(executor, "id", None) or getattr(executor, "name", None) or f"node_{len(self.nodes)}"

        if node_id in self.nodes:
            raise ValueError(f"Node '{node_id}' already exists")

        node = GraphNode(node_id=node_id, executor=executor)
        self.nodes[node_id] = node
        return node

    def add_edge(
        self,
        from_node: str | GraphNode,
        to_node: str | GraphNode,
        condition: Callable[[GraphState], bool] | None = None,
    ) -> GraphEdge:
        """Add an edge between two nodes with optional condition function that receives full GraphState."""

        def resolve_node(node: str | GraphNode, node_type: str) -> GraphNode:
            if isinstance(node, str):
                if node not in self.nodes:
                    raise ValueError(f"{node_type} node '{node}' not found")
                return self.nodes[node]
            else:
                if node not in self.nodes.values():
                    raise ValueError(f"{node_type} node object has not been added to the graph, use graph.add_node")
                return node

        from_node_obj = resolve_node(from_node, "Source")
        to_node_obj = resolve_node(to_node, "Target")

        # Add edge and update dependencies
        edge = GraphEdge(from_node=from_node_obj, to_node=to_node_obj, condition=condition)
        self.edges.add(edge)
        to_node_obj.dependencies.add(from_node_obj)
        return edge

    def set_entry_point(self, node_id: str) -> "GraphBuilder":
        """Set a node as an entry point for graph execution."""
        if node_id not in self.nodes:
            raise ValueError(f"Node '{node_id}' not found")
        self.entry_points.add(self.nodes[node_id])
        return self

    def reset_on_revisit(self, enabled: bool = True) -> "GraphBuilder":
        """Control whether nodes reset their state when revisited.

        When enabled, nodes will reset their messages and state to initial values
        each time they are revisited (re-executed). This is useful for stateless
        behavior where nodes should start fresh on each revisit.

        Args:
            enabled: Whether to reset node state when revisited (default: True)
        """
        self._reset_on_revisit = enabled
        return self

    def set_max_node_executions(self, max_executions: int) -> "GraphBuilder":
        """Set maximum number of node executions allowed.

        Args:
            max_executions: Maximum total node executions (None for no limit)
        """
        self._max_node_executions = max_executions
        return self

    def set_execution_timeout(self, timeout: float) -> "GraphBuilder":
        """Set total execution timeout.

        Args:
            timeout: Total execution timeout in seconds (None for no limit)
        """
        self._execution_timeout = timeout
        return self

    def set_node_timeout(self, timeout: float) -> "GraphBuilder":
        """Set individual node execution timeout.

        Args:
            timeout: Individual node timeout in seconds (None for no limit)
        """
        self._node_timeout = timeout
        return self

    def set_graph_id(self, graph_id: str) -> "GraphBuilder":
        """Set graph id.

        Args:
            graph_id: Unique graph id
        """
        self._id = graph_id
        return self

    def set_session_manager(self, session_manager: SessionManager) -> "GraphBuilder":
        """Set session manager for the graph.

        Args:
            session_manager: SessionManager instance
        """
        self._session_manager = session_manager
        return self

    def set_hook_providers(self, hooks: list[HookProvider]) -> "GraphBuilder":
        """Set hook providers for the graph.

        Args:
            hooks: Customer hooks user passes in
        """
        self._hooks = hooks
        return self

    def build(self) -> "Graph":
        """Build and validate the graph with configured settings."""
        if not self.nodes:
            raise ValueError("Graph must contain at least one node")

        # Auto-detect entry points if none specified
        if not self.entry_points:
            self.entry_points = {node for node_id, node in self.nodes.items() if not node.dependencies}
            logger.debug(
                "entry_points=<%s> | auto-detected entrypoints", ", ".join(node.node_id for node in self.entry_points)
            )
            if not self.entry_points:
                raise ValueError("No entry points found - all nodes have dependencies")

        # Validate entry points and check for cycles
        self._validate_graph()

        return Graph(
            nodes=self.nodes.copy(),
            edges=self.edges.copy(),
            entry_points=self.entry_points.copy(),
            max_node_executions=self._max_node_executions,
            execution_timeout=self._execution_timeout,
            node_timeout=self._node_timeout,
            reset_on_revisit=self._reset_on_revisit,
            session_manager=self._session_manager,
            hooks=self._hooks,
            id=self._id,
        )

    def _validate_graph(self) -> None:
        """Validate graph structure."""
        # Validate entry points exist
        entry_point_ids = {node.node_id for node in self.entry_points}
        invalid_entries = entry_point_ids - set(self.nodes.keys())
        if invalid_entries:
            raise ValueError(f"Entry points not found in nodes: {invalid_entries}")

        # Warn about potential infinite loops if no execution limits are set
        if self._max_node_executions is None and self._execution_timeout is None:
            logger.warning("Graph without execution limits may run indefinitely if cycles exist")

__init__()

Initialize GraphBuilder with empty collections.

Source code in strands/multiagent/graph.py

def __init__(self) -> None:
    """Initialize GraphBuilder with empty collections."""
    self.nodes: dict[str, GraphNode] = {}
    self.edges: set[GraphEdge] = set()
    self.entry_points: set[GraphNode] = set()

    # Configuration options
    self._max_node_executions: Optional[int] = None
    self._execution_timeout: Optional[float] = None
    self._node_timeout: Optional[float] = None
    self._reset_on_revisit: bool = False
    self._id: str = _DEFAULT_GRAPH_ID
    self._session_manager: Optional[SessionManager] = None
    self._hooks: Optional[list[HookProvider]] = None

add_edge(from_node, to_node, condition=None)

Add an edge between two nodes with optional condition function that receives full GraphState.

Source code in strands/multiagent/graph.py

def add_edge(
    self,
    from_node: str | GraphNode,
    to_node: str | GraphNode,
    condition: Callable[[GraphState], bool] | None = None,
) -> GraphEdge:
    """Add an edge between two nodes with optional condition function that receives full GraphState."""

    def resolve_node(node: str | GraphNode, node_type: str) -> GraphNode:
        if isinstance(node, str):
            if node not in self.nodes:
                raise ValueError(f"{node_type} node '{node}' not found")
            return self.nodes[node]
        else:
            if node not in self.nodes.values():
                raise ValueError(f"{node_type} node object has not been added to the graph, use graph.add_node")
            return node

    from_node_obj = resolve_node(from_node, "Source")
    to_node_obj = resolve_node(to_node, "Target")

    # Add edge and update dependencies
    edge = GraphEdge(from_node=from_node_obj, to_node=to_node_obj, condition=condition)
    self.edges.add(edge)
    to_node_obj.dependencies.add(from_node_obj)
    return edge

add_node(executor, node_id=None)

Add an Agent or MultiAgentBase instance as a node to the graph.

Source code in strands/multiagent/graph.py

def add_node(self, executor: Agent | MultiAgentBase, node_id: str | None = None) -> GraphNode:
    """Add an Agent or MultiAgentBase instance as a node to the graph."""
    _validate_node_executor(executor, self.nodes)

    # Auto-generate node_id if not provided
    if node_id is None:
        node_id = getattr(executor, "id", None) or getattr(executor, "name", None) or f"node_{len(self.nodes)}"

    if node_id in self.nodes:
        raise ValueError(f"Node '{node_id}' already exists")

    node = GraphNode(node_id=node_id, executor=executor)
    self.nodes[node_id] = node
    return node

build()

Build and validate the graph with configured settings.

Source code in strands/multiagent/graph.py

def build(self) -> "Graph":
    """Build and validate the graph with configured settings."""
    if not self.nodes:
        raise ValueError("Graph must contain at least one node")

    # Auto-detect entry points if none specified
    if not self.entry_points:
        self.entry_points = {node for node_id, node in self.nodes.items() if not node.dependencies}
        logger.debug(
            "entry_points=<%s> | auto-detected entrypoints", ", ".join(node.node_id for node in self.entry_points)
        )
        if not self.entry_points:
            raise ValueError("No entry points found - all nodes have dependencies")

    # Validate entry points and check for cycles
    self._validate_graph()

    return Graph(
        nodes=self.nodes.copy(),
        edges=self.edges.copy(),
        entry_points=self.entry_points.copy(),
        max_node_executions=self._max_node_executions,
        execution_timeout=self._execution_timeout,
        node_timeout=self._node_timeout,
        reset_on_revisit=self._reset_on_revisit,
        session_manager=self._session_manager,
        hooks=self._hooks,
        id=self._id,
    )

reset_on_revisit(enabled=True)

Control whether nodes reset their state when revisited.

When enabled, nodes will reset their messages and state to initial values each time they are revisited (re-executed). This is useful for stateless behavior where nodes should start fresh on each revisit.

Parameters:

Name	Type	Description	Default
enabled	bool	Whether to reset node state when revisited (default: True)	True

Source code in strands/multiagent/graph.py

def reset_on_revisit(self, enabled: bool = True) -> "GraphBuilder":
    """Control whether nodes reset their state when revisited.

    When enabled, nodes will reset their messages and state to initial values
    each time they are revisited (re-executed). This is useful for stateless
    behavior where nodes should start fresh on each revisit.

    Args:
        enabled: Whether to reset node state when revisited (default: True)
    """
    self._reset_on_revisit = enabled
    return self

set_entry_point(node_id)

Set a node as an entry point for graph execution.

Source code in strands/multiagent/graph.py

def set_entry_point(self, node_id: str) -> "GraphBuilder":
    """Set a node as an entry point for graph execution."""
    if node_id not in self.nodes:
        raise ValueError(f"Node '{node_id}' not found")
    self.entry_points.add(self.nodes[node_id])
    return self

set_execution_timeout(timeout)

Set total execution timeout.

Parameters:

Name	Type	Description	Default
timeout	float	Total execution timeout in seconds (None for no limit)	required

Source code in strands/multiagent/graph.py

def set_execution_timeout(self, timeout: float) -> "GraphBuilder":
    """Set total execution timeout.

    Args:
        timeout: Total execution timeout in seconds (None for no limit)
    """
    self._execution_timeout = timeout
    return self

set_graph_id(graph_id)

Set graph id.

Parameters:

Name	Type	Description	Default
graph_id	str	Unique graph id	required

Source code in strands/multiagent/graph.py

def set_graph_id(self, graph_id: str) -> "GraphBuilder":
    """Set graph id.

    Args:
        graph_id: Unique graph id
    """
    self._id = graph_id
    return self

set_hook_providers(hooks)

Set hook providers for the graph.

Parameters:

Name	Type	Description	Default
hooks	list[HookProvider]	Customer hooks user passes in	required

Source code in strands/multiagent/graph.py

def set_hook_providers(self, hooks: list[HookProvider]) -> "GraphBuilder":
    """Set hook providers for the graph.

    Args:
        hooks: Customer hooks user passes in
    """
    self._hooks = hooks
    return self

set_max_node_executions(max_executions)

Set maximum number of node executions allowed.

Parameters:

Name	Type	Description	Default
max_executions	int	Maximum total node executions (None for no limit)	required

Source code in strands/multiagent/graph.py

def set_max_node_executions(self, max_executions: int) -> "GraphBuilder":
    """Set maximum number of node executions allowed.

    Args:
        max_executions: Maximum total node executions (None for no limit)
    """
    self._max_node_executions = max_executions
    return self

set_node_timeout(timeout)

Set individual node execution timeout.

Parameters:

Name	Type	Description	Default
timeout	float	Individual node timeout in seconds (None for no limit)	required

Source code in strands/multiagent/graph.py

def set_node_timeout(self, timeout: float) -> "GraphBuilder":
    """Set individual node execution timeout.

    Args:
        timeout: Individual node timeout in seconds (None for no limit)
    """
    self._node_timeout = timeout
    return self

set_session_manager(session_manager)

Set session manager for the graph.

Parameters:

Name	Type	Description	Default
session_manager	SessionManager	SessionManager instance	required

Source code in strands/multiagent/graph.py

def set_session_manager(self, session_manager: SessionManager) -> "GraphBuilder":
    """Set session manager for the graph.

    Args:
        session_manager: SessionManager instance
    """
    self._session_manager = session_manager
    return self

GraphEdge dataclass

Represents an edge in the graph with an optional condition.

Source code in strands/multiagent/graph.py

@dataclass
class GraphEdge:
    """Represents an edge in the graph with an optional condition."""

    from_node: "GraphNode"
    to_node: "GraphNode"
    condition: Callable[[GraphState], bool] | None = None

    def __hash__(self) -> int:
        """Return hash for GraphEdge based on from_node and to_node."""
        return hash((self.from_node.node_id, self.to_node.node_id))

    def should_traverse(self, state: GraphState) -> bool:
        """Check if this edge should be traversed based on condition."""
        if self.condition is None:
            return True
        return self.condition(state)

__hash__()

Return hash for GraphEdge based on from_node and to_node.

Source code in strands/multiagent/graph.py

def __hash__(self) -> int:
    """Return hash for GraphEdge based on from_node and to_node."""
    return hash((self.from_node.node_id, self.to_node.node_id))

should_traverse(state)

Check if this edge should be traversed based on condition.

Source code in strands/multiagent/graph.py

def should_traverse(self, state: GraphState) -> bool:
    """Check if this edge should be traversed based on condition."""
    if self.condition is None:
        return True
    return self.condition(state)

GraphNode dataclass

Represents a node in the graph.

The execution_status tracks the node's lifecycle within graph orchestration: - PENDING: Node hasn't started executing yet - EXECUTING: Node is currently running - COMPLETED/FAILED: Node finished executing (regardless of result quality)

Source code in strands/multiagent/graph.py

@dataclass
class GraphNode:
    """Represents a node in the graph.

    The execution_status tracks the node's lifecycle within graph orchestration:
    - PENDING: Node hasn't started executing yet
    - EXECUTING: Node is currently running
    - COMPLETED/FAILED: Node finished executing (regardless of result quality)
    """

    node_id: str
    executor: Agent | MultiAgentBase
    dependencies: set["GraphNode"] = field(default_factory=set)
    execution_status: Status = Status.PENDING
    result: NodeResult | None = None
    execution_time: int = 0
    _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
        if hasattr(self.executor, "messages"):
            self._initial_messages = copy.deepcopy(self.executor.messages)

        if hasattr(self.executor, "state") and hasattr(self.executor.state, "get"):
            self._initial_state = AgentState(self.executor.state.get())

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

        This is useful when nodes are executed multiple times and need to start
        fresh on each execution, providing stateless behavior.
        """
        if hasattr(self.executor, "messages"):
            self.executor.messages = copy.deepcopy(self._initial_messages)

        if hasattr(self.executor, "state"):
            self.executor.state = AgentState(self._initial_state.get())

        # Reset execution status
        self.execution_status = Status.PENDING
        self.result = None

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

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

__eq__(other)

Return equality for GraphNode based on node_id.

Source code in strands/multiagent/graph.py

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

__hash__()

Return hash for GraphNode based on node_id.

Source code in strands/multiagent/graph.py

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

__post_init__()

Capture initial executor state after initialization.

Source code in strands/multiagent/graph.py

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

    if hasattr(self.executor, "state") and hasattr(self.executor.state, "get"):
        self._initial_state = AgentState(self.executor.state.get())

reset_executor_state()

Reset GraphNode executor state to initial state when graph was created.

This is useful when nodes are executed multiple times and need to start fresh on each execution, providing stateless behavior.

Source code in strands/multiagent/graph.py

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

    This is useful when nodes are executed multiple times and need to start
    fresh on each execution, providing stateless behavior.
    """
    if hasattr(self.executor, "messages"):
        self.executor.messages = copy.deepcopy(self._initial_messages)

    if hasattr(self.executor, "state"):
        self.executor.state = AgentState(self._initial_state.get())

    # Reset execution status
    self.execution_status = Status.PENDING
    self.result = None

GraphResult dataclass

Bases: MultiAgentResult

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

Source code in strands/multiagent/graph.py

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

    total_nodes: int = 0
    completed_nodes: int = 0
    failed_nodes: int = 0
    execution_order: list["GraphNode"] = field(default_factory=list)
    edges: list[Tuple["GraphNode", "GraphNode"]] = field(default_factory=list)
    entry_points: list["GraphNode"] = field(default_factory=list)

GraphState dataclass

Graph execution state.

Attributes:

Name	Type	Description
status	Status	Current execution status of the graph.
completed_nodes	set[GraphNode]	Set of nodes that have completed execution.
failed_nodes	set[GraphNode]	Set of nodes that failed during execution.
execution_order	list[GraphNode]	List of nodes in the order they were executed.
task	MultiAgentInput	The original input prompt/query provided to the graph execution. This represents the actual work to be performed by the graph as a whole. Entry point nodes receive this task as their input if they have no dependencies.

Source code in strands/multiagent/graph.py

@dataclass
class GraphState:
    """Graph execution state.

    Attributes:
        status: Current execution status of the graph.
        completed_nodes: Set of nodes that have completed execution.
        failed_nodes: Set of nodes that failed during execution.
        execution_order: List of nodes in the order they were executed.
        task: The original input prompt/query provided to the graph execution.
              This represents the actual work to be performed by the graph as a whole.
              Entry point nodes receive this task as their input if they have no dependencies.
    """

    # Task (with default empty string)
    task: MultiAgentInput = ""

    # Execution state
    status: Status = Status.PENDING
    completed_nodes: set["GraphNode"] = field(default_factory=set)
    failed_nodes: set["GraphNode"] = field(default_factory=set)
    execution_order: list["GraphNode"] = field(default_factory=list)
    start_time: float = field(default_factory=time.time)

    # Results
    results: dict[str, NodeResult] = field(default_factory=dict)

    # Accumulated metrics
    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

    # Graph structure info
    total_nodes: int = 0
    edges: list[Tuple["GraphNode", "GraphNode"]] = field(default_factory=list)
    entry_points: list["GraphNode"] = field(default_factory=list)

    def should_continue(
        self,
        max_node_executions: Optional[int],
        execution_timeout: Optional[float],
    ) -> Tuple[bool, str]:
        """Check if the graph should continue execution.

        Returns: (should_continue, reason)
        """
        # Check node execution limit (only if set)
        if max_node_executions is not None and len(self.execution_order) >= max_node_executions:
            return False, f"Max node executions reached: {max_node_executions}"

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

        return True, "Continuing"

should_continue(max_node_executions, execution_timeout)

Check if the graph should continue execution.

Returns: (should_continue, reason)

Source code in strands/multiagent/graph.py

def should_continue(
    self,
    max_node_executions: Optional[int],
    execution_timeout: Optional[float],
) -> Tuple[bool, str]:
    """Check if the graph should continue execution.

    Returns: (should_continue, reason)
    """
    # Check node execution limit (only if set)
    if max_node_executions is not None and len(self.execution_order) >= max_node_executions:
        return False, f"Max node executions reached: {max_node_executions}"

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

    return True, "Continuing"

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

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,
        }
    )

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."
    )

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"

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

_validate_node_executor(executor, existing_nodes=None)

Validate a node executor for graph compatibility.

Parameters:

Name	Type	Description	Default
executor	Agent | MultiAgentBase	The executor to validate	required
existing_nodes	dict[str, GraphNode] | None	Optional dict of existing nodes to check for duplicates	None

Source code in strands/multiagent/graph.py

def _validate_node_executor(
    executor: Agent | MultiAgentBase, existing_nodes: dict[str, GraphNode] | None = None
) -> None:
    """Validate a node executor for graph compatibility.

    Args:
        executor: The executor to validate
        existing_nodes: Optional dict of existing nodes to check for duplicates
    """
    # Check for duplicate node instances
    if existing_nodes:
        seen_instances = {id(node.executor) for node in existing_nodes.values()}
        if id(executor) in seen_instances:
            raise ValueError("Duplicate node instance detected. Each node must have a unique object instance.")

    # Validate Agent-specific constraints
    if isinstance(executor, Agent):
        # Check for session persistence
        if executor._session_manager is not None:
            raise ValueError("Session persistence is not supported for Graph agents yet.")

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