strands.experimental.steering.handlers.llm.llm_handler

LLM-based steering handler that uses an LLM to provide contextual guidance.

ToolSteeringAction = Annotated[Proceed | Guide | Interrupt, Field(discriminator='type')] module-attribute

Steering actions valid for tool steering (steer_before_tool).

• Proceed: Allow tool execution to continue
• Guide: Cancel tool and provide feedback for alternative approaches
• Interrupt: Pause for human input before tool execution

logger = logging.getLogger(__name__) module-attribute

Agent

Core Agent implementation.

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 implementation.

    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: Model | str | None = None,
        messages: Messages | None = None,
        tools: list[Union[str, dict[str, str], "ToolProvider", Any]] | None = None,
        system_prompt: str | list[SystemContentBlock] | None = None,
        structured_output_model: type[BaseModel] | None = None,
        callback_handler: Callable[..., Any] | _DefaultCallbackHandlerSentinel | None = _DEFAULT_CALLBACK_HANDLER,
        conversation_manager: ConversationManager | None = None,
        record_direct_tool_call: bool = True,
        load_tools_from_directory: bool = False,
        trace_attributes: Mapping[str, AttributeValue] | None = None,
        *,
        agent_id: str | None = None,
        name: str | None = None,
        description: str | None = None,
        state: AgentState | dict | None = None,
        hooks: list[HookProvider] | None = None,
        session_manager: SessionManager | None = None,
        tool_executor: ToolExecutor | None = None,
        retry_strategy: ModelRetryStrategy | None = 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.).
            retry_strategy: Strategy for retrying model calls on throttling or other transient errors.
                Defaults to ModelRetryStrategy with max_attempts=6, initial_delay=4s, max_delay=240s.
                Implement a custom HookProvider for custom retry logic, or pass None to disable retries.

        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: 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: trace_api.Span | None = 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 lock for guarding concurrent invocations
        # Using threading.Lock instead of asyncio.Lock because run_async() creates
        # separate event loops in different threads, so asyncio.Lock wouldn't work
        self._invocation_lock = threading.Lock()

        # In the future, we'll have a RetryStrategy base class but until
        # that API is determined we only allow ModelRetryStrategy
        if retry_strategy and type(retry_strategy) is not ModelRetryStrategy:
            raise ValueError("retry_strategy must be an instance of ModelRetryStrategy")

        self._retry_strategy = (
            retry_strategy
            if retry_strategy is not None
            else ModelRetryStrategy(max_attempts=MAX_ATTEMPTS, max_delay=MAX_DELAY, initial_delay=INITIAL_DELAY)
        )

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

        # Register retry strategy as a hook
        self.hooks.add_hook(self._retry_strategy)

        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, invocation_state={}))
        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, invocation_state={}))

    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:
            ConcurrencyException: If another invocation is already in progress on this agent instance.
            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"]
            ```
        """
        # Acquire lock to prevent concurrent invocations
        # Using threading.Lock instead of asyncio.Lock because run_async() creates
        # separate event loops in different threads
        acquired = self._invocation_lock.acquire(blocking=False)
        if not acquired:
            raise ConcurrencyException(
                "Agent is already processing a request. Concurrent invocations are not supported."
            )

        try:
            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

        finally:
            self._invocation_lock.release()

    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.
        """
        before_invocation_event, _interrupts = await self.hooks.invoke_callbacks_async(
            BeforeInvocationEvent(agent=self, invocation_state=invocation_state, messages=messages)
        )
        messages = before_invocation_event.messages if before_invocation_event.messages is not None else messages

        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, invocation_state=invocation_state, 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: AgentResult | None = None,
        error: Exception | None = 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, retry_strategy=None)

Initialize the Agent with the specified configuration.

Parameters:

Name	Type	Description	Default
model	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	Messages | None	List of initial messages to pre-load into the conversation. Defaults to an empty list if None.	None
tools	list[Union[str, dict[str, str], ToolProvider, Any]] | None	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	str | list[SystemContentBlock] | None	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	type[BaseModel] | None	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	Callable[..., Any] | _DefaultCallbackHandlerSentinel | None	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	ConversationManager | None	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	Mapping[str, AttributeValue] | None	Custom trace attributes to apply to the agent's trace span.	None
agent_id	str | None	Optional ID for the agent, useful for session management and multi-agent scenarios. Defaults to "default".	None
name	str | None	name of the Agent Defaults to "Strands Agents".	None
description	str | None	description of what the Agent does Defaults to None.	None
state	AgentState | dict | None	stateful information for the agent. Can be either an AgentState object, or a json serializable dict. Defaults to an empty AgentState object.	None
hooks	list[HookProvider] | None	hooks to be added to the agent hook registry Defaults to None.	None
session_manager	SessionManager | None	Manager for handling agent sessions including conversation history and state. If provided, enables session-based persistence and state management.	None
tool_executor	ToolExecutor | None	Definition of tool execution strategy (e.g., sequential, concurrent, etc.).	None
retry_strategy	ModelRetryStrategy | None	Strategy for retrying model calls on throttling or other transient errors. Defaults to ModelRetryStrategy with max_attempts=6, initial_delay=4s, max_delay=240s. Implement a custom HookProvider for custom retry logic, or pass None to disable retries.	None

Raises:

Type	Description
ValueError	If agent id contains path separators.

Source code in strands/agent/agent.py

def __init__(
    self,
    model: Model | str | None = None,
    messages: Messages | None = None,
    tools: list[Union[str, dict[str, str], "ToolProvider", Any]] | None = None,
    system_prompt: str | list[SystemContentBlock] | None = None,
    structured_output_model: type[BaseModel] | None = None,
    callback_handler: Callable[..., Any] | _DefaultCallbackHandlerSentinel | None = _DEFAULT_CALLBACK_HANDLER,
    conversation_manager: ConversationManager | None = None,
    record_direct_tool_call: bool = True,
    load_tools_from_directory: bool = False,
    trace_attributes: Mapping[str, AttributeValue] | None = None,
    *,
    agent_id: str | None = None,
    name: str | None = None,
    description: str | None = None,
    state: AgentState | dict | None = None,
    hooks: list[HookProvider] | None = None,
    session_manager: SessionManager | None = None,
    tool_executor: ToolExecutor | None = None,
    retry_strategy: ModelRetryStrategy | None = 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.).
        retry_strategy: Strategy for retrying model calls on throttling or other transient errors.
            Defaults to ModelRetryStrategy with max_attempts=6, initial_delay=4s, max_delay=240s.
            Implement a custom HookProvider for custom retry logic, or pass None to disable retries.

    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: 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: trace_api.Span | None = 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 lock for guarding concurrent invocations
    # Using threading.Lock instead of asyncio.Lock because run_async() creates
    # separate event loops in different threads, so asyncio.Lock wouldn't work
    self._invocation_lock = threading.Lock()

    # In the future, we'll have a RetryStrategy base class but until
    # that API is determined we only allow ModelRetryStrategy
    if retry_strategy and type(retry_strategy) is not ModelRetryStrategy:
        raise ValueError("retry_strategy must be an instance of ModelRetryStrategy")

    self._retry_strategy = (
        retry_strategy
        if retry_strategy is not None
        else ModelRetryStrategy(max_attempts=MAX_ATTEMPTS, max_delay=MAX_DELAY, initial_delay=INITIAL_DELAY)
    )

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

    # Register retry strategy as a hook
    self.hooks.add_hook(self._retry_strategy)

    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
ConcurrencyException	If another invocation is already in progress on this agent instance.
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:
        ConcurrencyException: If another invocation is already in progress on this agent instance.
        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"]
        ```
    """
    # Acquire lock to prevent concurrent invocations
    # Using threading.Lock instead of asyncio.Lock because run_async() creates
    # separate event loops in different threads
    acquired = self._invocation_lock.acquire(blocking=False)
    if not acquired:
        raise ConcurrencyException(
            "Agent is already processing a request. Concurrent invocations are not supported."
        )

    try:
        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

    finally:
        self._invocation_lock.release()

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, invocation_state={}))
    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, invocation_state={}))

DefaultPromptMapper

Bases: LLMPromptMapper

Default prompt mapper for steering evaluation.

Source code in strands/experimental/steering/handlers/llm/mappers.py

class DefaultPromptMapper(LLMPromptMapper):
    """Default prompt mapper for steering evaluation."""

    def create_steering_prompt(
        self, steering_context: SteeringContext, tool_use: ToolUse | None = None, **kwargs: Any
    ) -> str:
        """Create default steering prompt using Agent SOP structure.

        Uses Agent SOP format for structured, constraint-based prompts.
        See: https://github.com/strands-agents/agent-sop
        """
        context_str = (
            json.dumps(steering_context.data.get(), indent=2) if steering_context.data.get() else "No context available"
        )

        if tool_use:
            event_description = (
                f"Tool: {tool_use['name']}\nArguments: {json.dumps(tool_use.get('input', {}), indent=2)}"
            )
            action_type = "tool call"
        else:
            event_description = "General evaluation"
            action_type = "action"

        return _STEERING_PROMPT_TEMPLATE.format(
            action_type=action_type,
            action_type_title=action_type.title(),
            context_str=context_str,
            event_description=event_description,
        )

create_steering_prompt(steering_context, tool_use=None, **kwargs)

Create default steering prompt using Agent SOP structure.

Uses Agent SOP format for structured, constraint-based prompts. See: https://github.com/strands-agents/agent-sop

Source code in strands/experimental/steering/handlers/llm/mappers.py

def create_steering_prompt(
    self, steering_context: SteeringContext, tool_use: ToolUse | None = None, **kwargs: Any
) -> str:
    """Create default steering prompt using Agent SOP structure.

    Uses Agent SOP format for structured, constraint-based prompts.
    See: https://github.com/strands-agents/agent-sop
    """
    context_str = (
        json.dumps(steering_context.data.get(), indent=2) if steering_context.data.get() else "No context available"
    )

    if tool_use:
        event_description = (
            f"Tool: {tool_use['name']}\nArguments: {json.dumps(tool_use.get('input', {}), indent=2)}"
        )
        action_type = "tool call"
    else:
        event_description = "General evaluation"
        action_type = "action"

    return _STEERING_PROMPT_TEMPLATE.format(
        action_type=action_type,
        action_type_title=action_type.title(),
        context_str=context_str,
        event_description=event_description,
    )

Guide

Bases: BaseModel

Provide contextual guidance to redirect the agent.

The agent receives the reason as contextual feedback to help guide its behavior. The specific handling depends on the steering context (e.g., tool call vs. model response).

Source code in strands/experimental/steering/core/action.py

class Guide(BaseModel):
    """Provide contextual guidance to redirect the agent.

    The agent receives the reason as contextual feedback to help guide
    its behavior. The specific handling depends on the steering context
    (e.g., tool call vs. model response).
    """

    type: Literal["guide"] = "guide"
    reason: str

Interrupt

Bases: BaseModel

Pause execution for human input via interrupt system.

Execution is paused and human input is requested through Strands' interrupt system. The human can approve or deny the operation, and their decision determines whether execution continues or is cancelled.

Source code in strands/experimental/steering/core/action.py

class Interrupt(BaseModel):
    """Pause execution for human input via interrupt system.

    Execution is paused and human input is requested through Strands'
    interrupt system. The human can approve or deny the operation, and their
    decision determines whether execution continues or is cancelled.
    """

    type: Literal["interrupt"] = "interrupt"
    reason: str

LLMPromptMapper

Bases: Protocol

Protocol for mapping context and events to LLM evaluation prompts.

Source code in strands/experimental/steering/handlers/llm/mappers.py

class LLMPromptMapper(Protocol):
    """Protocol for mapping context and events to LLM evaluation prompts."""

    def create_steering_prompt(
        self, steering_context: SteeringContext, tool_use: ToolUse | None = None, **kwargs: Any
    ) -> str:
        """Create steering prompt for LLM evaluation.

        Args:
            steering_context: Steering context with populated data
            tool_use: Tool use object for tool call events (None for other events)
            **kwargs: Additional event data for other steering events

        Returns:
            Formatted prompt string for LLM evaluation
        """
        ...

create_steering_prompt(steering_context, tool_use=None, **kwargs)

Create steering prompt for LLM evaluation.

Parameters:

Name	Type	Description	Default
steering_context	SteeringContext	Steering context with populated data	required
tool_use	ToolUse | None	Tool use object for tool call events (None for other events)	None
**kwargs	Any	Additional event data for other steering events	{}

Returns:

Type	Description
str	Formatted prompt string for LLM evaluation

Source code in strands/experimental/steering/handlers/llm/mappers.py

def create_steering_prompt(
    self, steering_context: SteeringContext, tool_use: ToolUse | None = None, **kwargs: Any
) -> str:
    """Create steering prompt for LLM evaluation.

    Args:
        steering_context: Steering context with populated data
        tool_use: Tool use object for tool call events (None for other events)
        **kwargs: Additional event data for other steering events

    Returns:
        Formatted prompt string for LLM evaluation
    """
    ...

LLMSteeringHandler

Bases: SteeringHandler

Steering handler that uses an LLM to provide contextual guidance.

Uses natural language prompts to evaluate tool calls and provide contextual steering guidance to help agents navigate complex workflows.

Source code in strands/experimental/steering/handlers/llm/llm_handler.py

class LLMSteeringHandler(SteeringHandler):
    """Steering handler that uses an LLM to provide contextual guidance.

    Uses natural language prompts to evaluate tool calls and provide
    contextual steering guidance to help agents navigate complex workflows.
    """

    def __init__(
        self,
        system_prompt: str,
        prompt_mapper: LLMPromptMapper | None = None,
        model: Model | None = None,
        context_providers: list[SteeringContextProvider] | None = None,
    ):
        """Initialize the LLMSteeringHandler.

        Args:
            system_prompt: System prompt defining steering guidance rules
            prompt_mapper: Custom prompt mapper for evaluation prompts
            model: Optional model override for steering evaluation
            context_providers: List of context providers for populating steering context.
                Defaults to [LedgerProvider()] if None. Pass an empty list to disable
                context providers.
        """
        providers: list[SteeringContextProvider] = (
            [LedgerProvider()] if context_providers is None else context_providers
        )
        super().__init__(context_providers=providers)
        self.system_prompt = system_prompt
        self.prompt_mapper = prompt_mapper or DefaultPromptMapper()
        self.model = model

    async def steer_before_tool(self, *, agent: Agent, tool_use: ToolUse, **kwargs: Any) -> ToolSteeringAction:
        """Provide contextual guidance for tool usage.

        Args:
            agent: The agent instance
            tool_use: The tool use object with name and arguments
            **kwargs: Additional keyword arguments for steering evaluation

        Returns:
            SteeringAction indicating how to guide the tool execution
        """
        # Generate steering prompt
        prompt = self.prompt_mapper.create_steering_prompt(self.steering_context, tool_use=tool_use)

        # Create isolated agent for steering evaluation (no shared conversation state)
        from .....agent import Agent

        steering_agent = Agent(system_prompt=self.system_prompt, model=self.model or agent.model, callback_handler=None)

        # Get LLM decision
        llm_result: _LLMSteering = cast(
            _LLMSteering, steering_agent(prompt, structured_output_model=_LLMSteering).structured_output
        )

        # Convert LLM decision to steering action
        match llm_result.decision:
            case "proceed":
                return Proceed(reason=llm_result.reason)
            case "guide":
                return Guide(reason=llm_result.reason)
            case "interrupt":
                return Interrupt(reason=llm_result.reason)
            case _:
                logger.warning("decision=<%s> | unknown llm decision, defaulting to proceed", llm_result.decision)  # type: ignore[unreachable]
                return Proceed(reason="Unknown LLM decision, defaulting to proceed")

__init__(system_prompt, prompt_mapper=None, model=None, context_providers=None)

Initialize the LLMSteeringHandler.

Parameters:

Name	Type	Description	Default
system_prompt	str	System prompt defining steering guidance rules	required
prompt_mapper	LLMPromptMapper | None	Custom prompt mapper for evaluation prompts	None
model	Model | None	Optional model override for steering evaluation	None
context_providers	list[SteeringContextProvider] | None	List of context providers for populating steering context. Defaults to [LedgerProvider()] if None. Pass an empty list to disable context providers.	None

Source code in strands/experimental/steering/handlers/llm/llm_handler.py

def __init__(
    self,
    system_prompt: str,
    prompt_mapper: LLMPromptMapper | None = None,
    model: Model | None = None,
    context_providers: list[SteeringContextProvider] | None = None,
):
    """Initialize the LLMSteeringHandler.

    Args:
        system_prompt: System prompt defining steering guidance rules
        prompt_mapper: Custom prompt mapper for evaluation prompts
        model: Optional model override for steering evaluation
        context_providers: List of context providers for populating steering context.
            Defaults to [LedgerProvider()] if None. Pass an empty list to disable
            context providers.
    """
    providers: list[SteeringContextProvider] = (
        [LedgerProvider()] if context_providers is None else context_providers
    )
    super().__init__(context_providers=providers)
    self.system_prompt = system_prompt
    self.prompt_mapper = prompt_mapper or DefaultPromptMapper()
    self.model = model

steer_before_tool(*, agent, tool_use, **kwargs) async

Provide contextual guidance for tool usage.

Parameters:

Name	Type	Description	Default
agent	Agent	The agent instance	required
tool_use	ToolUse	The tool use object with name and arguments	required
**kwargs	Any	Additional keyword arguments for steering evaluation	{}

Returns:

Type	Description
ToolSteeringAction	SteeringAction indicating how to guide the tool execution

Source code in strands/experimental/steering/handlers/llm/llm_handler.py

async def steer_before_tool(self, *, agent: Agent, tool_use: ToolUse, **kwargs: Any) -> ToolSteeringAction:
    """Provide contextual guidance for tool usage.

    Args:
        agent: The agent instance
        tool_use: The tool use object with name and arguments
        **kwargs: Additional keyword arguments for steering evaluation

    Returns:
        SteeringAction indicating how to guide the tool execution
    """
    # Generate steering prompt
    prompt = self.prompt_mapper.create_steering_prompt(self.steering_context, tool_use=tool_use)

    # Create isolated agent for steering evaluation (no shared conversation state)
    from .....agent import Agent

    steering_agent = Agent(system_prompt=self.system_prompt, model=self.model or agent.model, callback_handler=None)

    # Get LLM decision
    llm_result: _LLMSteering = cast(
        _LLMSteering, steering_agent(prompt, structured_output_model=_LLMSteering).structured_output
    )

    # Convert LLM decision to steering action
    match llm_result.decision:
        case "proceed":
            return Proceed(reason=llm_result.reason)
        case "guide":
            return Guide(reason=llm_result.reason)
        case "interrupt":
            return Interrupt(reason=llm_result.reason)
        case _:
            logger.warning("decision=<%s> | unknown llm decision, defaulting to proceed", llm_result.decision)  # type: ignore[unreachable]
            return Proceed(reason="Unknown LLM decision, defaulting to proceed")

LedgerProvider

Bases: SteeringContextProvider

Combined ledger context provider for both before and after tool calls.

Source code in strands/experimental/steering/context_providers/ledger_provider.py

class LedgerProvider(SteeringContextProvider):
    """Combined ledger context provider for both before and after tool calls."""

    def context_providers(self, **kwargs: Any) -> list[SteeringContextCallback]:
        """Return ledger context providers with shared state."""
        return [
            LedgerBeforeToolCall(),
            LedgerAfterToolCall(),
        ]

context_providers(**kwargs)

Return ledger context providers with shared state.

Source code in strands/experimental/steering/context_providers/ledger_provider.py

def context_providers(self, **kwargs: Any) -> list[SteeringContextCallback]:
    """Return ledger context providers with shared state."""
    return [
        LedgerBeforeToolCall(),
        LedgerAfterToolCall(),
    ]

Model

Bases: ABC

Abstract base class for Agent model providers.

This class defines the interface for all model implementations in the Strands Agents SDK. It provides a standardized way to configure and process requests for different AI model providers.

Source code in strands/models/model.py

class Model(abc.ABC):
    """Abstract base class for Agent model providers.

    This class defines the interface for all model implementations in the Strands Agents SDK. It provides a
    standardized way to configure and process requests for different AI model providers.
    """

    @abc.abstractmethod
    # pragma: no cover
    def update_config(self, **model_config: Any) -> None:
        """Update the model configuration with the provided arguments.

        Args:
            **model_config: Configuration overrides.
        """
        pass

    @abc.abstractmethod
    # pragma: no cover
    def get_config(self) -> Any:
        """Return the model configuration.

        Returns:
            The model's configuration.
        """
        pass

    @abc.abstractmethod
    # pragma: no cover
    def structured_output(
        self, output_model: type[T], prompt: Messages, system_prompt: str | None = None, **kwargs: Any
    ) -> AsyncGenerator[dict[str, T | Any], None]:
        """Get structured output from the model.

        Args:
            output_model: The output model to use for the agent.
            prompt: The prompt messages to use for the agent.
            system_prompt: System prompt to provide context to the model.
            **kwargs: Additional keyword arguments for future extensibility.

        Yields:
            Model events with the last being the structured output.

        Raises:
            ValidationException: The response format from the model does not match the output_model
        """
        pass

    @abc.abstractmethod
    # pragma: no cover
    def stream(
        self,
        messages: Messages,
        tool_specs: list[ToolSpec] | None = None,
        system_prompt: str | None = None,
        *,
        tool_choice: ToolChoice | None = None,
        system_prompt_content: list[SystemContentBlock] | None = None,
        invocation_state: dict[str, Any] | None = None,
        **kwargs: Any,
    ) -> AsyncIterable[StreamEvent]:
        """Stream conversation with the model.

        This method handles the full lifecycle of conversing with the model:

        1. Format the messages, tool specs, and configuration into a streaming request
        2. Send the request to the model
        3. Yield the formatted message chunks

        Args:
            messages: List of message objects to be processed by the model.
            tool_specs: List of tool specifications to make available to the model.
            system_prompt: System prompt to provide context to the model.
            tool_choice: Selection strategy for tool invocation.
            system_prompt_content: System prompt content blocks for advanced features like caching.
            invocation_state: Caller-provided state/context that was passed to the agent when it was invoked.
            **kwargs: Additional keyword arguments for future extensibility.

        Yields:
            Formatted message chunks from the model.

        Raises:
            ModelThrottledException: When the model service is throttling requests from the client.
        """
        pass

get_config() abstractmethod

Return the model configuration.

Returns:

Type	Description
Any	The model's configuration.

Source code in strands/models/model.py

@abc.abstractmethod
# pragma: no cover
def get_config(self) -> Any:
    """Return the model configuration.

    Returns:
        The model's configuration.
    """
    pass

stream(messages, tool_specs=None, system_prompt=None, *, tool_choice=None, system_prompt_content=None, invocation_state=None, **kwargs) abstractmethod

Stream conversation with the model.

This method handles the full lifecycle of conversing with the model:

1. Format the messages, tool specs, and configuration into a streaming request
2. Send the request to the model
3. Yield the formatted message chunks

Parameters:

Name	Type	Description	Default
messages	Messages	List of message objects to be processed by the model.	required
tool_specs	list[ToolSpec] | None	List of tool specifications to make available to the model.	None
system_prompt	str | None	System prompt to provide context to the model.	None
tool_choice	ToolChoice | None	Selection strategy for tool invocation.	None
system_prompt_content	list[SystemContentBlock] | None	System prompt content blocks for advanced features like caching.	None
invocation_state	dict[str, Any] | None	Caller-provided state/context that was passed to the agent when it was invoked.	None
**kwargs	Any	Additional keyword arguments for future extensibility.	{}

Yields:

Type	Description
AsyncIterable[StreamEvent]	Formatted message chunks from the model.

Raises:

Type	Description
ModelThrottledException	When the model service is throttling requests from the client.

Source code in strands/models/model.py

@abc.abstractmethod
# pragma: no cover
def stream(
    self,
    messages: Messages,
    tool_specs: list[ToolSpec] | None = None,
    system_prompt: str | None = None,
    *,
    tool_choice: ToolChoice | None = None,
    system_prompt_content: list[SystemContentBlock] | None = None,
    invocation_state: dict[str, Any] | None = None,
    **kwargs: Any,
) -> AsyncIterable[StreamEvent]:
    """Stream conversation with the model.

    This method handles the full lifecycle of conversing with the model:

    1. Format the messages, tool specs, and configuration into a streaming request
    2. Send the request to the model
    3. Yield the formatted message chunks

    Args:
        messages: List of message objects to be processed by the model.
        tool_specs: List of tool specifications to make available to the model.
        system_prompt: System prompt to provide context to the model.
        tool_choice: Selection strategy for tool invocation.
        system_prompt_content: System prompt content blocks for advanced features like caching.
        invocation_state: Caller-provided state/context that was passed to the agent when it was invoked.
        **kwargs: Additional keyword arguments for future extensibility.

    Yields:
        Formatted message chunks from the model.

    Raises:
        ModelThrottledException: When the model service is throttling requests from the client.
    """
    pass

structured_output(output_model, prompt, system_prompt=None, **kwargs) abstractmethod

Get structured output from the model.

Parameters:

Name	Type	Description	Default
output_model	type[T]	The output model to use for the agent.	required
prompt	Messages	The prompt messages to use for the agent.	required
system_prompt	str | None	System prompt to provide context to the model.	None
**kwargs	Any	Additional keyword arguments for future extensibility.	{}

Yields:

Type	Description
AsyncGenerator[dict[str, T | Any], None]	Model events with the last being the structured output.

Raises:

Type	Description
ValidationException	The response format from the model does not match the output_model

Source code in strands/models/model.py

@abc.abstractmethod
# pragma: no cover
def structured_output(
    self, output_model: type[T], prompt: Messages, system_prompt: str | None = None, **kwargs: Any
) -> AsyncGenerator[dict[str, T | Any], None]:
    """Get structured output from the model.

    Args:
        output_model: The output model to use for the agent.
        prompt: The prompt messages to use for the agent.
        system_prompt: System prompt to provide context to the model.
        **kwargs: Additional keyword arguments for future extensibility.

    Yields:
        Model events with the last being the structured output.

    Raises:
        ValidationException: The response format from the model does not match the output_model
    """
    pass

update_config(**model_config) abstractmethod

Update the model configuration with the provided arguments.

Parameters:

Name	Type	Description	Default
**model_config	Any	Configuration overrides.	{}

Source code in strands/models/model.py

@abc.abstractmethod
# pragma: no cover
def update_config(self, **model_config: Any) -> None:
    """Update the model configuration with the provided arguments.

    Args:
        **model_config: Configuration overrides.
    """
    pass

Proceed

Bases: BaseModel

Allow execution to continue without intervention.

The action proceeds as planned. The reason provides context for logging and debugging purposes.

Source code in strands/experimental/steering/core/action.py

class Proceed(BaseModel):
    """Allow execution to continue without intervention.

    The action proceeds as planned. The reason provides context
    for logging and debugging purposes.
    """

    type: Literal["proceed"] = "proceed"
    reason: str

SteeringContextProvider

Bases: ABC

Abstract base class for context providers that handle multiple event types.

Source code in strands/experimental/steering/core/context.py

class SteeringContextProvider(ABC):
    """Abstract base class for context providers that handle multiple event types."""

    @abstractmethod
    def context_providers(self, **kwargs: Any) -> list[SteeringContextCallback]:
        """Return list of context callbacks with event types extracted from generics."""
        ...

context_providers(**kwargs) abstractmethod

Return list of context callbacks with event types extracted from generics.

Source code in strands/experimental/steering/core/context.py

@abstractmethod
def context_providers(self, **kwargs: Any) -> list[SteeringContextCallback]:
    """Return list of context callbacks with event types extracted from generics."""
    ...

SteeringHandler

Bases: HookProvider, ABC

Base class for steering handlers that provide contextual guidance to agents.

Steering handlers maintain local context and register hook callbacks to populate context data as needed for guidance decisions.

Source code in strands/experimental/steering/core/handler.py

class SteeringHandler(HookProvider, ABC):
    """Base class for steering handlers that provide contextual guidance to agents.

    Steering handlers maintain local context and register hook callbacks
    to populate context data as needed for guidance decisions.
    """

    def __init__(self, context_providers: list[SteeringContextProvider] | None = None):
        """Initialize the steering handler.

        Args:
            context_providers: List of context providers for context updates
        """
        super().__init__()
        self.steering_context = SteeringContext()
        self._context_callbacks = []

        # Collect callbacks from all providers
        for provider in context_providers or []:
            self._context_callbacks.extend(provider.context_providers())

        logger.debug("handler_class=<%s> | initialized", self.__class__.__name__)

    def register_hooks(self, registry: HookRegistry, **kwargs: Any) -> None:
        """Register hooks for steering guidance and context updates."""
        # Register context update callbacks
        for callback in self._context_callbacks:
            registry.add_callback(
                callback.event_type, lambda event, callback=callback: callback(event, self.steering_context)
            )

        # Register tool steering guidance
        registry.add_callback(BeforeToolCallEvent, self._provide_tool_steering_guidance)

        # Register model steering guidance
        registry.add_callback(AfterModelCallEvent, self._provide_model_steering_guidance)

    async def _provide_tool_steering_guidance(self, event: BeforeToolCallEvent) -> None:
        """Provide steering guidance for tool call."""
        tool_name = event.tool_use["name"]
        logger.debug("tool_name=<%s> | providing tool steering guidance", tool_name)

        try:
            action = await self.steer_before_tool(agent=event.agent, tool_use=event.tool_use)
        except Exception as e:
            logger.debug("tool_name=<%s>, error=<%s> | tool steering handler guidance failed", tool_name, e)
            return

        self._handle_tool_steering_action(action, event, tool_name)

    def _handle_tool_steering_action(
        self, action: ToolSteeringAction, event: BeforeToolCallEvent, tool_name: str
    ) -> None:
        """Handle the steering action for tool calls by modifying tool execution flow.

        Proceed: Tool executes normally
        Guide: Tool cancelled with contextual feedback for agent to consider alternatives
        Interrupt: Tool execution paused for human input via interrupt system
        """
        if isinstance(action, Proceed):
            logger.debug("tool_name=<%s> | tool call proceeding", tool_name)
        elif isinstance(action, Guide):
            logger.debug("tool_name=<%s> | tool call guided: %s", tool_name, action.reason)
            event.cancel_tool = f"Tool call cancelled. {action.reason} You MUST follow this guidance immediately."
        elif isinstance(action, Interrupt):
            logger.debug("tool_name=<%s> | tool call requires human input: %s", tool_name, action.reason)
            can_proceed: bool = event.interrupt(name=f"steering_input_{tool_name}", reason={"message": action.reason})
            logger.debug("tool_name=<%s> | received human input for tool call", tool_name)

            if not can_proceed:
                event.cancel_tool = f"Manual approval denied: {action.reason}"
                logger.debug("tool_name=<%s> | tool call denied by manual approval", tool_name)
            else:
                logger.debug("tool_name=<%s> | tool call approved manually", tool_name)
        else:
            raise ValueError(f"Unknown steering action type for tool call: {action}")

    async def _provide_model_steering_guidance(self, event: AfterModelCallEvent) -> None:
        """Provide steering guidance for model response."""
        logger.debug("providing model steering guidance")

        # Only steer on successful model responses
        if event.stop_response is None:
            logger.debug("no stop response available | skipping model steering")
            return

        try:
            action = await self.steer_after_model(
                agent=event.agent, message=event.stop_response.message, stop_reason=event.stop_response.stop_reason
            )
        except Exception as e:
            logger.debug("error=<%s> | model steering handler guidance failed", e)
            return

        await self._handle_model_steering_action(action, event)

    async def _handle_model_steering_action(self, action: ModelSteeringAction, event: AfterModelCallEvent) -> None:
        """Handle the steering action for model responses by modifying response handling flow.

        Proceed: Model response accepted without modification
        Guide: Discard model response and retry with guidance message added to conversation
        """
        if isinstance(action, Proceed):
            logger.debug("model response proceeding")
        elif isinstance(action, Guide):
            logger.debug("model response guided (retrying): %s", action.reason)
            # Set retry flag to discard current response
            event.retry = True
            # Add guidance message to agent's conversation so model sees it on retry
            await event.agent._append_messages({"role": "user", "content": [{"text": action.reason}]})
            logger.debug("added guidance message to conversation for model retry")
        else:
            raise ValueError(f"Unknown steering action type for model response: {action}")

    async def steer_before_tool(self, *, agent: "Agent", tool_use: ToolUse, **kwargs: Any) -> ToolSteeringAction:
        """Provide contextual guidance before tool execution.

        This method is called before a tool is executed, allowing the handler to:
        - Proceed: Allow tool execution to continue
        - Guide: Cancel tool and provide feedback for alternative approaches
        - Interrupt: Pause for human input before tool execution

        Args:
            agent: The agent instance
            tool_use: The tool use object with name and arguments
            **kwargs: Additional keyword arguments for guidance evaluation

        Returns:
            ToolSteeringAction indicating how to guide the tool execution

        Note:
            Access steering context via self.steering_context
            Default implementation returns Proceed (allow tool execution)
            Override this method to implement custom tool steering logic
        """
        return Proceed(reason="Default implementation: allowing tool execution")

    async def steer_after_model(
        self, *, agent: "Agent", message: Message, stop_reason: StopReason, **kwargs: Any
    ) -> ModelSteeringAction:
        """Provide contextual guidance after model response.

        This method is called after the model generates a response, allowing the handler to:
        - Proceed: Accept the model response without modification
        - Guide: Discard the response and retry (message is dropped, model is called again)

        Note: Interrupt is not supported for model steering as the model has already responded.

        Args:
            agent: The agent instance
            message: The model's generated message
            stop_reason: The reason the model stopped generating
            **kwargs: Additional keyword arguments for guidance evaluation

        Returns:
            ModelSteeringAction indicating how to handle the model response

        Note:
            Access steering context via self.steering_context
            Default implementation returns Proceed (accept response as-is)
            Override this method to implement custom model steering logic
        """
        return Proceed(reason="Default implementation: accepting model response")

__init__(context_providers=None)

Initialize the steering handler.

Parameters:

Name	Type	Description	Default
context_providers	list[SteeringContextProvider] | None	List of context providers for context updates	None

Source code in strands/experimental/steering/core/handler.py

def __init__(self, context_providers: list[SteeringContextProvider] | None = None):
    """Initialize the steering handler.

    Args:
        context_providers: List of context providers for context updates
    """
    super().__init__()
    self.steering_context = SteeringContext()
    self._context_callbacks = []

    # Collect callbacks from all providers
    for provider in context_providers or []:
        self._context_callbacks.extend(provider.context_providers())

    logger.debug("handler_class=<%s> | initialized", self.__class__.__name__)

register_hooks(registry, **kwargs)

Register hooks for steering guidance and context updates.

Source code in strands/experimental/steering/core/handler.py

def register_hooks(self, registry: HookRegistry, **kwargs: Any) -> None:
    """Register hooks for steering guidance and context updates."""
    # Register context update callbacks
    for callback in self._context_callbacks:
        registry.add_callback(
            callback.event_type, lambda event, callback=callback: callback(event, self.steering_context)
        )

    # Register tool steering guidance
    registry.add_callback(BeforeToolCallEvent, self._provide_tool_steering_guidance)

    # Register model steering guidance
    registry.add_callback(AfterModelCallEvent, self._provide_model_steering_guidance)

steer_after_model(*, agent, message, stop_reason, **kwargs) async

Provide contextual guidance after model response.

This method is called after the model generates a response, allowing the handler to: - Proceed: Accept the model response without modification - Guide: Discard the response and retry (message is dropped, model is called again)

Note: Interrupt is not supported for model steering as the model has already responded.

Parameters:

Name	Type	Description	Default
agent	Agent	The agent instance	required
message	Message	The model's generated message	required
stop_reason	StopReason	The reason the model stopped generating	required
**kwargs	Any	Additional keyword arguments for guidance evaluation	{}

Returns:

Type	Description
ModelSteeringAction	ModelSteeringAction indicating how to handle the model response

Note

Access steering context via self.steering_context Default implementation returns Proceed (accept response as-is) Override this method to implement custom model steering logic

Source code in strands/experimental/steering/core/handler.py

async def steer_after_model(
    self, *, agent: "Agent", message: Message, stop_reason: StopReason, **kwargs: Any
) -> ModelSteeringAction:
    """Provide contextual guidance after model response.

    This method is called after the model generates a response, allowing the handler to:
    - Proceed: Accept the model response without modification
    - Guide: Discard the response and retry (message is dropped, model is called again)

    Note: Interrupt is not supported for model steering as the model has already responded.

    Args:
        agent: The agent instance
        message: The model's generated message
        stop_reason: The reason the model stopped generating
        **kwargs: Additional keyword arguments for guidance evaluation

    Returns:
        ModelSteeringAction indicating how to handle the model response

    Note:
        Access steering context via self.steering_context
        Default implementation returns Proceed (accept response as-is)
        Override this method to implement custom model steering logic
    """
    return Proceed(reason="Default implementation: accepting model response")

steer_before_tool(*, agent, tool_use, **kwargs) async

Provide contextual guidance before tool execution.

This method is called before a tool is executed, allowing the handler to: - Proceed: Allow tool execution to continue - Guide: Cancel tool and provide feedback for alternative approaches - Interrupt: Pause for human input before tool execution

Parameters:

Name	Type	Description	Default
agent	Agent	The agent instance	required
tool_use	ToolUse	The tool use object with name and arguments	required
**kwargs	Any	Additional keyword arguments for guidance evaluation	{}

Returns:

Type	Description
ToolSteeringAction	ToolSteeringAction indicating how to guide the tool execution

Note

Access steering context via self.steering_context Default implementation returns Proceed (allow tool execution) Override this method to implement custom tool steering logic

Source code in strands/experimental/steering/core/handler.py

async def steer_before_tool(self, *, agent: "Agent", tool_use: ToolUse, **kwargs: Any) -> ToolSteeringAction:
    """Provide contextual guidance before tool execution.

    This method is called before a tool is executed, allowing the handler to:
    - Proceed: Allow tool execution to continue
    - Guide: Cancel tool and provide feedback for alternative approaches
    - Interrupt: Pause for human input before tool execution

    Args:
        agent: The agent instance
        tool_use: The tool use object with name and arguments
        **kwargs: Additional keyword arguments for guidance evaluation

    Returns:
        ToolSteeringAction indicating how to guide the tool execution

    Note:
        Access steering context via self.steering_context
        Default implementation returns Proceed (allow tool execution)
        Override this method to implement custom tool steering logic
    """
    return Proceed(reason="Default implementation: allowing tool execution")

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

_LLMSteering

Bases: BaseModel

Structured output model for LLM steering decisions.

Source code in strands/experimental/steering/handlers/llm/llm_handler.py

class _LLMSteering(BaseModel):
    """Structured output model for LLM steering decisions."""

    decision: Literal["proceed", "guide", "interrupt"] = Field(
        description="Steering decision: 'proceed' to continue, 'guide' to provide feedback, 'interrupt' for human input"
    )
    reason: str = Field(description="Clear explanation of the steering decision and any guidance provided")