strands.event_loop.streaming

Utilities for handling streaming responses from language models.

Messages = List[Message] module-attribute

A list of messages representing a conversation.

StopReason = Literal['content_filtered', 'end_turn', 'guardrail_intervened', 'interrupt', 'max_tokens', 'stop_sequence', 'tool_use'] module-attribute

Reason for the model ending its response generation.

• "content_filtered": Content was filtered due to policy violation
• "end_turn": Normal completion of the response
• "guardrail_intervened": Guardrail system intervened
• "interrupt": Agent was interrupted for human input
• "max_tokens": Maximum token limit reached
• "stop_sequence": Stop sequence encountered
• "tool_use": Model requested to use a tool

logger = logging.getLogger(__name__) module-attribute

CitationStreamEvent

Bases: ModelStreamEvent

Event emitted during citation streaming.

Source code in strands/types/_events.py

class CitationStreamEvent(ModelStreamEvent):
    """Event emitted during citation streaming."""

    def __init__(self, delta: ContentBlockDelta, citation: Citation) -> None:
        """Initialize with delta and citation content."""
        super().__init__({"citation": citation, "delta": delta})

__init__(delta, citation)

Initialize with delta and citation content.

Source code in strands/types/_events.py

def __init__(self, delta: ContentBlockDelta, citation: Citation) -> None:
    """Initialize with delta and citation content."""
    super().__init__({"citation": citation, "delta": delta})

CitationsContentBlock

Bases: TypedDict

A content block containing generated text and associated citations.

This block type is returned when document citations are enabled, providing traceability between the generated content and the source documents that informed the response.

Attributes:

Name	Type	Description
citations	List[Citation]	An array of citations that reference the source documents used to generate the associated content.
content	List[CitationGeneratedContent]	The generated content that is supported by the associated citations.

Source code in strands/types/citations.py

class CitationsContentBlock(TypedDict, total=False):
    """A content block containing generated text and associated citations.

    This block type is returned when document citations are enabled, providing
    traceability between the generated content and the source documents that
    informed the response.

    Attributes:
        citations: An array of citations that reference the source documents
            used to generate the associated content.
        content: The generated content that is supported by the associated
            citations.
    """

    citations: List[Citation]
    content: List[CitationGeneratedContent]

ContentBlock

Bases: TypedDict

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

Attributes:

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

Source code in strands/types/content.py

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

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

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

ContentBlockDeltaEvent

Bases: TypedDict

Event containing a delta update for a content block in a streaming response.

Attributes:

Name	Type	Description
contentBlockIndex	Optional[int]	Index of the content block within the message. This is optional to accommodate different model providers.
delta	ContentBlockDelta	The incremental content update for the content block.

Source code in strands/types/streaming.py

class ContentBlockDeltaEvent(TypedDict, total=False):
    """Event containing a delta update for a content block in a streaming response.

    Attributes:
        contentBlockIndex: Index of the content block within the message.
            This is optional to accommodate different model providers.
        delta: The incremental content update for the content block.
    """

    contentBlockIndex: Optional[int]
    delta: ContentBlockDelta

ContentBlockStart

Bases: TypedDict

Content block start information.

Attributes:

Name	Type	Description
toolUse	Optional[ContentBlockStartToolUse]	Information about a tool that the model is requesting to use.

Source code in strands/types/content.py

class ContentBlockStart(TypedDict, total=False):
    """Content block start information.

    Attributes:
        toolUse: Information about a tool that the model is requesting to use.
    """

    toolUse: Optional[ContentBlockStartToolUse]

ContentBlockStartEvent

Bases: TypedDict

Event signaling the start of a content block in a streaming response.

Attributes:

Name	Type	Description
contentBlockIndex	Optional[int]	Index of the content block within the message. This is optional to accommodate different model providers.
start	ContentBlockStart	Information about the content block being started.

Source code in strands/types/streaming.py

class ContentBlockStartEvent(TypedDict, total=False):
    """Event signaling the start of a content block in a streaming response.

    Attributes:
        contentBlockIndex: Index of the content block within the message.
            This is optional to accommodate different model providers.
        start: Information about the content block being started.
    """

    contentBlockIndex: Optional[int]
    start: ContentBlockStart

InvalidToolUseNameException

Bases: Exception

Exception raised when a tool use has an invalid name.

Source code in strands/tools/tools.py

class InvalidToolUseNameException(Exception):
    """Exception raised when a tool use has an invalid name."""

    pass

Message

Bases: TypedDict

A message in a conversation with the agent.

Attributes:

Name	Type	Description
content	List[ContentBlock]	The message content.
role	Role	The role of the message sender.

Source code in strands/types/content.py

class Message(TypedDict):
    """A message in a conversation with the agent.

    Attributes:
        content: The message content.
        role: The role of the message sender.
    """

    content: List[ContentBlock]
    role: Role

MessageStartEvent

Bases: TypedDict

Event signaling the start of a message in a streaming response.

Attributes:

