strands.tools.executors.sequential

Sequential tool executor implementation.

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

SequentialToolExecutor

Bases: ToolExecutor

Sequential tool executor.

Source code in strands/tools/executors/sequential.py

class SequentialToolExecutor(ToolExecutor):
    """Sequential tool executor."""

    @override
    async def _execute(
        self,
        agent: "Agent",
        tool_uses: list[ToolUse],
        tool_results: list[ToolResult],
        cycle_trace: Trace,
        cycle_span: Any,
        invocation_state: dict[str, Any],
        structured_output_context: "StructuredOutputContext | None" = None,
    ) -> AsyncGenerator[TypedEvent, None]:
        """Execute tools sequentially.

        Breaks early if an interrupt is raised by the user.

        Args:
            agent: The agent for which tools are being executed.
            tool_uses: Metadata and inputs for the tools to be executed.
            tool_results: List of tool results from each tool execution.
            cycle_trace: Trace object for the current event loop cycle.
            cycle_span: Span object for tracing the cycle.
            invocation_state: Context for the tool invocation.
            structured_output_context: Context for structured output handling.

        Yields:
            Events from the tool execution stream.
        """
        interrupted = False

        for tool_use in tool_uses:
            events = ToolExecutor._stream_with_trace(
                agent, tool_use, tool_results, cycle_trace, cycle_span, invocation_state, structured_output_context
            )
            async for event in events:
                if isinstance(event, ToolInterruptEvent):
                    interrupted = True

                yield event

            if interrupted:
                break

StructuredOutputContext

Per-invocation context for structured output execution.

Source code in strands/tools/structured_output/_structured_output_context.py

class StructuredOutputContext:
    """Per-invocation context for structured output execution."""

    def __init__(self, structured_output_model: Type[BaseModel] | None = None):
        """Initialize a new structured output context.

        Args:
            structured_output_model: Optional Pydantic model type for structured output.
        """
        self.results: dict[str, BaseModel] = {}
        self.structured_output_model: Type[BaseModel] | None = structured_output_model
        self.structured_output_tool: StructuredOutputTool | None = None
        self.forced_mode: bool = False
        self.force_attempted: bool = False
        self.tool_choice: ToolChoice | None = None
        self.stop_loop: bool = False
        self.expected_tool_name: Optional[str] = None

        if structured_output_model:
            self.structured_output_tool = StructuredOutputTool(structured_output_model)
            self.expected_tool_name = self.structured_output_tool.tool_name

    @property
    def is_enabled(self) -> bool:
        """Check if structured output is enabled for this context.

        Returns:
            True if a structured output model is configured, False otherwise.
        """
        return self.structured_output_model is not None

    def store_result(self, tool_use_id: str, result: BaseModel) -> None:
        """Store a validated structured output result.

        Args:
            tool_use_id: Unique identifier for the tool use.
            result: Validated Pydantic model instance.
        """
        self.results[tool_use_id] = result

    def get_result(self, tool_use_id: str) -> BaseModel | None:
        """Retrieve a stored structured output result.

        Args:
            tool_use_id: Unique identifier for the tool use.

        Returns:
            The validated Pydantic model instance, or None if not found.
        """
        return self.results.get(tool_use_id)

    def set_forced_mode(self, tool_choice: dict | None = None) -> None:
        """Mark this context as being in forced structured output mode.

        Args:
            tool_choice: Optional tool choice configuration.
        """
        if not self.is_enabled:
            return
        self.forced_mode = True
        self.force_attempted = True
        self.tool_choice = tool_choice or {"any": {}}

    def has_structured_output_tool(self, tool_uses: list[ToolUse]) -> bool:
        """Check if any tool uses are for the structured output tool.

        Args:
            tool_uses: List of tool use dictionaries to check.

        Returns:
            True if any tool use matches the expected structured output tool name,
            False if no structured output tool is present or expected.
        """
        if not self.expected_tool_name:
            return False
        return any(tool_use.get("name") == self.expected_tool_name for tool_use in tool_uses)

    def get_tool_spec(self) -> Optional[ToolSpec]:
        """Get the tool specification for structured output.

        Returns:
            Tool specification, or None if no structured output model.
        """
        if self.structured_output_tool:
            return self.structured_output_tool.tool_spec
        return None

    def extract_result(self, tool_uses: list[ToolUse]) -> BaseModel | None:
        """Extract and remove structured output result from stored results.

        Args:
            tool_uses: List of tool use dictionaries from the current execution cycle.

        Returns:
            The structured output result if found, or None if no result available.
        """
        if not self.has_structured_output_tool(tool_uses):
            return None

        for tool_use in tool_uses:
            if tool_use.get("name") == self.expected_tool_name:
                tool_use_id = str(tool_use.get("toolUseId", ""))
                result = self.results.pop(tool_use_id, None)
                if result is not None:
                    logger.debug("Extracted structured output for %s", tool_use.get("name"))
                    return result
        return None

    def register_tool(self, registry: "ToolRegistry") -> None:
        """Register the structured output tool with the registry.

        Args:
            registry: The tool registry to register the tool with.
        """
        if self.structured_output_tool and self.structured_output_tool.tool_name not in registry.dynamic_tools:
            registry.register_dynamic_tool(self.structured_output_tool)
            logger.debug("Registered structured output tool: %s", self.structured_output_tool.tool_name)

    def cleanup(self, registry: "ToolRegistry") -> None:
        """Clean up the registered structured output tool from the registry.

        Args:
            registry: The tool registry to clean up the tool from.
        """
        if self.structured_output_tool and self.structured_output_tool.tool_name in registry.dynamic_tools:
            del registry.dynamic_tools[self.structured_output_tool.tool_name]
            logger.debug("Cleaned up structured output tool: %s", self.structured_output_tool.tool_name)

