strands.models.openai

OpenAI model provider.

• Docs: https://platform.openai.com/docs/overview

Messages = List[Message] module-attribute

A list of messages representing a conversation.

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

ToolChoice = Union[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

logger = logging.getLogger(__name__) module-attribute

Client

Bases: Protocol

Protocol defining the OpenAI-compatible interface for the underlying provider client.

Source code in strands/models/openai.py

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

    @property
    # pragma: no cover
    def chat(self) -> Any:
        """Chat completions interface."""
        ...

chat property

Chat completions 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: 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

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)

OpenAIModel

Bases: Model

OpenAI model provider implementation.

Source code in strands/models/openai.py

class OpenAIModel(Model):
    """OpenAI model provider implementation."""

    client: Client

    class OpenAIConfig(TypedDict, total=False):
        """Configuration options for OpenAI 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_tokens).
                For a complete list of supported parameters, see
                https://platform.openai.com/docs/api-reference/chat/create.
        """

        model_id: str
        params: Optional[dict[str, Any]]

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

        Args:
            client: Pre-configured OpenAI-compatible client to reuse across requests.
                When provided, this client will be reused for all requests and will NOT be closed
                by the model. The caller is responsible for managing the client lifecycle.
                This is useful for:
                - Injecting custom client wrappers (e.g., GuardrailsAsyncOpenAI)
                - Reusing connection pools within a single event loop/worker
                - Centralizing observability, retries, and networking policy
                - Pointing to custom model gateways
                Note: The client should not be shared across different asyncio event loops.
            client_args: Arguments for the OpenAI client (legacy approach).
                For a complete list of supported arguments, see https://pypi.org/project/openai/.
            **model_config: Configuration options for the OpenAI model.

        Raises:
            ValueError: If both `client` and `client_args` are provided.
        """
        validate_config_keys(model_config, self.OpenAIConfig)
        self.config = dict(model_config)

        # Validate that only one client configuration method is provided
        if client is not None and client_args is not None and len(client_args) > 0:
            raise ValueError("Only one of 'client' or 'client_args' should be provided, not both.")

        self._custom_client = client
        self.client_args = client_args or {}

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

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

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

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

        Returns:
            The OpenAI model configuration.
        """
        return cast(OpenAIModel.OpenAIConfig, self.config)

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

        Args:
            content: Message content.
            **kwargs: Additional keyword arguments for future extensibility.

        Returns:
            OpenAI compatible content block.

        Raises:
            TypeError: If the content block type cannot be converted to an OpenAI-compatible format.
        """
        if "document" in content:
            mime_type = mimetypes.types_map.get(f".{content['document']['format']}", "application/octet-stream")
            file_data = base64.b64encode(content["document"]["source"]["bytes"]).decode("utf-8")
            return {
                "file": {
                    "file_data": f"data:{mime_type};base64,{file_data}",
                    "filename": content["document"]["name"],
                },
                "type": "file",
            }

        if "image" in content:
            mime_type = mimetypes.types_map.get(f".{content['image']['format']}", "application/octet-stream")
            image_data = base64.b64encode(content["image"]["source"]["bytes"]).decode("utf-8")

            return {
                "image_url": {
                    "detail": "auto",
                    "format": mime_type,
                    "url": f"data:{mime_type};base64,{image_data}",
                },
                "type": "image_url",
            }

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

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

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

        Args:
            tool_use: Tool use requested by the model.
            **kwargs: Additional keyword arguments for future extensibility.

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

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

        Args:
            tool_result: Tool result collected from a tool execution.
            **kwargs: Additional keyword arguments for future extensibility.

        Returns:
            OpenAI compatible tool message.
        """
        contents = cast(
            list[ContentBlock],
            [
                {"text": json.dumps(content["json"])} if "json" in content else content
                for content in tool_result["content"]
            ],
        )

        return {
            "role": "tool",
            "tool_call_id": tool_result["toolUseId"],
            "content": [cls.format_request_message_content(content) for content in contents],
        }

    @classmethod
    def _split_tool_message_images(
        cls, tool_message: dict[str, Any]
    ) -> tuple[dict[str, Any], Optional[dict[str, Any]]]:
        """Split a tool message into text-only tool message and optional user message with images.

        OpenAI API restricts images to user role messages only. This method extracts any image
        content from a tool message and returns it separately as a user message.

        Args:
            tool_message: A formatted tool message that may contain images.

        Returns:
            A tuple of (tool_message_without_images, user_message_with_images_or_None).
        """
        if tool_message.get("role") != "tool":
            return tool_message, None

        content = tool_message.get("content", [])
        if not isinstance(content, list):
            return tool_message, None

        # Separate image and non-image content
        text_content = []
        image_content = []

        for item in content:
            if isinstance(item, dict) and item.get("type") == "image_url":
                image_content.append(item)
            else:
                text_content.append(item)

        # If no images found, return original message
        if not image_content:
            return tool_message, None

        # Let the user know that we are modifying the messages for OpenAI compatibility
        logger.warning(
            "tool_call_id=<%s> | Moving image from tool message to a new user message for OpenAI compatibility",
            tool_message["tool_call_id"],
        )

        # Append a message to the text content to inform the model about the upcoming image
        text_content.append(
            {
                "type": "text",
                "text": (
                    "Tool successfully returned an image. The image is being provided in the following user message."
                ),
            }
        )

        # Create the clean tool message with the updated text content
        tool_message_clean = {
            "role": "tool",
            "tool_call_id": tool_message["tool_call_id"],
            "content": text_content,
        }

        # Create user message with only images
        user_message_with_images = {"role": "user", "content": image_content}

        return tool_message_clean, user_message_with_images

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

        Args:
            tool_choice: Tool choice configuration in Bedrock format.

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

        match tool_choice:
            case {"auto": _}:
                return {"tool_choice": "auto"}  # OpenAI SDK doesn't define constants for these values
            case {"any": _}:
                return {"tool_choice": "required"}
            case {"tool": {"name": tool_name}}:
                return {"tool_choice": {"type": "function", "function": {"name": tool_name}}}
            case _:
                # This should not happen with proper typing, but handle gracefully
                return {"tool_choice": "auto"}

    @classmethod
    def _format_system_messages(
        cls,
        system_prompt: Optional[str] = None,
        *,
        system_prompt_content: Optional[list[SystemContentBlock]] = None,
        **kwargs: Any,
    ) -> list[dict[str, Any]]:
        """Format system messages for OpenAI-compatible providers.

        Args:
            system_prompt: System prompt to provide context to the model.
            system_prompt_content: System prompt content blocks to provide context to the model.
            **kwargs: Additional keyword arguments for future extensibility.

        Returns:
            List of formatted system messages.
        """
        # Handle backward compatibility: if system_prompt is provided but system_prompt_content is None
        if system_prompt and system_prompt_content is None:
            system_prompt_content = [{"text": system_prompt}]

        # TODO: Handle caching blocks https://github.com/strands-agents/sdk-python/issues/1140
        return [
            {"role": "system", "content": content["text"]}
            for content in system_prompt_content or []
            if "text" in content
        ]

    @classmethod
    def _format_regular_messages(cls, messages: Messages, **kwargs: Any) -> list[dict[str, Any]]:
        """Format regular messages for OpenAI-compatible providers.

        Args:
            messages: List of message objects to be processed by the model.
            **kwargs: Additional keyword arguments for future extensibility.

        Returns:
            List of formatted messages.
        """
        formatted_messages = []

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

            # Check for reasoningContent and warn user
            if any("reasoningContent" in content for content in contents):
                logger.warning(
                    "reasoningContent is not supported in multi-turn conversations with the Chat Completions API."
                )

            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", "reasoningContent"])
            ]
            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
            ]

            formatted_message = {
                "role": message["role"],
                "content": formatted_contents,
                **({"tool_calls": formatted_tool_calls} if formatted_tool_calls else {}),
            }
            formatted_messages.append(formatted_message)

            # Process tool messages to extract images into separate user messages
            # OpenAI API requires images to be in user role messages only
            for tool_msg in formatted_tool_messages:
                tool_msg_clean, user_msg_with_images = cls._split_tool_message_images(tool_msg)
                formatted_messages.append(tool_msg_clean)
                if user_msg_with_images:
                    formatted_messages.append(user_msg_with_images)

        return formatted_messages

    @classmethod
    def format_request_messages(
        cls,
        messages: Messages,
        system_prompt: Optional[str] = None,
        *,
        system_prompt_content: Optional[list[SystemContentBlock]] = None,
        **kwargs: Any,
    ) -> list[dict[str, Any]]:
        """Format an OpenAI compatible messages array.

        Args:
            messages: List of message objects to be processed by the model.
            system_prompt: System prompt to provide context to the model.
            system_prompt_content: System prompt content blocks to provide context to the model.
            **kwargs: Additional keyword arguments for future extensibility.

        Returns:
            An OpenAI compatible messages array.
        """
        formatted_messages = cls._format_system_messages(system_prompt, system_prompt_content=system_prompt_content)
        formatted_messages.extend(cls._format_regular_messages(messages))

        return [message for message in formatted_messages if message["content"] or "tool_calls" in message]

    def format_request(
        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,
        **kwargs: Any,
    ) -> dict[str, Any]:
        """Format an OpenAI compatible chat 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.
            system_prompt_content: System prompt content blocks to provide context to the model.
            **kwargs: Additional keyword arguments for future extensibility.

        Returns:
            An OpenAI compatible chat streaming request.

        Raises:
            TypeError: If a message contains a content block type that cannot be converted to an OpenAI-compatible
                format.
        """
        return {
            "messages": self.format_request_messages(
                messages, system_prompt, system_prompt_content=system_prompt_content
            ),
            "model": self.config["model_id"],
            "stream": True,
            "stream_options": {"include_usage": True},
            "tools": [
                {
                    "type": "function",
                    "function": {
                        "name": tool_spec["name"],
                        "description": tool_spec["description"],
                        "parameters": tool_spec["inputSchema"]["json"],
                    },
                }
                for tool_spec in tool_specs or []
            ],
            **(self._format_request_tool_choice(tool_choice)),
            **cast(dict[str, Any], self.config.get("params", {})),
        }

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

        Args:
            event: A response event from the OpenAI compatible model.
            **kwargs: Additional keyword arguments for future extensibility.

        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":
                return {
                    "metadata": {
                        "usage": {
                            "inputTokens": event["data"].prompt_tokens,
                            "outputTokens": event["data"].completion_tokens,
                            "totalTokens": event["data"].total_tokens,
                        },
                        "metrics": {
                            "latencyMs": 0,  # TODO
                        },
                    },
                }

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

    @asynccontextmanager
    async def _get_client(self) -> AsyncIterator[Any]:
        """Get an OpenAI client for making requests.

        This context manager handles client lifecycle management:
        - If an injected client was provided during initialization, it yields that client
          without closing it (caller manages lifecycle).
        - Otherwise, creates a new AsyncOpenAI client from client_args and automatically
          closes it when the context exits.

        Note: We create a new client per request to avoid connection sharing in the underlying
        httpx client, as the asyncio event loop does not allow connections to be shared.
        For more details, see https://github.com/encode/httpx/discussions/2959.

        Yields:
            Client: An OpenAI-compatible client instance.
        """
        if self._custom_client is not None:
            # Use the injected client (caller manages lifecycle)
            yield self._custom_client
        else:
            # Create a new client from client_args
            # We initialize an OpenAI context on every request so as to avoid connection sharing in the underlying
            # httpx client. The asyncio event loop does not allow connections to be shared. For more details, please
            # refer to https://github.com/encode/httpx/discussions/2959.
            async with openai.AsyncOpenAI(**self.client_args) as client:
                yield client

    @override
    async def stream(
        self,
        messages: Messages,
        tool_specs: Optional[list[ToolSpec]] = None,
        system_prompt: Optional[str] = None,
        *,
        tool_choice: ToolChoice | None = None,
        **kwargs: Any,
    ) -> AsyncGenerator[StreamEvent, None]:
        """Stream conversation with the OpenAI 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")
        request = self.format_request(messages, tool_specs, system_prompt, tool_choice)
        logger.debug("formatted request=<%s>", request)

        logger.debug("invoking model")

        # We initialize an OpenAI context on every request so as to avoid connection sharing in the underlying httpx
        # client. The asyncio event loop does not allow connections to be shared. For more details, please refer to
        # https://github.com/encode/httpx/discussions/2959.
        async with self._get_client() as client:
            try:
                response = await client.chat.completions.create(**request)
            except openai.BadRequestError as e:
                # Check if this is a context length exceeded error
                if hasattr(e, "code") and e.code == "context_length_exceeded":
                    logger.warning("OpenAI threw context window overflow error")
                    raise ContextWindowOverflowException(str(e)) from e
                # Re-raise other BadRequestError exceptions
                raise
            except openai.RateLimitError as e:
                # All rate limit errors should be treated as throttling, not context overflow
                # Rate limits (including TPM) require waiting/retrying, not context reduction
                logger.warning("OpenAI threw rate limit error")
                raise ModelThrottledException(str(e)) from e

            logger.debug("got response from model")
            yield self.format_chunk({"chunk_type": "message_start"})
            tool_calls: dict[int, list[Any]] = {}
            data_type = None
            finish_reason = None  # Store finish_reason for later use
            event = None  # Initialize for scope safety

            async for event in response:
                # Defensive: skip events with empty or missing choices
                if not getattr(event, "choices", None):
                    continue
                choice = event.choices[0]

                if hasattr(choice.delta, "reasoning_content") and choice.delta.reasoning_content:
                    chunks, data_type = self._stream_switch_content("reasoning_content", data_type)
                    for chunk in chunks:
                        yield chunk
                    yield self.format_chunk(
                        {
                            "chunk_type": "content_delta",
                            "data_type": data_type,
                            "data": choice.delta.reasoning_content,
                        }
                    )

                if choice.delta.content:
                    chunks, data_type = self._stream_switch_content("text", data_type)
                    for chunk in chunks:
                        yield chunk
                    yield self.format_chunk(
                        {"chunk_type": "content_delta", "data_type": data_type, "data": choice.delta.content}
                    )

                for tool_call in choice.delta.tool_calls or []:
                    tool_calls.setdefault(tool_call.index, []).append(tool_call)

                if choice.finish_reason:
                    finish_reason = choice.finish_reason  # Store for use outside loop
                    if data_type:
                        yield self.format_chunk({"chunk_type": "content_stop", "data_type": data_type})
                    break

            for tool_deltas in tool_calls.values():
                yield self.format_chunk({"chunk_type": "content_start", "data_type": "tool", "data": tool_deltas[0]})

                for tool_delta in tool_deltas:
                    yield self.format_chunk({"chunk_type": "content_delta", "data_type": "tool", "data": tool_delta})

                yield self.format_chunk({"chunk_type": "content_stop", "data_type": "tool"})

            yield self.format_chunk({"chunk_type": "message_stop", "data": finish_reason or "end_turn"})

            # Skip remaining events as we don't have use for anything except the final usage payload
            async for event in response:
                _ = event

            if event and hasattr(event, "usage") and event.usage:
                yield self.format_chunk({"chunk_type": "metadata", "data": event.usage})

        logger.debug("finished streaming response from model")

    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 = []
        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

    @override
    async 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:
            ContextWindowOverflowException: If the input exceeds the model's context window.
            ModelThrottledException: If the request is throttled by OpenAI (rate limits).
        """
        # We initialize an OpenAI context on every request so as to avoid connection sharing in the underlying httpx
        # client. The asyncio event loop does not allow connections to be shared. For more details, please refer to
        # https://github.com/encode/httpx/discussions/2959.
        async with self._get_client() as client:
            try:
                response: ParsedChatCompletion = await client.beta.chat.completions.parse(
                    model=self.get_config()["model_id"],
                    messages=self.format_request(prompt, system_prompt=system_prompt)["messages"],
                    response_format=output_model,
                )
            except openai.BadRequestError as e:
                # Check if this is a context length exceeded error
                if hasattr(e, "code") and e.code == "context_length_exceeded":
                    logger.warning("OpenAI threw context window overflow error")
                    raise ContextWindowOverflowException(str(e)) from e
                # Re-raise other BadRequestError exceptions
                raise
            except openai.RateLimitError as e:
                # All rate limit errors should be treated as throttling, not context overflow
                # Rate limits (including TPM) require waiting/retrying, not context reduction
                logger.warning("OpenAI threw rate limit error")
                raise ModelThrottledException(str(e)) from e

        parsed: T | None = None
        # Find the first choice with tool_calls
        if len(response.choices) > 1:
            raise ValueError("Multiple choices found in the OpenAI response.")

        for choice in response.choices:
            if isinstance(choice.message.parsed, output_model):
                parsed = choice.message.parsed
                break

        if parsed:
            yield {"output": parsed}
        else:
            raise ValueError("No valid tool use or tool use input was found in the OpenAI response.")

