strands.models.openai_responses

OpenAI model provider using the Responses API.

The Responses API is OpenAI's newer API that differs from the Chat Completions API in several key ways:

1. The Responses API can maintain conversation state server-side through "previous_response_id", while Chat Completions is stateless and requires sending full conversation history each time. Note: This implementation currently only implements the stateless approach.

2. Responses API uses "input" (list of items) instead of "messages", and system prompts are passed as "instructions" rather than a system role message.

3. Responses API supports built-in tools (web search, code interpreter, file search) Note: These are not yet implemented in this provider.

4. Docs: https://platform.openai.com/docs/api-reference/responses

Messages = list[Message] module-attribute

A list of messages representing a conversation.

T = TypeVar('T', bound=BaseModel) module-attribute

ToolChoice = ToolChoiceAutoDict | ToolChoiceAnyDict | ToolChoiceToolDict module-attribute

Configuration for how the model should choose tools.

• "auto": The model decides whether to use tools based on the context
• "any": The model must use at least one tool (any tool)
• "tool": The model must use the specified tool

_CONTEXT_WINDOW_OVERFLOW_MSG = 'OpenAI Responses API threw context window overflow error' module-attribute

_DEFAULT_MIME_TYPE = 'application/octet-stream' module-attribute

_MAX_MEDIA_SIZE_BYTES = 20 * 1024 * 1024 module-attribute

_MAX_MEDIA_SIZE_LABEL = '20MB' module-attribute

_MIN_OPENAI_VERSION = Version('2.0.0') module-attribute

_RATE_LIMIT_MSG = 'OpenAI Responses API threw rate limit error' module-attribute

_openai_version = Version(get_package_version('openai')) module-attribute

logger = logging.getLogger(__name__) module-attribute

Client

Bases: Protocol

Protocol defining the OpenAI Responses API interface for the underlying provider client.

Source code in strands/models/openai_responses.py

class Client(Protocol):
    """Protocol defining the OpenAI Responses API interface for the underlying provider client."""

    @property
    # pragma: no cover
    def responses(self) -> Any:
        """Responses interface."""
        ...

responses property

Responses interface.

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

ContextWindowOverflowException

Bases: Exception

Exception raised when the context window is exceeded.

This exception is raised when the input to a model exceeds the maximum context window size that the model can handle. This typically occurs when the combined length of the conversation history, system prompt, and current message is too large for the model to process.

Source code in strands/types/exceptions.py

class ContextWindowOverflowException(Exception):
    """Exception raised when the context window is exceeded.

    This exception is raised when the input to a model exceeds the maximum context window size that the model can
    handle. This typically occurs when the combined length of the conversation history, system prompt, and current
    message is too large for the model to process.
    """

    pass

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

ModelThrottledException

Bases: Exception

Exception raised when the model is throttled.

This exception is raised when the model is throttled by the service. This typically occurs when the service is throttling the requests from the client.

Source code in strands/types/exceptions.py

class ModelThrottledException(Exception):
    """Exception raised when the model is throttled.

    This exception is raised when the model is throttled by the service. This typically occurs when the service is
    throttling the requests from the client.
    """

    def __init__(self, message: str) -> None:
        """Initialize exception.

        Args:
            message: The message from the service that describes the throttling.
        """
        self.message = message
        super().__init__(message)

    pass

__init__(message)

Initialize exception.

Parameters:

Name	Type	Description	Default
message	str	The message from the service that describes the throttling.	required

Source code in strands/types/exceptions.py

def __init__(self, message: str) -> None:
    """Initialize exception.

    Args:
        message: The message from the service that describes the throttling.
    """
    self.message = message
    super().__init__(message)

OpenAIResponsesModel

Bases: Model

OpenAI Responses API model provider implementation.

Note

This implementation currently only supports function tools (custom tools defined via tool_specs). OpenAI's built-in system tools are not yet supported.

Source code in strands/models/openai_responses.py