is_enabled property

Check if structured output is enabled for this context.

Returns:

Type	Description
bool	True if a structured output model is configured, False otherwise.

__init__(structured_output_model=None)

Initialize a new structured output context.

Parameters:

Name	Type	Description	Default
structured_output_model	Type[BaseModel] | None	Optional Pydantic model type for structured output.	None

Source code in strands/tools/structured_output/_structured_output_context.py

def __init__(self, structured_output_model: Type[BaseModel] | None = None):
    """Initialize a new structured output context.

    Args:
        structured_output_model: Optional Pydantic model type for structured output.
    """
    self.results: dict[str, BaseModel] = {}
    self.structured_output_model: Type[BaseModel] | None = structured_output_model
    self.structured_output_tool: StructuredOutputTool | None = None
    self.forced_mode: bool = False
    self.force_attempted: bool = False
    self.tool_choice: ToolChoice | None = None
    self.stop_loop: bool = False
    self.expected_tool_name: Optional[str] = None

    if structured_output_model:
        self.structured_output_tool = StructuredOutputTool(structured_output_model)
        self.expected_tool_name = self.structured_output_tool.tool_name

cleanup(registry)

Clean up the registered structured output tool from the registry.

Parameters:

Name	Type	Description	Default
registry	ToolRegistry	The tool registry to clean up the tool from.	required

Source code in strands/tools/structured_output/_structured_output_context.py

def cleanup(self, registry: "ToolRegistry") -> None:
    """Clean up the registered structured output tool from the registry.

    Args:
        registry: The tool registry to clean up the tool from.
    """
    if self.structured_output_tool and self.structured_output_tool.tool_name in registry.dynamic_tools:
        del registry.dynamic_tools[self.structured_output_tool.tool_name]
        logger.debug("Cleaned up structured output tool: %s", self.structured_output_tool.tool_name)

extract_result(tool_uses)

Extract and remove structured output result from stored results.

Parameters:

Name	Type	Description	Default
tool_uses	list[ToolUse]	List of tool use dictionaries from the current execution cycle.	required

Returns:

Type	Description
BaseModel | None	The structured output result if found, or None if no result available.

Source code in strands/tools/structured_output/_structured_output_context.py

def extract_result(self, tool_uses: list[ToolUse]) -> BaseModel | None:
    """Extract and remove structured output result from stored results.

    Args:
        tool_uses: List of tool use dictionaries from the current execution cycle.

    Returns:
        The structured output result if found, or None if no result available.
    """
    if not self.has_structured_output_tool(tool_uses):
        return None

    for tool_use in tool_uses:
        if tool_use.get("name") == self.expected_tool_name:
            tool_use_id = str(tool_use.get("toolUseId", ""))
            result = self.results.pop(tool_use_id, None)
            if result is not None:
                logger.debug("Extracted structured output for %s", tool_use.get("name"))
                return result
    return None