OpenAIConfig

Bases: TypedDict

Configuration options for OpenAI 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	Optional[dict[str, Any]]	Model parameters (e.g., max_tokens). For a complete list of supported parameters, see https://platform.openai.com/docs/api-reference/chat/create.

Source code in strands/models/openai.py

class OpenAIConfig(TypedDict, total=False):
    """Configuration options for OpenAI 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_tokens).
            For a complete list of supported parameters, see
            https://platform.openai.com/docs/api-reference/chat/create.
    """

    model_id: str
    params: Optional[dict[str, Any]]

__init__(client=None, client_args=None, **model_config)

Initialize provider instance.

Parameters:

Name	Type	Description	Default
client	Optional[Client]	Pre-configured OpenAI-compatible client to reuse across requests. When provided, this client will be reused for all requests and will NOT be closed by the model. The caller is responsible for managing the client lifecycle. This is useful for: - Injecting custom client wrappers (e.g., GuardrailsAsyncOpenAI) - Reusing connection pools within a single event loop/worker - Centralizing observability, retries, and networking policy - Pointing to custom model gateways Note: The client should not be shared across different asyncio event loops.	None
client_args	Optional[dict[str, Any]]	Arguments for the OpenAI client (legacy approach). For a complete list of supported arguments, see https://pypi.org/project/openai/.	None
**model_config	Unpack[OpenAIConfig]	Configuration options for the OpenAI model.	{}