class OpenAIResponsesModel(Model):
    """OpenAI Responses API model provider implementation.

    Note:
        This implementation currently only supports function tools (custom tools defined via tool_specs).
        OpenAI's built-in system tools are not yet supported.
    """

    client: Client
    client_args: dict[str, Any]

    class OpenAIResponsesConfig(TypedDict, total=False):
        """Configuration options for OpenAI Responses API models.

        Attributes:
            model_id: Model ID (e.g., "gpt-4o").
                For a complete list of supported models, see https://platform.openai.com/docs/models.
            params: Model parameters (e.g., max_output_tokens, temperature, etc.).
                For a complete list of supported parameters, see
                https://platform.openai.com/docs/api-reference/responses/create.
        """

        model_id: str
        params: dict[str, Any] | None

    def __init__(
        self, client_args: dict[str, Any] | None = None, **model_config: Unpack[OpenAIResponsesConfig]
    ) -> None:
        """Initialize provider instance.

        Args:
            client_args: Arguments for the OpenAI client.
                For a complete list of supported arguments, see https://pypi.org/project/openai/.
            **model_config: Configuration options for the OpenAI Responses API model.
        """
        validate_config_keys(model_config, self.OpenAIResponsesConfig)
        self.config = dict(model_config)
        self.client_args = client_args or {}

        logger.debug("config=<%s> | initializing", self.config)

    @override
    def update_config(self, **model_config: Unpack[OpenAIResponsesConfig]) -> None:  # type: ignore[override]
        """Update the OpenAI Responses API model configuration with the provided arguments.

        Args:
            **model_config: Configuration overrides.
        """
        validate_config_keys(model_config, self.OpenAIResponsesConfig)
        self.config.update(model_config)

    @override
    def get_config(self) -> OpenAIResponsesConfig:
        """Get the OpenAI Responses API model configuration.

        Returns:
            The OpenAI Responses API model configuration.
        """
        return cast(OpenAIResponsesModel.OpenAIResponsesConfig, self.config)

    @override
    async def stream(
        self,
        messages: Messages,
        tool_specs: list[ToolSpec] | None = None,
        system_prompt: str | None = None,
        *,
        tool_choice: ToolChoice | None = None,
        **kwargs: Any,
    ) -> AsyncGenerator[StreamEvent, None]:
        """Stream conversation with the OpenAI Responses API model.

        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.
            **kwargs: Additional keyword arguments for future extensibility.

        Yields:
            Formatted message chunks from the model.

        Raises:
            ContextWindowOverflowException: If the input exceeds the model's context window.
            ModelThrottledException: If the request is throttled by OpenAI (rate limits).
        """
        logger.debug("formatting request for OpenAI Responses API")
        request = self._format_request(messages, tool_specs, system_prompt, tool_choice)
        logger.debug("formatted request=<%s>", request)

        logger.debug("invoking OpenAI Responses API model")

        async with openai.AsyncOpenAI(**self.client_args) as client:
            try:
                response = await client.responses.create(**request)

                logger.debug("streaming response from OpenAI Responses API model")

                yield self._format_chunk({"chunk_type": "message_start"})

                tool_calls: dict[str, _ToolCallInfo] = {}
                final_usage = None
                data_type: str | None = None
                stop_reason: str | None = None

                async for event in response:
                    if hasattr(event, "type"):
                        if event.type == "response.reasoning_text.delta":
                            # Reasoning content streaming (for o1/o3 reasoning models)
                            chunks, data_type = self._stream_switch_content("reasoning_content", data_type)
                            for chunk in chunks:
                                yield chunk
                            if hasattr(event, "delta") and isinstance(event.delta, str):
                                yield self._format_chunk(
                                    {
                                        "chunk_type": "content_delta",
                                        "data_type": "reasoning_content",
                                        "data": event.delta,
                                    }
                                )

                        elif event.type == "response.output_text.delta":
                            # Text content streaming
                            chunks, data_type = self._stream_switch_content("text", data_type)
                            for chunk in chunks:
                                yield chunk
                            if hasattr(event, "delta") and isinstance(event.delta, str):
                                yield self._format_chunk(
                                    {"chunk_type": "content_delta", "data_type": "text", "data": event.delta}
                                )

                        elif event.type == "response.output_item.added":
                            # Tool call started
                            if (
                                hasattr(event, "item")
                                and hasattr(event.item, "type")
                                and event.item.type == "function_call"
                            ):
                                call_id = getattr(event.item, "call_id", "unknown")
                                tool_calls[call_id] = {
                                    "name": getattr(event.item, "name", ""),
                                    "arguments": "",
                                    "call_id": call_id,
                                    "item_id": getattr(event.item, "id", ""),
                                }

                        elif event.type == "response.function_call_arguments.delta":
                            # Tool arguments streaming - accumulate deltas by item_id
                            if hasattr(event, "delta") and hasattr(event, "item_id"):
                                for _call_id, call_info in tool_calls.items():
                                    if call_info["item_id"] == event.item_id:
                                        call_info["arguments"] += event.delta
                                        break

                        elif event.type == "response.function_call_arguments.done":
                            # Tool arguments complete - use final arguments as source of truth
                            if hasattr(event, "arguments") and hasattr(event, "item_id"):
                                for _call_id, call_info in tool_calls.items():
                                    if call_info["item_id"] == event.item_id:
                                        call_info["arguments"] = event.arguments
                                        break

                        elif event.type == "response.incomplete":
                            # Response stopped early (e.g., max tokens reached)
                            if hasattr(event, "response"):
                                if hasattr(event.response, "usage"):
                                    final_usage = event.response.usage
                                # Check if stopped due to max_output_tokens
                                if (
                                    hasattr(event.response, "incomplete_details")
                                    and event.response.incomplete_details
                                    and getattr(event.response.incomplete_details, "reason", None)
                                    == "max_output_tokens"
                                ):
                                    stop_reason = "length"
                            break

                        elif event.type == "response.completed":
                            # Response complete
                            if hasattr(event, "response") and hasattr(event.response, "usage"):
                                final_usage = event.response.usage
                            break
            except openai.BadRequestError as e:
                if hasattr(e, "code") and e.code == "context_length_exceeded":
                    logger.warning(_CONTEXT_WINDOW_OVERFLOW_MSG)
                    raise ContextWindowOverflowException(str(e)) from e
                raise
            except openai.RateLimitError as e:
                logger.warning(_RATE_LIMIT_MSG)
                raise ModelThrottledException(str(e)) from e

            # Close current content block if we had any
            if data_type:
                yield self._format_chunk({"chunk_type": "content_stop", "data_type": data_type})

            # Emit tool calls with complete arguments.
            # We emit a single delta per tool containing the full arguments rather than streaming
            # incremental argument deltas. The Responses API streams argument chunks via separate
            # events (response.function_call_arguments.delta) which we accumulate above, then use
            # the final arguments from response.function_call_arguments.done. This approach ensures
            # we emit valid, complete JSON arguments rather than partial fragments.
            for call_info in tool_calls.values():
                tool_call = SimpleNamespace(
                    function=SimpleNamespace(name=call_info["name"], arguments=call_info["arguments"]),
                    id=call_info["call_id"],
                )

                yield self._format_chunk({"chunk_type": "content_start", "data_type": "tool", "data": tool_call})
                yield self._format_chunk({"chunk_type": "content_delta", "data_type": "tool", "data": tool_call})
                yield self._format_chunk({"chunk_type": "content_stop", "data_type": "tool"})

            # Determine finish reason: tool_calls > max_tokens (length) > normal stop
            if tool_calls:
                finish_reason = "tool_calls"
            elif stop_reason == "length":
                finish_reason = "length"
            else:
                finish_reason = "stop"
            yield self._format_chunk({"chunk_type": "message_stop", "data": finish_reason})

            if final_usage:
                yield self._format_chunk({"chunk_type": "metadata", "data": final_usage})

        logger.debug("finished streaming response from OpenAI Responses API model")

    @override
    async 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 OpenAI Responses API 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:
            ContextWindowOverflowException: If the input exceeds the model's context window.
            ModelThrottledException: If the request is throttled by OpenAI (rate limits).
        """
        async with openai.AsyncOpenAI(**self.client_args) as client:
            try:
                response = await client.responses.parse(
                    model=self.get_config()["model_id"],
                    input=self._format_request(prompt, system_prompt=system_prompt)["input"],
                    text_format=output_model,
                )
            except openai.BadRequestError as e:
                if hasattr(e, "code") and e.code == "context_length_exceeded":
                    logger.warning(_CONTEXT_WINDOW_OVERFLOW_MSG)
                    raise ContextWindowOverflowException(str(e)) from e
                raise
            except openai.RateLimitError as e:
                logger.warning(_RATE_LIMIT_MSG)
                raise ModelThrottledException(str(e)) from e

        if response.output_parsed:
            yield {"output": response.output_parsed}
        else:
            raise ValueError("No valid parsed output found in the OpenAI Responses API response.")

    def _format_request(
        self,
        messages: Messages,
        tool_specs: list[ToolSpec] | None = None,
        system_prompt: str | None = None,
        tool_choice: ToolChoice | None = None,
    ) -> dict[str, Any]:
        """Format an OpenAI Responses API compatible response streaming request.

        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.

        Returns:
            An OpenAI Responses API compatible response streaming request.

        Raises:
            TypeError: If a message contains a content block type that cannot be converted to an OpenAI-compatible
                format.
        """
        input_items = self._format_request_messages(messages)
        request = {
            "model": self.config["model_id"],
            "input": input_items,
            "stream": True,
            **cast(dict[str, Any], self.config.get("params", {})),
        }

        if system_prompt:
            request["instructions"] = system_prompt

        # Add tools if provided
        if tool_specs:
            request["tools"] = [
                {
                    "type": "function",
                    "name": tool_spec["name"],
                    "description": tool_spec.get("description", ""),
                    "parameters": tool_spec["inputSchema"]["json"],
                }
                for tool_spec in tool_specs
            ]
            # Add tool_choice if provided
            request.update(self._format_request_tool_choice(tool_choice))

        return request

    @classmethod
    def _format_request_tool_choice(cls, tool_choice: ToolChoice | None) -> dict[str, Any]:
        """Format a tool choice for OpenAI Responses API compatibility.

        Args:
            tool_choice: Tool choice configuration.

        Returns:
            OpenAI Responses API compatible tool choice format.
        """
        if not tool_choice:
            return {}

        match tool_choice:
            case {"auto": _}:
                return {"tool_choice": "auto"}
            case {"any": _}:
                return {"tool_choice": "required"}
            case {"tool": {"name": tool_name}}:
                return {"tool_choice": {"type": "function", "name": tool_name}}
            case _:
                # Default to auto for unknown formats
                return {"tool_choice": "auto"}

    @classmethod
    def _format_request_messages(cls, messages: Messages) -> list[dict[str, Any]]:
        """Format an OpenAI compatible messages array.

        Args:
            messages: List of message objects to be processed by the model.

        Returns:
            An OpenAI compatible messages array.
        """
        formatted_messages: list[dict[str, Any]] = []

        for message in messages:
            role = message["role"]
            contents = message["content"]

            formatted_contents = [
                cls._format_request_message_content(content)
                for content in contents
                if not any(block_type in content for block_type in ["toolResult", "toolUse"])
            ]

            formatted_tool_calls = [
                cls._format_request_message_tool_call(content["toolUse"])
                for content in contents
                if "toolUse" in content
            ]

            formatted_tool_messages = [
                cls._format_request_tool_message(content["toolResult"])
                for content in contents
                if "toolResult" in content
            ]

            if formatted_contents:
                formatted_messages.append(
                    {
                        "role": role,  # "user" | "assistant"
                        "content": formatted_contents,
                    }
                )

            formatted_messages.extend(formatted_tool_calls)
            formatted_messages.extend(formatted_tool_messages)

        return [
            message
            for message in formatted_messages
            if message.get("content") or message.get("type") in ["function_call", "function_call_output"]
        ]

    @classmethod
    def _format_request_message_content(cls, content: ContentBlock) -> dict[str, Any]:
        """Format an OpenAI compatible content block.

        Args:
            content: Message content.

        Returns:
            OpenAI compatible content block.

        Raises:
            TypeError: If the content block type cannot be converted to an OpenAI-compatible format.
            ValueError: If the image or document size exceeds the maximum allowed size (20MB).
        """
        if "document" in content:
            doc = content["document"]
            data_url = _encode_media_to_data_url(doc["source"]["bytes"], doc["format"], "document")
            return {"type": "input_file", "file_url": data_url}

        if "image" in content:
            img = content["image"]
            data_url = _encode_media_to_data_url(img["source"]["bytes"], img["format"], "image")
            return {"type": "input_image", "image_url": data_url}

        if "text" in content:
            return {"type": "input_text", "text": content["text"]}

        raise TypeError(f"content_type=<{next(iter(content))}> | unsupported type")

    @classmethod
    def _format_request_message_tool_call(cls, tool_use: ToolUse) -> dict[str, Any]:
        """Format an OpenAI compatible tool call.

        Args:
            tool_use: Tool use requested by the model.

        Returns:
            OpenAI compatible tool call.
        """
        return {
            "type": "function_call",
            "call_id": tool_use["toolUseId"],
            "name": tool_use["name"],
            "arguments": json.dumps(tool_use["input"]),
        }

    @classmethod
    def _format_request_tool_message(cls, tool_result: ToolResult) -> dict[str, Any]:
        """Format an OpenAI compatible tool message.

        Args:
            tool_result: Tool result collected from a tool execution.

        Returns:
            OpenAI compatible tool message.

        Raises:
            ValueError: If the image or document size exceeds the maximum allowed size (20MB).

        Note:
            The Responses API's function_call_output can be either a string (typically JSON encoded)
            or an array of content objects when returning images/files.
            See: https://platform.openai.com/docs/guides/function-calling
        """
        output_parts: list[dict[str, Any]] = []
        has_media = False

        for content in tool_result["content"]:
            if "json" in content:
                output_parts.append({"type": "input_text", "text": json.dumps(content["json"])})
            elif "text" in content:
                output_parts.append({"type": "input_text", "text": content["text"]})
            elif "image" in content:
                has_media = True
                img = content["image"]
                data_url = _encode_media_to_data_url(img["source"]["bytes"], img["format"], "image")
                output_parts.append({"type": "input_image", "image_url": data_url})
            elif "document" in content:
                has_media = True
                doc = content["document"]
                data_url = _encode_media_to_data_url(doc["source"]["bytes"], doc["format"], "document")
                output_parts.append({"type": "input_file", "file_url": data_url})

        # Return array if has media content, otherwise join as string for simpler text-only cases
        output: list[dict[str, Any]] | str
        if has_media:
            output = output_parts
        else:
            output = "\n".join(part.get("text", "") for part in output_parts) if output_parts else ""

        return {
            "type": "function_call_output",
            "call_id": tool_result["toolUseId"],
            "output": output,
        }

    def _stream_switch_content(self, data_type: str, prev_data_type: str | None) -> tuple[list[StreamEvent], str]:
        """Handle switching to a new content stream.

        Args:
            data_type: The next content data type.
            prev_data_type: The previous content data type.

        Returns:
            Tuple containing:
            - Stop block for previous content and the start block for the next content.
            - Next content data type.
        """
        chunks: list[StreamEvent] = []
        if data_type != prev_data_type:
            if prev_data_type is not None:
                chunks.append(self._format_chunk({"chunk_type": "content_stop", "data_type": prev_data_type}))
            chunks.append(self._format_chunk({"chunk_type": "content_start", "data_type": data_type}))

        return chunks, data_type

    def _format_chunk(self, event: dict[str, Any]) -> StreamEvent:
        """Format an OpenAI response event into a standardized message chunk.

        Args:
            event: A response event from the OpenAI compatible model.

        Returns:
            The formatted chunk.

        Raises:
            RuntimeError: If chunk_type is not recognized.
                This error should never be encountered as chunk_type is controlled in the stream method.
        """
        match event["chunk_type"]:
            case "message_start":
                return {"messageStart": {"role": "assistant"}}

            case "content_start":
                if event["data_type"] == "tool":
                    return {
                        "contentBlockStart": {
                            "start": {
                                "toolUse": {
                                    "name": event["data"].function.name,
                                    "toolUseId": event["data"].id,
                                }
                            }
                        }
                    }

                return {"contentBlockStart": {"start": {}}}

            case "content_delta":
                if event["data_type"] == "tool":
                    return {
                        "contentBlockDelta": {"delta": {"toolUse": {"input": event["data"].function.arguments or ""}}}
                    }

                if event["data_type"] == "reasoning_content":
                    return {"contentBlockDelta": {"delta": {"reasoningContent": {"text": event["data"]}}}}

                return {"contentBlockDelta": {"delta": {"text": event["data"]}}}

            case "content_stop":
                return {"contentBlockStop": {}}

            case "message_stop":
                match event["data"]:
                    case "tool_calls":
                        return {"messageStop": {"stopReason": "tool_use"}}
                    case "length":
                        return {"messageStop": {"stopReason": "max_tokens"}}
                    case _:
                        return {"messageStop": {"stopReason": "end_turn"}}

            case "metadata":
                # Responses API uses input_tokens/output_tokens naming convention
                return {
                    "metadata": {
                        "usage": {
                            "inputTokens": getattr(event["data"], "input_tokens", 0),
                            "outputTokens": getattr(event["data"], "output_tokens", 0),
                            "totalTokens": getattr(event["data"], "total_tokens", 0),
                        },
                        "metrics": {
                            "latencyMs": 0,  # TODO
                        },
                    },
                }

            case _:
                raise RuntimeError(f"chunk_type=<{event['chunk_type']}> | unknown type")

OpenAIResponsesConfig

Bases: TypedDict

Configuration options for OpenAI Responses API models.

Attributes:

Name	Type	Description
model_id	str	Model ID (e.g., "gpt-4o"). For a complete list of supported models, see https://platform.openai.com/docs/models.
params	dict[str, Any] | None	Model parameters (e.g., max_output_tokens, temperature, etc.). For a complete list of supported parameters, see https://platform.openai.com/docs/api-reference/responses/create.

Source code in strands/models/openai_responses.py

class OpenAIResponsesConfig(TypedDict, total=False):
    """Configuration options for OpenAI Responses API models.

    Attributes:
        model_id: Model ID (e.g., "gpt-4o").
            For a complete list of supported models, see https://platform.openai.com/docs/models.
        params: Model parameters (e.g., max_output_tokens, temperature, etc.).
            For a complete list of supported parameters, see
            https://platform.openai.com/docs/api-reference/responses/create.
    """

    model_id: str
    params: dict[str, Any] | None

__init__(client_args=None, **model_config)

Initialize provider instance.

Parameters:

Name	Type	Description	Default
client_args	dict[str, Any] | None	Arguments for the OpenAI client. For a complete list of supported arguments, see https://pypi.org/project/openai/.	None
**model_config	Unpack[OpenAIResponsesConfig]	Configuration options for the OpenAI Responses API model.	{}

Source code in strands/models/openai_responses.py

def __init__(
    self, client_args: dict[str, Any] | None = None, **model_config: Unpack[OpenAIResponsesConfig]
) -> None:
    """Initialize provider instance.

    Args:
        client_args: Arguments for the OpenAI client.
            For a complete list of supported arguments, see https://pypi.org/project/openai/.
        **model_config: Configuration options for the OpenAI Responses API model.
    """
    validate_config_keys(model_config, self.OpenAIResponsesConfig)
    self.config = dict(model_config)
    self.client_args = client_args or {}

    logger.debug("config=<%s> | initializing", self.config)

get_config()

Get the OpenAI Responses API model configuration.

Returns:

Type	Description
OpenAIResponsesConfig	The OpenAI Responses API model configuration.

Source code in strands/models/openai_responses.py

@override
def get_config(self) -> OpenAIResponsesConfig:
    """Get the OpenAI Responses API model configuration.

    Returns:
        The OpenAI Responses API model configuration.
    """
    return cast(OpenAIResponsesModel.OpenAIResponsesConfig, self.config)

stream(messages, tool_specs=None, system_prompt=None, *, tool_choice=None, **kwargs) async

Stream conversation with the OpenAI Responses API model.

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
**kwargs	Any	Additional keyword arguments for future extensibility.	{}

Yields:

Type	Description
AsyncGenerator[StreamEvent, None]	Formatted message chunks from the model.

Raises:

Type	Description
ContextWindowOverflowException	If the input exceeds the model's context window.
ModelThrottledException	If the request is throttled by OpenAI (rate limits).

Source code in strands/models/openai_responses.py

@override
async def stream(
    self,
    messages: Messages,
    tool_specs: list[ToolSpec] | None = None,
    system_prompt: str | None = None,
    *,
    tool_choice: ToolChoice | None = None,
    **kwargs: Any,
) -> AsyncGenerator[StreamEvent, None]:
    """Stream conversation with the OpenAI Responses API model.

    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.
        **kwargs: Additional keyword arguments for future extensibility.

    Yields:
        Formatted message chunks from the model.

    Raises:
        ContextWindowOverflowException: If the input exceeds the model's context window.
        ModelThrottledException: If the request is throttled by OpenAI (rate limits).
    """
    logger.debug("formatting request for OpenAI Responses API")
    request = self._format_request(messages, tool_specs, system_prompt, tool_choice)
    logger.debug("formatted request=<%s>", request)

    logger.debug("invoking OpenAI Responses API model")

    async with openai.AsyncOpenAI(**self.client_args) as client:
        try:
            response = await client.responses.create(**request)

            logger.debug("streaming response from OpenAI Responses API model")

            yield self._format_chunk({"chunk_type": "message_start"})

            tool_calls: dict[str, _ToolCallInfo] = {}
            final_usage = None
            data_type: str | None = None
            stop_reason: str | None = None

            async for event in response:
                if hasattr(event, "type"):
                    if event.type == "response.reasoning_text.delta":
                        # Reasoning content streaming (for o1/o3 reasoning models)
                        chunks, data_type = self._stream_switch_content("reasoning_content", data_type)
                        for chunk in chunks:
                            yield chunk
                        if hasattr(event, "delta") and isinstance(event.delta, str):
                            yield self._format_chunk(
                                {
                                    "chunk_type": "content_delta",
                                    "data_type": "reasoning_content",
                                    "data": event.delta,
                                }
                            )

                    elif event.type == "response.output_text.delta":
                        # Text content streaming
                        chunks, data_type = self._stream_switch_content("text", data_type)
                        for chunk in chunks:
                            yield chunk
                        if hasattr(event, "delta") and isinstance(event.delta, str):
                            yield self._format_chunk(
                                {"chunk_type": "content_delta", "data_type": "text", "data": event.delta}
                            )

                    elif event.type == "response.output_item.added":
                        # Tool call started
                        if (
                            hasattr(event, "item")
                            and hasattr(event.item, "type")
                            and event.item.type == "function_call"
                        ):
                            call_id = getattr(event.item, "call_id", "unknown")
                            tool_calls[call_id] = {
                                "name": getattr(event.item, "name", ""),
                                "arguments": "",
                                "call_id": call_id,
                                "item_id": getattr(event.item, "id", ""),
                            }

                    elif event.type == "response.function_call_arguments.delta":
                        # Tool arguments streaming - accumulate deltas by item_id
                        if hasattr(event, "delta") and hasattr(event, "item_id"):
                            for _call_id, call_info in tool_calls.items():
                                if call_info["item_id"] == event.item_id:
                                    call_info["arguments"] += event.delta
                                    break

                    elif event.type == "response.function_call_arguments.done":
                        # Tool arguments complete - use final arguments as source of truth
                        if hasattr(event, "arguments") and hasattr(event, "item_id"):
                            for _call_id, call_info in tool_calls.items():
                                if call_info["item_id"] == event.item_id:
                                    call_info["arguments"] = event.arguments
                                    break

                    elif event.type == "response.incomplete":
                        # Response stopped early (e.g., max tokens reached)
                        if hasattr(event, "response"):
                            if hasattr(event.response, "usage"):
                                final_usage = event.response.usage
                            # Check if stopped due to max_output_tokens
                            if (
                                hasattr(event.response, "incomplete_details")
                                and event.response.incomplete_details
                                and getattr(event.response.incomplete_details, "reason", None)
                                == "max_output_tokens"
                            ):
                                stop_reason = "length"
                        break

                    elif event.type == "response.completed":
                        # Response complete
                        if hasattr(event, "response") and hasattr(event.response, "usage"):
                            final_usage = event.response.usage
                        break
        except openai.BadRequestError as e:
            if hasattr(e, "code") and e.code == "context_length_exceeded":
                logger.warning(_CONTEXT_WINDOW_OVERFLOW_MSG)
                raise ContextWindowOverflowException(str(e)) from e
            raise
        except openai.RateLimitError as e:
            logger.warning(_RATE_LIMIT_MSG)
            raise ModelThrottledException(str(e)) from e

        # Close current content block if we had any
        if data_type:
            yield self._format_chunk({"chunk_type": "content_stop", "data_type": data_type})

        # Emit tool calls with complete arguments.
        # We emit a single delta per tool containing the full arguments rather than streaming
        # incremental argument deltas. The Responses API streams argument chunks via separate
        # events (response.function_call_arguments.delta) which we accumulate above, then use
        # the final arguments from response.function_call_arguments.done. This approach ensures
        # we emit valid, complete JSON arguments rather than partial fragments.
        for call_info in tool_calls.values():
            tool_call = SimpleNamespace(
                function=SimpleNamespace(name=call_info["name"], arguments=call_info["arguments"]),
                id=call_info["call_id"],
            )

            yield self._format_chunk({"chunk_type": "content_start", "data_type": "tool", "data": tool_call})
            yield self._format_chunk({"chunk_type": "content_delta", "data_type": "tool", "data": tool_call})
            yield self._format_chunk({"chunk_type": "content_stop", "data_type": "tool"})

        # Determine finish reason: tool_calls > max_tokens (length) > normal stop
        if tool_calls:
            finish_reason = "tool_calls"
        elif stop_reason == "length":
            finish_reason = "length"
        else:
            finish_reason = "stop"
        yield self._format_chunk({"chunk_type": "message_stop", "data": finish_reason})

        if final_usage:
            yield self._format_chunk({"chunk_type": "metadata", "data": final_usage})

    logger.debug("finished streaming response from OpenAI Responses API model")

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

Get structured output from the OpenAI Responses API 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
ContextWindowOverflowException	If the input exceeds the model's context window.
ModelThrottledException	If the request is throttled by OpenAI (rate limits).

Source code in strands/models/openai_responses.py

@override
async 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 OpenAI Responses API 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:
        ContextWindowOverflowException: If the input exceeds the model's context window.
        ModelThrottledException: If the request is throttled by OpenAI (rate limits).
    """
    async with openai.AsyncOpenAI(**self.client_args) as client:
        try:
            response = await client.responses.parse(
                model=self.get_config()["model_id"],
                input=self._format_request(prompt, system_prompt=system_prompt)["input"],
                text_format=output_model,
            )
        except openai.BadRequestError as e:
            if hasattr(e, "code") and e.code == "context_length_exceeded":
                logger.warning(_CONTEXT_WINDOW_OVERFLOW_MSG)
                raise ContextWindowOverflowException(str(e)) from e
            raise
        except openai.RateLimitError as e:
            logger.warning(_RATE_LIMIT_MSG)
            raise ModelThrottledException(str(e)) from e

    if response.output_parsed:
        yield {"output": response.output_parsed}
    else:
        raise ValueError("No valid parsed output found in the OpenAI Responses API response.")