get_result(tool_use_id)

Retrieve a stored structured output result.

Parameters:

Name	Type	Description	Default
tool_use_id	str	Unique identifier for the tool use.	required

Returns:

Type	Description
BaseModel | None	The validated Pydantic model instance, or None if not found.

Source code in strands/tools/structured_output/_structured_output_context.py

def get_result(self, tool_use_id: str) -> BaseModel | None:
    """Retrieve a stored structured output result.

    Args:
        tool_use_id: Unique identifier for the tool use.

    Returns:
        The validated Pydantic model instance, or None if not found.
    """
    return self.results.get(tool_use_id)

get_tool_spec()

Get the tool specification for structured output.

Returns:

Type	Description
Optional[ToolSpec]	Tool specification, or None if no structured output model.

Source code in strands/tools/structured_output/_structured_output_context.py

def get_tool_spec(self) -> Optional[ToolSpec]:
    """Get the tool specification for structured output.

    Returns:
        Tool specification, or None if no structured output model.
    """
    if self.structured_output_tool:
        return self.structured_output_tool.tool_spec
    return None

has_structured_output_tool(tool_uses)

Check if any tool uses are for the structured output tool.

Parameters:

Name	Type	Description	Default
tool_uses	list[ToolUse]	List of tool use dictionaries to check.	required

Returns:

Type	Description
bool	True if any tool use matches the expected structured output tool name,
bool	False if no structured output tool is present or expected.

Source code in strands/tools/structured_output/_structured_output_context.py

def has_structured_output_tool(self, tool_uses: list[ToolUse]) -> bool:
    """Check if any tool uses are for the structured output tool.

    Args:
        tool_uses: List of tool use dictionaries to check.

    Returns:
        True if any tool use matches the expected structured output tool name,
        False if no structured output tool is present or expected.
    """
    if not self.expected_tool_name:
        return False
    return any(tool_use.get("name") == self.expected_tool_name for tool_use in tool_uses)

register_tool(registry)

Register the structured output tool with the registry.

Parameters:

Name	Type	Description	Default
registry	ToolRegistry	The tool registry to register the tool with.	required

Source code in strands/tools/structured_output/_structured_output_context.py

def register_tool(self, registry: "ToolRegistry") -> None:
    """Register the structured output tool with the registry.

    Args:
        registry: The tool registry to register the tool with.
    """
    if self.structured_output_tool and self.structured_output_tool.tool_name not in registry.dynamic_tools:
        registry.register_dynamic_tool(self.structured_output_tool)
        logger.debug("Registered structured output tool: %s", self.structured_output_tool.tool_name)

set_forced_mode(tool_choice=None)

Mark this context as being in forced structured output mode.

Parameters:

Name	Type	Description	Default
tool_choice	dict | None	Optional tool choice configuration.	None

Source code in strands/tools/structured_output/_structured_output_context.py

def set_forced_mode(self, tool_choice: dict | None = None) -> None:
    """Mark this context as being in forced structured output mode.

    Args:
        tool_choice: Optional tool choice configuration.
    """
    if not self.is_enabled:
        return
    self.forced_mode = True
    self.force_attempted = True
    self.tool_choice = tool_choice or {"any": {}}

store_result(tool_use_id, result)

Store a validated structured output result.

Parameters:

Name	Type	Description	Default
tool_use_id	str	Unique identifier for the tool use.	required
result	BaseModel	Validated Pydantic model instance.	required

Source code in strands/tools/structured_output/_structured_output_context.py

def store_result(self, tool_use_id: str, result: BaseModel) -> None:
    """Store a validated structured output result.

    Args:
        tool_use_id: Unique identifier for the tool use.
        result: Validated Pydantic model instance.
    """
    self.results[tool_use_id] = result

ToolExecutor

Bases: ABC

Abstract base class for tool executors.

Source code in strands/tools/executors/_executor.py

