Python: Support prompt cache breakpoints for GPT-5.6 models in OpenAI clients

python/packages/openai/agent_framework_openai/_chat_client.py

@@ -88,6 +88,7 @@
 from ._exceptions import OpenAIContentFilterException
 from ._shared import (
     AzureTokenProvider,
+    _attach_prompt_cache_breakpoint,  # pyright: ignore[reportPrivateUsage]
     load_openai_service_settings,
     maybe_append_azure_endpoint_guidance,
 )
@@ -105,6 +106,17 @@
 else:
     from typing_extensions import TypedDict  # pragma: no cover
 
+try:
+    from openai.types.responses.response_create_params import PromptCacheOptions
+except ImportError:  # pragma: no cover
+
+    class PromptCacheOptions(TypedDict, total=False):
+        """Fallback for openai versions that predate prompt cache options."""
+
+        mode: Literal["implicit", "explicit"]
+        ttl: Literal["30m"]
+
+
 if TYPE_CHECKING:
     from azure.core.credentials import TokenCredential
     from azure.core.credentials_async import AsyncTokenCredential
@@ -232,6 +244,13 @@ class OpenAIChatOptions(ChatOptions[ResponseFormatT], Generic[ResponseFormatT],
     prompt_cache_retention: Literal["24h"]
     """Retention policy for prompt cache. Set to '24h' for extended caching."""
 
+    prompt_cache_options: PromptCacheOptions
+    """Request-wide prompt cache policy for GPT-5.6 and later models.
+    Set mode to 'explicit' to use only the breakpoints set on content parts via
+    ``Content.additional_properties["prompt_cache_breakpoint"]``.
+    Sending this option requires openai 2.45.0 or later.
+    See: https://developers.openai.com/api/docs/guides/prompt-caching#prompt-cache-breakpoints"""
+
     reasoning: ReasoningOptions
     """Configuration for reasoning models (gpt-5, o-series).
     See: https://platform.openai.com/docs/guides/reasoning"""
@@ -1655,10 +1674,13 @@ def _prepare_content_for_openai(
                         "text": content.text,
                         "annotations": _annotations_to_output_text(getattr(content, "annotations", None)),
                     }
-                return {
-                    "type": "input_text",
-                    "text": content.text,
-                }
+                return _attach_prompt_cache_breakpoint(
+                    {
+                        "type": "input_text",
+                        "text": content.text,
+                    },
+                    content,
+                )
             case "text_reasoning":
                 ret: dict[str, Any] = {"type": "reasoning", "summary": []}
                 if content.id:
@@ -1686,7 +1708,7 @@ def _prepare_content_for_openai(
                     file_id = content.additional_properties.get("file_id") if content.additional_properties else None
                     if file_id is not None:
                         result["file_id"] = file_id
-                    return result
+                    return _attach_prompt_cache_breakpoint(result, content)
                 if content.has_top_level_media_type("audio"):
                     if content.media_type and "wav" in content.media_type:
                         format = "wav"
@@ -1714,7 +1736,7 @@ def _prepare_content_for_openai(
                     }
                     if filename:
                         file_obj["filename"] = filename
-                    return file_obj
+                    return _attach_prompt_cache_breakpoint(file_obj, content)
                 return {}
             case "function_call":
                 if not content.call_id:

python/packages/openai/agent_framework_openai/_chat_completion_client.py

@@ -59,7 +59,9 @@
 
 from ._exceptions import OpenAIContentFilterException
 from ._shared import (
+    PROMPT_CACHE_BREAKPOINT_KEY,
     AzureTokenProvider,
+    _attach_prompt_cache_breakpoint,  # pyright: ignore[reportPrivateUsage]
     load_openai_service_settings,
     maybe_append_azure_endpoint_guidance,
 )
@@ -77,6 +79,17 @@
 else:
     from typing_extensions import TypedDict  # pragma: no cover
 
+try:
+    from openai.types.chat.completion_create_params import PromptCacheOptions
+except ImportError:  # pragma: no cover
+
+    class PromptCacheOptions(TypedDict, total=False):
+        """Fallback for openai versions that predate prompt cache options."""
+
+        mode: Literal["implicit", "explicit"]
+        ttl: Literal["30m"]
+
+
 if TYPE_CHECKING:
     from azure.core.credentials import TokenCredential
     from azure.core.credentials_async import AsyncTokenCredential
@@ -157,6 +170,13 @@ class OpenAIChatCompletionOptions(ChatOptions[ResponseModelT], Generic[ResponseM
     """Output verbosity for GPT-5 family models. Lower values yield shorter responses.
     See: https://developers.openai.com/cookbook/examples/gpt-5/gpt-5_new_params_and_tools#1-verbosity-parameter"""
 
+    prompt_cache_options: PromptCacheOptions
+    """Request-wide prompt cache policy for GPT-5.6 and later models.
+    Set mode to 'explicit' to use only the breakpoints set on content parts via
+    ``Content.additional_properties["prompt_cache_breakpoint"]``.
+    Sending this option requires openai 2.45.0 or later.
+    See: https://developers.openai.com/api/docs/guides/prompt-caching#prompt-cache-breakpoints"""
+
 
 OpenAIChatCompletionOptionsT = TypeVar(
     "OpenAIChatCompletionOptionsT",
@@ -873,12 +893,24 @@ def _prepare_messages_for_openai(
 
     def _prepare_message_for_openai(self, message: Message) -> list[dict[str, Any]]:
         """Prepare a chat message for OpenAI."""
-        # System/developer messages must use plain string content because some
-        # OpenAI-compatible endpoints reject list content for non-user roles.
+        # System/developer messages default to plain string content because some
+        # OpenAI-compatible endpoints reject list content for non-user roles. The
+        # exception is a prompt cache breakpoint on a text part: it can only live on
+        # a typed part, so opting in switches that message to list content.
         if message.role in ("system", "developer"):
-            texts = [content.text for content in message.contents if content.type == "text" and content.text]
-            if texts:
-                sys_args: dict[str, Any] = {"role": message.role, "content": "\n".join(texts)}
+            text_contents = [content for content in message.contents if content.type == "text" and content.text]
+            if text_contents:
+                # A prompt cache breakpoint lives on a typed part; plain-string content
+                # cannot carry it, so keep the typed list form when one is present.
+                content_value: str | list[dict[str, Any]]
+                if any(
+                    (content.additional_properties or {}).get(PROMPT_CACHE_BREAKPOINT_KEY) is not None
+                    for content in text_contents
+                ):
+                    content_value = [self._prepare_content_for_openai(content) for content in text_contents]
+                else:
+                    content_value = "\n".join(content.text for content in text_contents if content.text)
+                sys_args: dict[str, Any] = {"role": message.role, "content": content_value}
                 if message.author_name:
                     sys_args["name"] = message.author_name
                 return [sys_args]
@@ -969,6 +1001,10 @@ def _prepare_message_for_openai(self, message: Message) -> list[dict[str, Any]]:
                     text_item = cast(Mapping[str, Any], item)
                     if text_item.get("type") != "text":
                         break
+                    if PROMPT_CACHE_BREAKPOINT_KEY in text_item:
+                        # A plain string cannot carry a prompt cache breakpoint;
+                        # keep the typed part form for this message.
+                        break
                     text_items.append(text_item)
                 else:
                     msg["content"] = "\n".join(
@@ -981,6 +1017,11 @@ def _prepare_message_for_openai(self, message: Message) -> list[dict[str, Any]]:
     def _prepare_content_for_openai(self, content: Content) -> dict[str, Any]:
         """Prepare content for OpenAI."""
         match content.type:
+            case "text":
+                return _attach_prompt_cache_breakpoint(
+                    {"type": "text", "text": content.text},
+                    content,
+                )
             case "function_call":
                 args = json.dumps(content.arguments) if isinstance(content.arguments, Mapping) else content.arguments
                 return {
@@ -998,10 +1039,13 @@ def _prepare_content_for_openai(self, content: Content) -> dict[str, Any]:
                 detail = content.additional_properties.get("detail")
                 if isinstance(detail, str):
                     image_url_obj["detail"] = detail
-                return {
-                    "type": "image_url",
-                    "image_url": image_url_obj,
-                }
+                return _attach_prompt_cache_breakpoint(
+                    {
+                        "type": "image_url",
+                        "image_url": image_url_obj,
+                    },
+                    content,
+                )
             case "data" | "uri" if content.has_top_level_media_type("audio"):
                 if content.media_type and "wav" in content.media_type:
                     audio_format = "wav"
@@ -1017,13 +1061,16 @@ def _prepare_content_for_openai(self, content: Content) -> dict[str, Any]:
                     # Extract just the base64 part after "data:audio/format;base64,"
                     audio_data = audio_data.split(",", 1)[-1]  # type: ignore[union-attr]
 
-                return {
-                    "type": "input_audio",
-                    "input_audio": {
-                        "data": audio_data,
-                        "format": audio_format,
+                return _attach_prompt_cache_breakpoint(
+                    {
+                        "type": "input_audio",
+                        "input_audio": {
+                            "data": audio_data,
+                            "format": audio_format,
+                        },
                     },
-                }
+                    content,
+                )
             case "data" | "uri" if content.has_top_level_media_type("application") and content.uri.startswith("data:"):  # type: ignore[union-attr]
                 # All application/* media types should be treated as files for OpenAI
                 filename = getattr(content, "filename", None) or (
@@ -1034,10 +1081,13 @@ def _prepare_content_for_openai(self, content: Content) -> dict[str, Any]:
                 file_obj = {"file_data": content.uri}
                 if filename:
                     file_obj["filename"] = filename
-                return {
-                    "type": "file",
-                    "file": file_obj,
-                }
+                return _attach_prompt_cache_breakpoint(
+                    {
+                        "type": "file",
+                        "file": file_obj,
+                    },
+                    content,
+                )
             case _:
                 # Default fallback for all other content types
                 return content.to_dict(exclude_none=True)

python/packages/openai/agent_framework_openai/_shared.py

@@ -5,7 +5,7 @@
 import sys
 from collections.abc import Awaitable, Callable, Mapping, Sequence
 from copy import copy
-from typing import TYPE_CHECKING, Any, Literal, Union
+from typing import TYPE_CHECKING, Any, Literal, Union, cast
 
 from agent_framework._settings import SecretString, load_settings
 from agent_framework._telemetry import APP_INFO, prepend_agent_framework_to_user_agent
@@ -24,6 +24,7 @@
     from typing_extensions import TypedDict  # pragma: no cover
 
 if TYPE_CHECKING:
+    from agent_framework import Content
     from azure.core.credentials import TokenCredential
     from azure.core.credentials_async import AsyncTokenCredential
 
@@ -49,6 +50,25 @@
 AzureTokenProvider = Callable[[], str | Awaitable[str]]
 
 
+PROMPT_CACHE_BREAKPOINT_KEY = "prompt_cache_breakpoint"
+
+
+def _attach_prompt_cache_breakpoint(  # pyright: ignore[reportUnusedFunction]
+    part: dict[str, Any], content: Content
+) -> dict[str, Any]:
+    """Copy a prompt cache breakpoint from content metadata onto an outgoing part.
+
+    GPT-5.6 and later models accept an explicit cache breakpoint on supported content
+    blocks; users opt in per part via
+    ``Content.additional_properties["prompt_cache_breakpoint"]``.
+    """
+    props = content.additional_properties
+    breakpoint_value = props.get(PROMPT_CACHE_BREAKPOINT_KEY) if props else None
+    if isinstance(breakpoint_value, Mapping):
+        part[PROMPT_CACHE_BREAKPOINT_KEY] = dict(cast("Mapping[str, Any]", breakpoint_value))
+    return part
+
+
 class OpenAISettings(TypedDict, total=False):
     """OpenAI environment settings.
 

python/packages/openai/tests/openai/test_openai_chat_client.py

@@ -7019,3 +7019,71 @@ def test_prepare_messages_strips_mcp_items_under_storage() -> None:
 
 
 # endregion
+
+
+# region Prompt cache breakpoints and options
+
+
+def _breakpoint_text_content() -> Content:
+    return Content.from_text(
+        "This is a stable prefix that should be cached.",
+        additional_properties={"prompt_cache_breakpoint": {"mode": "explicit"}},
+    )
+
+
+def test_prepare_messages_for_openai_text_prompt_cache_breakpoint() -> None:
+    """A text part carries an explicit prompt cache breakpoint onto the request."""
+    client = OpenAIChatClient(api_key="test-api-key", model="test-model")
+    items = client._prepare_messages_for_openai(
+        [Message(role="user", contents=[_breakpoint_text_content()])],
+        request_uses_service_side_storage=False,
+    )
+    part = items[0]["content"][0]
+    assert part["type"] == "input_text"
+    assert part["prompt_cache_breakpoint"] == {"mode": "explicit"}
+
+
+def test_prepare_content_for_openai_image_prompt_cache_breakpoint() -> None:
+    """An image part carries an explicit prompt cache breakpoint onto the request."""
+    client = OpenAIChatClient(api_key="test-api-key", model="test-model")
+    image = Content.from_uri(
+        uri="https://example.com/x.png",
+        media_type="image/png",
+        additional_properties={"prompt_cache_breakpoint": {"mode": "explicit"}},
+    )
+    part = client._prepare_content_for_openai("user", image)
+    assert part["type"] == "input_image"
+    assert part["prompt_cache_breakpoint"] == {"mode": "explicit"}
+
+
+def test_prepare_content_for_openai_file_prompt_cache_breakpoint() -> None:
+    """A file part carries an explicit prompt cache breakpoint onto the request."""
+    client = OpenAIChatClient(api_key="test-api-key", model="test-model")
+    file_content = Content.from_uri(
+        uri="data:application/pdf;base64,AAAA",
+        media_type="application/pdf",
+        additional_properties={"prompt_cache_breakpoint": {"mode": "explicit"}},
+    )
+    part = client._prepare_content_for_openai("user", file_content)
+    assert part["type"] == "input_file"
+    assert part["prompt_cache_breakpoint"] == {"mode": "explicit"}
+
+
+def test_prepare_content_for_openai_no_prompt_cache_breakpoint_by_default() -> None:
+    """Parts without the property keep their existing shape."""
+    client = OpenAIChatClient(api_key="test-api-key", model="test-model")
+    part = client._prepare_content_for_openai("user", Content.from_text("hello"))
+    assert part == {"type": "input_text", "text": "hello"}
+
+
+async def test_prepare_options_prompt_cache_options_passthrough() -> None:
+    """Request-level prompt_cache_options reaches the Responses API run options."""
+    client = OpenAIChatClient(api_key="test-api-key", model="test-model")
+    run_options = await client._prepare_options(
+        [Message(role="user", contents=[Content.from_text("hi")])],
+        {"model": "test-model", "prompt_cache_options": {"mode": "explicit", "ttl": "30m"}},
+    )
+    assert run_options["prompt_cache_options"] == {"mode": "explicit", "ttl": "30m"}
+
+
+# endregion