Name	Type	Description
role	Role	The role of the message sender (e.g., "assistant", "user").

Source code in strands/types/streaming.py

class MessageStartEvent(TypedDict):
    """Event signaling the start of a message in a streaming response.

    Attributes:
        role: The role of the message sender (e.g., "assistant", "user").
    """

    role: Role

MessageStopEvent

Bases: TypedDict

Event signaling the end of a message in a streaming response.

Attributes:

Name	Type	Description
additionalModelResponseFields	Optional[Union[dict, list, int, float, str, bool, None]]	Additional fields to include in model response.
stopReason	StopReason	The reason why the model stopped generating content.

Source code in strands/types/streaming.py

class MessageStopEvent(TypedDict, total=False):
    """Event signaling the end of a message in a streaming response.

    Attributes:
        additionalModelResponseFields: Additional fields to include in model response.
        stopReason: The reason why the model stopped generating content.
    """

    additionalModelResponseFields: Optional[Union[dict, list, int, float, str, bool, None]]
    stopReason: StopReason

MetadataEvent

Bases: TypedDict

Event containing metadata about the streaming response.

Attributes:

Name	Type	Description
metrics	Metrics	Performance metrics related to the model invocation.
trace	Optional[Trace]	Trace information for debugging and monitoring.
usage	Usage	Resource usage information for the model invocation.

Source code in strands/types/streaming.py

class MetadataEvent(TypedDict, total=False):
    """Event containing metadata about the streaming response.

    Attributes:
        metrics: Performance metrics related to the model invocation.
        trace: Trace information for debugging and monitoring.
        usage: Resource usage information for the model invocation.
    """

    metrics: Metrics
    trace: Optional[Trace]
    usage: Usage

Metrics

Bases: TypedDict

Performance metrics for model interactions.

Attributes:

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