class ToolExecutor(abc.ABC):
    """Abstract base class for tool executors."""

    @staticmethod
    def _is_agent(agent: "Agent | BidiAgent") -> bool:
        """Check if the agent is an Agent instance, otherwise we assume BidiAgent.

        Note, we use a runtime import to avoid a circular dependency error.
        """
        from ...agent import Agent

        return isinstance(agent, Agent)

    @staticmethod
    async def _invoke_before_tool_call_hook(
        agent: "Agent | BidiAgent",
        tool_func: Any,
        tool_use: ToolUse,
        invocation_state: dict[str, Any],
    ) -> tuple[BeforeToolCallEvent | BidiBeforeToolCallEvent, list[Interrupt]]:
        """Invoke the appropriate before tool call hook based on agent type."""
        kwargs = {
            "selected_tool": tool_func,
            "tool_use": tool_use,
            "invocation_state": invocation_state,
        }
        event = (
            BeforeToolCallEvent(agent=cast("Agent", agent), **kwargs)
            if ToolExecutor._is_agent(agent)
            else BidiBeforeToolCallEvent(agent=cast("BidiAgent", agent), **kwargs)
        )

        return await agent.hooks.invoke_callbacks_async(event)

    @staticmethod
    async def _invoke_after_tool_call_hook(
        agent: "Agent | BidiAgent",
        selected_tool: Any,
        tool_use: ToolUse,
        invocation_state: dict[str, Any],
        result: ToolResult,
        exception: Exception | None = None,
        cancel_message: str | None = None,
    ) -> tuple[AfterToolCallEvent | BidiAfterToolCallEvent, list[Interrupt]]:
        """Invoke the appropriate after tool call hook based on agent type."""
        kwargs = {
            "selected_tool": selected_tool,
            "tool_use": tool_use,
            "invocation_state": invocation_state,
            "result": result,
            "exception": exception,
            "cancel_message": cancel_message,
        }
        event = (
            AfterToolCallEvent(agent=cast("Agent", agent), **kwargs)
            if ToolExecutor._is_agent(agent)
            else BidiAfterToolCallEvent(agent=cast("BidiAgent", agent), **kwargs)
        )

        return await agent.hooks.invoke_callbacks_async(event)

    @staticmethod
    async def _stream(
        agent: "Agent | BidiAgent",
        tool_use: ToolUse,
        tool_results: list[ToolResult],
        invocation_state: dict[str, Any],
        structured_output_context: StructuredOutputContext | None = None,
        **kwargs: Any,
    ) -> AsyncGenerator[TypedEvent, None]:
        """Stream tool events.

        This method adds additional logic to the stream invocation including:

        - Tool lookup and validation
        - Before/after hook execution
        - Tracing and metrics collection
        - Error handling and recovery
        - Interrupt handling for human-in-the-loop workflows

        Args:
            agent: The agent (Agent or BidiAgent) for which the tool is being executed.
            tool_use: Metadata and inputs for the tool to be executed.
            tool_results: List of tool results from each tool execution.
            invocation_state: Context for the tool invocation.
            structured_output_context: Context for structured output management.
            **kwargs: Additional keyword arguments for future extensibility.

        Yields:
            Tool events with the last being the tool result.
        """
        logger.debug("tool_use=<%s> | streaming", tool_use)
        tool_name = tool_use["name"]
        structured_output_context = structured_output_context or StructuredOutputContext()

        tool_info = agent.tool_registry.dynamic_tools.get(tool_name)
        tool_func = tool_info if tool_info is not None else agent.tool_registry.registry.get(tool_name)
        tool_spec = tool_func.tool_spec if tool_func is not None else None

        current_span = trace_api.get_current_span()
        if current_span and tool_spec is not None:
            current_span.set_attribute("gen_ai.tool.description", tool_spec["description"])
            input_schema = tool_spec["inputSchema"]
            if "json" in input_schema:
                current_span.set_attribute("gen_ai.tool.json_schema", serialize(input_schema["json"]))

        invocation_state.update(
            {
                "agent": agent,
                "model": agent.model,
                "messages": agent.messages,
                "system_prompt": agent.system_prompt,
                "tool_config": ToolConfig(  # for backwards compatibility
                    tools=[{"toolSpec": tool_spec} for tool_spec in agent.tool_registry.get_all_tool_specs()],
                    toolChoice=cast(ToolChoice, {"auto": ToolChoiceAuto()}),
                ),
            }
        )

        before_event, interrupts = await ToolExecutor._invoke_before_tool_call_hook(
            agent, tool_func, tool_use, invocation_state
        )

        if interrupts:
            yield ToolInterruptEvent(tool_use, interrupts)
            return

        if before_event.cancel_tool:
            cancel_message = (
                before_event.cancel_tool if isinstance(before_event.cancel_tool, str) else "tool cancelled by user"
            )
            yield ToolCancelEvent(tool_use, cancel_message)

            cancel_result: ToolResult = {
                "toolUseId": str(tool_use.get("toolUseId")),
                "status": "error",
                "content": [{"text": cancel_message}],
            }

            after_event, _ = await ToolExecutor._invoke_after_tool_call_hook(
                agent, None, tool_use, invocation_state, cancel_result, cancel_message=cancel_message
            )
            yield ToolResultEvent(after_event.result)
            tool_results.append(after_event.result)
            return

        try:
            selected_tool = before_event.selected_tool
            tool_use = before_event.tool_use
            invocation_state = before_event.invocation_state

            if not selected_tool:
                if tool_func == selected_tool:
                    logger.error(
                        "tool_name=<%s>, available_tools=<%s> | tool not found in registry",
                        tool_name,
                        list(agent.tool_registry.registry.keys()),
                    )
                else:
                    logger.debug(
                        "tool_name=<%s>, tool_use_id=<%s> | a hook resulted in a non-existing tool call",
                        tool_name,
                        str(tool_use.get("toolUseId")),
                    )

                result: ToolResult = {
                    "toolUseId": str(tool_use.get("toolUseId")),
                    "status": "error",
                    "content": [{"text": f"Unknown tool: {tool_name}"}],
                }

                after_event, _ = await ToolExecutor._invoke_after_tool_call_hook(
                    agent, selected_tool, tool_use, invocation_state, result
                )
                yield ToolResultEvent(after_event.result)
                tool_results.append(after_event.result)
                return
            if structured_output_context.is_enabled:
                kwargs["structured_output_context"] = structured_output_context
            async for event in selected_tool.stream(tool_use, invocation_state, **kwargs):
                # Internal optimization; for built-in AgentTools, we yield TypedEvents out of .stream()
                # so that we don't needlessly yield ToolStreamEvents for non-generator callbacks.
                # In which case, as soon as we get a ToolResultEvent we're done and for ToolStreamEvent
                # we yield it directly; all other cases (non-sdk AgentTools), we wrap events in
                # ToolStreamEvent and the last event is just the result.

                if isinstance(event, ToolInterruptEvent):
                    yield event
                    return

                if isinstance(event, ToolResultEvent):
                    # below the last "event" must point to the tool_result
                    event = event.tool_result
                    break

                if isinstance(event, ToolStreamEvent):
                    yield event
                else:
                    yield ToolStreamEvent(tool_use, event)

            result = cast(ToolResult, event)

            after_event, _ = await ToolExecutor._invoke_after_tool_call_hook(
                agent, selected_tool, tool_use, invocation_state, result
            )

            yield ToolResultEvent(after_event.result)
            tool_results.append(after_event.result)

        except Exception as e:
            logger.exception("tool_name=<%s> | failed to process tool", tool_name)
            error_result: ToolResult = {
                "toolUseId": str(tool_use.get("toolUseId")),
                "status": "error",
                "content": [{"text": f"Error: {str(e)}"}],
            }

            after_event, _ = await ToolExecutor._invoke_after_tool_call_hook(
                agent, selected_tool, tool_use, invocation_state, error_result, exception=e
            )
            yield ToolResultEvent(after_event.result)
            tool_results.append(after_event.result)

    @staticmethod
    async def _stream_with_trace(
        agent: "Agent",
        tool_use: ToolUse,
        tool_results: list[ToolResult],
        cycle_trace: Trace,
        cycle_span: Any,
        invocation_state: dict[str, Any],
        structured_output_context: StructuredOutputContext | None = None,
        **kwargs: Any,
    ) -> AsyncGenerator[TypedEvent, None]:
        """Execute tool with tracing and metrics collection.

        Args:
            agent: The agent for which the tool is being executed.
            tool_use: Metadata and inputs for the tool to be executed.
            tool_results: List of tool results from each tool execution.
            cycle_trace: Trace object for the current event loop cycle.
            cycle_span: Span object for tracing the cycle.
            invocation_state: Context for the tool invocation.
            structured_output_context: Context for structured output management.
            **kwargs: Additional keyword arguments for future extensibility.

        Yields:
            Tool events with the last being the tool result.
        """
        tool_name = tool_use["name"]
        structured_output_context = structured_output_context or StructuredOutputContext()

        tracer = get_tracer()

        tool_call_span = tracer.start_tool_call_span(
            tool_use, cycle_span, custom_trace_attributes=agent.trace_attributes
        )
        tool_trace = Trace(f"Tool: {tool_name}", parent_id=cycle_trace.id, raw_name=tool_name)
        tool_start_time = time.time()

        with trace_api.use_span(tool_call_span):
            async for event in ToolExecutor._stream(
                agent, tool_use, tool_results, invocation_state, structured_output_context, **kwargs
            ):
                yield event

            if isinstance(event, ToolInterruptEvent):
                tracer.end_tool_call_span(tool_call_span, tool_result=None)
                return

            result_event = cast(ToolResultEvent, event)
            result = result_event.tool_result

            tool_success = result.get("status") == "success"
            tool_duration = time.time() - tool_start_time
            message = Message(role="user", content=[{"toolResult": result}])
            if ToolExecutor._is_agent(agent):
                agent.event_loop_metrics.add_tool_usage(tool_use, tool_duration, tool_trace, tool_success, message)
            cycle_trace.add_child(tool_trace)

            tracer.end_tool_call_span(tool_call_span, result)

    @abc.abstractmethod
    # pragma: no cover
    def _execute(
        self,
        agent: "Agent",
        tool_uses: list[ToolUse],
        tool_results: list[ToolResult],
        cycle_trace: Trace,
        cycle_span: Any,
        invocation_state: dict[str, Any],
        structured_output_context: "StructuredOutputContext | None" = None,
    ) -> AsyncGenerator[TypedEvent, None]:
        """Execute the given tools according to this executor's strategy.

        Args:
            agent: The agent for which tools are being executed.
            tool_uses: Metadata and inputs for the tools to be executed.
            tool_results: List of tool results from each tool execution.
            cycle_trace: Trace object for the current event loop cycle.
            cycle_span: Span object for tracing the cycle.
            invocation_state: Context for the tool invocation.
            structured_output_context: Context for structured output management.

        Yields:
            Events from the tool execution stream.
        """
        pass

