strands.agent.base

Agent Interface.

Defines the minimal interface that all agent types must implement.

AgentInput = str | list[ContentBlock] | list[InterruptResponseContent] | Messages | None module-attribute

AgentBase

Bases: Protocol

Protocol defining the interface for all agent types in Strands.

This protocol defines the minimal contract that all agent implementations must satisfy.

Source code in strands/agent/base.py

@runtime_checkable
class AgentBase(Protocol):
    """Protocol defining the interface for all agent types in Strands.

    This protocol defines the minimal contract that all agent implementations
    must satisfy.
    """

    async def invoke_async(
        self,
        prompt: AgentInput = None,
        **kwargs: Any,
    ) -> AgentResult:
        """Asynchronously invoke the agent with the given prompt.

        Args:
            prompt: Input to the agent.
            **kwargs: Additional arguments.

        Returns:
            AgentResult containing the agent's response.
        """
        ...

    def __call__(
        self,
        prompt: AgentInput = None,
        **kwargs: Any,
    ) -> AgentResult:
        """Synchronously invoke the agent with the given prompt.

        Args:
            prompt: Input to the agent.
            **kwargs: Additional arguments.

        Returns:
            AgentResult containing the agent's response.
        """
        ...

    def stream_async(
        self,
        prompt: AgentInput = None,
        **kwargs: Any,
    ) -> AsyncIterator[Any]:
        """Stream agent execution asynchronously.

        Args:
            prompt: Input to the agent.
            **kwargs: Additional arguments.

        Yields:
            Events representing the streaming execution.
        """
        ...

__call__(prompt=None, **kwargs)

Synchronously invoke the agent with the given prompt.

Parameters:

Name	Type	Description	Default
prompt	AgentInput	Input to the agent.	None
**kwargs	Any	Additional arguments.	{}

Returns:

Type	Description
AgentResult	AgentResult containing the agent's response.

Source code in strands/agent/base.py

def __call__(
    self,
    prompt: AgentInput = None,
    **kwargs: Any,
) -> AgentResult:
    """Synchronously invoke the agent with the given prompt.

    Args:
        prompt: Input to the agent.
        **kwargs: Additional arguments.

    Returns:
        AgentResult containing the agent's response.
    """
    ...

invoke_async(prompt=None, **kwargs) async

Asynchronously invoke the agent with the given prompt.

Parameters:

Name	Type	Description	Default
prompt	AgentInput	Input to the agent.	None
**kwargs	Any	Additional arguments.	{}

Returns:

Type	Description
AgentResult	AgentResult containing the agent's response.

Source code in strands/agent/base.py

async def invoke_async(
    self,
    prompt: AgentInput = None,
    **kwargs: Any,
) -> AgentResult:
    """Asynchronously invoke the agent with the given prompt.

    Args:
        prompt: Input to the agent.
        **kwargs: Additional arguments.

    Returns:
        AgentResult containing the agent's response.
    """
    ...

stream_async(prompt=None, **kwargs)

Stream agent execution asynchronously.

Parameters:

Name	Type	Description	Default
prompt	AgentInput	Input to the agent.	None
**kwargs	Any	Additional arguments.	{}

Yields:

Type	Description
AsyncIterator[Any]	Events representing the streaming execution.

Source code in strands/agent/base.py

def stream_async(
    self,
    prompt: AgentInput = None,
    **kwargs: Any,
) -> AsyncIterator[Any]:
    """Stream agent execution asynchronously.

    Args:
        prompt: Input to the agent.
        **kwargs: Additional arguments.

    Yields:
        Events representing the streaming execution.
    """
    ...

AgentResult dataclass

Represents the last result of invoking an agent with a prompt.

Attributes:

Name	Type	Description
stop_reason	StopReason	The reason why the agent's processing stopped.
message	Message	The last message generated by the agent.
metrics	EventLoopMetrics	Performance metrics collected during processing.
state	Any	Additional state information from the event loop.
interrupts	Sequence[Interrupt] | None	List of interrupts if raised by user.
structured_output	BaseModel | None	Parsed structured output when structured_output_model was specified.

Source code in strands/agent/agent_result.py