Source code in strands/types/event_loop.py

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

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

    latencyMs: Required[int]
    timeToFirstByteMs: int

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: Optional[str] = None, **kwargs: Any
    ) -> AsyncGenerator[dict[str, Union[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: Optional[list[ToolSpec]] = None,
        system_prompt: Optional[str] = None,
        *,
        tool_choice: ToolChoice | None = None,
        system_prompt_content: list[SystemContentBlock] | 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.
            **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, **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	Optional[list[ToolSpec]]	List of tool specifications to make available to the model.	None
system_prompt	Optional[str]	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
**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: Optional[list[ToolSpec]] = None,
    system_prompt: Optional[str] = None,
    *,
    tool_choice: ToolChoice | None = None,
    system_prompt_content: list[SystemContentBlock] | 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.
        **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	Optional[str]	System prompt to provide context to the model.	None
**kwargs	Any	Additional keyword arguments for future extensibility.	{}

Yields:

Type	Description
AsyncGenerator[dict[str, Union[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: Optional[str] = None, **kwargs: Any
) -> AsyncGenerator[dict[str, Union[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

ModelStopReason

Bases: TypedEvent

Event emitted during reasoning signature streaming.

Source code in strands/types/_events.py

class ModelStopReason(TypedEvent):
    """Event emitted during reasoning signature streaming."""

    def __init__(
        self,
        stop_reason: StopReason,
        message: Message,
        usage: Usage,
        metrics: Metrics,
    ) -> None:
        """Initialize with the final execution results.

        Args:
            stop_reason: Why the agent execution stopped
            message: Final message from the model
            usage: Usage information from the model
            metrics: Execution metrics and performance data
        """
        super().__init__({"stop": (stop_reason, message, usage, metrics)})

    @property
    @override
    def is_callback_event(self) -> bool:
        return False

__init__(stop_reason, message, usage, metrics)

Initialize with the final execution results.

Parameters:

Name	Type	Description	Default
stop_reason	StopReason	Why the agent execution stopped	required
message	Message	Final message from the model	required
usage	Usage	Usage information from the model	required
metrics	Metrics	Execution metrics and performance data	required

Source code in strands/types/_events.py

def __init__(
    self,
    stop_reason: StopReason,
    message: Message,
    usage: Usage,
    metrics: Metrics,
) -> None:
    """Initialize with the final execution results.

    Args:
        stop_reason: Why the agent execution stopped
        message: Final message from the model
        usage: Usage information from the model
        metrics: Execution metrics and performance data
    """
    super().__init__({"stop": (stop_reason, message, usage, metrics)})

ModelStreamChunkEvent

Bases: TypedEvent

Event emitted during model response streaming for each raw chunk.

Source code in strands/types/_events.py

class ModelStreamChunkEvent(TypedEvent):
    """Event emitted during model response streaming for each raw chunk."""

    def __init__(self, chunk: StreamEvent) -> None:
        """Initialize with streaming delta data from the model.

        Args:
            chunk: Incremental streaming data from the model response
        """
        super().__init__({"event": chunk})

    @property
    def chunk(self) -> StreamEvent:
        return cast(StreamEvent, self.get("event"))

__init__(chunk)

Initialize with streaming delta data from the model.

Parameters:

Name	Type	Description	Default
chunk	StreamEvent	Incremental streaming data from the model response	required

Source code in strands/types/_events.py

def __init__(self, chunk: StreamEvent) -> None:
    """Initialize with streaming delta data from the model.

    Args:
        chunk: Incremental streaming data from the model response
    """
    super().__init__({"event": chunk})

ModelStreamEvent

Bases: TypedEvent

Event emitted during model response streaming.

This event is fired when the model produces streaming output during response generation.

Source code in strands/types/_events.py

class ModelStreamEvent(TypedEvent):
    """Event emitted during model response streaming.

    This event is fired when the model produces streaming output during response
    generation.
    """

    def __init__(self, delta_data: dict[str, Any]) -> None:
        """Initialize with streaming delta data from the model.

        Args:
            delta_data: Incremental streaming data from the model response
        """
        super().__init__(delta_data)

    @property
    def is_callback_event(self) -> bool:
        # Only invoke a callback if we're non-empty
        return len(self.keys()) > 0

    @override
    def prepare(self, invocation_state: dict) -> None:
        if "delta" in self:
            self.update(invocation_state)

__init__(delta_data)

Initialize with streaming delta data from the model.

Parameters:

Name	Type	Description	Default
delta_data	dict[str, Any]	Incremental streaming data from the model response	required

Source code in strands/types/_events.py

def __init__(self, delta_data: dict[str, Any]) -> None:
    """Initialize with streaming delta data from the model.

    Args:
        delta_data: Incremental streaming data from the model response
    """
    super().__init__(delta_data)

ReasoningRedactedContentStreamEvent

Bases: ModelStreamEvent

Event emitted during redacted content streaming.

Source code in strands/types/_events.py

class ReasoningRedactedContentStreamEvent(ModelStreamEvent):
    """Event emitted during redacted content streaming."""

    def __init__(self, delta: ContentBlockDelta, redacted_content: bytes | None) -> None:
        """Initialize with delta and redacted content."""
        super().__init__({"reasoningRedactedContent": redacted_content, "delta": delta, "reasoning": True})

__init__(delta, redacted_content)

Initialize with delta and redacted content.

Source code in strands/types/_events.py

def __init__(self, delta: ContentBlockDelta, redacted_content: bytes | None) -> None:
    """Initialize with delta and redacted content."""
    super().__init__({"reasoningRedactedContent": redacted_content, "delta": delta, "reasoning": True})

ReasoningSignatureStreamEvent

Bases: ModelStreamEvent

Event emitted during reasoning signature streaming.

Source code in strands/types/_events.py

class ReasoningSignatureStreamEvent(ModelStreamEvent):
    """Event emitted during reasoning signature streaming."""

    def __init__(self, delta: ContentBlockDelta, reasoning_signature: str | None) -> None:
        """Initialize with delta and reasoning signature."""
        super().__init__({"reasoning_signature": reasoning_signature, "delta": delta, "reasoning": True})

__init__(delta, reasoning_signature)

Initialize with delta and reasoning signature.

Source code in strands/types/_events.py

def __init__(self, delta: ContentBlockDelta, reasoning_signature: str | None) -> None:
    """Initialize with delta and reasoning signature."""
    super().__init__({"reasoning_signature": reasoning_signature, "delta": delta, "reasoning": True})

ReasoningTextStreamEvent

Bases: ModelStreamEvent

Event emitted during reasoning text streaming.

Source code in strands/types/_events.py

class ReasoningTextStreamEvent(ModelStreamEvent):
    """Event emitted during reasoning text streaming."""

    def __init__(self, delta: ContentBlockDelta, reasoning_text: str | None) -> None:
        """Initialize with delta and reasoning text."""
        super().__init__({"reasoningText": reasoning_text, "delta": delta, "reasoning": True})

__init__(delta, reasoning_text)

Initialize with delta and reasoning text.

Source code in strands/types/_events.py

def __init__(self, delta: ContentBlockDelta, reasoning_text: str | None) -> None:
    """Initialize with delta and reasoning text."""
    super().__init__({"reasoningText": reasoning_text, "delta": delta, "reasoning": True})

RedactContentEvent

Bases: TypedDict

Event for redacting content.

Attributes:

Name	Type	Description
redactUserContentMessage	Optional[str]	The string to overwrite the users input with.
redactAssistantContentMessage	Optional[str]	The string to overwrite the assistants output with.

Source code in strands/types/streaming.py

class RedactContentEvent(TypedDict, total=False):
    """Event for redacting content.

    Attributes:
        redactUserContentMessage: The string to overwrite the users input with.
        redactAssistantContentMessage: The string to overwrite the assistants output with.

    """

    redactUserContentMessage: Optional[str]
    redactAssistantContentMessage: Optional[str]

StreamEvent

Bases: TypedDict

The messages output stream.

Attributes:

Name	Type	Description
contentBlockDelta	ContentBlockDeltaEvent	Delta content for a content block.
contentBlockStart	ContentBlockStartEvent	Start of a content block.
contentBlockStop	ContentBlockStopEvent	End of a content block.
internalServerException	ExceptionEvent	Internal server error information.
messageStart	MessageStartEvent	Start of a message.
messageStop	MessageStopEvent	End of a message.
metadata	MetadataEvent	Metadata about the streaming response.
modelStreamErrorException	ModelStreamErrorEvent	Model streaming error information.
serviceUnavailableException	ExceptionEvent	Service unavailable error information.
throttlingException	ExceptionEvent	Throttling error information.
validationException	ExceptionEvent	Validation error information.

Source code in strands/types/streaming.py

class StreamEvent(TypedDict, total=False):
    """The messages output stream.

    Attributes:
        contentBlockDelta: Delta content for a content block.
        contentBlockStart: Start of a content block.
        contentBlockStop: End of a content block.
        internalServerException: Internal server error information.
        messageStart: Start of a message.
        messageStop: End of a message.
        metadata: Metadata about the streaming response.
        modelStreamErrorException: Model streaming error information.
        serviceUnavailableException: Service unavailable error information.
        throttlingException: Throttling error information.
        validationException: Validation error information.
    """

    contentBlockDelta: ContentBlockDeltaEvent
    contentBlockStart: ContentBlockStartEvent
    contentBlockStop: ContentBlockStopEvent
    internalServerException: ExceptionEvent
    messageStart: MessageStartEvent
    messageStop: MessageStopEvent
    metadata: MetadataEvent
    redactContent: RedactContentEvent
    modelStreamErrorException: ModelStreamErrorEvent
    serviceUnavailableException: ExceptionEvent
    throttlingException: ExceptionEvent
    validationException: ExceptionEvent

SystemContentBlock

Bases: TypedDict

Contains configurations for instructions to provide the model for how to handle input.

Attributes:

Name	Type	Description
cachePoint	CachePoint	A cache point configuration to optimize conversation history.
text	str	A system prompt for the model.

Source code in strands/types/content.py

class SystemContentBlock(TypedDict, total=False):
    """Contains configurations for instructions to provide the model for how to handle input.

    Attributes:
        cachePoint: A cache point configuration to optimize conversation history.
        text: A system prompt for the model.
    """

    cachePoint: CachePoint
    text: str

TextStreamEvent

Bases: ModelStreamEvent

Event emitted during text content streaming.

Source code in strands/types/_events.py

class TextStreamEvent(ModelStreamEvent):
    """Event emitted during text content streaming."""

    def __init__(self, delta: ContentBlockDelta, text: str) -> None:
        """Initialize with delta and text content."""
        super().__init__({"data": text, "delta": delta})

__init__(delta, text)

Initialize with delta and text content.

Source code in strands/types/_events.py

def __init__(self, delta: ContentBlockDelta, text: str) -> None:
    """Initialize with delta and text content."""
    super().__init__({"data": text, "delta": delta})

ToolSpec

Bases: TypedDict

Specification for a tool that can be used by an agent.

Attributes:

Name	Type	Description
description	str	A human-readable description of what the tool does.
inputSchema	JSONSchema	JSON Schema defining the expected input parameters.
name	str	The unique name of the tool.
outputSchema	NotRequired[JSONSchema]	Optional JSON Schema defining the expected output format. Note: Not all model providers support this field. Providers that don't support it should filter it out before sending to their API.

Source code in strands/types/tools.py

class ToolSpec(TypedDict):
    """Specification for a tool that can be used by an agent.

    Attributes:
        description: A human-readable description of what the tool does.
        inputSchema: JSON Schema defining the expected input parameters.
        name: The unique name of the tool.
        outputSchema: Optional JSON Schema defining the expected output format.
            Note: Not all model providers support this field. Providers that don't
            support it should filter it out before sending to their API.
    """

    description: str
    inputSchema: JSONSchema
    name: str
    outputSchema: NotRequired[JSONSchema]

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

ToolUseStreamEvent

Bases: ModelStreamEvent

Event emitted during tool use input streaming.

Source code in strands/types/_events.py

class ToolUseStreamEvent(ModelStreamEvent):
    """Event emitted during tool use input streaming."""

    def __init__(self, delta: ContentBlockDelta, current_tool_use: dict[str, Any]) -> None:
        """Initialize with delta and current tool use state."""
        super().__init__({"type": "tool_use_stream", "delta": delta, "current_tool_use": current_tool_use})

__init__(delta, current_tool_use)

Initialize with delta and current tool use state.

Source code in strands/types/_events.py

def __init__(self, delta: ContentBlockDelta, current_tool_use: dict[str, Any]) -> None:
    """Initialize with delta and current tool use state."""
    super().__init__({"type": "tool_use_stream", "delta": delta, "current_tool_use": current_tool_use})

TypedEvent

Bases: dict

Base class for all typed events in the agent system.

Source code in strands/types/_events.py

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

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

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

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

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

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

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

is_callback_event property

True if this event should trigger the callback_handler to fire.

__init__(data=None)

Initialize the typed event with optional data.

Parameters:

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

Source code in strands/types/_events.py

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

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

as_dict()

Convert this event to a raw dictionary for emitting purposes.

Source code in strands/types/_events.py

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

prepare(invocation_state)

Prepare the event for emission by adding invocation state.

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

Source code in strands/types/_events.py

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

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

Usage

Bases: TypedDict

Token usage information for model interactions.

Attributes:

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

Source code in strands/types/event_loop.py

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

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

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

_normalize_messages(messages)

Remove or replace blank text in message content.

Parameters:

Name	Type	Description	Default
messages	Messages	Conversation messages to update.	required

Returns:

Type	Description
Messages	Updated messages.

Source code in strands/event_loop/streaming.py

def _normalize_messages(messages: Messages) -> Messages:
    """Remove or replace blank text in message content.

    Args:
        messages: Conversation messages to update.

    Returns:
        Updated messages.
    """
    removed_blank_message_content_text = False
    replaced_blank_message_content_text = False
    replaced_tool_names = False

    for message in messages:
        # only modify assistant messages
        if "role" in message and message["role"] != "assistant":
            continue
        if "content" in message:
            content = message["content"]
            if len(content) == 0:
                content.append({"text": "[blank text]"})
                continue

            has_tool_use = False

            # Ensure the tool-uses always have valid names before sending
            # https://github.com/strands-agents/sdk-python/issues/1069
            for item in content:
                if "toolUse" in item:
                    has_tool_use = True
                    tool_use: ToolUse = item["toolUse"]

                    try:
                        validate_tool_use_name(tool_use)
                    except InvalidToolUseNameException:
                        tool_use["name"] = "INVALID_TOOL_NAME"
                        replaced_tool_names = True

            if has_tool_use:
                # Remove blank 'text' items for assistant messages
                before_len = len(content)
                content[:] = [item for item in content if "text" not in item or item["text"].strip()]
                if not removed_blank_message_content_text and before_len != len(content):
                    removed_blank_message_content_text = True
            else:
                # Replace blank 'text' with '[blank text]' for assistant messages
                for item in content:
                    if "text" in item and not item["text"].strip():
                        replaced_blank_message_content_text = True
                        item["text"] = "[blank text]"

    if removed_blank_message_content_text:
        logger.debug("removed blank message context text")
    if replaced_blank_message_content_text:
        logger.debug("replaced blank message context text")
    if replaced_tool_names:
        logger.debug("replaced invalid tool name")

    return messages

extract_usage_metrics(event, time_to_first_byte_ms=None)

Extracts usage metrics from the metadata chunk.

Parameters:

Name	Type	Description	Default
event	MetadataEvent	metadata.	required
time_to_first_byte_ms	int | None	time to get the first byte from the model in milliseconds	None

Returns:

Type	Description
tuple[Usage, Metrics]	The extracted usage metrics and latency.

Source code in strands/event_loop/streaming.py

def extract_usage_metrics(event: MetadataEvent, time_to_first_byte_ms: int | None = None) -> tuple[Usage, Metrics]:
    """Extracts usage metrics from the metadata chunk.

    Args:
        event: metadata.
        time_to_first_byte_ms: time to get the first byte from the model in milliseconds

    Returns:
        The extracted usage metrics and latency.
    """
    # MetadataEvent has total=False, making all fields optional, but Usage and Metrics types
    # have Required fields. Provide defaults to handle cases where custom models don't
    # provide usage/metrics (e.g., when latency info is unavailable).
    usage = Usage(**{"inputTokens": 0, "outputTokens": 0, "totalTokens": 0, **event.get("usage", {})})
    metrics = Metrics(**{"latencyMs": 0, **event.get("metrics", {})})
    if time_to_first_byte_ms:
        metrics["timeToFirstByteMs"] = time_to_first_byte_ms

    return usage, metrics

handle_content_block_delta(event, state)

Handles content block delta updates by appending text, tool input, or reasoning content to the state.

Parameters:

Name	Type	Description	Default
event	ContentBlockDeltaEvent	Delta event.	required
state	dict[str, Any]	The current state of message processing.	required

Returns:

Type	Description
tuple[dict[str, Any], ModelStreamEvent]	Updated state with appended text or tool input.

Source code in strands/event_loop/streaming.py

def handle_content_block_delta(
    event: ContentBlockDeltaEvent, state: dict[str, Any]
) -> tuple[dict[str, Any], ModelStreamEvent]:
    """Handles content block delta updates by appending text, tool input, or reasoning content to the state.

    Args:
        event: Delta event.
        state: The current state of message processing.

    Returns:
        Updated state with appended text or tool input.
    """
    delta_content = event["delta"]

    typed_event: ModelStreamEvent = ModelStreamEvent({})

    if "toolUse" in delta_content:
        if "input" not in state["current_tool_use"]:
            state["current_tool_use"]["input"] = ""

        state["current_tool_use"]["input"] += delta_content["toolUse"]["input"]
        typed_event = ToolUseStreamEvent(delta_content, state["current_tool_use"])

    elif "text" in delta_content:
        state["text"] += delta_content["text"]
        typed_event = TextStreamEvent(text=delta_content["text"], delta=delta_content)

    elif "citation" in delta_content:
        if "citationsContent" not in state:
            state["citationsContent"] = []

        state["citationsContent"].append(delta_content["citation"])
        typed_event = CitationStreamEvent(delta=delta_content, citation=delta_content["citation"])

    elif "reasoningContent" in delta_content:
        if "text" in delta_content["reasoningContent"]:
            if "reasoningText" not in state:
                state["reasoningText"] = ""

            state["reasoningText"] += delta_content["reasoningContent"]["text"]
            typed_event = ReasoningTextStreamEvent(
                reasoning_text=delta_content["reasoningContent"]["text"],
                delta=delta_content,
            )

        elif "signature" in delta_content["reasoningContent"]:
            if "signature" not in state:
                state["signature"] = ""

            state["signature"] += delta_content["reasoningContent"]["signature"]
            typed_event = ReasoningSignatureStreamEvent(
                reasoning_signature=delta_content["reasoningContent"]["signature"],
                delta=delta_content,
            )

        elif redacted_content := delta_content["reasoningContent"].get("redactedContent"):
            state["redactedContent"] = state.get("redactedContent", b"") + redacted_content
            typed_event = ReasoningRedactedContentStreamEvent(redacted_content=redacted_content, delta=delta_content)

    return state, typed_event

handle_content_block_start(event)

Handles the start of a content block by extracting tool usage information if any.

Parameters:

Name	Type	Description	Default
event	ContentBlockStartEvent	Start event.	required

Returns:

Type	Description
dict[str, Any]	Dictionary with tool use id and name if tool use request, empty dictionary otherwise.

Source code in strands/event_loop/streaming.py

def handle_content_block_start(event: ContentBlockStartEvent) -> dict[str, Any]:
    """Handles the start of a content block by extracting tool usage information if any.

    Args:
        event: Start event.

    Returns:
        Dictionary with tool use id and name if tool use request, empty dictionary otherwise.
    """
    start: ContentBlockStart = event["start"]
    current_tool_use = {}

    if "toolUse" in start and start["toolUse"]:
        tool_use_data = start["toolUse"]
        current_tool_use["toolUseId"] = tool_use_data["toolUseId"]
        current_tool_use["name"] = tool_use_data["name"]
        current_tool_use["input"] = ""

    return current_tool_use

handle_content_block_stop(state)

Handles the end of a content block by finalizing tool usage, text content, or reasoning content.

Parameters:

Name	Type	Description	Default
state	dict[str, Any]	The current state of message processing.	required

Returns:

Type	Description
dict[str, Any]	Updated state with finalized content block.

Source code in strands/event_loop/streaming.py

def handle_content_block_stop(state: dict[str, Any]) -> dict[str, Any]:
    """Handles the end of a content block by finalizing tool usage, text content, or reasoning content.

    Args:
        state: The current state of message processing.

    Returns:
        Updated state with finalized content block.
    """
    content: list[ContentBlock] = state["content"]

    current_tool_use = state["current_tool_use"]
    text = state["text"]
    reasoning_text = state["reasoningText"]
    citations_content = state["citationsContent"]
    redacted_content = state.get("redactedContent")

    if current_tool_use:
        if "input" not in current_tool_use:
            current_tool_use["input"] = ""

        try:
            current_tool_use["input"] = json.loads(current_tool_use["input"])
        except ValueError:
            current_tool_use["input"] = {}

        tool_use_id = current_tool_use["toolUseId"]
        tool_use_name = current_tool_use["name"]

        tool_use = ToolUse(
            toolUseId=tool_use_id,
            name=tool_use_name,
            input=current_tool_use["input"],
        )
        content.append({"toolUse": tool_use})
        state["current_tool_use"] = {}

    elif text:
        if citations_content:
            citations_block: CitationsContentBlock = {"citations": citations_content, "content": [{"text": text}]}
            content.append({"citationsContent": citations_block})
            state["citationsContent"] = []
        else:
            content.append({"text": text})
        state["text"] = ""

    elif reasoning_text:
        content_block: ContentBlock = {
            "reasoningContent": {
                "reasoningText": {
                    "text": state["reasoningText"],
                }
            }
        }

        if "signature" in state:
            content_block["reasoningContent"]["reasoningText"]["signature"] = state["signature"]

        content.append(content_block)
        state["reasoningText"] = ""
    elif redacted_content:
        content.append({"reasoningContent": {"redactedContent": redacted_content}})
        state["redactedContent"] = b""

    return state

handle_message_start(event, message)

Handles the start of a message by setting the role in the message dictionary.

Parameters:

Name	Type	Description	Default
event	MessageStartEvent	A message start event.	required
message	Message	The message dictionary being constructed.	required

Returns:

Type	Description
Message	Updated message dictionary with the role set.

Source code in strands/event_loop/streaming.py

def handle_message_start(event: MessageStartEvent, message: Message) -> Message:
    """Handles the start of a message by setting the role in the message dictionary.

    Args:
        event: A message start event.
        message: The message dictionary being constructed.

    Returns:
        Updated message dictionary with the role set.
    """
    message["role"] = event["role"]
    return message

handle_message_stop(event)

Handles the end of a message by returning the stop reason.

Parameters:

Name	Type	Description	Default
event	MessageStopEvent	Stop event.	required

Returns:

Type	Description
StopReason	The reason for stopping the stream.

Source code in strands/event_loop/streaming.py

def handle_message_stop(event: MessageStopEvent) -> StopReason:
    """Handles the end of a message by returning the stop reason.

    Args:
        event: Stop event.

    Returns:
        The reason for stopping the stream.
    """
    return event["stopReason"]

handle_redact_content(event, state)

Handles redacting content from the input or output.

Parameters:

Name	Type	Description	Default
event	RedactContentEvent	Redact Content Event.	required
state	dict[str, Any]	The current state of message processing.	required

Source code in strands/event_loop/streaming.py

def handle_redact_content(event: RedactContentEvent, state: dict[str, Any]) -> None:
    """Handles redacting content from the input or output.

    Args:
        event: Redact Content Event.
        state: The current state of message processing.
    """
    if event.get("redactAssistantContentMessage") is not None:
        state["message"]["content"] = [{"text": event["redactAssistantContentMessage"]}]

process_stream(chunks, start_time=None) async

Processes the response stream from the API, constructing the final message and extracting usage metrics.

Parameters:

Name	Type	Description	Default
chunks	AsyncIterable[StreamEvent]	The chunks of the response stream from the model.	required
start_time	float | None	Time when the model request is initiated	None

Yields:

Type	Description
AsyncGenerator[TypedEvent, None]	The reason for stopping, the constructed message, and the usage metrics.

Source code in strands/event_loop/streaming.py

async def process_stream(
    chunks: AsyncIterable[StreamEvent], start_time: float | None = None
) -> AsyncGenerator[TypedEvent, None]:
    """Processes the response stream from the API, constructing the final message and extracting usage metrics.

    Args:
        chunks: The chunks of the response stream from the model.
        start_time: Time when the model request is initiated

    Yields:
        The reason for stopping, the constructed message, and the usage metrics.
    """
    stop_reason: StopReason = "end_turn"
    first_byte_time = None

    state: dict[str, Any] = {
        "message": {"role": "assistant", "content": []},
        "text": "",
        "current_tool_use": {},
        "reasoningText": "",
        "citationsContent": [],
    }
    state["content"] = state["message"]["content"]

    usage: Usage = Usage(inputTokens=0, outputTokens=0, totalTokens=0)
    metrics: Metrics = Metrics(latencyMs=0, timeToFirstByteMs=0)

    async for chunk in chunks:
        # Track first byte time when we get first content
        if first_byte_time is None and ("contentBlockDelta" in chunk or "contentBlockStart" in chunk):
            first_byte_time = time.time()
        yield ModelStreamChunkEvent(chunk=chunk)

        if "messageStart" in chunk:
            state["message"] = handle_message_start(chunk["messageStart"], state["message"])
        elif "contentBlockStart" in chunk:
            state["current_tool_use"] = handle_content_block_start(chunk["contentBlockStart"])
        elif "contentBlockDelta" in chunk:
            state, typed_event = handle_content_block_delta(chunk["contentBlockDelta"], state)
            yield typed_event
        elif "contentBlockStop" in chunk:
            state = handle_content_block_stop(state)
        elif "messageStop" in chunk:
            stop_reason = handle_message_stop(chunk["messageStop"])
        elif "metadata" in chunk:
            time_to_first_byte_ms = (
                int(1000 * (first_byte_time - start_time)) if (start_time and first_byte_time) else None
            )
            usage, metrics = extract_usage_metrics(chunk["metadata"], time_to_first_byte_ms)
        elif "redactContent" in chunk:
            handle_redact_content(chunk["redactContent"], state)

    yield ModelStopReason(stop_reason=stop_reason, message=state["message"], usage=usage, metrics=metrics)

remove_blank_messages_content_text(messages)

Remove or replace blank text in message content.

!!deprecated!! This function is deprecated and will be removed in a future version.

Parameters:

Name	Type	Description	Default
messages	Messages	Conversation messages to update.	required

Returns:

Type	Description
Messages	Updated messages.

Source code in strands/event_loop/streaming.py

def remove_blank_messages_content_text(messages: Messages) -> Messages:
    """Remove or replace blank text in message content.

    !!deprecated!!
        This function is deprecated and will be removed in a future version.

    Args:
        messages: Conversation messages to update.

    Returns:
        Updated messages.
    """
    warnings.warn(
        "remove_blank_messages_content_text is deprecated and will be removed in a future version.",
        DeprecationWarning,
        stacklevel=2,
    )
    removed_blank_message_content_text = False
    replaced_blank_message_content_text = False

    for message in messages:
        # only modify assistant messages
        if "role" in message and message["role"] != "assistant":
            continue
        if "content" in message:
            content = message["content"]
            has_tool_use = any("toolUse" in item for item in content)
            if len(content) == 0:
                content.append({"text": "[blank text]"})
                continue

            if has_tool_use:
                # Remove blank 'text' items for assistant messages
                before_len = len(content)
                content[:] = [item for item in content if "text" not in item or item["text"].strip()]
                if not removed_blank_message_content_text and before_len != len(content):
                    removed_blank_message_content_text = True
            else:
                # Replace blank 'text' with '[blank text]' for assistant messages
                for item in content:
                    if "text" in item and not item["text"].strip():
                        replaced_blank_message_content_text = True
                        item["text"] = "[blank text]"

    if removed_blank_message_content_text:
        logger.debug("removed blank message context text")
    if replaced_blank_message_content_text:
        logger.debug("replaced blank message context text")

    return messages

stream_messages(model, system_prompt, messages, tool_specs, *, tool_choice=None, system_prompt_content=None, **kwargs) async

Streams messages to the model and processes the response.

Parameters:

Name	Type	Description	Default
model	Model	Model provider.	required
system_prompt	Optional[str]	The system prompt string, used for backwards compatibility with models that expect it.	required
messages	Messages	List of messages to send.	required
tool_specs	list[ToolSpec]	The list of tool specs.	required
tool_choice	Optional[Any]	Optional tool choice constraint for forcing specific tool usage.	None
system_prompt_content	Optional[list[SystemContentBlock]]	The authoritative system prompt content blocks that always contains the system prompt data.	None
**kwargs	Any	Additional keyword arguments for future extensibility.	{}

Yields:

Type	Description
AsyncGenerator[TypedEvent, None]	The reason for stopping, the final message, and the usage metrics

Source code in strands/event_loop/streaming.py

async def stream_messages(
    model: Model,
    system_prompt: Optional[str],
    messages: Messages,
    tool_specs: list[ToolSpec],
    *,
    tool_choice: Optional[Any] = None,
    system_prompt_content: Optional[list[SystemContentBlock]] = None,
    **kwargs: Any,
) -> AsyncGenerator[TypedEvent, None]:
    """Streams messages to the model and processes the response.

    Args:
        model: Model provider.
        system_prompt: The system prompt string, used for backwards compatibility with models that expect it.
        messages: List of messages to send.
        tool_specs: The list of tool specs.
        tool_choice: Optional tool choice constraint for forcing specific tool usage.
        system_prompt_content: The authoritative system prompt content blocks that always contains the
            system prompt data.
        **kwargs: Additional keyword arguments for future extensibility.

    Yields:
        The reason for stopping, the final message, and the usage metrics
    """
    logger.debug("model=<%s> | streaming messages", model)

    messages = _normalize_messages(messages)
    start_time = time.time()

    chunks = model.stream(
        messages,
        tool_specs if tool_specs else None,
        system_prompt,
        tool_choice=tool_choice,
        system_prompt_content=system_prompt_content,
    )

    async for event in process_stream(chunks, start_time):
        yield event

validate_tool_use_name(tool)

Validate the name of a tool use.

Parameters:

Name	Type	Description	Default
tool	ToolUse	The tool use to validate.	required

Raises:

Type	Description
InvalidToolUseNameException	If the tool name is invalid.

Source code in strands/tools/tools.py

def validate_tool_use_name(tool: ToolUse) -> None:
    """Validate the name of a tool use.

    Args:
        tool: The tool use to validate.

    Raises:
        InvalidToolUseNameException: If the tool name is invalid.
    """
    # We need to fix some typing here, because we don't actually expect a ToolUse, but dict[str, Any]
    if "name" not in tool:
        message = "tool name missing"  # type: ignore[unreachable]
        logger.warning(message)
        raise InvalidToolUseNameException(message)

    tool_name = tool["name"]
    tool_name_pattern = r"^[a-zA-Z0-9_\-]{1,}$"
    tool_name_max_length = 64
    valid_name_pattern = bool(re.match(tool_name_pattern, tool_name))
    tool_name_len = len(tool_name)

    if not valid_name_pattern:
        message = f"tool_name=<{tool_name}> | invalid tool name pattern"
        logger.warning(message)
        raise InvalidToolUseNameException(message)

    if tool_name_len > tool_name_max_length:
        message = f"tool_name=<{tool_name}>, tool_name_max_length=<{tool_name_max_length}> | invalid tool name length"
        logger.warning(message)
        raise InvalidToolUseNameException(message)