ToolInterruptEvent

Bases: TypedEvent

Event emitted when a tool is interrupted.

Source code in strands/types/_events.py

class ToolInterruptEvent(TypedEvent):
    """Event emitted when a tool is interrupted."""

    def __init__(self, tool_use: ToolUse, interrupts: list[Interrupt]) -> None:
        """Set interrupt in the event payload."""
        super().__init__({"tool_interrupt_event": {"tool_use": tool_use, "interrupts": interrupts}})

    @property
    def tool_use_id(self) -> str:
        """The id of the tool interrupted."""
        return cast(ToolUse, cast(dict, self.get("tool_interrupt_event")).get("tool_use"))["toolUseId"]

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

interrupts property

The interrupt instances.

tool_use_id property

The id of the tool interrupted.

__init__(tool_use, interrupts)

Set interrupt in the event payload.

Source code in strands/types/_events.py

def __init__(self, tool_use: ToolUse, interrupts: list[Interrupt]) -> None:
    """Set interrupt in the event payload."""
    super().__init__({"tool_interrupt_event": {"tool_use": tool_use, "interrupts": interrupts}})

ToolResult

Bases: TypedDict

Result of a tool execution.

Attributes:

Name	Type	Description
content	list[ToolResultContent]	List of result content returned by the tool.
status	ToolResultStatus	The status of the tool execution ("success" or "error").
toolUseId	str	The unique identifier of the tool use request that produced this result.

