strands.types.streaming

Streaming-related type definitions for the SDK.

These types are modeled after the Bedrock API.

• Bedrock docs: https://docs.aws.amazon.com/bedrock/latest/APIReference/API_Types_Amazon_Bedrock_Runtime.html

CitationLocation = Union[DocumentCharLocationDict, DocumentPageLocationDict, DocumentChunkLocationDict, SearchResultLocationDict, WebLocationDict] module-attribute

Role = Literal['user', 'assistant'] module-attribute

Role of a message sender.

• "user": Messages from the user to the assistant
• "assistant": Messages from the assistant to the user

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

CitationSourceContentDelta

Bases: TypedDict

Contains incremental updates to source content text during streaming.

Allows clients to build up the cited content progressively during streaming responses.

Attributes:

Name	Type	Description
text	str	An incremental update to the text content from the source document that is being cited.

Source code in strands/types/streaming.py

class CitationSourceContentDelta(TypedDict, total=False):
    """Contains incremental updates to source content text during streaming.

    Allows clients to build up the cited content progressively during
    streaming responses.

    Attributes:
        text: An incremental update to the text content from the source
            document that is being cited.
    """

    text: str

CitationsDelta

Bases: TypedDict

Contains incremental updates to citation information during streaming.

This allows clients to build up citation data progressively as the response is generated.

Attributes:

Name	Type	Description
location	CitationLocation	Specifies the precise location within a source document where cited content can be found. This can include character-level positions, page numbers, or document chunks depending on the document type and indexing method.
sourceContent	list[CitationSourceContentDelta]	The specific content from the source document that was referenced or cited in the generated response.
title	str	The title or identifier of the source document being cited.

Source code in strands/types/streaming.py

class CitationsDelta(TypedDict, total=False):
    """Contains incremental updates to citation information during streaming.

    This allows clients to build up citation data progressively as the
    response is generated.

    Attributes:
        location: Specifies the precise location within a source document
            where cited content can be found. This can include character-level
            positions, page numbers, or document chunks depending on the
            document type and indexing method.
        sourceContent: The specific content from the source document that was
            referenced or cited in the generated response.
        title: The title or identifier of the source document being cited.
    """

    location: CitationLocation
    sourceContent: list[CitationSourceContentDelta]
    title: str

ContentBlockDelta

Bases: TypedDict

A block of content in a streaming response.

Attributes:

Name	Type	Description
reasoningContent	ReasoningContentBlockDelta	Contains content regarding the reasoning that is carried out by the model.
text	str	Text fragment being streamed.
toolUse	ContentBlockDeltaToolUse	Tool use input fragment being streamed.

Source code in strands/types/streaming.py

class ContentBlockDelta(TypedDict, total=False):
    """A block of content in a streaming response.

    Attributes:
        reasoningContent: Contains content regarding the reasoning that is carried out by the model.
        text: Text fragment being streamed.
        toolUse: Tool use input fragment being streamed.
    """

    reasoningContent: ReasoningContentBlockDelta
    text: str
    toolUse: ContentBlockDeltaToolUse
    citation: CitationsDelta

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

ContentBlockDeltaText

Bases: TypedDict

Text content delta in a streaming response.

Attributes:

Name	Type	Description
text	str	The text fragment being streamed.

Source code in strands/types/streaming.py

class ContentBlockDeltaText(TypedDict):
    """Text content delta in a streaming response.

    Attributes:
        text: The text fragment being streamed.
    """

    text: str

ContentBlockDeltaToolUse

Bases: TypedDict

Tool use input delta in a streaming response.

Attributes:

Name	Type	Description
input	str	The tool input fragment being streamed.

Source code in strands/types/streaming.py

class ContentBlockDeltaToolUse(TypedDict):
    """Tool use input delta in a streaming response.

    Attributes:
        input: The tool input fragment being streamed.
    """

    input: str

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

ContentBlockStopEvent

Bases: TypedDict

Event signaling the end 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.

Source code in strands/types/streaming.py

class ContentBlockStopEvent(TypedDict, total=False):
    """Event signaling the end 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.
    """

    contentBlockIndex: Optional[int]

ExceptionEvent

Bases: TypedDict

Base event for exceptions in a streaming response.

Attributes:

Name	Type	Description
message	str	The error message describing what went wrong.

Source code in strands/types/streaming.py

class ExceptionEvent(TypedDict):
    """Base event for exceptions in a streaming response.

    Attributes:
        message: The error message describing what went wrong.
    """

    message: str

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

ModelStreamErrorEvent

Bases: ExceptionEvent

Event for model streaming errors.

Attributes:

Name	Type	Description
originalMessage	str	The original error message from the model provider.
originalStatusCode	int	The HTTP status code returned by the model provider.

Source code in strands/types/streaming.py

class ModelStreamErrorEvent(ExceptionEvent):
    """Event for model streaming errors.

    Attributes:
        originalMessage: The original error message from the model provider.
        originalStatusCode: The HTTP status code returned by the model provider.
    """

    originalMessage: str
    originalStatusCode: int

ReasoningContentBlockDelta

Bases: TypedDict

Delta for reasoning content block in a streaming response.

Attributes:

Name	Type	Description
redactedContent	Optional[bytes]	The content in the reasoning that was encrypted by the model provider for safety reasons.
signature	Optional[str]	A token that verifies that the reasoning text was generated by the model.
text	Optional[str]	The reasoning that the model used to return the output.

Source code in strands/types/streaming.py

class ReasoningContentBlockDelta(TypedDict, total=False):
    """Delta for reasoning content block in a streaming response.

    Attributes:
        redactedContent: The content in the reasoning that was encrypted by the model provider for safety reasons.
        signature: A token that verifies that the reasoning text was generated by the model.
        text: The reasoning that the model used to return the output.
    """

    redactedContent: Optional[bytes]
    signature: Optional[str]
    text: Optional[str]

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

Trace

Bases: TypedDict

A Top level guardrail trace object.

Attributes:

Name	Type	Description
guardrail	GuardrailTrace	Trace information from guardrail processing.

Source code in strands/types/guardrails.py

class Trace(TypedDict):
    """A Top level guardrail trace object.

    Attributes:
        guardrail: Trace information from guardrail processing.
    """

    guardrail: GuardrailTrace

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