update_config(**model_config)

Update the OpenAI Responses API model configuration with the provided arguments.

Parameters:

Name	Type	Description	Default
**model_config	Unpack[OpenAIResponsesConfig]	Configuration overrides.	{}

Source code in strands/models/openai_responses.py

@override
def update_config(self, **model_config: Unpack[OpenAIResponsesConfig]) -> None:  # type: ignore[override]
    """Update the OpenAI Responses API model configuration with the provided arguments.

    Args:
        **model_config: Configuration overrides.
    """
    validate_config_keys(model_config, self.OpenAIResponsesConfig)
    self.config.update(model_config)

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

ToolResult

Bases: TypedDict

Result of a tool execution.

Attributes:

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

Source code in strands/types/tools.py

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

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

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

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.
reasoningSignature	NotRequired[str]	Token that ties the model's reasoning to this tool call.

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.
        reasoningSignature: Token that ties the model's reasoning to this tool call.
    """

    input: Any
    name: str
    toolUseId: str
    reasoningSignature: NotRequired[str]

_ToolCallInfo

Bases: TypedDict

Internal type for tracking tool call information during streaming.

Source code in strands/models/openai_responses.py

class _ToolCallInfo(TypedDict):
    """Internal type for tracking tool call information during streaming."""

    name: str
    arguments: str
    call_id: str
    item_id: str

_encode_media_to_data_url(data, format_ext, media_type='image')

Encode media bytes to a base64 data URL with size validation.

Parameters:

Name	Type	Description	Default
data	bytes	Raw bytes of the media content.	required
format_ext	str	File format extension (e.g., "png", "pdf").	required
media_type	str	Type of media for error messages ("image" or "document").	'image'

Returns:

Type	Description
str	Base64-encoded data URL string.

Raises:

Type	Description
ValueError	If the media size exceeds the maximum allowed size.

Source code in strands/models/openai_responses.py

def _encode_media_to_data_url(data: bytes, format_ext: str, media_type: str = "image") -> str:
    """Encode media bytes to a base64 data URL with size validation.

    Args:
        data: Raw bytes of the media content.
        format_ext: File format extension (e.g., "png", "pdf").
        media_type: Type of media for error messages ("image" or "document").

    Returns:
        Base64-encoded data URL string.

    Raises:
        ValueError: If the media size exceeds the maximum allowed size.
    """
    if len(data) > _MAX_MEDIA_SIZE_BYTES:
        raise ValueError(
            f"{media_type.capitalize()} size {len(data)} bytes exceeds maximum of"
            f" {_MAX_MEDIA_SIZE_BYTES} bytes ({_MAX_MEDIA_SIZE_LABEL})"
        )
    mime_type = mimetypes.types_map.get(f".{format_ext}", _DEFAULT_MIME_TYPE)
    encoded_data = base64.b64encode(data).decode("utf-8")
    return f"data:{mime_type};base64,{encoded_data}"

validate_config_keys(config_dict, config_class)

Validate that config keys match the TypedDict fields.

Parameters:

Name	Type	Description	Default
config_dict	Mapping[str, Any]	Dictionary of configuration parameters	required
config_class	type	TypedDict class to validate against	required

Source code in strands/models/_validation.py

def validate_config_keys(config_dict: Mapping[str, Any], config_class: type) -> None:
    """Validate that config keys match the TypedDict fields.

    Args:
        config_dict: Dictionary of configuration parameters
        config_class: TypedDict class to validate against
    """
    valid_keys = set(get_type_hints(config_class).keys())
    provided_keys = set(config_dict.keys())
    invalid_keys = provided_keys - valid_keys

    if invalid_keys:
        warnings.warn(
            f"Invalid configuration parameters: {sorted(invalid_keys)}."
            f"\nValid parameters are: {sorted(valid_keys)}."
            f"\n"
            f"\nSee https://github.com/strands-agents/sdk-python/issues/815",
            stacklevel=4,
        )