Raises:

Type	Description
ValueError	If both client and client_args are provided.

Source code in strands/models/openai.py

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

    Args:
        client: Pre-configured OpenAI-compatible client to reuse across requests.
            When provided, this client will be reused for all requests and will NOT be closed
            by the model. The caller is responsible for managing the client lifecycle.
            This is useful for:
            - Injecting custom client wrappers (e.g., GuardrailsAsyncOpenAI)
            - Reusing connection pools within a single event loop/worker
            - Centralizing observability, retries, and networking policy
            - Pointing to custom model gateways
            Note: The client should not be shared across different asyncio event loops.
        client_args: Arguments for the OpenAI client (legacy approach).
            For a complete list of supported arguments, see https://pypi.org/project/openai/.
        **model_config: Configuration options for the OpenAI model.

    Raises:
        ValueError: If both `client` and `client_args` are provided.
    """
    validate_config_keys(model_config, self.OpenAIConfig)
    self.config = dict(model_config)

    # Validate that only one client configuration method is provided
    if client is not None and client_args is not None and len(client_args) > 0:
        raise ValueError("Only one of 'client' or 'client_args' should be provided, not both.")

    self._custom_client = client
    self.client_args = client_args or {}

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

format_chunk(event, **kwargs)

Format an OpenAI response event into a standardized message chunk.

Parameters:

Name	Type	Description	Default
event	dict[str, Any]	A response event from the OpenAI compatible model.	required
**kwargs	Any	Additional keyword arguments for future extensibility.	{}

Returns:

Type	Description
StreamEvent	The formatted chunk.

Raises:

Type	Description
RuntimeError	If chunk_type is not recognized. This error should never be encountered as chunk_type is controlled in the stream method.

Source code in strands/models/openai.py

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

    Args:
        event: A response event from the OpenAI compatible model.
        **kwargs: Additional keyword arguments for future extensibility.

    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":
            return {
                "metadata": {
                    "usage": {
                        "inputTokens": event["data"].prompt_tokens,
                        "outputTokens": event["data"].completion_tokens,
                        "totalTokens": event["data"].total_tokens,
                    },
                    "metrics": {
                        "latencyMs": 0,  # TODO
                    },
                },
            }

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

format_request(messages, tool_specs=None, system_prompt=None, tool_choice=None, *, system_prompt_content=None, **kwargs)

Format an OpenAI compatible chat streaming request.

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 to provide context to the model.	None
**kwargs	Any	Additional keyword arguments for future extensibility.	{}

Returns:

Type	Description
dict[str, Any]	An OpenAI compatible chat streaming request.

Raises:

Type	Description
TypeError	If a message contains a content block type that cannot be converted to an OpenAI-compatible format.

Source code in strands/models/openai.py

def format_request(
    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,
    **kwargs: Any,
) -> dict[str, Any]:
    """Format an OpenAI compatible chat 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.
        system_prompt_content: System prompt content blocks to provide context to the model.
        **kwargs: Additional keyword arguments for future extensibility.

    Returns:
        An OpenAI compatible chat streaming request.

    Raises:
        TypeError: If a message contains a content block type that cannot be converted to an OpenAI-compatible
            format.
    """
    return {
        "messages": self.format_request_messages(
            messages, system_prompt, system_prompt_content=system_prompt_content
        ),
        "model": self.config["model_id"],
        "stream": True,
        "stream_options": {"include_usage": True},
        "tools": [
            {
                "type": "function",
                "function": {
                    "name": tool_spec["name"],
                    "description": tool_spec["description"],
                    "parameters": tool_spec["inputSchema"]["json"],
                },
            }
            for tool_spec in tool_specs or []
        ],
        **(self._format_request_tool_choice(tool_choice)),
        **cast(dict[str, Any], self.config.get("params", {})),
    }

format_request_message_content(content, **kwargs) classmethod

Format an OpenAI compatible content block.

Parameters:

Name	Type	Description	Default
content	ContentBlock	Message content.	required
**kwargs	Any	Additional keyword arguments for future extensibility.	{}

Returns:

Type	Description
dict[str, Any]	OpenAI compatible content block.

Raises:

Type	Description
TypeError	If the content block type cannot be converted to an OpenAI-compatible format.

Source code in strands/models/openai.py

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

    Args:
        content: Message content.
        **kwargs: Additional keyword arguments for future extensibility.

    Returns:
        OpenAI compatible content block.

    Raises:
        TypeError: If the content block type cannot be converted to an OpenAI-compatible format.
    """
    if "document" in content:
        mime_type = mimetypes.types_map.get(f".{content['document']['format']}", "application/octet-stream")
        file_data = base64.b64encode(content["document"]["source"]["bytes"]).decode("utf-8")
        return {
            "file": {
                "file_data": f"data:{mime_type};base64,{file_data}",
                "filename": content["document"]["name"],
            },
            "type": "file",
        }

    if "image" in content:
        mime_type = mimetypes.types_map.get(f".{content['image']['format']}", "application/octet-stream")
        image_data = base64.b64encode(content["image"]["source"]["bytes"]).decode("utf-8")

        return {
            "image_url": {
                "detail": "auto",
                "format": mime_type,
                "url": f"data:{mime_type};base64,{image_data}",
            },
            "type": "image_url",
        }

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

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