Source code in strands/types/tools.py

class ToolResult(TypedDict):
    """Result of a tool execution.

    Attributes:
        content: List of result content returned by the tool.
        status: The status of the tool execution ("success" or "error").
        toolUseId: The unique identifier of the tool use request that produced this result.
    """

    content: list[ToolResultContent]
    status: ToolResultStatus
    toolUseId: str

ToolUse

Bases: TypedDict

A request from the model to use a specific tool with the provided input.

Attributes:

Name	Type	Description
input	Any	The input parameters for the tool. Can be any JSON-serializable type.
name	str	The name of the tool to invoke.
toolUseId	str	A unique identifier for this specific tool use request.

Source code in strands/types/tools.py

class ToolUse(TypedDict):
    """A request from the model to use a specific tool with the provided input.

    Attributes:
        input: The input parameters for the tool.
            Can be any JSON-serializable type.
        name: The name of the tool to invoke.
        toolUseId: A unique identifier for this specific tool use request.
    """

    input: Any
    name: str
    toolUseId: str

Trace

A trace representing a single operation or step in the execution flow.

Source code in strands/telemetry/metrics.py

class Trace:
    """A trace representing a single operation or step in the execution flow."""

    def __init__(
        self,
        name: str,
        parent_id: Optional[str] = None,
        start_time: Optional[float] = None,
        raw_name: Optional[str] = None,
        metadata: Optional[Dict[str, Any]] = None,
        message: Optional[Message] = None,
    ) -> None:
        """Initialize a new trace.

        Args:
            name: Human-readable name of the operation being traced.
            parent_id: ID of the parent trace, if this is a child operation.
            start_time: Timestamp when the trace started.
                If not provided, the current time will be used.
            raw_name: System level name.
            metadata: Additional contextual information about the trace.
            message: Message associated with the trace.
        """
        self.id: str = str(uuid.uuid4())
        self.name: str = name
        self.raw_name: Optional[str] = raw_name
        self.parent_id: Optional[str] = parent_id
        self.start_time: float = start_time if start_time is not None else time.time()
        self.end_time: Optional[float] = None
        self.children: List["Trace"] = []
        self.metadata: Dict[str, Any] = metadata or {}
        self.message: Optional[Message] = message

    def end(self, end_time: Optional[float] = None) -> None:
        """Mark the trace as complete with the given or current timestamp.

        Args:
            end_time: Timestamp to use as the end time.
                If not provided, the current time will be used.
        """
        self.end_time = end_time if end_time is not None else time.time()

    def add_child(self, child: "Trace") -> None:
        """Add a child trace to this trace.

        Args:
            child: The child trace to add.
        """
        self.children.append(child)

    def duration(self) -> Optional[float]:
        """Calculate the duration of this trace.

        Returns:
            The duration in seconds, or None if the trace hasn't ended yet.
        """
        return None if self.end_time is None else self.end_time - self.start_time

    def add_message(self, message: Message) -> None:
        """Add a message to the trace.

        Args:
            message: The message to add.
        """
        self.message = message

    def to_dict(self) -> Dict[str, Any]:
        """Convert the trace to a dictionary representation.

        Returns:
            A dictionary containing all trace information, suitable for serialization.
        """
        return {
            "id": self.id,
            "name": self.name,
            "raw_name": self.raw_name,
            "parent_id": self.parent_id,
            "start_time": self.start_time,
            "end_time": self.end_time,
            "duration": self.duration(),
            "children": [child.to_dict() for child in self.children],
            "metadata": self.metadata,
            "message": self.message,
        }