@dataclass
class AgentResult:
    """Represents the last result of invoking an agent with a prompt.

    Attributes:
        stop_reason: The reason why the agent's processing stopped.
        message: The last message generated by the agent.
        metrics: Performance metrics collected during processing.
        state: Additional state information from the event loop.
        interrupts: List of interrupts if raised by user.
        structured_output: Parsed structured output when structured_output_model was specified.
    """

    stop_reason: StopReason
    message: Message
    metrics: EventLoopMetrics
    state: Any
    interrupts: Sequence[Interrupt] | None = None
    structured_output: BaseModel | None = None

    def __str__(self) -> str:
        """Get the agent's last message as a string.

        This method extracts and concatenates all text content from the final message, ignoring any non-text content
        like images or structured data. If there's no text content but structured output is present, it serializes
        the structured output instead.

        Returns:
            The agent's last message as a string.
        """
        content_array = self.message.get("content", [])

        result = ""
        for item in content_array:
            if isinstance(item, dict) and "text" in item:
                result += item.get("text", "") + "\n"

        if not result and self.structured_output:
            result = self.structured_output.model_dump_json()

        return result

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "AgentResult":
        """Rehydrate an AgentResult from persisted JSON.

        Args:
            data: Dictionary containing the serialized AgentResult data
        Returns:
            AgentResult instance
        Raises:
            TypeError: If the data format is invalid@
        """
        if data.get("type") != "agent_result":
            raise TypeError(f"AgentResult.from_dict: unexpected type {data.get('type')!r}")

        message = cast(Message, data.get("message"))
        stop_reason = cast(StopReason, data.get("stop_reason"))

        return cls(message=message, stop_reason=stop_reason, metrics=EventLoopMetrics(), state={})

    def to_dict(self) -> dict[str, Any]:
        """Convert this AgentResult to JSON-serializable dictionary.

        Returns:
            Dictionary containing serialized AgentResult data
        """
        return {
            "type": "agent_result",
            "message": self.message,
            "stop_reason": self.stop_reason,
        }

__str__()

Get the agent's last message as a string.

This method extracts and concatenates all text content from the final message, ignoring any non-text content like images or structured data. If there's no text content but structured output is present, it serializes the structured output instead.

Returns:

Type	Description
str	The agent's last message as a string.

Source code in strands/agent/agent_result.py

def __str__(self) -> str:
    """Get the agent's last message as a string.

    This method extracts and concatenates all text content from the final message, ignoring any non-text content
    like images or structured data. If there's no text content but structured output is present, it serializes
    the structured output instead.

    Returns:
        The agent's last message as a string.
    """
    content_array = self.message.get("content", [])

    result = ""
    for item in content_array:
        if isinstance(item, dict) and "text" in item:
            result += item.get("text", "") + "\n"

    if not result and self.structured_output:
        result = self.structured_output.model_dump_json()

    return result

from_dict(data) classmethod

Rehydrate an AgentResult from persisted JSON.

Parameters:

Name	Type	Description	Default
data	dict[str, Any]	Dictionary containing the serialized AgentResult data	required

Returns: AgentResult instance Raises: TypeError: If the data format is invalid@

Source code in strands/agent/agent_result.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "AgentResult":
    """Rehydrate an AgentResult from persisted JSON.

    Args:
        data: Dictionary containing the serialized AgentResult data
    Returns:
        AgentResult instance
    Raises:
        TypeError: If the data format is invalid@
    """
    if data.get("type") != "agent_result":
        raise TypeError(f"AgentResult.from_dict: unexpected type {data.get('type')!r}")

    message = cast(Message, data.get("message"))
    stop_reason = cast(StopReason, data.get("stop_reason"))

    return cls(message=message, stop_reason=stop_reason, metrics=EventLoopMetrics(), state={})

to_dict()

Convert this AgentResult to JSON-serializable dictionary.

Returns:

Type	Description
dict[str, Any]	Dictionary containing serialized AgentResult data

Source code in strands/agent/agent_result.py

def to_dict(self) -> dict[str, Any]:
    """Convert this AgentResult to JSON-serializable dictionary.

    Returns:
        Dictionary containing serialized AgentResult data
    """
    return {
        "type": "agent_result",
        "message": self.message,
        "stop_reason": self.stop_reason,
    }