format_request_message_tool_call(tool_use, **kwargs) classmethod

Format an OpenAI compatible tool call.

Parameters:

Name	Type	Description	Default
tool_use	ToolUse	Tool use requested by the model.	required
**kwargs	Any	Additional keyword arguments for future extensibility.	{}

Returns:

Type	Description
dict[str, Any]	OpenAI compatible tool call.

Source code in strands/models/openai.py

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

    Args:
        tool_use: Tool use requested by the model.
        **kwargs: Additional keyword arguments for future extensibility.

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

format_request_messages(messages, system_prompt=None, *, system_prompt_content=None, **kwargs) classmethod

Format an OpenAI compatible messages array.

Parameters:

Name	Type	Description	Default
messages	Messages	List of message objects to be processed by the model.	required
system_prompt	Optional[str]	System prompt to provide context to the model.	None
system_prompt_content	Optional[list[SystemContentBlock]]	System prompt content blocks to provide context to the model.	None
**kwargs	Any	Additional keyword arguments for future extensibility.	{}

Returns:

Type	Description
list[dict[str, Any]]	An OpenAI compatible messages array.

Source code in strands/models/openai.py

@classmethod
def format_request_messages(
    cls,
    messages: Messages,
    system_prompt: Optional[str] = None,
    *,
    system_prompt_content: Optional[list[SystemContentBlock]] = None,
    **kwargs: Any,
) -> list[dict[str, Any]]:
    """Format an OpenAI compatible messages array.

    Args:
        messages: List of message objects to be processed by the model.
        system_prompt: System prompt to provide context to the model.
        system_prompt_content: System prompt content blocks to provide context to the model.
        **kwargs: Additional keyword arguments for future extensibility.

    Returns:
        An OpenAI compatible messages array.
    """
    formatted_messages = cls._format_system_messages(system_prompt, system_prompt_content=system_prompt_content)
    formatted_messages.extend(cls._format_regular_messages(messages))

    return [message for message in formatted_messages if message["content"] or "tool_calls" in message]