__init__(name, parent_id=None, start_time=None, raw_name=None, metadata=None, message=None)

Initialize a new trace.

Parameters:

Name	Type	Description	Default
name	str	Human-readable name of the operation being traced.	required
parent_id	Optional[str]	ID of the parent trace, if this is a child operation.	None
start_time	Optional[float]	Timestamp when the trace started. If not provided, the current time will be used.	None
raw_name	Optional[str]	System level name.	None
metadata	Optional[Dict[str, Any]]	Additional contextual information about the trace.	None
message	Optional[Message]	Message associated with the trace.	None

Source code in strands/telemetry/metrics.py

def __init__(
    self,
    name: str,
    parent_id: Optional[str] = None,
    start_time: Optional[float] = None,
    raw_name: Optional[str] = None,
    metadata: Optional[Dict[str, Any]] = None,
    message: Optional[Message] = None,
) -> None:
    """Initialize a new trace.

    Args:
        name: Human-readable name of the operation being traced.
        parent_id: ID of the parent trace, if this is a child operation.
        start_time: Timestamp when the trace started.
            If not provided, the current time will be used.
        raw_name: System level name.
        metadata: Additional contextual information about the trace.
        message: Message associated with the trace.
    """
    self.id: str = str(uuid.uuid4())
    self.name: str = name
    self.raw_name: Optional[str] = raw_name
    self.parent_id: Optional[str] = parent_id
    self.start_time: float = start_time if start_time is not None else time.time()
    self.end_time: Optional[float] = None
    self.children: List["Trace"] = []
    self.metadata: Dict[str, Any] = metadata or {}
    self.message: Optional[Message] = message