format_request_tool_message(tool_result, **kwargs) classmethod

Format an OpenAI compatible tool message.

Parameters:

Name	Type	Description	Default
tool_result	ToolResult	Tool result collected from a tool execution.	required
**kwargs	Any	Additional keyword arguments for future extensibility.	{}

Returns:

Type	Description
dict[str, Any]	OpenAI compatible tool message.

Source code in strands/models/openai.py

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

    Args:
        tool_result: Tool result collected from a tool execution.
        **kwargs: Additional keyword arguments for future extensibility.

    Returns:
        OpenAI compatible tool message.
    """
    contents = cast(
        list[ContentBlock],
        [
            {"text": json.dumps(content["json"])} if "json" in content else content
            for content in tool_result["content"]
        ],
    )

    return {
        "role": "tool",
        "tool_call_id": tool_result["toolUseId"],
        "content": [cls.format_request_message_content(content) for content in contents],
    }

get_config()

Get the OpenAI model configuration.

Returns:

Type	Description
OpenAIConfig	The OpenAI model configuration.

Source code in strands/models/openai.py

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

    Returns:
        The OpenAI model configuration.
    """
    return cast(OpenAIModel.OpenAIConfig, self.config)

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

Stream conversation with the OpenAI model.

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
**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.py

@override
async def stream(
    self,
    messages: Messages,
    tool_specs: Optional[list[ToolSpec]] = None,
    system_prompt: Optional[str] = None,
    *,
    tool_choice: ToolChoice | None = None,
    **kwargs: Any,
) -> AsyncGenerator[StreamEvent, None]:
    """Stream conversation with the OpenAI 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")
    request = self.format_request(messages, tool_specs, system_prompt, tool_choice)
    logger.debug("formatted request=<%s>", request)

    logger.debug("invoking model")

    # We initialize an OpenAI context on every request so as to avoid connection sharing in the underlying httpx
    # client. The asyncio event loop does not allow connections to be shared. For more details, please refer to
    # https://github.com/encode/httpx/discussions/2959.
    async with self._get_client() as client:
        try:
            response = await client.chat.completions.create(**request)
        except openai.BadRequestError as e:
            # Check if this is a context length exceeded error
            if hasattr(e, "code") and e.code == "context_length_exceeded":
                logger.warning("OpenAI threw context window overflow error")
                raise ContextWindowOverflowException(str(e)) from e
            # Re-raise other BadRequestError exceptions
            raise
        except openai.RateLimitError as e:
            # All rate limit errors should be treated as throttling, not context overflow
            # Rate limits (including TPM) require waiting/retrying, not context reduction
            logger.warning("OpenAI threw rate limit error")
            raise ModelThrottledException(str(e)) from e

        logger.debug("got response from model")
        yield self.format_chunk({"chunk_type": "message_start"})
        tool_calls: dict[int, list[Any]] = {}
        data_type = None
        finish_reason = None  # Store finish_reason for later use
        event = None  # Initialize for scope safety

        async for event in response:
            # Defensive: skip events with empty or missing choices
            if not getattr(event, "choices", None):
                continue
            choice = event.choices[0]

            if hasattr(choice.delta, "reasoning_content") and choice.delta.reasoning_content:
                chunks, data_type = self._stream_switch_content("reasoning_content", data_type)
                for chunk in chunks:
                    yield chunk
                yield self.format_chunk(
                    {
                        "chunk_type": "content_delta",
                        "data_type": data_type,
                        "data": choice.delta.reasoning_content,
                    }
                )

            if choice.delta.content:
                chunks, data_type = self._stream_switch_content("text", data_type)
                for chunk in chunks:
                    yield chunk
                yield self.format_chunk(
                    {"chunk_type": "content_delta", "data_type": data_type, "data": choice.delta.content}
                )

            for tool_call in choice.delta.tool_calls or []:
                tool_calls.setdefault(tool_call.index, []).append(tool_call)

            if choice.finish_reason:
                finish_reason = choice.finish_reason  # Store for use outside loop
                if data_type:
                    yield self.format_chunk({"chunk_type": "content_stop", "data_type": data_type})
                break

        for tool_deltas in tool_calls.values():
            yield self.format_chunk({"chunk_type": "content_start", "data_type": "tool", "data": tool_deltas[0]})

            for tool_delta in tool_deltas:
                yield self.format_chunk({"chunk_type": "content_delta", "data_type": "tool", "data": tool_delta})

            yield self.format_chunk({"chunk_type": "content_stop", "data_type": "tool"})

        yield self.format_chunk({"chunk_type": "message_stop", "data": finish_reason or "end_turn"})

        # Skip remaining events as we don't have use for anything except the final usage payload
        async for event in response:
            _ = event

        if event and hasattr(event, "usage") and event.usage:
            yield self.format_chunk({"chunk_type": "metadata", "data": event.usage})

    logger.debug("finished streaming response from model")

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

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

@override
async 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:
        ContextWindowOverflowException: If the input exceeds the model's context window.
        ModelThrottledException: If the request is throttled by OpenAI (rate limits).
    """
    # We initialize an OpenAI context on every request so as to avoid connection sharing in the underlying httpx
    # client. The asyncio event loop does not allow connections to be shared. For more details, please refer to
    # https://github.com/encode/httpx/discussions/2959.
    async with self._get_client() as client:
        try:
            response: ParsedChatCompletion = await client.beta.chat.completions.parse(
                model=self.get_config()["model_id"],
                messages=self.format_request(prompt, system_prompt=system_prompt)["messages"],
                response_format=output_model,
            )
        except openai.BadRequestError as e:
            # Check if this is a context length exceeded error
            if hasattr(e, "code") and e.code == "context_length_exceeded":
                logger.warning("OpenAI threw context window overflow error")
                raise ContextWindowOverflowException(str(e)) from e
            # Re-raise other BadRequestError exceptions
            raise
        except openai.RateLimitError as e:
            # All rate limit errors should be treated as throttling, not context overflow
            # Rate limits (including TPM) require waiting/retrying, not context reduction
            logger.warning("OpenAI threw rate limit error")
            raise ModelThrottledException(str(e)) from e

    parsed: T | None = None
    # Find the first choice with tool_calls
    if len(response.choices) > 1:
        raise ValueError("Multiple choices found in the OpenAI response.")

    for choice in response.choices:
        if isinstance(choice.message.parsed, output_model):
            parsed = choice.message.parsed
            break

    if parsed:
        yield {"output": parsed}
    else:
        raise ValueError("No valid tool use or tool use input was found in the OpenAI response.")

update_config(**model_config)

Update the OpenAI model configuration with the provided arguments.

Parameters:

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

Source code in strands/models/openai.py

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

    Args:
        **model_config: Configuration overrides.
    """
    validate_config_keys(model_config, self.OpenAIConfig)
    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

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

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.

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

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