add_child(child)

Add a child trace to this trace.

Parameters:

Name	Type	Description	Default
child	Trace	The child trace to add.	required

Source code in strands/telemetry/metrics.py

def add_child(self, child: "Trace") -> None:
    """Add a child trace to this trace.

    Args:
        child: The child trace to add.
    """
    self.children.append(child)

add_message(message)

Add a message to the trace.

Parameters:

Name	Type	Description	Default
message	Message	The message to add.	required

Source code in strands/telemetry/metrics.py

def add_message(self, message: Message) -> None:
    """Add a message to the trace.

    Args:
        message: The message to add.
    """
    self.message = message

duration()

Calculate the duration of this trace.

Returns:

Type	Description
Optional[float]	The duration in seconds, or None if the trace hasn't ended yet.

Source code in strands/telemetry/metrics.py

def duration(self) -> Optional[float]:
    """Calculate the duration of this trace.

    Returns:
        The duration in seconds, or None if the trace hasn't ended yet.
    """
    return None if self.end_time is None else self.end_time - self.start_time

end(end_time=None)

Mark the trace as complete with the given or current timestamp.

Parameters:

Name	Type	Description	Default
end_time	Optional[float]	Timestamp to use as the end time. If not provided, the current time will be used.	None

Source code in strands/telemetry/metrics.py

def end(self, end_time: Optional[float] = None) -> None:
    """Mark the trace as complete with the given or current timestamp.

    Args:
        end_time: Timestamp to use as the end time.
            If not provided, the current time will be used.
    """
    self.end_time = end_time if end_time is not None else time.time()

to_dict()

Convert the trace to a dictionary representation.

Returns:

Type	Description
Dict[str, Any]	A dictionary containing all trace information, suitable for serialization.

Source code in strands/telemetry/metrics.py

def to_dict(self) -> Dict[str, Any]:
    """Convert the trace to a dictionary representation.

    Returns:
        A dictionary containing all trace information, suitable for serialization.
    """
    return {
        "id": self.id,
        "name": self.name,
        "raw_name": self.raw_name,
        "parent_id": self.parent_id,
        "start_time": self.start_time,
        "end_time": self.end_time,
        "duration": self.duration(),
        "children": [child.to_dict() for child in self.children],
        "metadata": self.metadata,
        "message": self.message,
    }

TypedEvent

Bases: dict

Base class for all typed events in the agent system.

Source code in strands/types/_events.py

class TypedEvent(dict):
    """Base class for all typed events in the agent system."""

    def __init__(self, data: dict[str, Any] | None = None) -> None:
        """Initialize the typed event with optional data.

        Args:
            data: Optional dictionary of event data to initialize with
        """
        super().__init__(data or {})

    @property
    def is_callback_event(self) -> bool:
        """True if this event should trigger the callback_handler to fire."""
        return True

    def as_dict(self) -> dict:
        """Convert this event to a raw dictionary for emitting purposes."""
        return {**self}

    def prepare(self, invocation_state: dict) -> None:
        """Prepare the event for emission by adding invocation state.

        This allows a subset of events to merge with the invocation_state without needing to
        pass around the invocation_state throughout the system.
        """
        ...

is_callback_event property

True if this event should trigger the callback_handler to fire.

__init__(data=None)

Initialize the typed event with optional data.

Parameters:

Name	Type	Description	Default
data	dict[str, Any] | None	Optional dictionary of event data to initialize with	None

Source code in strands/types/_events.py

def __init__(self, data: dict[str, Any] | None = None) -> None:
    """Initialize the typed event with optional data.

    Args:
        data: Optional dictionary of event data to initialize with
    """
    super().__init__(data or {})

as_dict()

Convert this event to a raw dictionary for emitting purposes.

Source code in strands/types/_events.py

def as_dict(self) -> dict:
    """Convert this event to a raw dictionary for emitting purposes."""
    return {**self}

prepare(invocation_state)

Prepare the event for emission by adding invocation state.

This allows a subset of events to merge with the invocation_state without needing to pass around the invocation_state throughout the system.

Source code in strands/types/_events.py

def prepare(self, invocation_state: dict) -> None:
    """Prepare the event for emission by adding invocation state.

    This allows a subset of events to merge with the invocation_state without needing to
    pass around the invocation_state throughout the system.
    """
    ...