from __future__ import annotations as _annotations import base64 import itertools import json import warnings from collections import deque from collections.abc import ( AsyncGenerator, AsyncIterable, AsyncIterator, Callable, Generator, Iterable, Sequence, ) from contextlib import asynccontextmanager, contextmanager from dataclasses import dataclass, field, replace from datetime import datetime, timedelta from functools import cached_property from typing import Any, Literal, cast, get_args, overload from httpx2 import Timeout as HTTPX2Timeout from pydantic import BaseModel, TypeAdapter, ValidationError from pydantic_core import to_json from typing_extensions import Never, Protocol, TypedDict, assert_never from .. import ModelAPIError, ModelHTTPError, UnexpectedModelBehavior, _utils, usage from .._http import to_httpx2_timeout from .._instrumentation import get_instructions from .._output import DEFAULT_OUTPUT_TOOL_NAME from .._run_context import RunContext from .._thinking_part import split_content_into_text_and_thinking from .._utils import ( format_inlined_text_file as _format_inlined_text_file, guard_tool_call_id as _guard_tool_call_id, is_str_dict as _is_str_dict, is_text_like_media_type as _is_text_like_media_type, now_utc as _now_utc, number_to_datetime, ) from ..capabilities.abstract import AbstractCapability from ..exceptions import SuspendedResponseExpired, UserError from ..messages import ( STANDING_PROMPT_PLANTED_KEY, AudioUrl, BinaryContent, BinaryImage, CachePoint, CompactionPart, DocumentUrl, FilePart, FinishReason, ImageUrl, ModelMessage, ModelRequest, ModelResponse, ModelResponsePart, ModelResponseState, ModelResponseStreamEvent, NativeToolCallPart, NativeToolReturnPart, NativeToolSearchCallPart, NativeToolSearchReturnPart, PartStartEvent, RetryPromptPart, SpeechPart, SystemPromptPart, TextContent, TextPart, ThinkingPart, ToolAvailabilityDeltaPart, ToolCallPart, ToolReturnPart, ToolSearchCallPart, ToolSearchReturnPart, UploadedFile, UserContent, UserPromptPart, VideoUrl, is_multi_modal_content, ) from ..native_tools import ( SUPPORTED_NATIVE_TOOLS, AbstractNativeTool, CodeExecutionTool, FileSearchTool, ImageAspectRatio, ImageGenerationTool, MCPServerTool, WebSearchTool, ) from ..native_tools._tool_search import ( TOOL_SEARCH_FUNCTION_TOOL_NAME, ToolSearchArgs, ToolSearchMatch, ToolSearchTool, ) from ..output import OutputObjectDefinition from ..profiles import DEFAULT_THINKING_TAGS, ModelProfile, ModelProfileSpec, merge_profile from ..profiles.openai import ( OPENAI_REASONING_EFFORT_MAP, SAMPLING_PARAMS, OpenAIModelProfile, validate_openai_profile, ) from ..providers import Provider, infer_provider from ..settings import ModelSettings, ThinkingLevel, ToolOrOutput, merge_model_settings from ..tools import AgentDepsT, ToolDefinition from . import ( Model, ModelRequestContext, ModelRequestParameters, OpenAIChatCompatibleProvider, OpenAIResponsesCompatibleProvider, StreamedResponse, ToolVisibility, _suggest_known_model_id_from_provider_error, # pyright: ignore[reportPrivateUsage] _unconverted_speech_part_error, # pyright: ignore[reportPrivateUsage] _unsynthesized_tool_availability_delta_error, # pyright: ignore[reportPrivateUsage] check_allow_model_requests, download_item, get_user_agent, ) from ._tool_choice import ResolvedToolChoice, resolve_tool_choice _OPENAI_BACKGROUND_POLL_INTERVAL = 2.0 try: from openai import ( NOT_GIVEN, APIConnectionError, APIStatusError, AsyncAzureOpenAI, AsyncOpenAI, AsyncStream, NotGiven, Omit, omit, ) from openai.types import AllModels, chat, responses from openai.types.chat import ( ChatCompletionChunk, ChatCompletionContentPartImageParam, ChatCompletionContentPartInputAudioParam, ChatCompletionContentPartParam, ChatCompletionContentPartTextParam, chat_completion, chat_completion_chunk, chat_completion_token_logprob, ) from openai.types.chat.chat_completion_content_part_image_param import ImageURL from openai.types.chat.chat_completion_content_part_input_audio_param import InputAudio from openai.types.chat.chat_completion_content_part_param import File, FileFile from openai.types.chat.chat_completion_message_custom_tool_call import ChatCompletionMessageCustomToolCall from openai.types.chat.chat_completion_message_function_tool_call import ChatCompletionMessageFunctionToolCall from openai.types.chat.chat_completion_message_function_tool_call_param import ( ChatCompletionMessageFunctionToolCallParam, ) from openai.types.chat.chat_completion_prediction_content_param import ChatCompletionPredictionContentParam from openai.types.chat.chat_completion_tool_choice_option_param import ChatCompletionToolChoiceOptionParam from openai.types.chat.completion_create_params import ( Moderation, WebSearchOptions, WebSearchOptionsUserLocation, WebSearchOptionsUserLocationApproximate, ) from openai.types.responses import ( ComputerToolParam, FileSearchToolParam, ResponseCompactionItem, ToolSearchToolParam, WebSearchToolParam, ) from openai.types.responses.response_compaction_item_param_param import ResponseCompactionItemParamParam from openai.types.responses.response_create_params import ( ContextManagement, ToolChoice as ResponsesToolChoice, ) from openai.types.responses.response_input_file_content_param import ResponseInputFileContentParam from openai.types.responses.response_input_image_content_param import ResponseInputImageContentParam from openai.types.responses.response_input_item_param import ToolSearchCall as ToolSearchCallParam from openai.types.responses.response_input_param import FunctionCallOutput, Message from openai.types.responses.response_input_text_content_param import ResponseInputTextContentParam from openai.types.responses.response_reasoning_item_param import ( Content as ReasoningContent, Summary as ReasoningSummary, ) from openai.types.responses.response_status import ResponseStatus from openai.types.responses.response_tool_search_call import ResponseToolSearchCall from openai.types.responses.response_tool_search_output_item_param_param import ( ResponseToolSearchOutputItemParamParam, ) from openai.types.responses.tool_choice_allowed_param import ToolChoiceAllowedParam from openai.types.responses.tool_choice_function_param import ToolChoiceFunctionParam from openai.types.shared import ReasoningEffort from openai.types.shared_params import Reasoning OMIT = omit except ImportError as _import_error: raise ImportError( 'Please install `openai` to use the OpenAI model, ' 'you can use the `openai` optional group — `pip install "pydantic-ai-slim[openai]"`' ) from _import_error def _preload_openai_sdk_resource_modules(model: OpenAIChatModel | OpenAIResponsesModel, client: AsyncOpenAI) -> None: """Load deferred OpenAI SDK modules before request handling. The provider client only triggers process-wide module loading; it need not be the exact client later used by an OpenAI-compatible model subclass. """ if isinstance(model, OpenAIChatModel): _ = client.chat.completions else: _ = client.responses @contextmanager def _map_api_errors(model_name: str, model_id_namespace: str = 'openai') -> Generator[None]: try: yield except APIStatusError as e: if (status_code := e.status_code) >= 400: body: object | None = e.body suggested_model_id = None if _utils.is_str_dict(body) and body.get('code') == 'model_not_found': suggested_model_id = _suggest_known_model_id_from_provider_error(model_id_namespace, model_name) raise ModelHTTPError( status_code=status_code, model_name=model_name, body=body, headers=dict(e.response.headers), suggested_model_id=suggested_model_id, ) from e raise ModelAPIError(model_name=model_name, message=e.message) from e # pragma: lax no cover except APIConnectionError as e: raise ModelAPIError(model_name=model_name, message=e.message) from e __all__ = ( 'DEPRECATED_OPENAI_MODELS', 'OpenAIChatModel', 'OpenAIResponsesModel', 'OpenAIChatModelSettings', 'OpenAIResponsesModelSettings', 'OpenAIPromptCacheOptions', 'OpenAIModelName', ) DEPRECATED_OPENAI_MODELS: frozenset[str] = frozenset( { # https://developers.openai.com/api/docs/deprecations#2025-11-18-chatgpt-4o-latest-snapshot 'chatgpt-4o-latest', # https://developers.openai.com/api/docs/deprecations#2025-11-17-codex-mini-latest-model-snapshot 'codex-mini-latest', # https://developers.openai.com/api/docs/deprecations#2023-11-06-chat-model-updates 'gpt-3.5-turbo-0613', 'gpt-3.5-turbo-16k-0613', # https://developers.openai.com/api/docs/deprecations#2025-09-26-legacy-gpt-model-snapshots 'gpt-4-0125-preview', 'gpt-4-1106-preview', 'gpt-4-turbo-preview', # https://developers.openai.com/api/docs/deprecations#2024-06-06-gpt-4-32k-and-vision-preview-models 'gpt-4-32k', 'gpt-4-32k-0314', 'gpt-4-32k-0613', 'gpt-4-vision-preview', # https://developers.openai.com/api/docs/deprecations#2025-06-10-gpt-4o-audio-preview-2024-10-01 'gpt-4o-audio-preview-2024-10-01', # Does not exist 'gpt-5.1-mini', # https://developers.openai.com/api/docs/deprecations#2025-04-28-o1-preview-and-o1-mini 'o1-mini', 'o1-mini-2024-09-12', 'o1-preview', 'o1-preview-2024-09-12', } ) """Models that are deprecated or don't exist but are still present in the OpenAI SDK's type definitions.""" _DEFAULT_CLIENT_TOOL_SEARCH_DESCRIPTION = 'Search for relevant tools.' OpenAIModelName = str | AllModels | Literal['gpt-5.5-2026-04-23', 'gpt-5.5-pro', 'gpt-5.5-pro-2026-04-23'] """ Possible OpenAI model names. Since OpenAI supports a variety of date-stamped models, we explicitly list the latest models but allow any name in the type hints. See [the OpenAI docs](https://platform.openai.com/docs/models) for a full list. Using this more broad type for the model name instead of the ChatModel definition allows this model to be used more easily with other model types (ie, Ollama, Deepseek). The ids in the local `Literal` are bridged because `AllModels` doesn't list them at the floor the `openai` extra declares; they arrived in `openai` 3.1.0 (https://github.com/openai/openai-python/pull/3617). Drop them once the floor is bumped past it. """ MCP_SERVER_TOOL_CONNECTOR_URI_SCHEME: Literal['x-openai-connector'] = 'x-openai-connector' """ Prefix for OpenAI connector IDs. OpenAI supports either a URL or a connector ID when passing MCP configuration to a model, by using that prefix like `x-openai-connector:` in a URL, you can pass a connector ID to a model. """ _CHAT_FINISH_REASON_MAP: dict[ Literal['stop', 'length', 'tool_calls', 'content_filter', 'function_call'], FinishReason ] = { 'stop': 'stop', 'length': 'length', 'tool_calls': 'tool_call', 'content_filter': 'content_filter', 'function_call': 'tool_call', } _RESPONSES_FINISH_REASON_MAP: dict[ Literal['max_output_tokens', 'max_messages', 'content_filter', 'steered'] | ResponseStatus, FinishReason ] = { 'max_output_tokens': 'length', 'content_filter': 'content_filter', 'completed': 'stop', 'cancelled': 'error', 'failed': 'error', } def _response_status_to_state(status: ResponseStatus | None, *, background: bool) -> ModelResponseState: """Map a Responses API status to a `ModelResponseState`. Only genuine background jobs (`response.background`) suspend on a pending status: they can be resumed server-side via `retrieve(..., starting_after=...)`. A foreground stream also emits `in_progress`/`queued` statuses while it runs, but it isn't resumable, so a pending status there means the response is merely `'incomplete'` (still streaming, or ended without a terminal event) — marking it `'suspended'` would send the continuation loop off to `retrieve` a non-background job. """ if status in ('queued', 'in_progress'): return 'suspended' if background else 'incomplete' return 'complete' class _OpenAIResponsesContinuationDetails(TypedDict, total=False): """Provider details for OpenAI Responses API continuation.""" last_sequence_number: int _OPENAI_ASPECT_RATIO_TO_SIZE: dict[ImageAspectRatio, Literal['1024x1024', '1024x1536', '1536x1024']] = { '1:1': '1024x1024', '2:3': '1024x1536', '3:2': '1536x1024', } _OPENAI_IMAGE_SIZE = Literal['auto', '1024x1024', '1024x1536', '1536x1024'] _OPENAI_IMAGE_SIZES: tuple[_OPENAI_IMAGE_SIZE, ...] = get_args(_OPENAI_IMAGE_SIZE) class _ChatCompletion(chat.ChatCompletion): """Relaxes strict Literal validation on fields that OpenAI-compatible providers may return non-standard values for.""" model_config = {'title': 'ChatCompletion'} service_tier: str | None = None # type: ignore[reportIncompatibleVariableOverride] """OpenAI-compatible providers can return arbitrary `service_tier` values (e.g. `"standard"`, `"on_demand"`).""" class _ChatCompletionChunk(ChatCompletionChunk): # pyright: ignore[reportUnusedClass] — subclassed in openrouter.py """Relaxes strict Literal validation on fields that OpenAI-compatible providers may return non-standard values for.""" model_config = {'title': 'ChatCompletionChunk'} service_tier: str | None = None # type: ignore[reportIncompatibleVariableOverride] """OpenAI-compatible providers can return arbitrary `service_tier` values (e.g. `"standard"`, `"on_demand"`).""" class _AzureContentFilterResultDetail(BaseModel): filtered: bool severity: str | None = None detected: bool | None = None class _AzureContentFilterResult(BaseModel): hate: _AzureContentFilterResultDetail | None = None self_harm: _AzureContentFilterResultDetail | None = None sexual: _AzureContentFilterResultDetail | None = None violence: _AzureContentFilterResultDetail | None = None jailbreak: _AzureContentFilterResultDetail | None = None profanity: _AzureContentFilterResultDetail | None = None class _AzureInnerError(BaseModel): code: str content_filter_result: _AzureContentFilterResult class _AzureError(BaseModel): code: str message: str innererror: _AzureInnerError | None = None def _resolve_openai_image_generation_size( tool: ImageGenerationTool, ) -> _OPENAI_IMAGE_SIZE: """Map `ImageGenerationTool.aspect_ratio` to an OpenAI size string when provided.""" aspect_ratio = tool.aspect_ratio if aspect_ratio is None: if tool.size is None: return 'auto' # default if tool.size not in _OPENAI_IMAGE_SIZES: raise UserError( f'OpenAI image generation only supports `size` values: {_OPENAI_IMAGE_SIZES}. ' f'Got: {tool.size}. Omit `size` to use the default (auto).' ) return tool.size mapped_size = _OPENAI_ASPECT_RATIO_TO_SIZE.get(aspect_ratio) if mapped_size is None: supported = ', '.join(_OPENAI_ASPECT_RATIO_TO_SIZE) raise UserError( f'OpenAI image generation only supports `aspect_ratio` values: {supported}. Specify one of those values or omit `aspect_ratio`.' ) # When aspect_ratio is set, size must be None, 'auto', or match the mapped size if tool.size not in (None, 'auto', mapped_size): raise UserError( '`ImageGenerationTool` cannot combine `aspect_ratio` with a conflicting `size` when using OpenAI.' ) return mapped_size def _map_openai_image_generation_tool(tool: ImageGenerationTool) -> responses.tool_param.ImageGeneration: size = _resolve_openai_image_generation_size(tool) output_compression = tool.output_compression if tool.output_compression is not None else 100 image_generation_tool = responses.tool_param.ImageGeneration( type='image_generation', action=tool.action, background=tool.background, moderation=tool.moderation, output_compression=output_compression, output_format=tool.output_format or 'png', partial_images=tool.partial_images, quality=tool.quality, size=size, ) if tool.model is not None: image_generation_tool['model'] = tool.model if tool.input_fidelity is not None: image_generation_tool['input_fidelity'] = tool.input_fidelity return image_generation_tool def _is_azure(client: AsyncOpenAI, system: str) -> bool: return system == 'azure' or isinstance(client, AsyncAzureOpenAI) def _check_azure_content_filter( e: APIStatusError, client: AsyncOpenAI, system: str, model_name: str ) -> ModelResponse | None: """Check if the error is an Azure content filter error.""" # Assign to Any to avoid 'dict[Unknown, Unknown]' inference in strict mode body_any: Any = e.body if _is_azure(client, system) and e.status_code == 400 and _is_str_dict(body_any): try: error_data = _AzureError.model_validate(body_any.get('error', body_any)) if error_data.code == 'content_filter': provider_details: dict[str, Any] = {'finish_reason': 'content_filter'} if error_data.innererror: provider_details['content_filter_result'] = error_data.innererror.content_filter_result.model_dump( exclude_none=True ) return ModelResponse( parts=[], # Empty parts to trigger content filter error in agent graph model_name=model_name, timestamp=_utils.now_utc(), provider_name=system, finish_reason='content_filter', provider_details=provider_details, ) except ValidationError: pass return None def _merge_leading_system_messages( openai_messages: list[chat.ChatCompletionMessageParam], system_prompt_role: str ) -> list[chat.ChatCompletionMessageParam]: """Merge consecutive messages with `system_prompt_role` at the start of the list into a single message. Required by strict OpenAI-compatible backends (some LiteLLM/vLLM deployments) that reject more than one initial system message with `System message must be at the beginning.` Only applies when system prompts are sent as `'system'` or `'developer'` role; with `'user'` role we can't distinguish a system prompt cast as user from an actual user turn. """ if system_prompt_role not in ('system', 'developer'): return openai_messages leading_count = next( (i for i, m in enumerate(openai_messages) if m.get('role') != system_prompt_role), len(openai_messages), ) if leading_count < 2: return openai_messages # Content is always `str` here: it originates from `SystemPromptPart.content` or instruction # `TextPart.content`, both of which are typed as `str`. merged_content = '\n\n'.join(m['content'] for m in openai_messages[:leading_count]) # type: ignore[misc] merged: chat.ChatCompletionMessageParam = {**openai_messages[0], 'content': merged_content} # type: ignore[typeddict-item] return [merged, *openai_messages[leading_count:]] def _resolve_openai_thinking_effort(thinking: ThinkingLevel, profile: OpenAIModelProfile) -> ReasoningEffort: """Map unified thinking to the closest reasoning effort the model supports.""" if thinking == 'minimal' and not profile.get('openai_supports_minimal_reasoning_effort', True): return 'low' return OPENAI_REASONING_EFFORT_MAP[thinking] # type: ignore[return-value] def _reasoning_active( profile: OpenAIModelProfile, model_settings: OpenAIChatModelSettings | OpenAIResponsesModelSettings, model_request_parameters: ModelRequestParameters, ) -> bool: """Whether reasoning is effectively on for this request. Explicit effort wins; then the unified `thinking` setting; otherwise the model's default. """ reasoning_effort = model_settings.get('openai_reasoning_effort') if reasoning_effort is not None: return reasoning_effort != 'none' thinking = model_request_parameters.thinking if thinking is not None: return thinking is not False return profile.get('openai_reasoning_enabled_by_default', False) def _drop_sampling_params_for_reasoning( profile: OpenAIModelProfile, model_settings: OpenAIChatModelSettings, model_request_parameters: ModelRequestParameters, ) -> None: """Drop sampling params when reasoning is enabled on models that support it. Mutates `model_settings`. Reasoning models don't support sampling parameters while reasoning is active. For models that can turn reasoning off (`openai_supports_reasoning_effort_none`), sampling params are allowed when reasoning is off. Whether reasoning is on when no effort is set depends on the model's default (`openai_reasoning_enabled_by_default`): the GPT-5.1..5.4 mainline models default to off, while the o-series, the original GPT-5, and GPT-5.5+ default to on. """ if not profile.get('openai_supports_reasoning', False): return # When the model can turn reasoning off, sampling params are allowed while reasoning is inactive. if profile.get('openai_supports_reasoning_effort_none', False) and not _reasoning_active( profile, model_settings, model_request_parameters ): return if dropped := [k for k in SAMPLING_PARAMS if k in model_settings]: warnings.warn( f'Sampling parameters {dropped} are not supported when reasoning is enabled. ' 'These settings will be ignored.', UserWarning, ) for k in SAMPLING_PARAMS: model_settings.pop(k, None) def _drop_unsupported_params(profile: OpenAIModelProfile, model_settings: OpenAIChatModelSettings) -> None: """Drop unsupported parameters based on model profile. Mutates `model_settings`. Used currently only by Cerebras """ for setting in profile.get('openai_unsupported_model_settings', ()): model_settings.pop(setting, None) @dataclass class _ResponsesRequestParams: """Typed request parameters shared by Responses API calls.""" model: OpenAIModelName input: list[responses.ResponseInputItemParam] instructions: str | Omit parallel_tool_calls: bool | Omit tools: list[responses.ToolParam] | Omit tool_choice: ResponsesToolChoice | Omit previous_response_id: str | Omit conversation: str | Omit reasoning: Reasoning | Omit text: responses.ResponseTextConfigParam | Omit truncation: Literal['auto', 'disabled'] | Omit context_management: list[ContextManagement] | Omit class OpenAIPromptCacheOptions(TypedDict, total=False): """Options for OpenAI prompt caching on GPT-5.6 models.""" mode: Literal['implicit', 'explicit'] """Whether OpenAI may create an implicit cache breakpoint. Defaults to `implicit`.""" ttl: Literal['30m'] """The minimum lifetime for cache breakpoints. Defaults to `30m`, the only currently supported value.""" class _OpenAIPromptCacheBreakpoint(TypedDict): mode: Literal['explicit'] def _add_openai_prompt_cache_breakpoint( content: Sequence[ChatCompletionContentPartParam | responses.ResponseInputContentParam], ) -> None: if not content: raise UserError( 'CachePoint cannot be the first content in a user message - ' 'there must be previous content to attach the cache breakpoint to.' ) cache_breakpoint: _OpenAIPromptCacheBreakpoint = {'mode': 'explicit'} content[-1]['prompt_cache_breakpoint'] = cache_breakpoint class OpenAIChatModelSettings(ModelSettings, total=False): """Settings used for an OpenAI model request.""" # ALL FIELDS MUST BE `openai_` PREFIXED SO YOU CAN MERGE THEM WITH OTHER MODELS. openai_reasoning_effort: ReasoningEffort """Constrains effort on reasoning for [reasoning models](https://platform.openai.com/docs/guides/reasoning). Currently supported values are `low`, `medium`, and `high`. Reducing reasoning effort can result in faster responses and fewer tokens used on reasoning in a response. """ openai_logprobs: bool """Include log probabilities in the response. For Chat models, these will be included in `ModelResponse.provider_details['logprobs']`. For Responses models, these will be included in the response output parts `TextPart.provider_details['logprobs']`. """ openai_top_logprobs: int """Include log probabilities of the top n tokens in the response.""" openai_store: bool | None """Whether or not to store the output of this request in OpenAI's systems. If `False`, OpenAI will not store the request for its own internal review or training. See [OpenAI API reference](https://platform.openai.com/docs/api-reference/chat/create#chat-create-store). When used with `OpenAIResponsesModel`, stored responses appear in OpenAI's dashboard and can be referenced via [`openai_previous_response_id`][pydantic_ai.models.openai.OpenAIResponsesModelSettings.openai_previous_response_id]. Pair this with `openai_previous_response_id='auto'` to avoid storing duplicate copies of the conversation history across retries and subsequent requests within the same run. """ openai_user: str """A unique identifier representing the end-user, which can help OpenAI monitor and detect abuse. See [OpenAI's safety best practices](https://platform.openai.com/docs/guides/safety-best-practices#end-user-ids) for more details. """ openai_moderation: Moderation """Run moderation on the input and output of the request, e.g. `{'model': 'omni-moderation-latest'}`. Supported by both the Chat Completions API and the Responses API. In both cases, the moderation results returned by the API are exposed in [`ModelResponse.provider_details`][pydantic_ai.messages.ModelResponse.provider_details] under the `'moderation'` key. See the [OpenAI moderation documentation](https://platform.openai.com/docs/guides/moderation) for more details. """ openai_service_tier: Literal['auto', 'default', 'flex', 'priority'] """The service tier to use for the model request. Currently supported values are `auto`, `default`, `flex`, and `priority`. For more information, see [OpenAI's service tiers documentation](https://platform.openai.com/docs/api-reference/chat/object#chat/object-service_tier). """ openai_prediction: ChatCompletionPredictionContentParam """Enables [predictive outputs](https://platform.openai.com/docs/guides/predicted-outputs). This feature is currently only supported for some OpenAI models. """ openai_prompt_cache_key: str """Used by OpenAI to cache responses for similar requests to optimize your cache hit rates. See the [OpenAI Prompt Caching documentation](https://platform.openai.com/docs/guides/prompt-caching#how-it-works) for more information. """ openai_prompt_cache_retention: Literal['in_memory', '24h'] """The retention policy for the prompt cache. Set to 24h to enable extended prompt caching, which keeps cached prefixes active for longer, up to a maximum of 24 hours. For GPT-5.6 and later models, OpenAI deprecates this field in favor of the `ttl` in `openai_prompt_cache_options`; earlier models keep using this field. The two are independent and do not interact: this field expresses a maximum retention policy, while `ttl` expresses a minimum cache lifetime. See the [OpenAI Prompt Caching documentation](https://platform.openai.com/docs/guides/prompt-caching#how-it-works) for more information. """ openai_prompt_cache_options: OpenAIPromptCacheOptions """Controls implicit and explicit prompt cache breakpoints, supported by GPT-5.6 and later models. Explicit breakpoints are added to user content with [`CachePoint`][pydantic_ai.messages.CachePoint]. OpenAI applies the request-wide `ttl` to every breakpoint and ignores `CachePoint.ttl`. The `ttl` here is independent of the `openai_prompt_cache_retention` setting, which OpenAI deprecates for GPT-5.6 and later models. See the [OpenAI prompt caching documentation](https://developers.openai.com/api/docs/guides/prompt-caching) for more information. """ openai_continuous_usage_stats: bool """When True, enables continuous usage statistics in streaming responses. When enabled, the API returns cumulative usage data with each chunk rather than only at the end. This setting correctly handles the cumulative nature of these stats by using only the final usage values rather than summing all intermediate values. See [OpenAI's streaming documentation](https://platform.openai.com/docs/api-reference/chat/create#stream_options) for more information. """ class OpenAIResponsesModelSettings(OpenAIChatModelSettings, total=False): """Settings used for an OpenAI Responses model request. ALL FIELDS MUST BE `openai_` PREFIXED SO YOU CAN MERGE THEM WITH OTHER MODELS. """ openai_native_tools: Sequence[FileSearchToolParam | WebSearchToolParam | ComputerToolParam] """The provided OpenAI built-in tools to use. See [OpenAI's built-in tools](https://platform.openai.com/docs/guides/tools?api-mode=responses) for more details. """ openai_reasoning_mode: Literal['standard', 'pro'] """The reasoning mode to use, for models that support it (currently the GPT-5.6 family). `standard` is the default. `pro` performs more model work to improve reliability on difficult tasks, at the cost of higher latency and token usage. Reasoning mode is independent of [`openai_reasoning_effort`][pydantic_ai.models.openai.OpenAIChatModelSettings.openai_reasoning_effort], and the unified [`thinking`][pydantic_ai.settings.ModelSettings.thinking] setting only influences the effort, never the mode. This setting is ignored when the resolved model profile does not support reasoning mode ([`OpenAIModelProfile.openai_responses_supports_reasoning_mode`][pydantic_ai.profiles.openai.OpenAIModelProfile.openai_responses_supports_reasoning_mode]). See [OpenAI's reasoning mode documentation](https://developers.openai.com/api/docs/guides/reasoning#reasoning-mode) for more details. """ openai_reasoning_context: Literal['auto', 'current_turn', 'all_turns'] """The reasoning context to use, for models that support it. Controls which prior-turn reasoning items the model can use when sampling: `auto` defers to the model's own default (OpenAI treats it exactly like not sending the field), `current_turn` makes only the active turn's reasoning available, and `all_turns` renders compatible reasoning items from earlier turns into the next sample (requires access to earlier response items via `previous_response_id`, a conversation, or replayed history). When this setting is omitted, Pydantic AI sends `all_turns` on models that support it, so that earlier-turn reasoning stays available by default. Set `auto` explicitly to defer to OpenAI's own per-model default instead. `auto` and `current_turn` are sent to any model that supports reasoning. `all_turns` is sent only to models whose profile sets [`OpenAIModelProfile.openai_responses_supports_reasoning_context`][pydantic_ai.profiles.openai.OpenAIModelProfile.openai_responses_supports_reasoning_context] (currently the GPT-5.4, GPT-5.5, and GPT-5.6 families). A value the resolved profile doesn't support is ignored. See [OpenAI's reasoning context documentation](https://developers.openai.com/api/docs/guides/reasoning#preserve-reasoning-across-calls) for more details. """ openai_reasoning_summary: Literal['detailed', 'concise', 'auto'] """A summary of the reasoning performed by the model. This can be useful for debugging and understanding the model's reasoning process. One of `concise`, `detailed`, or `auto`. Check the [OpenAI Reasoning documentation](https://platform.openai.com/docs/guides/reasoning?api-mode=responses#reasoning-summaries) for more details. """ # TODO(v3): rename to `openai_send_item_ids`; this gates far more than reasoning IDs (see docstring). openai_send_reasoning_ids: bool """Whether to send the unique IDs of reasoning, text, and function call parts from the message history to the model. Enabled by default for reasoning models. This can result in errors like `"Item 'rs_123' of type 'reasoning' was provided without its required following item."` if the message history you're sending does not match exactly what was received from the Responses API in a previous response, for example if you're using a [history processor](../../message-history.md#processing-message-history). In that case, you'll want to disable this. Most server-side tool items (web search, code interpreter, image generation) are replayed *by* their ID, so disabling this also stops them from being sent back entirely. Hosted tool-search items are the exception: they carry their state (the query and discovered tools) inline, so they are still replayed with the IDs omitted, and previously discovered tools stay callable. """ openai_truncation: Literal['disabled', 'auto'] """The truncation strategy to use for the model response. It can be either: - `disabled` (default): If a model response will exceed the context window size for a model, the request will fail with a 400 error. - `auto`: If the context of this response and previous ones exceeds the model's context window size, the model will truncate the response to fit the context window by dropping input items in the middle of the conversation. """ openai_text_verbosity: Literal['low', 'medium', 'high'] """Constrains the verbosity of the model's text response. Lower values will result in more concise responses, while higher values will result in more verbose responses. Currently supported values are `low`, `medium`, and `high`. """ openai_previous_response_id: Literal['auto'] | str """Reference a prior OpenAI response to continue a conversation server-side, omitting already-stored messages from the input. - `'auto'`: chain to the most recent `provider_response_id` in the message history. If the history contains no such response, no `previous_response_id` is sent. - A concrete response ID string: use it as the seed for the first request in the run (e.g. to continue from a prior turn). On subsequent in-run requests (retries, tool-call continuations), the most recent `provider_response_id` from the message history takes precedence so the chain extends correctly without re-sending messages that are already server-side. In both cases, messages that precede the chosen response in the history are omitted from the input, since OpenAI reconstructs them from server-side state. Requires the referenced response to have been stored (see [`openai_store`][pydantic_ai.models.openai.OpenAIResponsesModelSettings.openai_store], which defaults to `True` on OpenAI's side). Not compatible with Zero Data Retention. See the [OpenAI Responses API documentation](https://platform.openai.com/docs/guides/reasoning#keeping-reasoning-items-in-context) for more information. """ openai_conversation_id: Literal['auto'] | str """Reference an OpenAI conversation to continue durable conversation state server-side. - `'auto'`: use the most recent OpenAI conversation ID from `ModelResponse.provider_details['conversation_id']` in the message history with the same Pydantic AI `conversation_id`, when available. If the history contains no such response, no `conversation` is sent. - A concrete conversation ID string: use it as the OpenAI Responses API `conversation` parameter. When a matching conversation ID is found in message history, messages that precede that response are omitted from the input, since OpenAI reconstructs them from the server-side conversation. Not compatible with [`openai_previous_response_id`][pydantic_ai.models.openai.OpenAIResponsesModelSettings.openai_previous_response_id]. See the [OpenAI conversation state documentation](https://platform.openai.com/docs/guides/conversation-state) for more information. """ openai_include_code_execution_outputs: bool """Whether to include the code execution results in the response. Corresponds to the `code_interpreter_call.outputs` value of the `include` parameter in the Responses API. """ openai_include_web_search_sources: bool """Whether to include the web search results in the response. Corresponds to the `web_search_call.action.sources` value of the `include` parameter in the Responses API. """ openai_include_file_search_results: bool """Whether to include the file search results in the response. Corresponds to the `file_search_call.results` value of the `include` parameter in the Responses API. """ openai_include_raw_annotations: bool """Whether to include the raw annotations in `TextPart.provider_details`. When enabled, any annotations (e.g., citations from web search) will be available in the `provider_details['annotations']` field of text parts. This is opt-in since there may be overlap with native annotation support once added via https://github.com/pydantic/pydantic-ai/issues/3126. """ openai_context_management: list[ContextManagement] """Context management configuration for the request. This enables OpenAI's server-side automatic compaction inside the regular `/responses` call, as opposed to the standalone `/responses/compact` endpoint. See [OpenAI's compaction guide](https://developers.openai.com/api/docs/guides/compaction) for details. The [`OpenAICompaction`][pydantic_ai.models.openai.OpenAICompaction] capability sets this automatically in its default (stateful) mode. """ openai_background: bool """Enable background mode for long-running requests. When enabled, this setting passes `background=True` to the Responses API and opts into automatic polling for completion. If the response is still pending (`'queued'` or `'in_progress'`), the agent automatically polls for completion using `retrieve()`. """ def _resolve_openai_service_tier( model_settings: OpenAIChatModelSettings, ) -> Literal['auto', 'default', 'flex', 'priority'] | Omit: """Resolve the value to send as `service_tier` on the OpenAI request. Per-provider [`openai_service_tier`][pydantic_ai.models.openai.OpenAIChatModelSettings.openai_service_tier] wins; otherwise the top-level [`service_tier`][pydantic_ai.settings.ModelSettings.service_tier] maps 1:1 to OpenAI's accepted values. """ if openai_tier := model_settings.get('openai_service_tier'): return openai_tier if unified := model_settings.get('service_tier'): return unified return OMIT def _resolve_prompt_cache_retention( default_settings: ModelSettings | None, model_settings: ModelSettings | None ) -> timedelta | None: settings = merge_model_settings(default_settings, model_settings) or {} if settings.get('openai_prompt_cache_retention') == '24h': return timedelta(hours=24) return None @dataclass(init=False) class OpenAIChatModel(Model[AsyncOpenAI]): """A model that uses the OpenAI API. Internally, this uses the [OpenAI Python client](https://github.com/openai/openai-python) to interact with the API. Apart from `__init__`, all methods are private or match those of the base class. """ _model_name: OpenAIModelName = field(repr=False) _provider: Provider[AsyncOpenAI] = field(repr=False) def __init__( self, model_name: OpenAIModelName, *, provider: OpenAIChatCompatibleProvider | Literal[ 'openai', 'openai-chat', 'gateway', ] | Provider[AsyncOpenAI] = 'openai', profile: ModelProfileSpec | None = None, settings: ModelSettings | None = None, ): """Initialize an OpenAI model. Args: model_name: The name of the OpenAI model to use. List of model names available [here](https://github.com/openai/openai-python/blob/v1.54.3/src/openai/types/chat_model.py#L7) (Unfortunately, despite being ask to do so, OpenAI do not provide `.inv` files for their API). provider: The provider to use. Defaults to `'openai'`. profile: The model profile to use. Defaults to a profile picked by the provider based on the model name. settings: Default model settings for this model instance. """ self._model_name = model_name if isinstance(provider, str): provider = infer_provider('gateway/openai' if provider == 'gateway' else provider) self._provider = provider super().__init__(settings=settings, profile=profile) validate_openai_profile(self.profile) _preload_openai_sdk_resource_modules(self, self._provider.client) @property def client(self) -> AsyncOpenAI: return self._provider.client @property def base_url(self) -> str: return str(self.client.base_url) @property def model_name(self) -> OpenAIModelName: """The model name.""" return self._model_name def resolve_prompt_cache_retention(self, model_settings: ModelSettings | None) -> timedelta | None: """Resolve the extended prompt cache retention requested by OpenAI settings.""" return _resolve_prompt_cache_retention(self.settings, model_settings) @property def system(self) -> str: """The model provider.""" return self._provider.name @classmethod def supported_native_tools(cls) -> frozenset[type[AbstractNativeTool]]: """Return the set of builtin tool types this model can handle.""" return frozenset({WebSearchTool}) @cached_property def profile(self) -> OpenAIModelProfile: """The model profile. WebSearchTool is only supported if openai_chat_supports_web_search is True. """ _profile = super().profile if (provider := self.provider) is not None and _is_azure(provider.client, provider.name): user = self._profile if user is None or (not callable(user) and 'openai_chat_supports_document_input' not in user): _profile = merge_profile(_profile, OpenAIModelProfile(openai_chat_supports_document_input=False)) if not _profile.get('openai_chat_supports_web_search', False): new_tools = _profile.get('supported_native_tools', SUPPORTED_NATIVE_TOOLS) - {WebSearchTool} _profile = merge_profile(_profile, ModelProfile(supported_native_tools=new_tools)) return cast(OpenAIModelProfile, _profile) def prepare_request( self, model_settings: ModelSettings | None, model_request_parameters: ModelRequestParameters, ) -> tuple[ModelSettings | None, ModelRequestParameters]: # Check for WebSearchTool before base validation to provide a helpful error message if ( any(isinstance(tool, WebSearchTool) for tool in model_request_parameters.native_tools) and not self.profile.get('openai_chat_supports_web_search', False) and not any(t.unless_native == 'web_search' for t in model_request_parameters.function_tools) ): raise UserError( f'WebSearchTool is not supported with `OpenAIChatModel` and model {self.model_name!r}. ' f'Please use `OpenAIResponsesModel` instead.' ) return super().prepare_request(model_settings, model_request_parameters) async def request( self, messages: list[ModelMessage], model_settings: ModelSettings | None, model_request_parameters: ModelRequestParameters, ) -> ModelResponse: check_allow_model_requests() model_settings, model_request_parameters = self.prepare_request( model_settings, model_request_parameters, ) response = await self._completions_create( messages, False, cast(OpenAIChatModelSettings, model_settings or {}), model_request_parameters ) # Handle ModelResponse returned directly (for content filters) if isinstance(response, ModelResponse): return response model_response = self._process_response(response) return model_response def _translate_thinking( self, model_settings: OpenAIChatModelSettings, model_request_parameters: ModelRequestParameters, ) -> ReasoningEffort | Omit: """Get reasoning effort, falling back to unified thinking when provider-specific setting is not set.""" if effort := model_settings.get('openai_reasoning_effort'): return effort thinking = model_request_parameters.thinking if thinking is None: return OMIT return _resolve_openai_thinking_effort(thinking, self.profile) @asynccontextmanager async def request_stream( self, messages: list[ModelMessage], model_settings: ModelSettings | None, model_request_parameters: ModelRequestParameters, run_context: RunContext[Any] | None = None, ) -> AsyncGenerator[StreamedResponse]: check_allow_model_requests() model_settings, model_request_parameters = self.prepare_request( model_settings, model_request_parameters, ) model_settings_cast = cast(OpenAIChatModelSettings, model_settings or {}) response = await self._completions_create(messages, True, model_settings_cast, model_request_parameters) async with response: yield await self._process_streamed_response(response, model_request_parameters, model_settings_cast) @overload async def _completions_create( self, messages: list[ModelMessage], stream: Literal[True], model_settings: OpenAIChatModelSettings, model_request_parameters: ModelRequestParameters, ) -> AsyncStream[ChatCompletionChunk]: ... @overload async def _completions_create( self, messages: list[ModelMessage], stream: Literal[False], model_settings: OpenAIChatModelSettings, model_request_parameters: ModelRequestParameters, ) -> chat.ChatCompletion | ModelResponse: ... async def _completions_create( self, messages: list[ModelMessage], stream: bool, model_settings: OpenAIChatModelSettings, model_request_parameters: ModelRequestParameters, ) -> chat.ChatCompletion | AsyncStream[ChatCompletionChunk] | ModelResponse: tool_choice: ChatCompletionToolChoiceOptionParam | None tools, tool_choice = self._get_tool_choice(model_settings, model_request_parameters) web_search_options = self._get_web_search_options(model_request_parameters) profile = self.profile openai_messages = await self._map_messages(messages, model_request_parameters, model_settings=model_settings) response_format: chat.completion_create_params.ResponseFormat | None = None if model_request_parameters.output_mode == 'native': output_object = model_request_parameters.output_object assert output_object is not None response_format = self._map_json_schema(output_object) elif model_request_parameters.output_mode == 'prompted' and self.profile.get( 'supports_json_object_output', False ): # pragma: no branch response_format = {'type': 'json_object'} # Both helpers mutate the settings they receive. model_settings = OpenAIChatModelSettings(**model_settings) _drop_sampling_params_for_reasoning(profile, model_settings, model_request_parameters) _drop_unsupported_params(profile, model_settings) with _map_api_errors(self.model_name, self._provider.model_id_namespace): try: extra_headers = dict(model_settings.get('extra_headers', {})) extra_headers.setdefault('User-Agent', get_user_agent()) # OpenAI SDK type stubs incorrectly use 'in-memory' but API requires 'in_memory', so we have to use `Any` to not hit type errors prompt_cache_retention: Any = model_settings.get('openai_prompt_cache_retention', OMIT) # Most providers only accept one of `max_completion_tokens` (OpenAI, incl. o-series) or # `max_tokens` (e.g. OpenRouter), so the profile decides which field the `max_tokens` setting maps to. max_tokens = model_settings.get('max_tokens', OMIT) supports_max_completion_tokens = profile.get('openai_chat_supports_max_completion_tokens', True) return await self.client.chat.completions.create( model=self.model_name, messages=openai_messages, parallel_tool_calls=model_settings.get('parallel_tool_calls', OMIT) if tools else OMIT, tools=tools or OMIT, tool_choice=tool_choice or OMIT, stream=stream, stream_options=self._get_stream_options(model_settings) if stream else OMIT, stop=model_settings.get('stop_sequences', OMIT), max_completion_tokens=max_tokens if supports_max_completion_tokens else OMIT, max_tokens=OMIT if supports_max_completion_tokens else max_tokens, timeout=to_httpx2_timeout(model_settings.get('timeout', NOT_GIVEN)), response_format=response_format or OMIT, seed=model_settings.get('seed', OMIT), reasoning_effort=self._translate_thinking(model_settings, model_request_parameters), user=model_settings.get('openai_user', OMIT), web_search_options=web_search_options or OMIT, service_tier=_resolve_openai_service_tier(model_settings), prediction=model_settings.get('openai_prediction', OMIT), temperature=model_settings.get('temperature', OMIT), top_p=model_settings.get('top_p', OMIT), presence_penalty=model_settings.get('presence_penalty', OMIT), frequency_penalty=model_settings.get('frequency_penalty', OMIT), logit_bias=model_settings.get('logit_bias', OMIT), logprobs=model_settings.get('openai_logprobs', OMIT), top_logprobs=model_settings.get('openai_top_logprobs', OMIT), store=model_settings.get('openai_store', OMIT), moderation=model_settings.get('openai_moderation', OMIT), prompt_cache_key=model_settings.get('openai_prompt_cache_key', OMIT), prompt_cache_retention=prompt_cache_retention, prompt_cache_options=model_settings.get('openai_prompt_cache_options', OMIT), extra_headers=extra_headers, extra_body=model_settings.get('extra_body'), ) except APIStatusError as e: if model_response := _check_azure_content_filter(e, self.client, self.system, self.model_name): return model_response raise def _validate_completion(self, response: chat.ChatCompletion) -> _ChatCompletion: """Hook that validates chat completions before processing. This method may be overridden by subclasses of `OpenAIChatModel` to apply custom completion validations. """ return _ChatCompletion.model_validate(response.model_dump()) def _process_provider_details(self, response: chat.ChatCompletion) -> dict[str, Any] | None: """Hook that response content to provider details. This method may be overridden by subclasses of `OpenAIChatModel` to apply custom mappings. """ return _map_provider_details(response.choices[0]) def _process_response(self, response: chat.ChatCompletion | str) -> ModelResponse: """Process a non-streamed response, and prepare a message to return.""" # Although the OpenAI SDK claims to return a Pydantic model (`ChatCompletion`) from the chat completions function: # * it hasn't actually performed validation (presumably they're creating the model with `model_construct` or something?!) # * if the endpoint returns plain text, the return type is a string # Thus we validate it fully here. if not isinstance(response, chat.ChatCompletion): raise UnexpectedModelBehavior( f'Invalid response from {self.system} chat completions endpoint, expected JSON data' ) timestamp = _now_utc() if not response.created: response.created = int(timestamp.timestamp()) # Workaround for local Ollama which sometimes returns a `None` finish reason. if response.choices and (choice := response.choices[0]) and choice.finish_reason is None: # pyright: ignore[reportUnnecessaryComparison] choice.finish_reason = 'stop' try: response = self._validate_completion(response) except ValidationError as e: raise UnexpectedModelBehavior(f'Invalid response from {self.system} chat completions endpoint: {e}') from e choice = response.choices[0] # Moderation is a top-level field, so it's read here rather than in the choice-scoped # `_process_provider_details` hook that subclasses may override. provider_details = self._process_provider_details(response) or {} if response.moderation: provider_details['moderation'] = response.moderation.model_dump() # Handle refusal responses (structured output safety filter) if choice.message.refusal: provider_details.pop('finish_reason', None) provider_details['refusal'] = choice.message.refusal if response.created: # pragma: no branch provider_details['timestamp'] = number_to_datetime(response.created) return ModelResponse( parts=[], usage=self._map_usage(response), model_name=response.model, timestamp=_now_utc(), provider_details=provider_details or None, provider_response_id=response.id, provider_name=self._provider.name, provider_url=self._provider.base_url, finish_reason='content_filter', ) items: list[ModelResponsePart] = [] if thinking_parts := self._process_thinking(choice.message): items.extend(thinking_parts) if choice.message.content: items.extend( (replace(part, id='content', provider_name=self.system) if isinstance(part, ThinkingPart) else part) for part in split_content_into_text_and_thinking( choice.message.content, self.profile.get('thinking_tags', DEFAULT_THINKING_TAGS) ) ) if choice.message.tool_calls is not None: for c in choice.message.tool_calls: if isinstance(c, ChatCompletionMessageFunctionToolCall): part = ToolCallPart(c.function.name, c.function.arguments, tool_call_id=c.id) elif isinstance(c, ChatCompletionMessageCustomToolCall): # pragma: no cover # NOTE: Custom tool calls are not supported. # See for more details. raise RuntimeError('Custom tool calls are not supported') else: assert_never(c) part.tool_call_id = _guard_tool_call_id(part) items.append(part) if response.created: # pragma: no branch provider_details['timestamp'] = number_to_datetime(response.created) return ModelResponse( parts=items, usage=self._map_usage(response), model_name=response.model, timestamp=timestamp, provider_details=provider_details or None, provider_response_id=response.id, provider_name=self._provider.name, provider_url=self._provider.base_url, finish_reason=self._map_finish_reason(choice.finish_reason), ) def _process_thinking(self, message: chat.ChatCompletionMessage) -> list[ThinkingPart] | None: """Hook that maps reasoning tokens to thinking parts. This method may be overridden by subclasses of `OpenAIChatModel` to apply custom mappings. """ profile = self.profile custom_field = profile.get('openai_chat_thinking_field', None) items: list[ThinkingPart] = [] # Prefer the configured custom reasoning field, if present in profile. # Fall back to built-in fields if no custom field result was found. # The `reasoning_content` field is typically present in DeepSeek and Moonshot models. # https://api-docs.deepseek.com/guides/reasoning_model # The `reasoning` field is typically present in gpt-oss via Ollama and OpenRouter. # - https://cookbook.openai.com/articles/gpt-oss/handle-raw-cot#chat-completions-api # - https://openrouter.ai/docs/use-cases/reasoning-tokens#basic-usage-with-reasoning-tokens for field_name in (custom_field, 'reasoning', 'reasoning_content'): if not field_name: continue reasoning: object = getattr(message, field_name, None) if not reasoning: continue if not isinstance(reasoning, str): warnings.warn( f'Unexpected non-string value for {field_name!r}: {type(reasoning).__name__}. ' 'Please open an issue at https://github.com/pydantic/pydantic-ai/issues.', UserWarning, ) continue items.append(ThinkingPart(id=field_name, content=reasoning, provider_name=self.system)) return items return items or None async def _process_streamed_response( self, response: AsyncStream[ChatCompletionChunk], model_request_parameters: ModelRequestParameters, model_settings: OpenAIChatModelSettings | None = None, ) -> OpenAIStreamedResponse: """Process a streamed response, and prepare a streaming response to return.""" peekable_response: _utils.PeekableAsyncStream[ChatCompletionChunk, AsyncStream[ChatCompletionChunk]] = ( _utils.PeekableAsyncStream(response) ) with _map_api_errors(self.model_name, self._provider.model_id_namespace): first_chunk = await peekable_response.peek() if isinstance(first_chunk, _utils.Unset): raise UnexpectedModelBehavior( # pragma: no cover 'Streamed response ended without content or tool calls' ) # When using Azure OpenAI and a content filter is enabled, the first chunk will contain a `''` model name, # so we set it from a later chunk in `OpenAIChatStreamedResponse`. model_name = first_chunk.model or self.model_name return self._streamed_response_cls( model_request_parameters=model_request_parameters, _model_name=model_name, _model_profile=self.profile, _response=peekable_response, _provider_name=self._provider.name, _model_id_namespace=self._provider.model_id_namespace, _provider_url=self._provider.base_url, _model_settings=model_settings, ) @property def _streamed_response_cls(self) -> type[OpenAIStreamedResponse]: """Returns the `StreamedResponse` type that will be used for streamed responses. This method may be overridden by subclasses of `OpenAIChatModel` to provide their own `StreamedResponse` type. """ return OpenAIStreamedResponse def _map_usage(self, response: chat.ChatCompletion) -> usage.RequestUsage: return _map_usage(response, self._provider.name, self._provider.base_url, self.model_name) def _get_tool_choice( self, model_settings: OpenAIChatModelSettings, model_request_parameters: ModelRequestParameters, ) -> tuple[list[chat.ChatCompletionToolParam], ChatCompletionToolChoiceOptionParam | None]: """Determine which tools to send and the API tool_choice value. Returns: A tuple of (filtered_tools, tool_choice). """ resolved_tool_choice = resolve_tool_choice(model_settings, model_request_parameters) tool_defs = model_request_parameters.declared_tool_defs tool_choice: ChatCompletionToolChoiceOptionParam if resolved_tool_choice in ('auto', 'none'): tool_choice = resolved_tool_choice elif resolved_tool_choice == 'required': supports = self._supports_tool_forcing( model_settings, model_request_parameters, resolved_tool_choice, "tool_choice='required'", ) tool_choice = 'required' if supports else 'auto' elif isinstance(resolved_tool_choice, tuple): tool_choice_mode, tool_names = resolved_tool_choice supports = self._supports_tool_forcing(model_settings, model_request_parameters, resolved_tool_choice) if tool_choice_mode == 'required' and len(tool_names) == 1: if supports: tool_choice = {'type': 'function', 'function': {'name': next(iter(tool_names))}} else: # Forcing not supported: filter so the model can only see the requested tool. # Breaks caching, but OpenAI Chat doesn't support limiting tools via API arg. tool_defs = {k: v for k, v in tool_defs.items() if k in tool_names} tool_choice = 'auto' else: # Breaks caching, but OpenAI Chat doesn't support limiting tools via API arg tool_defs = {k: v for k, v in tool_defs.items() if k in tool_names} tool_choice = 'auto' if tool_choice_mode == 'auto' or not supports else tool_choice_mode else: assert_never(resolved_tool_choice) tools: list[chat.ChatCompletionToolParam] = [ self._map_tool_definition(t, model_settings) for t in tool_defs.values() ] if not tools: return tools, None return tools, tool_choice def _supports_tool_forcing( self, model_settings: OpenAIChatModelSettings, model_request_parameters: ModelRequestParameters, resolved_tool_choice: ResolvedToolChoice, context: str = 'forcing specific tools', ) -> bool: """Allow provider subclasses to express conditional forcing support. Overrides should raise `UserError` when the user explicitly requested forcing. """ return _support_tool_forcing(self.model_name, self.profile, model_settings, model_request_parameters) def _get_stream_options(self, model_settings: OpenAIChatModelSettings) -> chat.ChatCompletionStreamOptionsParam: """Build stream_options for the API request. Returns a dict with include_usage=True and optionally continuous_usage_stats if configured. """ options: dict[str, bool] = {'include_usage': True} if model_settings.get('openai_continuous_usage_stats'): options['continuous_usage_stats'] = True return cast(chat.ChatCompletionStreamOptionsParam, options) def _get_web_search_options(self, model_request_parameters: ModelRequestParameters) -> WebSearchOptions | None: for tool in model_request_parameters.native_tools: if isinstance(tool, WebSearchTool): # pragma: no branch if tool.user_location: return WebSearchOptions( search_context_size=tool.search_context_size, user_location=WebSearchOptionsUserLocation( type='approximate', approximate=WebSearchOptionsUserLocationApproximate(**tool.user_location), ), ) return WebSearchOptions(search_context_size=tool.search_context_size) return None @dataclass class _MapModelResponseContext: """Context object for mapping a `ModelResponse` to OpenAI chat completion parameters. This class is designed to be subclassed to add new fields for custom logic, collecting various parts of the model response (like text and tool calls) to form a single assistant message. """ _model: OpenAIChatModel texts: list[str] = field(default_factory=list[str]) thinkings: dict[str, list[str]] = field(default_factory=dict[str, list[str]]) tool_calls: list[ChatCompletionMessageFunctionToolCallParam] = field( default_factory=list[ChatCompletionMessageFunctionToolCallParam] ) def map_assistant_message(self, message: ModelResponse) -> chat.ChatCompletionAssistantMessageParam | None: for item in message.parts: if isinstance(item, TextPart): self._map_response_text_part(item) elif isinstance(item, ThinkingPart): self._map_response_thinking_part(item) elif isinstance(item, ToolCallPart): self._map_response_tool_call_part(item) elif isinstance(item, NativeToolCallPart | NativeToolReturnPart): # pragma: no cover self._map_response_builtin_part(item) elif isinstance(item, FilePart): # pragma: no cover self._map_response_file_part(item) elif isinstance(item, CompactionPart): # pragma: no cover # Compaction parts are not sent back to the Chat Completions API. pass elif isinstance(item, SpeechPart): # pragma: no cover # Unconverted realtime speech; `prepare_messages` turns these into `TextPart`s in `Model.prepare_messages`. raise _unconverted_speech_part_error() else: assert_never(item) return self._into_message_param() def _into_message_param(self) -> chat.ChatCompletionAssistantMessageParam | None: """Converts the collected texts and tool calls into a single OpenAI `ChatCompletionAssistantMessageParam`. This method serves as a hook that can be overridden by subclasses to implement custom logic for how collected parts are transformed into the final message parameter. Returns: An OpenAI `ChatCompletionAssistantMessageParam` representing the assistant's response, or `None` if there is nothing to send (e.g. the `ModelResponse` had no parts because the model returned an empty response). Returning `None` ensures we don't emit an assistant message with `content=None` and no `tool_calls`, which the Chat Completions API rejects with a 400 error. """ if not self.texts and not self.tool_calls: return None message_param = chat.ChatCompletionAssistantMessageParam(role='assistant') # Chat Completions stores one `content` string per assistant message, so multiple # TextParts in a single response (text after a tool call in the same stream, or # history produced by a different model) are joined here: if self.thinkings: for field_name, contents in self.thinkings.items(): message_param[field_name] = '\n\n'.join(contents) if self.texts: message_param['content'] = '\n\n'.join(self.texts) else: message_param['content'] = None if self.tool_calls: message_param['tool_calls'] = self.tool_calls return message_param def _map_response_text_part(self, item: TextPart) -> None: """Maps a `TextPart` to the response context. This method serves as a hook that can be overridden by subclasses to implement custom logic for handling text parts. """ self.texts.append(item.content) def _map_response_thinking_part(self, item: ThinkingPart) -> None: """Maps a `ThinkingPart` to the response context. This method serves as a hook that can be overridden by subclasses to implement custom logic for handling thinking parts. """ profile = self._model.profile include_method = profile.get('openai_chat_send_back_thinking_parts', 'auto') # Auto-detect: if thinking came from a custom field and from the same provider, use field mode # id='content' means it came from tags in content, not a custom field if include_method == 'auto': # Check if thinking came from a custom field from the same provider custom_field = profile.get('openai_chat_thinking_field', None) matches_custom_field = (not custom_field) or (item.id == custom_field) if ( item.id and item.id != 'content' and item.provider_name == self._model.system and matches_custom_field ): # Store both content and field name for later use in _into_message_param self.thinkings.setdefault(item.id, []).append(item.content) else: # Fall back to tags mode start_tag, end_tag = self._model.profile.get('thinking_tags', DEFAULT_THINKING_TAGS) self.texts.append('\n'.join([start_tag, item.content, end_tag])) elif include_method == 'tags': start_tag, end_tag = self._model.profile.get('thinking_tags', DEFAULT_THINKING_TAGS) self.texts.append('\n'.join([start_tag, item.content, end_tag])) elif include_method == 'field': field = profile.get('openai_chat_thinking_field', None) if field: # pragma: no branch self.thinkings.setdefault(field, []).append(item.content) def _map_response_tool_call_part(self, item: ToolCallPart) -> None: """Maps a `ToolCallPart` to the response context. This method serves as a hook that can be overridden by subclasses to implement custom logic for handling tool call parts. """ self.tool_calls.append(self._model._map_tool_call(item)) def _map_response_builtin_part(self, item: NativeToolCallPart | NativeToolReturnPart) -> None: """Maps a built-in tool call or return part to the response context. This method serves as a hook that can be overridden by subclasses to implement custom logic for handling built-in tool parts. """ # OpenAI doesn't return built-in tool calls pass def _map_response_file_part(self, item: FilePart) -> None: """Maps a `FilePart` to the response context. This method serves as a hook that can be overridden by subclasses to implement custom logic for handling file parts. """ # Files generated by models are not sent back to models that don't themselves generate files. pass def _map_model_response(self, message: ModelResponse) -> chat.ChatCompletionMessageParam | None: """Hook that determines how `ModelResponse` is mapped into `ChatCompletionMessageParam` objects before sending. Returns `None` to skip emitting any message for this `ModelResponse` (e.g. when it has no parts). Subclasses of `OpenAIChatModel` may override this method to provide their own mapping logic. """ return self._MapModelResponseContext(self).map_assistant_message(message) def _map_finish_reason( self, key: Literal['stop', 'length', 'tool_calls', 'content_filter', 'function_call'] ) -> FinishReason | None: """Hooks that maps a finish reason key to a [FinishReason][pydantic_ai.messages.FinishReason]. This method may be overridden by subclasses of `OpenAIChatModel` to accommodate custom keys. """ return _CHAT_FINISH_REASON_MAP.get(key) async def _map_messages( self, messages: Sequence[ModelMessage], model_request_parameters: ModelRequestParameters, *, model_settings: ModelSettings | None = None, ) -> list[chat.ChatCompletionMessageParam]: """Just maps a `pydantic_ai.Message` to a `openai.types.ChatCompletionMessageParam`.""" openai_messages: list[chat.ChatCompletionMessageParam] = [] profile = self.profile # DeepSeek-style `'field'` providers 400 on a tool-calling assistant turn that omits their # thinking field while the run is reasoning. The framework-synthesized deferred-capability # tool-search turn is the common trigger (it carries no thinking), but the backfill below # deliberately covers any such turn missing the field. Resolve the field name to set empty below. thinking_active = any( isinstance(part, ThinkingPart) for message in messages if isinstance(message, ModelResponse) for part in message.parts ) backfill_field = ( profile.get('openai_chat_thinking_field', None) if thinking_active and profile.get('openai_chat_send_back_thinking_parts', 'auto') == 'field' else None ) for message in messages: if isinstance(message, ModelRequest): async for item in self._map_user_message(message): openai_messages.append(item) elif isinstance(message, ModelResponse): if (mapped := self._map_model_response(message)) is not None: if ( backfill_field and mapped['role'] == 'assistant' and mapped.get('tool_calls') and backfill_field not in mapped ): mapped[backfill_field] = '' openai_messages.append(mapped) else: assert_never(message) system_prompt_role = profile.get('openai_system_prompt_role', None) or 'system' if instruction_parts := self._get_instruction_parts(messages, model_request_parameters): system_prompt_count = next( (i for i, m in enumerate(openai_messages) if m.get('role') != system_prompt_role), len(openai_messages) ) if system_prompt_role == 'developer': instruction_messages: list[chat.ChatCompletionMessageParam] = [ chat.ChatCompletionDeveloperMessageParam(role='developer', content=part.content) for part in instruction_parts ] elif system_prompt_role == 'user': instruction_messages = [ chat.ChatCompletionUserMessageParam(role='user', content=part.content) for part in instruction_parts ] else: instruction_messages = [ chat.ChatCompletionSystemMessageParam(role='system', content=part.content) for part in instruction_parts ] openai_messages[system_prompt_count:system_prompt_count] = instruction_messages if not self.profile.get('openai_chat_supports_multiple_system_messages', True): openai_messages = _merge_leading_system_messages(openai_messages, system_prompt_role) return openai_messages @staticmethod def _map_tool_call(t: ToolCallPart) -> ChatCompletionMessageFunctionToolCallParam: return ChatCompletionMessageFunctionToolCallParam( id=_guard_tool_call_id(t=t), type='function', function={'name': t.tool_name, 'arguments': t.args_as_json_str()}, ) def _map_json_schema(self, o: OutputObjectDefinition) -> chat.completion_create_params.ResponseFormat: response_format_param: chat.completion_create_params.ResponseFormatJSONSchema = { # pyright: ignore[reportPrivateImportUsage] 'type': 'json_schema', 'json_schema': {'name': o.name or DEFAULT_OUTPUT_TOOL_NAME, 'schema': o.json_schema}, } if o.description: response_format_param['json_schema']['description'] = o.description if self.profile.get('openai_supports_strict_tool_definition', True): # pragma: no branch response_format_param['json_schema']['strict'] = o.strict return response_format_param def _map_tool_definition(self, f: ToolDefinition, model_settings: ModelSettings) -> chat.ChatCompletionToolParam: """Map a tool definition to an OpenAI tool parameter. This method may be overridden by subclasses to apply custom tool mappings. `model_settings` is typed as the base `ModelSettings` so subclass overrides can read provider-specific keys without narrowing the type. """ tool_param: chat.ChatCompletionToolParam = { 'type': 'function', 'function': { 'name': f.name, 'description': f.description or '', 'parameters': f.parameters_json_schema, }, } if f.strict and self.profile.get('openai_supports_strict_tool_definition', True): tool_param['function']['strict'] = f.strict return tool_param async def _map_user_message(self, message: ModelRequest) -> AsyncIterable[chat.ChatCompletionMessageParam]: file_content: list[UserContent] = [] for part in message.parts: if isinstance(part, SystemPromptPart): system_prompt_role = self.profile.get('openai_system_prompt_role', None) if system_prompt_role == 'developer': yield chat.ChatCompletionDeveloperMessageParam(role='developer', content=part.content) elif system_prompt_role == 'user': yield chat.ChatCompletionUserMessageParam(role='user', content=part.content) else: yield chat.ChatCompletionSystemMessageParam(role='system', content=part.content) elif isinstance(part, UserPromptPart): yield await self._map_user_prompt(part) elif isinstance(part, ToolReturnPart): tool_text, tool_file_content = part.model_response_str_and_user_content() file_content.extend(tool_file_content) yield chat.ChatCompletionToolMessageParam( role='tool', tool_call_id=_guard_tool_call_id(t=part), content=tool_text, ) elif isinstance(part, RetryPromptPart): if part.tool_name is None: yield chat.ChatCompletionUserMessageParam(role='user', content=part.model_response()) else: yield chat.ChatCompletionToolMessageParam( role='tool', tool_call_id=_guard_tool_call_id(t=part), content=part.model_response(), ) elif isinstance(part, ToolAvailabilityDeltaPart): # pragma: no cover raise _unsynthesized_tool_availability_delta_error() elif isinstance(part, SpeechPart): # Unconverted realtime speech; `prepare_messages` turns these into `UserPromptPart`s in `Model.prepare_messages`. raise _unconverted_speech_part_error() else: assert_never(part) if file_content: yield await self._map_user_prompt(UserPromptPart(content=file_content)) async def _map_image_url_item(self, item: ImageUrl) -> ChatCompletionContentPartImageParam: """Map an ImageUrl to a chat completion image content part.""" image_url: ImageURL = {'url': item.url} if metadata := item.vendor_metadata: image_url['detail'] = metadata.get('detail', 'auto') if item.force_download: image_content = await download_item(item, data_format='base64_uri', type_format='extension') image_url['url'] = image_content['data'] return ChatCompletionContentPartImageParam(image_url=image_url, type='image_url') async def _map_binary_content_item(self, item: BinaryContent) -> ChatCompletionContentPartParam: """Map a BinaryContent item to a chat completion content part.""" profile = self.profile if _is_text_like_media_type(item.media_type): # Inline text-like binary content as a text block return self._inline_text_file_part( item.data.decode('utf-8'), media_type=item.media_type, identifier=item.identifier, ) elif item.is_image: image_url = ImageURL(url=item.data_uri) if metadata := item.vendor_metadata: image_url['detail'] = metadata.get('detail', 'auto') return ChatCompletionContentPartImageParam(image_url=image_url, type='image_url') elif item.is_audio: assert item.format in ('wav', 'mp3') if profile.get('openai_chat_audio_input_encoding', 'base64') == 'uri': audio = InputAudio(data=item.data_uri, format=item.format) else: audio = InputAudio(data=item.base64, format=item.format) return ChatCompletionContentPartInputAudioParam(input_audio=audio, type='input_audio') elif item.is_document: if not profile.get('openai_chat_supports_document_input', True): self._raise_document_input_not_supported_error() return File( file=FileFile( file_data=item.data_uri, filename=f'filename.{item.format}', ), type='file', ) elif item.is_video: raise NotImplementedError('VideoUrl is not supported in OpenAI Chat Completions user prompts') else: # pragma: no cover raise RuntimeError(f'Unsupported binary content type: {item.media_type}') async def _map_audio_url_item(self, item: AudioUrl) -> ChatCompletionContentPartInputAudioParam: """Map an AudioUrl to a chat completion audio content part.""" profile = self.profile data_format = 'base64_uri' if profile.get('openai_chat_audio_input_encoding', 'base64') == 'uri' else 'base64' downloaded_item = await download_item(item, data_format=data_format, type_format='extension') assert downloaded_item['data_type'] in ( 'wav', 'mp3', ), f'Unsupported audio format: {downloaded_item["data_type"]}' audio = InputAudio(data=downloaded_item['data'], format=downloaded_item['data_type']) return ChatCompletionContentPartInputAudioParam(input_audio=audio, type='input_audio') async def _map_document_url_item(self, item: DocumentUrl) -> ChatCompletionContentPartParam: """Map a DocumentUrl to a chat completion content part.""" profile = self.profile # OpenAI Chat API's FileFile only supports base64-encoded data, not URLs. # Some providers (e.g., OpenRouter) support URLs via the profile flag. if not item.force_download and profile.get('openai_chat_supports_file_urls', False): return File( file=FileFile( file_data=item.url, filename=f'filename.{item.format}', ), type='file', ) if _is_text_like_media_type(item.media_type): downloaded_text = await download_item(item, data_format='text') return self._inline_text_file_part( downloaded_text['data'], media_type=item.media_type, identifier=item.identifier, ) else: if not profile.get('openai_chat_supports_document_input', True): self._raise_document_input_not_supported_error() downloaded_item = await download_item(item, data_format='base64_uri', type_format='extension') return File( file=FileFile( file_data=downloaded_item['data'], filename=f'filename.{downloaded_item["data_type"]}', ), type='file', ) async def _map_video_url_item(self, item: VideoUrl) -> ChatCompletionContentPartParam: """Map a VideoUrl to a chat completion content part.""" raise NotImplementedError('VideoUrl is not supported in OpenAI Chat Completions user prompts') async def _map_content_item( self, item: str | TextContent | ImageUrl | BinaryContent | AudioUrl | DocumentUrl | VideoUrl | UploadedFile | CachePoint, ) -> ChatCompletionContentPartParam | None: """Map a single content item to a chat completion content part, or None to filter it out.""" if isinstance(item, str | TextContent): text = item if isinstance(item, str) else item.content return ChatCompletionContentPartTextParam(text=text, type='text') elif isinstance(item, ImageUrl): return await self._map_image_url_item(item) elif isinstance(item, BinaryContent): return await self._map_binary_content_item(item) elif isinstance(item, AudioUrl): return await self._map_audio_url_item(item) elif isinstance(item, DocumentUrl): return await self._map_document_url_item(item) elif isinstance(item, VideoUrl): return await self._map_video_url_item(item) elif isinstance(item, UploadedFile): self._validate_uploaded_file_provider(item) if item.media_type.startswith('image/'): # Chat Completions can only reference an uploaded file as a document `file` part; # `image_url` parts take a URL/data URI, not a `file_id`, so images can't be referenced by id. raise UserError( 'Referencing an uploaded image by `file_id` is not supported by OpenAIChatModel. ' 'Use `ImageUrl` or `BinaryContent` for images, or use `OpenAIResponsesModel`.' ) return File( file=FileFile(file_id=item.file_id), type='file', ) elif isinstance(item, CachePoint): # Cache points are handled by `_map_user_prompt_content_item()` when supported. return None else: assert_never(item) async def _map_user_prompt_content_item( self, item: UserContent, content: list[ChatCompletionContentPartParam] ) -> None: """Map a single user-prompt content item onto the outgoing `content` list. Stable protected hook: subclasses override this to intercept user-prompt content items before the default mapping (e.g. `OpenRouterModel` translates `CachePoint` into a `cache_control` breakpoint on the preceding part). """ if isinstance(item, CachePoint) and self.profile.get('openai_supports_prompt_cache_breakpoints', False): _add_openai_prompt_cache_breakpoint(content) else: mapped_item = await self._map_content_item(item) if mapped_item is not None: content.append(mapped_item) async def _map_user_prompt(self, part: UserPromptPart) -> chat.ChatCompletionUserMessageParam: content: str | list[ChatCompletionContentPartParam] if isinstance(part.content, str): content = part.content else: content = [] for item in part.content: await self._map_user_prompt_content_item(item, content) return chat.ChatCompletionUserMessageParam(role='user', content=content) def _raise_document_input_not_supported_error(self) -> Never: if _is_azure(self.client, self.system): raise UserError( "Azure's Chat Completions API does not support document input. " 'Use `OpenAIResponsesModel` with `AzureProvider` instead.' ) raise UserError( f'The {self._provider.name!r} provider does not support document input via the Chat Completions API.' ) @staticmethod def _inline_text_file_part(text: str, *, media_type: str, identifier: str) -> ChatCompletionContentPartTextParam: return ChatCompletionContentPartTextParam( text=_format_inlined_text_file(text, media_type=media_type, identifier=identifier), type='text', ) responses_output_text_annotations_ta = TypeAdapter(list[responses.response_output_text.Annotation]) @dataclass(init=False) class OpenAIResponsesModel(Model[AsyncOpenAI]): """A model that uses the OpenAI Responses API. The [OpenAI Responses API](https://platform.openai.com/docs/api-reference/responses) is the new API for OpenAI models. If you are interested in the differences between the Responses API and the Chat Completions API, see the [OpenAI API docs](https://platform.openai.com/docs/guides/responses-vs-chat-completions). """ supported_tool_deferral_modes = frozenset({'with_tool_search'}) supported_tool_addition_modes = frozenset({'with_definitions'}) # Responses only: `OpenAIChatModel` shares this model family's profile but has no compaction # surface, and the Responses API bills replayed pre-boundary items, so the trim is what makes # compaction actually compact here. # # The compaction item is an opaque blob the API decrypts; without it there is nothing to send in # place of the history the boundary would hide. compaction_requires_encrypted_content = True # `SystemPromptPart`s render as `system` input items *inside* the compacted window, and the item # keeps serving them: a latent directive that never fired before the boundary still governs # post-compaction replies without being re-sent (live-verified). compaction_retains_standing_prompt = True _model_name: OpenAIModelName = field(repr=False) _provider: Provider[AsyncOpenAI] = field(repr=False) def __init__( self, model_name: OpenAIModelName, *, provider: OpenAIResponsesCompatibleProvider | Literal[ 'openai', 'gateway', ] | Provider[AsyncOpenAI] = 'openai', profile: ModelProfileSpec | None = None, settings: ModelSettings | None = None, ): """Initialize an OpenAI Responses model. Args: model_name: The name of the OpenAI model to use. provider: The provider to use. Defaults to `'openai'`. profile: The model profile to use. Defaults to a profile picked by the provider based on the model name. settings: Default model settings for this model instance. """ self._model_name = model_name if isinstance(provider, str): provider = infer_provider('gateway/openai' if provider == 'gateway' else provider) self._provider = provider super().__init__(settings=settings, profile=profile) _preload_openai_sdk_resource_modules(self, self._provider.client) @property def client(self) -> AsyncOpenAI: return self._provider.client @property def base_url(self) -> str: return str(self.client.base_url) @property def model_name(self) -> OpenAIModelName: """The model name.""" return self._model_name def resolve_prompt_cache_retention(self, model_settings: ModelSettings | None) -> timedelta | None: """Resolve the extended prompt cache retention requested by OpenAI settings.""" return _resolve_prompt_cache_retention(self.settings, model_settings) @property def system(self) -> str: """The model provider.""" return self._provider.name async def cancel_suspended_response(self, response: ModelResponse) -> None: """Cancel a suspended background response by cancelling its server-side job. `responses.cancel` only applies to background responses; calling it on an ordinary (foreground) response returns a 400. The `provider_details['background']` marker is stamped from the API's own `response.background` field (see `_process_response` and `OpenAIResponsesStreamedResponse._track_background`), so it explicitly and reliably distinguishes a cancellable background job from a normal streamed response that happens to be interrupted by `cancel()` — no need to infer background mode from the `continuation_delay` poll interval. """ if ( (response.provider_details or {}).get('background') and response.provider_response_id and response.provider_name == self.system ): with _map_api_errors(self.model_name, self._provider.model_id_namespace): await self.client.responses.cancel(response.provider_response_id) def continuation_delay(self, response: ModelResponse) -> float | None: if response.state == 'suspended' and (response.provider_details or {}).get('background'): return _OPENAI_BACKGROUND_POLL_INTERVAL return None @cached_property def profile(self) -> OpenAIModelProfile: profile = cast(OpenAIModelProfile, super().profile) # A model can be more capable on the Responses API than on Chat Completions, and the two # share one profile: DeepSeek honors `text.format` of type `json_schema` here while its # Chat Completions endpoint rejects it, so the generic flag can only be raised per model class. if profile.get('openai_responses_supports_json_schema_output', False): return cast(OpenAIModelProfile, merge_profile(profile, ModelProfile(supports_json_schema_output=True))) return profile @classmethod def supported_native_tools(cls) -> frozenset[type[AbstractNativeTool]]: """Return the set of builtin tool types this model can handle.""" return frozenset( {WebSearchTool, CodeExecutionTool, FileSearchTool, MCPServerTool, ImageGenerationTool, ToolSearchTool} ) async def compact_messages( self, request_context: ModelRequestContext, *, instructions: str | None = None, ) -> ModelResponse: """Compact messages using the OpenAI Responses compaction endpoint. This calls OpenAI's `responses.compact` API to produce an encrypted compaction that summarizes the conversation history. The returned `ModelResponse` contains a single `CompactionPart` that must be round-tripped in subsequent requests. Args: request_context: The model request context containing messages, settings, and parameters. instructions: Optional custom instructions for the compaction summarization. If provided, these override the agent-level instructions. Returns: A `ModelResponse` with a single `CompactionPart` containing the encrypted compaction data. """ check_allow_model_requests() model_settings, model_request_parameters = self.prepare_request( request_context.model_settings, request_context.model_request_parameters, ) response = await self._responses_compact( request_context.messages, cast(OpenAIResponsesModelSettings, model_settings or {}), model_request_parameters, instructions_override=instructions, ) # Handle ModelResponse (e.g. from content filter) if isinstance(response, ModelResponse): # pragma: no cover return response if not response.output: # pragma: no cover raise UnexpectedModelBehavior('CompactedResponse returned with no output items') compaction = response.output[-1] if not isinstance(compaction, ResponseCompactionItem): # pragma: no cover raise UnexpectedModelBehavior(f'Last item in response is not a compaction, got: {compaction.type}') # This compact call's input window explicitly planted the standing prompt (see # `_responses_compact`), so the minted part carries the provenance stamp that lets the # trim rely on the item's retention instead of re-sending the standing prompt. part = _map_compaction_item(compaction, self.system, standing_prompt_planted=True) return ModelResponse( parts=[part], usage=_map_usage(response, self._provider.name, self._provider.base_url, self.model_name), model_name=self._model_name, provider_response_id=response.id, # Marks this ModelResponse as coming from the stateless `/compact` endpoint. # `_get_previous_response_id_and_new_messages` uses this to break the auto-chain, # since compaction response ids cannot be used as `previous_response_id`. provider_details={'compaction': True}, timestamp=_now_utc(), provider_name=self._provider.name, provider_url=self._provider.base_url, ) async def _responses_compact( self, messages: list[ModelMessage], model_settings: OpenAIResponsesModelSettings, model_request_parameters: ModelRequestParameters, *, instructions_override: str | None = None, ) -> responses.CompactedResponse | ModelResponse: """Call the OpenAI Responses compaction endpoint.""" previous_response_id, messages = self._resolve_previous_response_id( model_settings.get('openai_previous_response_id'), messages, allow_no_new_messages=True ) # Same ordering rule as `_build_responses_request_params`: the introduced-tools derivation # and the mapping must both see the trimmed history, and re-compacting only compacts the # current effective window rather than content an earlier compaction already replaced. # `standing_prompt_retained=False` (here and in the mapping below): blob-of-blob retention # decayed in probing, so the window sent for re-compaction plants the standing prompt # explicitly rather than relying on the previous compaction item to carry it forward. messages = self._trim_before_compaction(messages, standing_prompt_retained=False) instructions, openai_messages = await self._map_messages( messages, model_settings, model_request_parameters, previous_response_id=previous_response_id, standing_prompt_retained=False, ) if instructions_override is not None: instructions = instructions_override try: return await self.client.responses.compact( input=openai_messages, model=self.model_name, instructions=instructions, previous_response_id=previous_response_id or OMIT, ) except APIStatusError as e: # pragma: no cover if model_response := _check_azure_content_filter(e, self.client, self.system, self.model_name): return model_response if (status_code := e.status_code) >= 400: raise ModelHTTPError( status_code=status_code, model_name=self.model_name, body=e.body, headers=dict(e.response.headers), ) from e raise except APIConnectionError as e: # pragma: no cover raise ModelAPIError(model_name=self.model_name, message=e.message) from e async def request( self, messages: list[ModelRequest | ModelResponse], model_settings: ModelSettings | None, model_request_parameters: ModelRequestParameters, ) -> ModelResponse: check_allow_model_requests() model_settings, model_request_parameters = self.prepare_request( model_settings, model_request_parameters, ) settings = cast(OpenAIResponsesModelSettings, model_settings or {}) if info := self._get_continuation_info(messages, settings): # Non-streaming retrieve: on `store=false` backends (Codex, which is also stream-only) # `_get_continuation_info` already rejected the continuation with `UserError`. response_id, _, _ = info response = await self._responses_retrieve(response_id, settings) elif self.profile.get('openai_responses_requires_streaming', False): # Stream-only backend (e.g. Codex subscription auth): drain a forced stream via the # streamed-response path, which handles `response.completed` arriving with an empty `output`. # Note: if a higher-level path to enforce streaming is ever added, this branch belongs there instead. stream = await self._responses_create( messages, stream=True, model_settings=settings, model_request_parameters=model_request_parameters ) if isinstance(stream, ModelResponse): # A handled rejection (e.g. an Azure content filter) arrives as a finished # response, not a stream; same guard as `request_stream` below. return stream async with stream: streamed_response = await self._process_streamed_response(stream, settings, model_request_parameters) async for _ in streamed_response: pass return streamed_response.get() else: response = await self._responses_create( messages, stream=False, model_settings=settings, model_request_parameters=model_request_parameters ) if isinstance(response, ModelResponse): return response return self._process_response(response, settings, model_request_parameters) async def count_tokens( self, messages: list[ModelRequest | ModelResponse], model_settings: ModelSettings | None, model_request_parameters: ModelRequestParameters, ) -> usage.RequestUsage: check_allow_model_requests() if not self.profile.get('openai_supports_input_token_counting', True): raise UserError( f'Server-side token counting is not available for {self.system} models ' f'({self.model_name!r}); the provider does not expose the input-tokens endpoint.' ) model_settings, model_request_parameters = self.prepare_request( model_settings, model_request_parameters, ) settings = cast(OpenAIResponsesModelSettings, model_settings or {}) # Validate that we have something meaningful to count tokens for. if ( not messages and not settings.get('openai_previous_response_id') and not settings.get('openai_conversation_id') ): raise UserError('Cannot count tokens without any messages or a previous response ID.') profile = self.profile request_params = await self._build_responses_request_params( messages, settings, model_request_parameters, profile, ) extra_headers, timeout = self._build_request_options(settings) with _map_api_errors(self.model_name, self._provider.model_id_namespace): response = await self.client.responses.input_tokens.count( model=request_params.model, input=request_params.input, instructions=request_params.instructions, parallel_tool_calls=request_params.parallel_tool_calls, tools=request_params.tools, tool_choice=request_params.tool_choice, previous_response_id=request_params.previous_response_id, conversation=request_params.conversation, reasoning=request_params.reasoning, text=request_params.text, truncation=request_params.truncation, extra_headers=extra_headers, extra_body=settings.get('extra_body'), timeout=timeout, ) return usage.RequestUsage( input_tokens=response.input_tokens, ) @asynccontextmanager async def request_stream( self, messages: list[ModelMessage], model_settings: ModelSettings | None, model_request_parameters: ModelRequestParameters, run_context: RunContext[Any] | None = None, ) -> AsyncGenerator[StreamedResponse]: check_allow_model_requests() model_settings, model_request_parameters = self.prepare_request( model_settings, model_request_parameters, ) settings = cast(OpenAIResponsesModelSettings, model_settings or {}) if info := self._get_continuation_info(messages, settings): response_id, last_sequence_number, previous_model_name = info expected_response_id = response_id if last_sequence_number is None: # Some background responses were not previously streamed and have no resumable # sequence cursor. `retrieve(stream=True)` can block for a long time in this case, # so fall back to non-stream retrieve and return a static streamed wrapper. response = await self._responses_retrieve(response_id, settings) sr: StreamedResponse = _ModelResponseStreamedResponse( model_request_parameters=model_request_parameters, _model_response=self._process_response(response, settings, model_request_parameters), ) yield sr return response = await self._responses_retrieve( response_id, settings, stream=True, starting_after=last_sequence_number ) else: previous_model_name = None expected_response_id = None response = await self._responses_create( messages, stream=True, model_settings=settings, model_request_parameters=model_request_parameters ) if isinstance(response, ModelResponse): yield _ModelResponseStreamedResponse( model_request_parameters=model_request_parameters, _model_response=response, ) return async with response: sr = await self._process_streamed_response( response, settings, model_request_parameters, expected_model_name=previous_model_name, expected_response_id=expected_response_id, ) yield sr def _process_response( # noqa: C901 self, response: responses.Response, model_settings: OpenAIResponsesModelSettings, model_request_parameters: ModelRequestParameters, ) -> ModelResponse: """Process a non-streamed response, and prepare a message to return.""" items: list[ModelResponsePart] = [] refusal_text: str | None = None tool_search_output_call_ids = _tool_search_output_call_ids(response) tool_search_outputs = { tool_search_output_call_ids[item.id]: item for item in response.output if isinstance(item, responses.ResponseToolSearchOutputItem) and item.execution == 'server' and item.id in tool_search_output_call_ids } paired_tool_search_output_ids = {item.id for item in tool_search_outputs.values()} for item in response.output: if isinstance(item, responses.ResponseReasoningItem): signature = item.encrypted_content # Handle raw CoT content from gpt-oss models provider_details: dict[str, Any] = {} raw_content: list[str] | None = [c.text for c in item.content] if item.content else None if raw_content: provider_details['raw_content'] = raw_content if item.summary: for summary in item.summary: # We use the same id for all summaries so that we can merge them on the round trip. items.append( ThinkingPart( content=summary.text, id=item.id, signature=signature, provider_name=self.system, provider_details=provider_details or None, ) ) # We only need to store the signature and raw_content once. signature = None provider_details = {} elif signature or provider_details: items.append( ThinkingPart( content='', id=item.id, signature=signature, provider_name=self.system, provider_details=provider_details or None, ) ) elif isinstance(item, responses.ResponseOutputMessage): for content in item.content: if isinstance(content, responses.ResponseOutputRefusal): refusal_text = content.refusal elif isinstance(content, responses.ResponseOutputText): # pragma: no branch part_provider_details: dict[str, Any] | None = None if content.logprobs: part_provider_details = {'logprobs': _map_logprobs(content.logprobs)} if model_settings.get('openai_include_raw_annotations') and content.annotations: part_provider_details = part_provider_details or {} part_provider_details['annotations'] = responses_output_text_annotations_ta.dump_python( list(content.annotations), warnings=False ) if item.phase is not None: part_provider_details = part_provider_details or {} part_provider_details['phase'] = item.phase # Some OpenAI-compatible gateways (e.g. Bifrost) return text=null; # coalesce to '' so the part (and its ID) is preserved for round-tripping. items.append( TextPart( content.text or '', id=item.id, provider_name=self.system, provider_details=part_provider_details, ) ) elif isinstance(item, responses.ResponseFunctionToolCall): # When the Responses API wraps a discovered deferred tool in a namespace # (tool-search flow), it requires the namespace to be round-tripped on # subsequent turns. Preserve it via `provider_details`. fn_provider_details: dict[str, Any] | None = None if item.namespace: fn_provider_details = {'namespace': item.namespace} items.append( ToolCallPart( item.name, item.arguments, tool_call_id=_response_tool_call_id( item.call_id, response.id if self.profile.get('openai_responses_tool_call_ids_are_response_scoped', False) else None, ), id=item.id, provider_name=self.system, provider_details=fn_provider_details, ) ) elif isinstance(item, responses.ResponseCodeInterpreterToolCall): call_part, return_part, file_parts = _map_code_interpreter_tool_call(item, self.system) items.append(call_part) if file_parts: items.extend(file_parts) items.append(return_part) elif isinstance(item, responses.ResponseFunctionWebSearch): call_part, return_part = _map_web_search_tool_call(item, self.system) items.append(call_part) items.append(return_part) elif isinstance(item, responses.ResponseToolSearchCall): if item.execution == 'client': # Client-executed: emit a regular ToolCallPart so the standard agent-graph # tool-execution path runs the local `search_tools` function. The matching # `ToolReturnPart` is produced by the tool runner, not here. items.append(_map_client_tool_search_call(item, self.system)) else: call_part = _map_tool_search_call(item, self.system) items.append(call_part) if output_item := tool_search_outputs.get(call_part.tool_call_id): items.append(_build_tool_search_return_part(call_part.tool_call_id, output_item, self.system)) elif isinstance(item, responses.ResponseToolSearchOutputItem): if item.execution == 'server' and item.id not in paired_tool_search_output_ids: items.append(_build_tool_search_return_part(item.call_id or item.id, item, self.system)) elif isinstance(item, responses.response_output_item.ImageGenerationCall): call_part, return_part, file_part = _map_image_generation_tool_call(item, self.system) items.append(call_part) if file_part: # pragma: no branch items.append(file_part) items.append(return_part) elif isinstance(item, ResponseCompactionItem): items.append(_map_compaction_item(item, self.system)) elif isinstance(item, responses.ResponseComputerToolCall): # pragma: no cover # Pydantic AI doesn't yet support the ComputerUse built-in tool pass elif isinstance(item, responses.ResponseCustomToolCall): # pragma: no cover # Support is being implemented in https://github.com/pydantic/pydantic-ai/pull/2572 pass elif isinstance(item, responses.response_output_item.LocalShellCall): # pragma: no cover # Pydantic AI doesn't yet support the `codex-mini-latest` LocalShell built-in tool pass elif isinstance(item, responses.ResponseFileSearchToolCall): call_part, return_part = _map_file_search_tool_call(item, self.system) items.append(call_part) items.append(return_part) elif isinstance(item, responses.response_output_item.McpCall): call_part, return_part = _map_mcp_call(item, self.system) items.append(call_part) items.append(return_part) elif isinstance(item, responses.response_output_item.McpListTools): call_part, return_part = _map_mcp_list_tools(item, self.system) items.append(call_part) items.append(return_part) elif isinstance(item, responses.response_output_item.McpApprovalRequest): # pragma: no cover # Pydantic AI doesn't yet support McpApprovalRequest (explicit tool usage approval) pass finish_reason: FinishReason | None = None provider_details: dict[str, Any] = {} raw_finish_reason = details.reason if (details := response.incomplete_details) else response.status if raw_finish_reason: provider_details['finish_reason'] = raw_finish_reason finish_reason = _RESPONSES_FINISH_REASON_MAP.get(raw_finish_reason) if response.created_at: # pragma: no branch provider_details['timestamp'] = number_to_datetime(response.created_at) if response.conversation: provider_details['conversation_id'] = response.conversation.id if response.background: # Mark the response as a cancellable server-side background job (see # `cancel_suspended_response`), independent of the `continuation_delay` poll interval. provider_details['background'] = True if response.moderation: provider_details['moderation'] = response.moderation.model_dump() state = _response_status_to_state(response.status, background=bool(response.background)) if refusal_text is not None: items = [] finish_reason = 'content_filter' provider_details.pop('finish_reason', None) provider_details['refusal'] = refusal_text return ModelResponse( parts=items, usage=_map_usage(response, self._provider.name, self._provider.base_url, self.model_name), model_name=response.model, provider_response_id=response.id, timestamp=_now_utc(), provider_name=self._provider.name, provider_url=self._provider.base_url, finish_reason=finish_reason, state=state, provider_details=provider_details or None, ) async def _process_streamed_response( self, response: AsyncStream[responses.ResponseStreamEvent], model_settings: OpenAIResponsesModelSettings, model_request_parameters: ModelRequestParameters, *, expected_model_name: OpenAIModelName | None = None, expected_response_id: str | None = None, ) -> OpenAIResponsesStreamedResponse: """Process a streamed response, and prepare a streaming response to return.""" peekable_response: _utils.PeekableAsyncStream[ responses.ResponseStreamEvent, AsyncStream[responses.ResponseStreamEvent] ] = _utils.PeekableAsyncStream(response) with _map_api_errors(self.model_name, self._provider.model_id_namespace): first_chunk = await peekable_response.peek() if isinstance(first_chunk, _utils.Unset): # Covered by the Codex forced-stream path, which drains empty streams through here. raise UnexpectedModelBehavior('Streamed response ended without content or tool calls') if isinstance(first_chunk, responses.ResponseCreatedEvent): model_name = first_chunk.response.model provider_timestamp = ( number_to_datetime(first_chunk.response.created_at) if first_chunk.response.created_at else None ) background = bool(first_chunk.response.background) initial_state = _response_status_to_state(first_chunk.response.status, background=background) else: # When `starting_after` is used, OpenAI may omit `response.created` and start # directly with delta events. Keep the previous model name (if known) so # continuation merging doesn't accidentally replace earlier parts. model_name = expected_model_name or self.model_name provider_timestamp = None # We only resume via `retrieve(stream=True, starting_after=...)` for a background job that # was still in progress, so start `'suspended'`: a clean EOF here without a terminal event # means it's still running and should be continued, not treated as done. background = True initial_state = 'suspended' tool_call_ids_are_response_scoped = self.profile.get( 'openai_responses_tool_call_ids_are_response_scoped', False ) streamed_response = OpenAIResponsesStreamedResponse( model_request_parameters=model_request_parameters, _model_name=model_name, _model_settings=model_settings, _response=peekable_response, _provider_name=self._provider.name, _model_id_namespace=self._provider.model_id_namespace, _provider_url=self._provider.base_url, _provider_timestamp=provider_timestamp, _tool_call_ids_are_response_scoped=tool_call_ids_are_response_scoped, ) if tool_call_ids_are_response_scoped: # Response-scoped tool-call IDs are qualified with the response ID as chunks arrive, so the # ID must be known before iteration. This is only needed (and only set eagerly) for the # providers that opt in; other Responses streams keep resolving it during iteration. streamed_response.provider_response_id = ( first_chunk.response.id if isinstance(first_chunk, responses.ResponseCreatedEvent) else expected_response_id ) streamed_response.state = initial_state if background: # Stamp the background marker up front so the retry-delay gate (and `cancel_suspended_response`) # can tell a resumable background job from a foreground stream before any event is consumed; # `_track_background` keeps it stamped as status events arrive during iteration. streamed_response.provider_details = {**(streamed_response.provider_details or {}), 'background': True} return streamed_response async def _build_responses_request_params( self, messages: list[ModelRequest | ModelResponse], model_settings: OpenAIResponsesModelSettings, model_request_parameters: ModelRequestParameters, profile: OpenAIModelProfile, ) -> _ResponsesRequestParams: """Build typed request parameters shared by Responses API calls.""" # Deliberately a separate variable: `messages` itself stays untrimmed for # `_resolve_server_side_state`, which recovers conversation/response IDs from responses the # trim would drop; `_map_messages` applies the same idempotent trim to whatever slice that # resolution hands it. trimmed_messages = self._trim_before_compaction(messages) # Call-time import mirroring `models/__init__.py`'s `_tool_search` import: the toolsets # package imports `messages` while adapters load, so a module-level import would cycle. from ..toolsets._tool_search import parse_discovered_tools # A reveal whose `additional_tools` carrier sits before the compaction boundary is trimmed # with it, while the tool's `'via_history'` visibility still promises a history item — so it # would vanish from the request entirely. Until reveal state itself becomes boundary-aware # (#7225 derives it from the post-compaction window, which makes this branch inert), such a # tool is redeclared in `tools` with its schema: the compaction turn rebuilds the prefix # anyway, so the declaration costs nothing extra. post_trim_revealed = parse_discovered_tools(trimmed_messages) wire_request_parameters = model_request_parameters lost_reveals = { name for name, visibility in (model_request_parameters.tool_visibility or {}).items() if visibility == 'via_history' and name not in post_trim_revealed } if lost_reveals: tool_visibility: dict[str, ToolVisibility] = { **(model_request_parameters.tool_visibility or {}), **dict.fromkeys(sorted(lost_reveals), 'visible'), } wire_request_parameters = replace(model_request_parameters, tool_visibility=tool_visibility) history_declared_tool_names = { tool.name for tool in wire_request_parameters.function_tools if wire_request_parameters.visibility_of(tool.name) == 'via_history' } function_tools, tool_choice = self._get_responses_tool_choice(model_settings, wire_request_parameters) extra_native_tools = model_settings.get('openai_native_tools', ()) tools: list[responses.ToolParam] = ( self._get_native_tools(wire_request_parameters) + list(extra_native_tools) + function_tools ) if not tools and not history_declared_tool_names: tool_choice = None previous_response_id, conversation_id, messages = self._resolve_server_side_state(model_settings, messages) instructions, openai_messages = await self._map_messages( messages, model_settings, wire_request_parameters, previous_response_id=previous_response_id, ) reasoning = self._translate_thinking(model_settings, model_request_parameters) text: responses.ResponseTextConfigParam | Omit = OMIT if model_request_parameters.output_mode == 'native': output_object = model_request_parameters.output_object assert output_object is not None text = {'format': self._map_json_schema(output_object)} elif model_request_parameters.output_mode == 'prompted' and profile.get( 'supports_json_object_output', False ): # pragma: no branch text = {'format': {'type': 'json_object'}} # Without this trick, we'd hit this error: # > Response input messages must contain the word 'json' in some form to use 'text.format' of type 'json_object'. # Apparently they're only checking input messages for "JSON", not instructions. assert isinstance(instructions, str) system_prompt_role = profile.get('openai_system_prompt_role', None) or 'system' system_prompt_count = next( (i for i, m in enumerate(openai_messages) if m.get('role') != system_prompt_role), len(openai_messages) ) openai_messages.insert( system_prompt_count, responses.EasyInputMessageParam(role=system_prompt_role, content=instructions) ) instructions = OMIT if verbosity := model_settings.get('openai_text_verbosity'): text_with_verbosity: responses.ResponseTextConfigParam = text if isinstance(text, dict) else {} text_with_verbosity['verbosity'] = verbosity text = text_with_verbosity # When there are no input messages and we're not reusing server-side state, # the OpenAI API will reject a request without any input, # even if there are instructions. # To avoid this provide an explicit empty user message. if not openai_messages and not previous_response_id and not conversation_id: openai_messages.append( responses.EasyInputMessageParam( role='user', content='', ) ) # `parallel_tool_calls`, `truncation`, and `context_management` ride on the request params # rather than the create call, so they must honor the unsupported-settings seam here too — # `_drop_unsupported_params` runs too late for them (and never runs for `count_tokens`). unsupported_settings = profile.get('openai_unsupported_model_settings', ()) parallel_tool_calls = OMIT if tools and 'parallel_tool_calls' not in unsupported_settings: parallel_tool_calls = model_settings.get('parallel_tool_calls', OMIT) return _ResponsesRequestParams( model=self.model_name, input=openai_messages, instructions=instructions, parallel_tool_calls=parallel_tool_calls, tools=tools or OMIT, tool_choice=tool_choice or OMIT, previous_response_id=previous_response_id or OMIT, conversation=conversation_id or OMIT, reasoning=reasoning, text=text, truncation=OMIT if 'openai_truncation' in unsupported_settings else model_settings.get('openai_truncation', OMIT), context_management=OMIT if 'openai_context_management' in unsupported_settings else model_settings.get('openai_context_management', OMIT), ) @staticmethod def _build_request_options( model_settings: OpenAIResponsesModelSettings, ) -> tuple[dict[str, str], float | HTTPX2Timeout | NotGiven]: extra_headers = dict(model_settings.get('extra_headers', {})) extra_headers.setdefault('User-Agent', get_user_agent()) timeout = to_httpx2_timeout(model_settings.get('timeout', NOT_GIVEN)) return extra_headers, timeout def _prepare_responses_settings( self, messages: list[ModelRequest | ModelResponse], model_settings: OpenAIResponsesModelSettings, ) -> OpenAIResponsesModelSettings: """Adjust the settings of a single request; subclasses override this to derive settings from the messages.""" return model_settings @overload async def _responses_create( self, messages: list[ModelRequest | ModelResponse], stream: Literal[False], model_settings: OpenAIResponsesModelSettings, model_request_parameters: ModelRequestParameters, ) -> responses.Response: ... @overload async def _responses_create( self, messages: list[ModelRequest | ModelResponse], stream: Literal[True], model_settings: OpenAIResponsesModelSettings, model_request_parameters: ModelRequestParameters, ) -> AsyncStream[responses.ResponseStreamEvent]: ... async def _responses_create( self, messages: list[ModelRequest | ModelResponse], stream: bool, model_settings: OpenAIResponsesModelSettings, model_request_parameters: ModelRequestParameters, ) -> responses.Response | AsyncStream[responses.ResponseStreamEvent] | ModelResponse: profile = self.profile include = self._build_include(model_settings) request_params = await self._build_responses_request_params( messages, model_settings, model_request_parameters, profile, ) # Both helpers mutate the settings they receive. model_settings = self._prepare_responses_settings(messages, OpenAIResponsesModelSettings(**model_settings)) _drop_sampling_params_for_reasoning(profile, model_settings, model_request_parameters) _drop_unsupported_params(profile, model_settings) store: bool | Omit | None = model_settings.get('openai_store', OMIT) if profile.get('openai_responses_requires_store_false', False): store = False extra_headers, timeout = self._build_request_options(model_settings) # OpenAI SDK type stubs incorrectly use 'in-memory' but API requires 'in_memory', so we have to use `Any` to not hit type errors prompt_cache_retention: Any = model_settings.get('openai_prompt_cache_retention', OMIT) with _map_api_errors(self.model_name, self._provider.model_id_namespace): try: return await self.client.responses.create( model=request_params.model, input=request_params.input, instructions=request_params.instructions, parallel_tool_calls=request_params.parallel_tool_calls, tools=request_params.tools, tool_choice=request_params.tool_choice, previous_response_id=request_params.previous_response_id, reasoning=request_params.reasoning, text=request_params.text, truncation=request_params.truncation, context_management=request_params.context_management, max_output_tokens=model_settings.get('max_tokens', OMIT), stream=stream, temperature=model_settings.get('temperature', OMIT), top_p=model_settings.get('top_p', OMIT), service_tier=_resolve_openai_service_tier(model_settings), conversation=request_params.conversation, top_logprobs=model_settings.get('openai_top_logprobs', OMIT), store=store, user=model_settings.get('openai_user', OMIT), include=include or OMIT, prompt_cache_key=model_settings.get('openai_prompt_cache_key', OMIT), prompt_cache_retention=prompt_cache_retention, prompt_cache_options=model_settings.get('openai_prompt_cache_options', OMIT), background=model_settings.get('openai_background', OMIT), moderation=model_settings.get('openai_moderation', OMIT), timeout=timeout, extra_headers=extra_headers, extra_body=model_settings.get('extra_body'), ) except APIStatusError as e: if model_response := _check_azure_content_filter(e, self.client, self.system, self.model_name): return model_response raise def _get_continuation_info( self, messages: list[ModelMessage], model_settings: OpenAIResponsesModelSettings ) -> tuple[str, int | None, OpenAIModelName | None] | None: """If the last message is a suspended response from this provider, return continuation metadata.""" if not messages: # pragma: lax no cover return None last = messages[-1] if not isinstance(last, ModelResponse): return None if last.provider_name != self.system: # pragma: lax no cover return None if not (last.state == 'suspended' and last.provider_response_id): # pragma: lax no cover return None if self.profile.get('openai_responses_requires_store_false', False): # Continuation retrieves the suspended response server-side, but a `store=false` # backend (e.g. Codex subscription auth) never persisted it; without this guard the # retrieve fails at resume time as a misleading `SuspendedResponseExpired` on 404. raise UserError( f'Resuming a suspended run is not supported for {self.system} models ' f'({self.model_name!r}): the backend requires `store=false`, so the suspended ' 'response was never persisted server-side.' ) details: _OpenAIResponsesContinuationDetails = cast( _OpenAIResponsesContinuationDetails, last.provider_details or {} ) last_sequence_number = details.get('last_sequence_number') return ( last.provider_response_id, last_sequence_number, cast(OpenAIModelName | None, last.model_name), ) def _build_include( self, model_settings: OpenAIResponsesModelSettings, *, is_retrieve: bool = False ) -> list[responses.ResponseIncludable]: """Build the include list for retrieve/create requests.""" profile = self.profile include: list[responses.ResponseIncludable] = [] if profile.get('openai_supports_encrypted_reasoning_content', False) and not is_retrieve: # OpenAI rejects `reasoning.encrypted_content` on any retrieve of a background # response ('Encrypted content cannot be requested for persisted responses'), # so only request it on create. Nothing is lost: a retrieved background response # never carries encrypted content, and continuation works via `previous_response_id`. include.append('reasoning.encrypted_content') if model_settings.get('openai_include_code_execution_outputs'): include.append('code_interpreter_call.outputs') if model_settings.get('openai_include_web_search_sources'): include.append('web_search_call.action.sources') if model_settings.get('openai_include_file_search_results'): include.append('file_search_call.results') if model_settings.get('openai_logprobs'): include.append('message.output_text.logprobs') return include @overload async def _responses_retrieve( self, response_id: str, model_settings: OpenAIResponsesModelSettings, *, stream: Literal[False] = False, starting_after: int | None = None, ) -> responses.Response: ... @overload async def _responses_retrieve( self, response_id: str, model_settings: OpenAIResponsesModelSettings, *, stream: Literal[True], starting_after: int | None = None, ) -> AsyncStream[responses.ResponseStreamEvent]: ... async def _responses_retrieve( self, response_id: str, model_settings: OpenAIResponsesModelSettings, *, stream: bool = False, starting_after: int | None = None, ) -> responses.Response | AsyncStream[responses.ResponseStreamEvent]: """Retrieve a background response by ID, optionally streaming.""" include = self._build_include(model_settings, is_retrieve=True) extra_headers, timeout = self._build_request_options(model_settings) with _map_api_errors(self.model_name, self._provider.model_id_namespace): try: return await self.client.responses.retrieve( response_id=response_id, include=include or OMIT, starting_after=starting_after if starting_after is not None else OMIT, stream=stream, timeout=timeout, extra_headers=extra_headers, ) except APIStatusError as e: # A retrieve is only issued to resume a persisted suspended job, so a 404 means that job is # gone (past the provider's retention window). Surface a clear typed error instead of a # generic HTTP error; other statuses fall through to `_map_api_errors`. if e.status_code == 404: raise SuspendedResponseExpired( f'The suspended response {response_id!r} could not be resumed because its server-side ' 'job is no longer available (it may have expired).' ) from e raise def _translate_thinking( self, model_settings: OpenAIResponsesModelSettings, model_request_parameters: ModelRequestParameters, ) -> Reasoning | Omit: reasoning_effort = model_settings.get('openai_reasoning_effort', None) reasoning_mode = model_settings.get('openai_reasoning_mode', None) reasoning_context = model_settings.get('openai_reasoning_context', None) reasoning_summary = model_settings.get('openai_reasoning_summary', None) # Fall back to unified thinking when openai_reasoning_effort is not set if reasoning_effort is None and (thinking := model_request_parameters.thinking) is not None: reasoning_effort = _resolve_openai_thinking_effort(thinking, self.profile) reasoning: Reasoning = {} if reasoning_effort: reasoning['effort'] = reasoning_effort if reasoning_mode and self.profile.get('openai_responses_supports_reasoning_mode', False): reasoning['mode'] = reasoning_mode if reasoning_context is None: # Default to `all_turns` on models that support it, so earlier-turn reasoning stays # available without opting in — consistent with sending prior thinking (including # "foreign" thinking parts from another model) back to every other model. Models # without `all_turns` support get no `context` at all, never `auto`. if self.profile.get('openai_responses_supports_reasoning_context', False): reasoning['context'] = 'all_turns' else: # Support is per-value, not per-field: every reasoning model accepts `auto` and # `current_turn`, while `all_turns` is limited to the families that persist reasoning # across turns. Gating the whole field on the narrower fact would silently discard a # value the user explicitly set. if reasoning_context == 'all_turns': supports_reasoning_context = self.profile.get('openai_responses_supports_reasoning_context', False) else: supports_reasoning_context = self.profile.get('openai_supports_reasoning', False) if supports_reasoning_context: reasoning['context'] = reasoning_context if reasoning_summary: reasoning['summary'] = reasoning_summary return reasoning or OMIT def _get_responses_tool_choice( self, model_settings: OpenAIResponsesModelSettings, model_request_parameters: ModelRequestParameters, ) -> tuple[list[responses.FunctionToolParam], ResponsesToolChoice | None]: """Determine which tools to send and the API tool_choice value. Returns: A tuple of (filtered_function_tools, tool_choice). Note: builtin tools are handled separately and should be added to this list. """ openai_profile = self.profile resolved_tool_choice = resolve_tool_choice(model_settings, model_request_parameters) tool_choice: ResponsesToolChoice | None if resolved_tool_choice in ('auto', 'none'): tool_choice = resolved_tool_choice elif resolved_tool_choice == 'required': supports = _support_tool_forcing(self.model_name, openai_profile, model_settings, model_request_parameters) tool_choice = 'required' if supports else 'auto' elif isinstance(resolved_tool_choice, tuple): tool_choice_mode, tool_names = resolved_tool_choice supports = _support_tool_forcing(self.model_name, openai_profile, model_settings, model_request_parameters) if tool_choice_mode == 'required' and len(tool_names) == 1 and supports: tool_choice = ToolChoiceFunctionParam(type='function', name=next(iter(tool_names))) else: # `allowed_tools` filters via the API rather than the tools list, # so caching is preserved. Used for the multi-tool case and as the # cache-preserving fallback when forcing isn't supported. effective_mode = 'auto' if tool_choice_mode == 'auto' or not supports else tool_choice_mode tool_choice = ToolChoiceAllowedParam( type='allowed_tools', mode=effective_mode, tools=[{'type': 'function', 'name': n} for n in tool_names], ) else: assert_never(resolved_tool_choice) # In client-executed tool search mode the `search_tools` function tool is # represented on the wire by `ToolSearchToolParam(execution='client')` — sending # it as a regular function tool too makes OpenAI emit `function_call` items # instead of `tool_search_call` items. The local dispatch path still sees the # function tool via `ModelRequestParameters.function_tools` (unaffected by this # filtering), so `_search_tools` continues to run normally. client_tool_search = _has_tool_search(model_request_parameters) tools: list[responses.FunctionToolParam] = [ self._map_tool_definition(t, visibility=model_request_parameters.visibility_of(t.name)) for t in model_request_parameters.declared_tool_defs.values() if not (client_tool_search and t.name == TOOL_SEARCH_FUNCTION_TOOL_NAME) ] return tools, tool_choice def _get_native_tools( # noqa: C901 self, model_request_parameters: ModelRequestParameters ) -> list[responses.ToolParam]: tools: list[responses.ToolParam] = [] has_image_generating_tool = False for tool in model_request_parameters.native_tools: if isinstance(tool, WebSearchTool): web_search_tool = responses.WebSearchToolParam( type='web_search', search_context_size=tool.search_context_size ) if tool.user_location: web_search_tool['user_location'] = responses.web_search_tool_param.UserLocation( type='approximate', **tool.user_location ) filters = responses.web_search_tool_param.Filters() if tool.allowed_domains: filters['allowed_domains'] = tool.allowed_domains if tool.blocked_domains: # The OpenAI API supports this field, but the SDK's `Filters` does not include it yet. cast(dict[str, object], filters)['blocked_domains'] = tool.blocked_domains if filters: web_search_tool['filters'] = filters if tool.external_web_access is not None: # The OpenAI API supports this field, but the SDK's `WebSearchToolParam` does not include it yet. cast(dict[str, object], web_search_tool)['external_web_access'] = tool.external_web_access tools.append(web_search_tool) elif isinstance(tool, FileSearchTool): file_search_tool = cast( responses.FileSearchToolParam, {'type': 'file_search', 'vector_store_ids': list(tool.file_store_ids)}, ) tools.append(file_search_tool) elif isinstance(tool, CodeExecutionTool): has_image_generating_tool = True container: responses.tool_param.CodeInterpreterContainerCodeInterpreterToolAuto = {'type': 'auto'} if tool.files: # Cross-provider files are dropped silently here, not raised via # `_validate_uploaded_file_provider`; intentional per https://github.com/pydantic/pydantic-ai/issues/4338 (ignore over raise). provider_file_ids = [file.file_id for file in tool.files if file.provider_name == self.system] if provider_file_ids: container['file_ids'] = provider_file_ids tools.append({'type': 'code_interpreter', 'container': container}) elif isinstance(tool, MCPServerTool): mcp_tool = responses.tool_param.Mcp( type='mcp', server_label=tool.id, require_approval='never', ) if tool.authorization_token: # pragma: no branch mcp_tool['authorization'] = tool.authorization_token if tool.allowed_tools is not None: # pragma: no branch mcp_tool['allowed_tools'] = tool.allowed_tools if tool.description: # pragma: no branch mcp_tool['server_description'] = tool.description if tool.headers: # pragma: no branch mcp_tool['headers'] = tool.headers if tool.url.startswith(MCP_SERVER_TOOL_CONNECTOR_URI_SCHEME + ':'): _, connector_id = tool.url.split(':', maxsplit=1) mcp_tool['connector_id'] = connector_id # pyright: ignore[reportGeneralTypeIssues] else: mcp_tool['server_url'] = tool.url tools.append(mcp_tool) elif isinstance(tool, ImageGenerationTool): has_image_generating_tool = True tools.append(_map_openai_image_generation_tool(tool)) elif isinstance(tool, ToolSearchTool): # pragma: no branch # OpenAI's Responses API has no concept of named native strategies — the # `tool_search` builtin is one knob, decided server-side. Honor the user's # explicit choice or fail loudly; do not silently run a different algorithm. # `'custom'` is the internal marker for callable strategies and is handled # below — only named native strategies (`'bm25'`/`'regex'`) raise here. if tool.strategy is None: tools.append(ToolSearchToolParam(type='tool_search')) elif tool.strategy == 'custom': # With `strategy='custom'`, the search runs client-side via our local # `search_tools` function tool: the provider surfaces the search as a # `tool_search_call` with `execution='client'`, we dispatch it through the # normal function-call path, and reply with a `tool_search_output` that # carries the discovered tool definitions. search_tool_def = _find_search_tool_definition(model_request_parameters) parameters = dict(search_tool_def.parameters_json_schema) if search_tool_def else {} # OpenAI's strict JSON schema mode is opt-in for client tool search — the # `additionalProperties: False` is what it prefers for closed-world schemas. parameters.setdefault('additionalProperties', False) # OpenAI's generated schema marks this optional, but the live Responses API # rejects client-executed `tool_search` without a non-null description. description = ( search_tool_def.description if search_tool_def and search_tool_def.description else _DEFAULT_CLIENT_TOOL_SEARCH_DESCRIPTION ) tool_search_param: ToolSearchToolParam = { 'type': 'tool_search', 'execution': 'client', 'description': description, 'parameters': parameters, } tools.append(tool_search_param) else: raise UserError( f'`ToolSearch(strategy={tool.strategy!r})` is an Anthropic-native strategy ' 'and is not supported by OpenAI Responses. Use `strategy=None` (default, ' "provider-managed), `strategy='keywords'` (local keyword matching), or " 'a callable strategy.' ) else: raise UserError( # pragma: no cover f'`{tool.__class__.__name__}` is not supported by `OpenAIResponsesModel`. If it should be, please file an issue.' ) if model_request_parameters.allow_image_output and not has_image_generating_tool: tools.append({'type': 'image_generation'}) return tools def _map_tool_definition(self, f: ToolDefinition, *, visibility: ToolVisibility) -> responses.FunctionToolParam: tool_param: responses.FunctionToolParam = { 'name': f.name, 'parameters': f.parameters_json_schema, 'type': 'function', 'description': f.description, 'strict': bool(f.strict and self.profile.get('openai_supports_strict_tool_definition', True)), } # `'deferred'` visibility means "withhold this schema", and `prepare_request` only resolves # it where the API will accept it — which here means a # `tool_search` tool is along for the ride, without which this earns # `Invalid Value: 'tools.defer_loading'. Deferred tools require tools.tool_search.` if visibility == 'deferred': tool_param['defer_loading'] = True return tool_param def _resolve_previous_response_id( self, setting: str | None, messages: list[ModelMessage], *, allow_no_new_messages: bool = False, ) -> tuple[str | None, list[ModelMessage]]: # Resolve the effective `previous_response_id` and trim already-stored messages. # # A concrete ID in `setting` acts as a seed for the first request in a run (to # continue from a prior turn's stored response). On subsequent in-run requests # (retries, tool-call continuations), the most recent `provider_response_id` # from the message history takes precedence, so we chain to the latest stored # response instead of re-sending messages that are already server-side. # # A compaction response in the tail is a hard chain boundary even with a # concrete seed: crossing it would re-inject the context that compaction was # meant to replace. if setting is None: return None, messages auto_id, trimmed = self._get_previous_response_id_and_new_messages( messages, allow_no_new_messages=allow_no_new_messages ) if auto_id is not None: return auto_id, trimmed if setting == 'auto' or self._is_at_compaction_boundary(messages): return None, messages return setting, messages def _is_at_compaction_boundary(self, messages: list[ModelMessage]) -> bool: for m in reversed(messages): if isinstance(m, ModelResponse) and m.provider_name == self.system: return bool(m.provider_details and m.provider_details.get('compaction')) return False def _get_previous_response_id_and_new_messages( self, messages: list[ModelMessage], *, allow_no_new_messages: bool = False ) -> tuple[str | None, list[ModelMessage]]: # Find the most recent `provider_response_id` from messages produced by this # provider, and return it along with the messages that came after it (which # still need to be sent as new input). When nothing suitable is found, returns # `(None, messages)` with the full list unchanged. previous_response_id = None trimmed_messages: list[ModelMessage] = [] for m in reversed(messages): if isinstance(m, ModelResponse) and m.provider_name == self.system: # Responses from the stateless `/compact` endpoint can't be used as # `previous_response_id`, so the compaction acts as a hard chain boundary: # the next request must pass the `CompactionPart` via the `input` array # (handled by `_map_messages`) without a `previous_response_id`. if m.provider_details and m.provider_details.get('compaction'): return None, messages previous_response_id = m.provider_response_id break else: trimmed_messages.append(m) if previous_response_id and (allow_no_new_messages or trimmed_messages): return previous_response_id, list(reversed(trimmed_messages)) else: return None, messages def _resolve_server_side_state( self, model_settings: OpenAIResponsesModelSettings, messages: list[ModelMessage] ) -> tuple[str | None, str | None, list[ModelMessage]]: previous_response_id_setting = model_settings.get('openai_previous_response_id') conversation_id_setting = model_settings.get('openai_conversation_id') if previous_response_id_setting is not None and conversation_id_setting is not None: raise UserError( '`openai_previous_response_id` and `openai_conversation_id` cannot both be set because ' 'the OpenAI Responses API does not support `previous_response_id` with `conversation`.' ) if conversation_id_setting is not None: conversation_id, messages = self._resolve_conversation_id(conversation_id_setting, messages) return None, conversation_id, messages previous_response_id, messages = self._resolve_previous_response_id(previous_response_id_setting, messages) return previous_response_id, None, messages def _resolve_conversation_id( self, setting: Literal['auto'] | str, messages: list[ModelMessage] ) -> tuple[str | None, list[ModelMessage]]: if setting == 'auto': # Agent runs stamp the active Pydantic AI conversation ID on the final request. # Direct model calls may still pass an empty message list. pydantic_ai_conversation_id = next((m.conversation_id for m in messages[-1:]), None) return self._get_conversation_id_and_new_messages( messages, pydantic_ai_conversation_id=pydantic_ai_conversation_id ) conversation_id, trimmed = self._get_conversation_id_and_new_messages(messages, openai_conversation_id=setting) if conversation_id is not None: return conversation_id, trimmed return setting, messages def _get_conversation_id_and_new_messages( self, messages: list[ModelMessage], *, openai_conversation_id: str | None = None, pydantic_ai_conversation_id: str | None = None, ) -> tuple[str | None, list[ModelMessage]]: trimmed_messages: list[ModelMessage] = [] for m in reversed(messages): if isinstance(m, ModelResponse) and m.provider_name == self.system: candidate = m.provider_details and m.provider_details.get('conversation_id') if ( pydantic_ai_conversation_id is not None and m.conversation_id is not None and m.conversation_id != pydantic_ai_conversation_id ): trimmed_messages.append(m) continue if isinstance(candidate, str) and ( openai_conversation_id is None or candidate == openai_conversation_id ): return candidate, list(reversed(trimmed_messages)) trimmed_messages.append(m) return None, messages async def _map_messages( # noqa: C901 self, messages: list[ModelMessage], model_settings: OpenAIResponsesModelSettings, model_request_parameters: ModelRequestParameters, *, standing_prompt_retained: bool = True, previous_response_id: str | None = None, ) -> tuple[str | Omit, list[responses.ResponseInputItemParam]]: """Maps a `pydantic_ai.Message` to a `openai.types.responses.ResponseInputParam` i.e. the OpenAI Responses API input format. For `ThinkingParts`, this method: - Sends `signature` back as `encrypted_content` (for official OpenAI reasoning) - Sends `content` back as `summary` text - Sends `provider_details['raw_content']` back as `content` items (for gpt-oss raw CoT) Raw CoT is sent back to improve model performance in multi-turn conversations. """ # `standing_prompt_retained=True` on ordinary requests: the compaction item retains the # window's leading `system` items (single hop, live-verified), so re-sending them would # duplicate the standing prompt. The re-compaction path passes `False`: retention decayed # across a second compaction in probing, so each freshly built window plants it explicitly. messages = self._trim_before_compaction(messages, standing_prompt_retained=standing_prompt_retained) profile = self.profile response_scoped_tool_call_ids = profile.get('openai_responses_tool_call_ids_are_response_scoped', False) response_id = previous_response_id if response_scoped_tool_call_ids else None send_item_ids = model_settings.get( 'openai_send_reasoning_ids', profile.get('openai_supports_encrypted_reasoning_content', False) ) # With `execution='client'` tool search, `search_tools` calls/returns need to be # replayed as `tool_search_call` / `tool_search_output` items so the provider can # pair them with the builtin and unlock the discovered tools. The current # request's builtin configuration is the source of truth: a `ToolCallPart` for # `search_tools` belongs in the native replay flow whenever tool search is active. client_tool_search_active = _has_tool_search(model_request_parameters) client_replay_call_ids: set[str] = set() # Mirrors the Anthropic renderer's per-request `tool_addition` dedupe: several history # parts may name the same revealed tool (duplicated deltas from a UI round-trip, a delta # plus a replayed search return), and one declaration per request is enough. rendered_additional_tools: set[str] = set() openai_messages: list[responses.ResponseInputItemParam] = [] for message_index, message in enumerate(messages): if isinstance(message, ModelRequest): for part in message.parts: if isinstance(part, SystemPromptPart): openai_messages.append( responses.EasyInputMessageParam( role=profile.get('openai_system_prompt_role', None) or 'system', content=part.content ) ) elif isinstance(part, UserPromptPart): openai_messages.append(await self._map_user_prompt(part)) elif isinstance(part, ToolAvailabilityDeltaPart): if self.tool_addition_mode != 'with_definitions': # `prepare_messages` projects the delta onto the local tool-search exchange # for every model without native support, so arriving here means that # projection didn't run — reachable by calling `Model.request` directly, and # the same pipeline bug the other adapters raise on. # # Not because the API would reject the item: `gpt-5` and `gpt-4o`, both # outside the supported list, accept an `additional_tools` item and call the # tool it declares. So the raise is about the invariant, not the wire — but # silently rendering a shape whose support we haven't verified, for a tool # this same path has removed from `tools`, is how an availability change # goes missing with nothing to show for it. raise _unsynthesized_tool_availability_delta_error() additional_tools = self._map_additional_tools( part.tools_added, model_request_parameters, rendered=rendered_additional_tools ) if additional_tools['tools']: openai_messages.append(additional_tools) elif isinstance(part, ToolReturnPart): call_id = _guard_tool_call_id(t=part) call_id, _ = _split_combined_tool_call_id(call_id) call_id = _provider_response_tool_call_id(call_id, response_id) if call_id in client_replay_call_ids and isinstance(part, ToolSearchReturnPart): # Replay the local `search_tools` result as a # `tool_search_output` so the provider rehydrates the # discovered-tool state tied to the `tool_search_call`. openai_messages.append( _build_tool_search_output_param( part, call_id, 'client', 'completed', model_request_parameters, self._map_tool_definition, ) ) else: output = await self._map_tool_return_output(part) item = FunctionCallOutput( type='function_call_output', call_id=call_id, output=output, ) openai_messages.append(item) if ( isinstance(part, ToolSearchReturnPart) and not client_tool_search_active and self.tool_addition_mode == 'with_definitions' ): additional_tools = self._map_additional_tools( [match['name'] for match in part.discovered_tools], model_request_parameters, rendered=rendered_additional_tools, ) if additional_tools['tools']: openai_messages.append(additional_tools) elif isinstance(part, RetryPromptPart): if part.tool_name is None: openai_messages.append( Message(role='user', content=[{'type': 'input_text', 'text': part.model_response()}]) ) else: call_id = _guard_tool_call_id(t=part) call_id, _ = _split_combined_tool_call_id(call_id) call_id = _provider_response_tool_call_id(call_id, response_id) item = FunctionCallOutput( type='function_call_output', call_id=call_id, output=part.model_response(), ) openai_messages.append(item) elif isinstance(part, SpeechPart): # pragma: no cover # Unconverted realtime speech; `prepare_messages` turns these into `UserPromptPart`s in `Model.prepare_messages`. raise _unconverted_speech_part_error() else: assert_never(part) elif isinstance(message, ModelResponse): response_from_same_provider = message.provider_name == self.system message_item: responses.ResponseOutputMessageParam | None = None reasoning_item: responses.ResponseReasoningItemParam | None = None web_search_item: responses.ResponseFunctionWebSearchParam | None = None file_search_item: responses.ResponseFileSearchToolCallParam | None = None code_interpreter_item: responses.ResponseCodeInterpreterToolCallParam | None = None response_parts: Sequence[ModelResponsePart] = message.parts if not profile.get('openai_responses_supports_interleaved_function_calls', True): response_parts = _group_settled_portable_function_calls( messages, message_index, message, client_tool_search_active=client_tool_search_active, ) for item in response_parts: from_same_provider = item.provider_name == self.system or ( item.provider_name is None and message.provider_name == self.system ) response_id_from_same_provider = message.provider_name == self.system or ( message.provider_name is None and item.provider_name == self.system ) response_from_same_provider |= response_id_from_same_provider # Two distinct gates. For native tool items whose state lives server-side # (web search, code interpreter, image generation, MCP), the item ID *is* # the replay payload, so `should_send_item_id` decides whether those items # are sent at all. Hosted tool-search items carry their state inline, so # they replay on `from_same_provider` alone and this flag only strips their # IDs (see the `openai_send_reasoning_ids` docstring). should_send_item_id = send_item_ids and from_same_provider if isinstance(item, TextPart): phase = (item.provider_details or {}).get('phase') send_phase = ( profile.get('openai_supports_phase', False) and item.provider_name == self.system and phase in ('commentary', 'final_answer') ) if item.id and should_send_item_id: if message_item is None or message_item['id'] != item.id: # pragma: no branch message_item = responses.ResponseOutputMessageParam( role='assistant', id=item.id, content=[], type='message', status='completed', ) openai_messages.append(message_item) message_item['content'] = [ *message_item['content'], responses.ResponseOutputTextParam( text=item.content, type='output_text', annotations=[] ), ] if send_phase: message_item['phase'] = phase else: easy_message_item = responses.EasyInputMessageParam(role='assistant', content=item.content) if send_phase: easy_message_item['phase'] = phase openai_messages.append(easy_message_item) elif isinstance(item, ToolCallPart): call_id = _guard_tool_call_id(t=item) call_id, id = _split_combined_tool_call_id(call_id) call_id = _provider_response_tool_call_id( call_id, message.provider_response_id if response_scoped_tool_call_ids and response_id_from_same_provider else None, ) id = id or item.id if client_tool_search_active and item.tool_name == TOOL_SEARCH_FUNCTION_TOOL_NAME: # Replay the local `search_tools` call as a `tool_search_call` # with `execution='client'` so OpenAI re-attaches it to the # builtin and unlocks the discovered tools' schemas. Fires for # any `provider_name`, including `None` (cross-provider history # from a prior local-fallback turn) and other-provider names # (Google, etc.) — `execution='client'` is OpenAI's accepted # historical-replay shape regardless of the call's origin. client_replay_call_ids.add(call_id) tool_search_call: ToolSearchCallParam = { 'type': 'tool_search_call', 'execution': 'client', 'arguments': item.args_as_dict() or {}, 'call_id': call_id, 'status': 'completed', } if id and should_send_item_id: # pragma: no branch tool_search_call['id'] = id openai_messages.append(tool_search_call) else: param = responses.ResponseFunctionToolCallParam( name=item.tool_name, arguments=item.args_as_json_str(), call_id=call_id, type='function_call', ) if profile.get('openai_responses_requires_function_call_status_none', False): param['status'] = None # type: ignore[reportGeneralTypeIssues] if id and should_send_item_id: # pragma: no branch param['id'] = id # The Responses API attaches a `namespace` to function calls # for discovered deferred tools (tool-search flow). Round-trip # it on replay or the API rejects the call as missing namespace. if ( item.provider_name == self.system and item.provider_details and (ns := item.provider_details.get('namespace')) ): param['namespace'] = ns elif synthesized_ns := _tool_search_namespace_for_synthesis( item.tool_name, model_request_parameters ): # Cross-provider replay: prior turn ran on a non-OpenAI # provider, so no namespace was captured. See helper for the # evidence behind synthesizing `namespace = tool_name`. param['namespace'] = synthesized_ns openai_messages.append(param) elif isinstance(item, NativeToolCallPart): if from_same_provider and item.tool_name == ToolSearchTool.kind and item.tool_call_id: call_id, status, _ = _tool_search_replay_details(item) tool_search_call = responses.response_input_item_param.ToolSearchCall( call_id=call_id, arguments=item.args_as_dict() or {}, type='tool_search_call', execution='server', status=status, ) if should_send_item_id and item.id: tool_search_call['id'] = item.id openai_messages.append(tool_search_call) elif should_send_item_id: # pragma: no branch if ( item.tool_name == CodeExecutionTool.kind and item.tool_call_id and (args := item.args_as_dict()) and (container_id := args.get('container_id')) ): code_interpreter_item = responses.ResponseCodeInterpreterToolCallParam( id=item.tool_call_id, code=args.get('code'), container_id=container_id, outputs=None, # These can be read server-side status='completed', type='code_interpreter_call', ) openai_messages.append(code_interpreter_item) elif ( item.tool_name == WebSearchTool.kind and item.tool_call_id and (args := item.args_as_dict()) ): # We need to exclude None values because of https://github.com/pydantic/pydantic-ai/issues/3653 args = {k: v for k, v in args.items() if v is not None} web_search_item = responses.ResponseFunctionWebSearchParam( id=item.id or item.tool_call_id, action=cast(responses.response_function_web_search_param.Action, args), status='completed', type='web_search_call', ) openai_messages.append(web_search_item) elif ( # pragma: no cover item.tool_name == FileSearchTool.kind and item.tool_call_id and (args := item.args_as_dict()) ): file_search_item = cast( responses.ResponseFileSearchToolCallParam, { 'id': item.id or item.tool_call_id, 'queries': args.get('queries', []), 'status': 'completed', 'type': 'file_search_call', }, ) openai_messages.append(file_search_item) elif item.tool_name == ImageGenerationTool.kind and item.tool_call_id: # The cast is necessary because of https://github.com/openai/openai-python/issues/2648 image_generation_item = cast( responses.response_input_item_param.ImageGenerationCall, { 'id': item.tool_call_id, 'type': 'image_generation_call', }, ) openai_messages.append(image_generation_item) elif ( # pragma: no branch item.tool_name.startswith(MCPServerTool.kind) and item.tool_call_id and (server_id := item.tool_name.split(':', 1)[1]) and (args := item.args_as_dict()) and (action := args.get('action')) ): if action == 'list_tools': mcp_list_tools_item = responses.response_input_item_param.McpListTools( id=item.tool_call_id, type='mcp_list_tools', server_label=server_id, tools=[], # These can be read server-side ) openai_messages.append(mcp_list_tools_item) elif ( # pragma: no branch action == 'call_tool' and (tool_name := args.get('tool_name')) and (tool_args := args.get('tool_args')) is not None ): mcp_call_item = responses.response_input_item_param.McpCall( id=item.tool_call_id, server_label=server_id, name=tool_name, arguments=to_json(tool_args).decode(), error=None, # These can be read server-side output=None, # These can be read server-side type='mcp_call', ) openai_messages.append(mcp_call_item) elif isinstance(item, NativeToolReturnPart): if ( from_same_provider and isinstance(item, NativeToolSearchReturnPart) and not _lacks_tool_search_output_identity(item) ): call_id, status, provider_details = _tool_search_replay_details(item) tool_search_output = _build_tool_search_output_param( item, call_id, 'server', status, model_request_parameters, self._map_tool_definition, ) output_id = provider_details.get('id') if should_send_item_id and isinstance(output_id, str): tool_search_output['id'] = output_id openai_messages.append(tool_search_output) elif should_send_item_id: # pragma: no branch status = item.content.get('status') if _is_str_dict(item.content) else None kind_to_item = { CodeExecutionTool.kind: code_interpreter_item, WebSearchTool.kind: web_search_item, FileSearchTool.kind: file_search_item, } if status and (builtin_item := kind_to_item.get(item.tool_name)) is not None: builtin_item['status'] = status elif item.tool_name == ImageGenerationTool.kind: # Image generation result does not need to be sent back, just the `id` off of `NativeToolCallPart`. pass elif item.tool_name.startswith(MCPServerTool.kind): # MCP call result does not need to be sent back, just the fields off of `NativeToolCallPart`. pass elif isinstance(item, FilePart): # This was generated by the `ImageGenerationTool` or `CodeExecutionTool`, # and does not need to be sent back separately from the corresponding `NativeToolReturnPart`. # If `send_item_ids` is false, we won't send the `NativeToolReturnPart`, but OpenAI does not have a type for files from the assistant. pass elif isinstance(item, ThinkingPart): # Get raw CoT content from provider_details if present and from this provider raw_content: list[str] | None = None if item.provider_name == self.system: raw_content = (item.provider_details or {}).get('raw_content') # Chat Completions stores its content/reasoning field name as a synthetic # part ID. Preserve opaque IDs from compatible Responses APIs unless they # collide with one of those sentinels without carrying native wire data. synthetic_chat_reasoning = item.id in ('content', 'reasoning', 'reasoning_content') native_reasoning = not synthetic_chat_reasoning or bool(item.signature or raw_content) if item.id and native_reasoning and (should_send_item_id or raw_content): signature: str | None = None if ( item.signature and item.provider_name == self.system and profile.get('openai_supports_encrypted_reasoning_content', False) ): signature = item.signature if (reasoning_item is None or reasoning_item['id'] != item.id) and ( signature or item.content or raw_content ): # pragma: no branch reasoning_item = responses.ResponseReasoningItemParam( id=item.id, summary=[], encrypted_content=signature, type='reasoning', ) openai_messages.append(reasoning_item) if item.content: # The check above guarantees that `reasoning_item` is not None assert reasoning_item is not None reasoning_item['summary'] = [ *reasoning_item['summary'], ReasoningSummary(text=item.content, type='summary_text'), ] if raw_content: # Send raw CoT back assert reasoning_item is not None reasoning_item['content'] = [ ReasoningContent(text=text, type='reasoning_text') for text in raw_content ] else: start_tag, end_tag = profile.get('thinking_tags', DEFAULT_THINKING_TAGS) openai_messages.append( responses.EasyInputMessageParam( role='assistant', content='\n'.join([start_tag, item.content, end_tag]) ) ) elif isinstance(item, CompactionPart): if ( item.provider_name == self.system and item.provider_details and 'encrypted_content' in item.provider_details ): openai_messages.append( ResponseCompactionItemParamParam( id=item.id, encrypted_content=item.provider_details['encrypted_content'], type='compaction', ) ) elif isinstance(item, SpeechPart): # pragma: no cover # Unconverted realtime speech; `prepare_messages` turns these into `TextPart`s in `Model.prepare_messages`. raise _unconverted_speech_part_error() else: assert_never(item) response_id = ( message.provider_response_id if response_scoped_tool_call_ids and response_from_same_provider else None ) else: assert_never(message) instructions = get_instructions(messages, model_request_parameters) or OMIT return instructions, openai_messages def _map_additional_tools( self, tool_names: list[str], model_request_parameters: ModelRequestParameters, *, rendered: set[str] ) -> responses.response_input_item_param.AdditionalTools: """Build one `additional_tools` item for the given names, deduped across the request. A `'via_history'` name renders because this item is the only place its definition travels; a `'deferred'` name renders because its `tools`-array entry stays schema-withheld for byte-stable caching and the item is what delivers the schema on reveal (`test_tool_availability_delta_and_the_tools_cache_section` measures that property). A `'visible'` name is fully declared in `tools` and has nothing to add, and a `'withheld'` name — a delta can name one when a direct `Model.request` caller authors `revealed_tool_names` that don't cover the history's deltas — must not have the schema the request just withheld smuggled on via the item. `rendered` accumulates the names already declared by an earlier history part (a duplicated delta from a UI round-trip, a delta plus a replayed search return) — one declaration per request is enough, mirroring the Anthropic renderer's `tool_addition` dedupe. """ tool_defs_by_name = {tool.name: tool for tool in model_request_parameters.function_tools} renderable: list[str] = [] for name in tool_names: # Marking `rendered` per accepted name (like the Anthropic loop) also dedupes a name # repeated within this part's own list, not just across parts. if ( name not in rendered and name in tool_defs_by_name and model_request_parameters.visibility_of(name) in ('deferred', 'via_history') ): renderable.append(name) rendered.add(name) return responses.response_input_item_param.AdditionalTools( type='additional_tools', role='developer', tools=[self._map_tool_definition(tool_defs_by_name[name], visibility='visible') for name in renderable], ) def _map_json_schema(self, o: OutputObjectDefinition) -> responses.ResponseFormatTextJSONSchemaConfigParam: response_format_param: responses.ResponseFormatTextJSONSchemaConfigParam = { 'type': 'json_schema', 'name': o.name or DEFAULT_OUTPUT_TOOL_NAME, 'schema': o.json_schema, } if o.description: response_format_param['description'] = o.description if self.profile.get('openai_supports_strict_tool_definition', True): # pragma: no branch response_format_param['strict'] = o.strict return response_format_param async def _map_user_prompt(self, part: UserPromptPart) -> responses.EasyInputMessageParam: content: str | list[responses.ResponseInputContentParam] if isinstance(part.content, str): content = part.content else: content = [] for item in part.content: if isinstance(item, str | TextContent): text = item if isinstance(item, str) else item.content content.append(responses.ResponseInputTextParam(text=text, type='input_text')) elif isinstance(item, UploadedFile): content.append(self._map_uploaded_file_to_response_content(item)) # pyright: ignore[reportArgumentType] elif isinstance(item, CachePoint): if self.profile.get('openai_supports_prompt_cache_breakpoints', False): _add_openai_prompt_cache_breakpoint(content) elif is_multi_modal_content(item): content.append(await OpenAIResponsesModel._map_file_to_response_content(item, 'user prompts')) # pyright: ignore[reportArgumentType] else: raise RuntimeError(f'Unsupported content type: {type(item)}') # pragma: no cover return responses.EasyInputMessageParam(role='user', content=content) def _map_uploaded_file_to_response_content( self, item: UploadedFile, ) -> ResponseInputImageContentParam | ResponseInputFileContentParam: """Map an `UploadedFile` to its OpenAI Responses API content param. Raises `UserError` if the file was uploaded to a different provider (`provider_name != self.system`). Image uploads (an `image/*` media type) map to `input_image`, carrying `detail` from `vendor_metadata` (default `'auto'`); everything else maps to `input_file`. Opaque OpenAI Files-API ids (e.g. `file-...`) report `application/octet-stream`, so an image referenced by such an id without an explicit `image/*` media type also maps to `input_file`. """ self._validate_uploaded_file_provider(item) if item.media_type.startswith('image/'): detail: Literal['auto', 'low', 'high'] = 'auto' if metadata := item.vendor_metadata: detail = metadata.get('detail', 'auto') return ResponseInputImageContentParam(type='input_image', file_id=item.file_id, detail=detail) return ResponseInputFileContentParam(type='input_file', file_id=item.file_id) @staticmethod async def _map_file_to_response_content( item: BinaryContent | ImageUrl | DocumentUrl | AudioUrl | VideoUrl, context: str, ) -> ResponseInputImageContentParam | ResponseInputFileContentParam: """Map a multimodal file item to its OpenAI Responses API content param.""" if isinstance(item, BinaryContent): if item.is_image: detail: Literal['auto', 'low', 'high'] = 'auto' if metadata := item.vendor_metadata: detail = metadata.get('detail', 'auto') return ResponseInputImageContentParam( image_url=item.data_uri, type='input_image', detail=detail, ) elif item.is_document: return ResponseInputFileContentParam( type='input_file', file_data=item.data_uri, filename=f'filename.{item.format}', ) elif item.is_audio: raise NotImplementedError(f'BinaryContent with audio is not supported in OpenAI Responses {context}') elif item.is_video: raise NotImplementedError(f'BinaryContent with video is not supported in OpenAI Responses {context}') else: # pragma: no cover raise RuntimeError(f'Unsupported binary content type: {item.media_type}') elif isinstance(item, ImageUrl): detail = 'auto' image_url = item.url if metadata := item.vendor_metadata: detail = metadata.get('detail', 'auto') if item.force_download: downloaded = await download_item(item, data_format='base64_uri', type_format='extension') image_url = downloaded['data'] return ResponseInputImageContentParam( image_url=image_url, type='input_image', detail=detail, ) elif isinstance(item, (AudioUrl, DocumentUrl)): if item.force_download: downloaded = await download_item(item, data_format='base64_uri', type_format='extension') return ResponseInputFileContentParam( type='input_file', file_data=downloaded['data'], filename=f'filename.{downloaded["data_type"]}', ) return ResponseInputFileContentParam( type='input_file', file_url=item.url, ) else: raise NotImplementedError(f'VideoUrl is not supported in OpenAI Responses {context}') async def _map_tool_return_output( self, part: ToolReturnPart, ) -> str | list[ResponseInputTextContentParam | ResponseInputImageContentParam | ResponseInputFileContentParam]: """Map a `ToolReturnPart` to OpenAI Responses API output format, supporting multimodal content. Iterates content directly to preserve order of mixed file/data content. """ if not part.files: return part.model_response_str() output: list[ ResponseInputTextContentParam | ResponseInputImageContentParam | ResponseInputFileContentParam ] = [] for item in part.content_items(mode='str'): if isinstance(item, UploadedFile): output.append(self._map_uploaded_file_to_response_content(item)) elif is_multi_modal_content(item): output.append(await OpenAIResponsesModel._map_file_to_response_content(item, 'tool returns')) # pyright: ignore[reportArgumentType] elif isinstance(item, str): # pragma: no branch output.append(ResponseInputTextContentParam(type='input_text', text=item)) return output @dataclass class OpenAIStreamedResponse(StreamedResponse): """Implementation of `StreamedResponse` for OpenAI models.""" _model_name: OpenAIModelName _model_profile: OpenAIModelProfile _response: _utils.PeekableAsyncStream[ChatCompletionChunk, AsyncStream[ChatCompletionChunk]] _provider_name: str _model_id_namespace: str _provider_url: str _provider_timestamp: datetime | None = None _timestamp: datetime = field(default_factory=_now_utc) _model_settings: OpenAIChatModelSettings | None = None _vendor_part_id: str = field(default='content', init=False) _has_refusal: bool = field(default=False, init=False) _refusal_text: str = field(default='', init=False) _has_finish_reason: bool = field(default=False, init=False) async def close_stream(self) -> None: await self._response.source.close() async def _get_event_iterator(self) -> AsyncIterator[ModelResponseStreamEvent]: # noqa: C901 with _map_api_errors(self._model_name, self._model_id_namespace): async for chunk in self._validate_response(): if self._provider_timestamp is None and chunk.created: self._provider_timestamp = number_to_datetime(chunk.created) self.provider_details = { **(self.provider_details or {}), 'timestamp': self._provider_timestamp, } chunk_usage = self._map_usage(chunk) if self._model_settings and self._model_settings.get('openai_continuous_usage_stats'): # When continuous_usage_stats is enabled, each chunk contains cumulative usage, # so we replace rather than increment to avoid double-counting. self._usage = chunk_usage else: self._usage += chunk_usage if chunk.id: # pragma: no branch self.provider_response_id = chunk.id if chunk.model: self._model_name = chunk.model # The moderation chunk carries no choices, so this has to happen before the guard below. if chunk.moderation: self.provider_details = { **(self.provider_details or {}), 'moderation': chunk.moderation.model_dump(), } # Empty on the final usage-only chunk; `None` from OpenAI-compatible providers emitting # malformed chunks that the openai SDK's loose constructor lets through (https://github.com/pydantic/pydantic-ai/issues/5165). if not chunk.choices: continue choice = chunk.choices[0] raw_finish_reason = choice.finish_reason self._has_finish_reason = self._has_finish_reason or bool(raw_finish_reason) # When using Azure OpenAI and an async content filter is enabled, the openai SDK can return None deltas. if choice.delta is None: # pyright: ignore[reportUnnecessaryComparison] continue # Handle refusal responses (structured output safety filter). # Note: OpenAI sends refusal instead of content (not alongside it), so in practice # text parts won't have been yielded before _has_refusal is set. if choice.delta.refusal: self._has_refusal = True self.finish_reason = 'content_filter' self._refusal_text += choice.delta.refusal continue if raw_finish_reason and not self._has_refusal: self.finish_reason = self._map_finish_reason(raw_finish_reason) if provider_details := self._map_provider_details(chunk): # pragma: no branch if self._has_refusal: provider_details.pop('finish_reason', None) self.provider_details = {**(self.provider_details or {}), **provider_details} for event in self._map_part_delta(choice): yield event if self._refusal_text: self.provider_details = {**(self.provider_details or {}), 'refusal': self._refusal_text} if ( self._model_profile.get('openai_chat_streaming_requires_finish_reason', False) and not self._has_finish_reason and not self.cancelled ): raise ModelAPIError( model_name=self.model_name, message='Streamed response ended without a `finish_reason`', ) def _validate_response(self) -> AsyncIterable[ChatCompletionChunk]: """Hook that validates incoming chunks. This method may be overridden by subclasses of `OpenAIStreamedResponse` to apply custom chunk validations. By default, this is a no-op since `ChatCompletionChunk` is already validated. """ return self._response def _map_part_delta(self, choice: chat_completion_chunk.Choice) -> Iterable[ModelResponseStreamEvent]: """Hook that determines the sequence of mappings that will be called to produce events. This method may be overridden by subclasses of `OpenAIStreamResponse` to customize the mapping. """ return itertools.chain( self._map_thinking_delta(choice), self._map_text_delta(choice), self._map_tool_call_delta(choice) ) def _map_thinking_delta(self, choice: chat_completion_chunk.Choice) -> Iterable[ModelResponseStreamEvent]: """Hook that maps thinking delta content to events. This method may be overridden by subclasses of `OpenAIStreamResponse` to customize the mapping. """ profile = self._model_profile custom_field = profile.get('openai_chat_thinking_field', None) # Prefer the configured custom reasoning field, if present in profile. # Fall back to built-in fields if no custom field result was found. # The `reasoning_content` field is typically present in DeepSeek and Moonshot models. # https://api-docs.deepseek.com/guides/reasoning_model # The `reasoning` field is typically present in gpt-oss via Ollama and OpenRouter. # - https://cookbook.openai.com/articles/gpt-oss/handle-raw-cot#chat-completions-api # - https://openrouter.ai/docs/use-cases/reasoning-tokens#basic-usage-with-reasoning-tokens for field_name in (custom_field, 'reasoning', 'reasoning_content'): if not field_name: continue reasoning: object = getattr(choice.delta, field_name, None) if not reasoning: continue if not isinstance(reasoning, str): warnings.warn( f'Unexpected non-string value for {field_name!r}: {type(reasoning).__name__}. ' 'Please open an issue at https://github.com/pydantic/pydantic-ai/issues.', UserWarning, ) continue yield from self._parts_manager.handle_thinking_delta( vendor_part_id=field_name, id=field_name, content=reasoning, provider_name=self.provider_name, ) break def _map_text_delta(self, choice: chat_completion_chunk.Choice) -> Iterable[ModelResponseStreamEvent]: """Hook that maps text delta content to events. This method may be overridden by subclasses of `OpenAIStreamResponse` to customize the mapping. """ # Handle the text part of the response content = choice.delta.content if content: for event in self._parts_manager.handle_text_delta( vendor_part_id=self._vendor_part_id, content=content, thinking_tags=self._model_profile.get('thinking_tags', DEFAULT_THINKING_TAGS), ignore_leading_whitespace=self._model_profile.get('ignore_streamed_leading_whitespace', False), ): if isinstance(event, PartStartEvent) and isinstance(event.part, ThinkingPart): event.part.id = 'content' event.part.provider_name = self.provider_name yield event def _map_tool_call_delta(self, choice: chat_completion_chunk.Choice) -> Iterable[ModelResponseStreamEvent]: """Hook that maps tool call delta content to events. This method may be overridden by subclasses of `OpenAIStreamResponse` to customize the mapping. """ for dtc in choice.delta.tool_calls or []: maybe_event = self._parts_manager.handle_tool_call_delta( vendor_part_id=dtc.index, tool_name=dtc.function.name if dtc.function is not None else None, args=dtc.function.arguments if dtc.function is not None else None, tool_call_id=dtc.id, ) if maybe_event is not None: if isinstance(maybe_event, PartStartEvent) and isinstance( self._parts_manager.get_part_by_vendor_id(self._vendor_part_id), TextPart ): self._vendor_part_id = f'{self._vendor_part_id}-{maybe_event.index}' yield maybe_event def _map_provider_details(self, chunk: ChatCompletionChunk) -> dict[str, Any] | None: """Hook that generates the provider details from chunk content. This method may be overridden by subclasses of `OpenAIStreamResponse` to customize the provider details. """ return _map_provider_details(chunk.choices[0]) def _map_usage(self, response: ChatCompletionChunk) -> usage.RequestUsage: return _map_usage(response, self._provider_name, self._provider_url, self.model_name) def _map_finish_reason( self, key: Literal['stop', 'length', 'tool_calls', 'content_filter', 'function_call'] ) -> FinishReason | None: """Hooks that maps a finish reason key to a [FinishReason](pydantic_ai.messages.FinishReason). This method may be overridden by subclasses of `OpenAIChatModel` to accommodate custom keys. """ return _CHAT_FINISH_REASON_MAP.get(key) @property def model_name(self) -> OpenAIModelName: """Get the model name of the response.""" return self._model_name @property def provider_name(self) -> str: """Get the provider name.""" return self._provider_name @property def provider_url(self) -> str: """Get the provider base URL.""" return self._provider_url @property def timestamp(self) -> datetime: """Get the timestamp of the response.""" return self._timestamp @dataclass class _ModelResponseStreamedResponse(StreamedResponse): """`StreamedResponse` wrapper for pre-built `ModelResponse` objects.""" _model_response: ModelResponse def __post_init__(self) -> None: self._usage = self._model_response.usage self.provider_response_id = self._model_response.provider_response_id self.provider_details = self._model_response.provider_details self.finish_reason = self._model_response.finish_reason self.state = self._model_response.state self.metadata = self._model_response.metadata for index, part in enumerate(self._model_response.parts): self._parts_manager.handle_part(vendor_part_id=index, part=part) async def _get_event_iterator(self) -> AsyncIterator[ModelResponseStreamEvent]: if False: # pragma: no cover yield cast(ModelResponseStreamEvent, None) async def close_stream(self) -> None: # No live connection to tear down: this wraps an already-retrieved `ModelResponse` (e.g. a # cursor-less background resume). `cancel()` and the continuation composite's teardown call this, # so it must no-op rather than inherit the base `NotImplementedError`; a server-side background # job is cancelled via `cancel_suspended_response`, not here. pass @property def model_name(self) -> str: # model_name is always set when _ModelResponseStreamedResponse is constructed assert self._model_response.model_name is not None return self._model_response.model_name @property def provider_name(self) -> str | None: return self._model_response.provider_name @property def provider_url(self) -> str | None: return self._model_response.provider_url @property def timestamp(self) -> datetime: return self._model_response.timestamp @dataclass class OpenAIResponsesStreamedResponse(StreamedResponse): """Implementation of `StreamedResponse` for OpenAI Responses API.""" _model_name: OpenAIModelName _model_settings: OpenAIResponsesModelSettings _response: _utils.PeekableAsyncStream[responses.ResponseStreamEvent, AsyncStream[responses.ResponseStreamEvent]] _provider_name: str _model_id_namespace: str _provider_url: str _tool_call_ids_are_response_scoped: bool _provider_timestamp: datetime | None = None _timestamp: datetime = field(default_factory=_now_utc) _has_refusal: bool = field(default=False, init=False) _refusal_text: str = field(default='', init=False) _last_sequence_number: int | None = field(default=None, init=False) def _set_state(self, status: ResponseStatus | None) -> None: # `_track_background` stamps the `background` marker from `response.background` on every status # event before this runs, so a pending status only suspends for a genuine (resumable) background job. background = bool((self.provider_details or {}).get('background')) self.state = _response_status_to_state(status, background=background) def _track_background(self, response: responses.Response) -> None: """Mark a background job so `cancel_suspended_response` can cancel it server-side. `responses.cancel` only works on background jobs; an explicit `background` marker (stamped from the API's own `response.background` field on every status event, so it survives the stream ending suspended, resuming via `retrieve`, or completing) keeps the cancel guard robust, rather than inferring background mode from the `continuation_delay` poll interval. """ if response.background: self.provider_details = {**(self.provider_details or {}), 'background': True} async def close_stream(self) -> None: await self._response.source.close() async def _get_event_iterator(self) -> AsyncIterator[ModelResponseStreamEvent]: # noqa: C901 with _map_api_errors(self._model_name, self._model_id_namespace): # Track annotations by item_id and content_index _annotations_by_item: dict[str, list[Any]] = {} # Track `phase` (commentary | final_answer) on assistant message items, captured # from the `output_item.added` event and merged into the corresponding # `TextPart.provider_details` on the first `output_text.delta` (so consumers can # filter commentary from final-answer text as it streams, rather than having to # buffer the whole part), or on `output_text.done` if no delta was received. _phase_by_item: dict[str, Literal['commentary', 'final_answer']] = {} mcp_list_tools_return_ids: set[str] = set() pending_tool_search_call_ids: deque[str] = deque() if self._provider_timestamp is not None: # pragma: no branch self.provider_details = {'timestamp': self._provider_timestamp} async for chunk in self._response: self._last_sequence_number = chunk.sequence_number if isinstance( chunk, ( responses.ResponseCreatedEvent, responses.ResponseInProgressEvent, responses.ResponseQueuedEvent, responses.ResponseCompletedEvent, responses.ResponseFailedEvent, responses.ResponseIncompleteEvent, ), ): # Stamp the background marker from any status event that carries it, so it survives a # stream that starts mid-way (a resumed `retrieve(stream=True)` whose first event is # `in_progress`/`queued`) or only reaches a terminal event. `cancel_suspended_response` # relies on it to cancel the server-side job. self._track_background(chunk.response) # NOTE: You can inspect the builtin tools used checking the `ResponseCompletedEvent`. if isinstance(chunk, responses.ResponseCompletedEvent): # Only the return part is backfilled; the call part is already emitted via `output_item.added`. # Backfill mcp_list_tools results missing from streamed output_item.done events (see https://github.com/pydantic/pydantic-ai/issues/5419). for item in chunk.response.output: if ( isinstance(item, responses.response_output_item.McpListTools) and item.id not in mcp_list_tools_return_ids ): _, return_part = _map_mcp_list_tools(item, self.provider_name) yield self._parts_manager.handle_part(vendor_part_id=f'{item.id}-return', part=return_part) mcp_list_tools_return_ids.add(item.id) self._usage += self._map_usage(chunk.response) self._store_conversation_id(chunk.response) self._set_state(chunk.response.status) raw_finish_reason = ( details.reason if (details := chunk.response.incomplete_details) else chunk.response.status ) if raw_finish_reason: # pragma: no branch if not self._has_refusal: self.provider_details = { **(self.provider_details or {}), 'finish_reason': raw_finish_reason, } self.finish_reason = _RESPONSES_FINISH_REASON_MAP.get(raw_finish_reason) if chunk.response.moderation: self.provider_details = { **(self.provider_details or {}), 'moderation': chunk.response.moderation.model_dump(), } elif isinstance(chunk, responses.ResponseContentPartAddedEvent): pass # there's nothing we need to do here elif isinstance(chunk, responses.ResponseContentPartDoneEvent): pass # there's nothing we need to do here elif isinstance(chunk, responses.ResponseCreatedEvent): self._set_state(chunk.response.status) if chunk.response.id: # pragma: no branch self.provider_response_id = chunk.response.id self._store_conversation_id(chunk.response) elif isinstance(chunk, responses.ResponseFailedEvent): self._usage += self._map_usage(chunk.response) # Parity with the non-streaming `_process_response`: a `failed` status maps to 'error'. if not self._has_refusal: self.provider_details = {**(self.provider_details or {}), 'finish_reason': 'failed'} self.finish_reason = _RESPONSES_FINISH_REASON_MAP.get('failed') self._set_state(chunk.response.status) elif isinstance(chunk, responses.ResponseFunctionCallArgumentsDeltaEvent): maybe_event = self._parts_manager.handle_tool_call_delta( vendor_part_id=chunk.item_id, args=chunk.delta, ) if maybe_event is not None: # pragma: no branch yield maybe_event elif isinstance(chunk, responses.ResponseFunctionCallArgumentsDoneEvent): pass # there's nothing we need to do here elif isinstance(chunk, (responses.ResponseInProgressEvent, responses.ResponseQueuedEvent)): self._usage += self._map_usage(chunk.response) self._set_state(chunk.response.status) elif isinstance(chunk, responses.ResponseIncompleteEvent): self._usage += self._map_usage(chunk.response) # Parity with the non-streaming `_process_response`: map the incomplete # reason when the provider sends one, leave the finish reason unset otherwise. raw_finish_reason = details.reason if (details := chunk.response.incomplete_details) else None if raw_finish_reason and not self._has_refusal: self.provider_details = {**(self.provider_details or {}), 'finish_reason': raw_finish_reason} self.finish_reason = _RESPONSES_FINISH_REASON_MAP.get(raw_finish_reason) self._set_state(chunk.response.status) elif isinstance(chunk, responses.ResponseOutputItemAddedEvent): if isinstance(chunk.item, responses.ResponseFunctionToolCall): # Preserve any `namespace` the Responses API attaches to a # discovered deferred tool so it can be round-tripped on replay. fn_provider_details: dict[str, Any] | None = None if chunk.item.namespace: fn_provider_details = {'namespace': chunk.item.namespace} yield self._parts_manager.handle_tool_call_part( vendor_part_id=chunk.item.id, tool_name=chunk.item.name, args=chunk.item.arguments, tool_call_id=_response_tool_call_id( chunk.item.call_id, self.provider_response_id if self._tool_call_ids_are_response_scoped else None, ), id=chunk.item.id, provider_name=self.provider_name, provider_details=fn_provider_details, ) elif isinstance(chunk.item, responses.ResponseReasoningItem): pass elif isinstance(chunk.item, responses.ResponseOutputMessage): if chunk.item.phase is not None: _phase_by_item[chunk.item.id] = chunk.item.phase elif isinstance(chunk.item, responses.ResponseFunctionWebSearch): call_part, _ = _map_web_search_tool_call(chunk.item, self.provider_name) yield self._parts_manager.handle_part( vendor_part_id=f'{chunk.item.id}-call', part=replace(call_part, args=None) ) elif isinstance(chunk.item, responses.ResponseToolSearchCall): if chunk.item.execution == 'client': # Emit a regular function-tool call so the standard # tool-execution path runs our local `search_tools`. client_call_part = _map_client_tool_search_call(chunk.item, self.provider_name) yield self._parts_manager.handle_tool_call_part( vendor_part_id=chunk.item.id, tool_name=client_call_part.tool_name, args=None, tool_call_id=client_call_part.tool_call_id, id=client_call_part.id, provider_name=self.provider_name, ) else: call_part = _map_tool_search_call(chunk.item, self.provider_name) if chunk.item.call_id is None: pending_tool_search_call_ids.append(call_part.tool_call_id) yield self._parts_manager.handle_part( vendor_part_id=f'{chunk.item.id}-call', part=replace(call_part, args=None) ) elif isinstance(chunk.item, responses.ResponseFileSearchToolCall): call_part, _ = _map_file_search_tool_call(chunk.item, self.provider_name) yield self._parts_manager.handle_part( vendor_part_id=f'{chunk.item.id}-call', part=replace(call_part, args=None) ) elif isinstance(chunk.item, responses.ResponseToolSearchOutputItem): # The completed result is emitted from `output_item.done` below. pass elif isinstance(chunk.item, responses.ResponseCodeInterpreterToolCall): call_part, _, _ = _map_code_interpreter_tool_call(chunk.item, self.provider_name) args_json = call_part.args_as_json_str() # Drop the final `"}` so that we can add code deltas args_json_delta = args_json[:-2] assert args_json_delta.endswith('"code":"'), ( f'Expected {args_json_delta!r} to end in `"code":"`' ) yield self._parts_manager.handle_part( vendor_part_id=f'{chunk.item.id}-call', part=replace(call_part, args=None) ) maybe_event = self._parts_manager.handle_tool_call_delta( vendor_part_id=f'{chunk.item.id}-call', args=args_json_delta, ) if maybe_event is not None: # pragma: no branch yield maybe_event elif isinstance(chunk.item, responses.response_output_item.ImageGenerationCall): call_part, _, _ = _map_image_generation_tool_call(chunk.item, self.provider_name) yield self._parts_manager.handle_part(vendor_part_id=f'{chunk.item.id}-call', part=call_part) elif isinstance(chunk.item, responses.response_output_item.McpCall): call_part, _ = _map_mcp_call(chunk.item, self.provider_name) args_json = call_part.args_as_json_str() # Drop the final `{}}` so that we can add tool args deltas args_json_delta = args_json[:-3] assert args_json_delta.endswith('"tool_args":'), ( f'Expected {args_json_delta!r} to end in `"tool_args":"`' ) yield self._parts_manager.handle_part( vendor_part_id=f'{chunk.item.id}-call', part=replace(call_part, args=None) ) maybe_event = self._parts_manager.handle_tool_call_delta( vendor_part_id=f'{chunk.item.id}-call', args=args_json_delta, ) if maybe_event is not None: # pragma: no branch yield maybe_event elif isinstance(chunk.item, responses.response_output_item.McpListTools): call_part, _ = _map_mcp_list_tools(chunk.item, self.provider_name) yield self._parts_manager.handle_part(vendor_part_id=f'{chunk.item.id}-call', part=call_part) elif isinstance(chunk.item, ResponseCompactionItem): # Emit a PartStartEvent so UIs can render compaction in progress. # The "done" event replaces this with the final encrypted_content. yield self._parts_manager.handle_part( vendor_part_id=chunk.item.id, part=_map_compaction_item(chunk.item, self.provider_name), ) else: warnings.warn( # pragma: no cover f'Handling of this item type is not yet implemented. Please report on our GitHub: {chunk}', UserWarning, ) elif isinstance(chunk, responses.ResponseOutputItemDoneEvent): if isinstance(chunk.item, responses.ResponseReasoningItem): if signature := chunk.item.encrypted_content: # pragma: no branch # Add the signature to the part corresponding to the first summary/raw CoT for event in self._parts_manager.handle_thinking_delta( vendor_part_id=chunk.item.id, id=chunk.item.id, signature=signature, provider_name=self.provider_name, ): yield event elif isinstance(chunk.item, responses.ResponseCodeInterpreterToolCall): _, return_part, file_parts = _map_code_interpreter_tool_call(chunk.item, self.provider_name) for i, file_part in enumerate(file_parts): yield self._parts_manager.handle_part( vendor_part_id=f'{chunk.item.id}-file-{i}', part=file_part ) yield self._parts_manager.handle_part( vendor_part_id=f'{chunk.item.id}-return', part=return_part ) elif isinstance(chunk.item, responses.ResponseFunctionWebSearch): call_part, return_part = _map_web_search_tool_call(chunk.item, self.provider_name) maybe_event = self._parts_manager.handle_tool_call_delta( vendor_part_id=f'{chunk.item.id}-call', args=call_part.args, ) if maybe_event is not None: # pragma: no branch yield maybe_event yield self._parts_manager.handle_part( vendor_part_id=f'{chunk.item.id}-return', part=return_part ) elif isinstance(chunk.item, responses.ResponseToolSearchCall): if chunk.item.execution == 'client': # Feed the final args through the tool-call delta path so the # part manager resolves its pending args; there's no paired # `tool_search_output` — the return part is produced by the # local tool runner after the stream completes. # # OpenAI Responses streaming sometimes assigns a different # `call_id` between the `output_item.added` and `output_item.done` # frames for client-executed `tool_search_call`s; the server's # log of record is the latter. Pass it through here so the # `tool_search_output` we replay on the next turn references the # call_id the API actually expects. client_call_part = _map_client_tool_search_call(chunk.item, self.provider_name) client_args_json = client_call_part.args_as_json_str() maybe_event = self._parts_manager.handle_tool_call_delta( vendor_part_id=chunk.item.id, args=client_args_json, tool_call_id=client_call_part.tool_call_id, ) if maybe_event is not None: # pragma: no branch yield maybe_event else: call_part = _map_tool_search_call(chunk.item, self.provider_name) maybe_event = self._parts_manager.handle_tool_call_delta( vendor_part_id=f'{chunk.item.id}-call', args=cast('str | dict[str, Any] | None', call_part.args), provider_details=call_part.provider_details, ) if maybe_event is not None: # pragma: no branch yield maybe_event elif isinstance(chunk.item, responses.ResponseToolSearchOutputItem): if chunk.item.execution == 'server': call_id = _match_tool_search_output_to_call( chunk.item.call_id, pending_tool_search_call_ids ) yield self._parts_manager.handle_part( vendor_part_id=f'{chunk.item.id}-return', part=_build_tool_search_return_part( call_id if call_id is not None else chunk.item.id, chunk.item, self.provider_name, ), ) elif isinstance(chunk.item, responses.ResponseFileSearchToolCall): call_part, return_part = _map_file_search_tool_call(chunk.item, self.provider_name) maybe_event = self._parts_manager.handle_tool_call_delta( vendor_part_id=f'{chunk.item.id}-call', args=call_part.args, ) if maybe_event is not None: # pragma: no branch yield maybe_event yield self._parts_manager.handle_part( vendor_part_id=f'{chunk.item.id}-return', part=return_part ) elif isinstance(chunk.item, responses.response_output_item.ImageGenerationCall): _, return_part, file_part = _map_image_generation_tool_call(chunk.item, self.provider_name) if file_part: # pragma: no branch yield self._parts_manager.handle_part( vendor_part_id=f'{chunk.item.id}-file', part=file_part ) yield self._parts_manager.handle_part( vendor_part_id=f'{chunk.item.id}-return', part=return_part ) elif isinstance(chunk.item, responses.response_output_item.McpCall): _, return_part = _map_mcp_call(chunk.item, self.provider_name) yield self._parts_manager.handle_part( vendor_part_id=f'{chunk.item.id}-return', part=return_part ) elif isinstance(chunk.item, responses.response_output_item.McpListTools): _, return_part = _map_mcp_list_tools(chunk.item, self.provider_name) yield self._parts_manager.handle_part( vendor_part_id=f'{chunk.item.id}-return', part=return_part ) mcp_list_tools_return_ids.add(chunk.item.id) elif isinstance(chunk.item, ResponseCompactionItem): # Replace the preliminary part from the "added" event with the # final encrypted_content for round-tripping. yield self._parts_manager.handle_part( vendor_part_id=chunk.item.id, part=_map_compaction_item(chunk.item, self.provider_name), ) elif isinstance(chunk, responses.ResponseReasoningSummaryPartAddedEvent): # Use same vendor_part_id as raw CoT for first summary (index 0) so they merge into one ThinkingPart vendor_id = chunk.item_id if chunk.summary_index == 0 else f'{chunk.item_id}-{chunk.summary_index}' for event in self._parts_manager.handle_thinking_delta( vendor_part_id=vendor_id, content=chunk.part.text, id=chunk.item_id, provider_name=self.provider_name, ): yield event elif isinstance(chunk, responses.ResponseReasoningSummaryPartDoneEvent): pass # there's nothing we need to do here elif isinstance(chunk, responses.ResponseReasoningSummaryTextDoneEvent): pass # there's nothing we need to do here elif isinstance(chunk, responses.ResponseReasoningSummaryTextDeltaEvent): # Use same vendor_part_id as raw CoT for first summary (index 0) so they merge into one ThinkingPart vendor_id = chunk.item_id if chunk.summary_index == 0 else f'{chunk.item_id}-{chunk.summary_index}' for event in self._parts_manager.handle_thinking_delta( vendor_part_id=vendor_id, content=chunk.delta, id=chunk.item_id, provider_name=self.provider_name, ): yield event elif isinstance(chunk, responses.ResponseReasoningTextDeltaEvent): # Handle raw CoT from gpt-oss models using callback pattern for event in self._parts_manager.handle_thinking_delta( vendor_part_id=chunk.item_id, id=chunk.item_id, provider_name=self.provider_name, provider_details=_make_raw_content_updater(chunk.delta, chunk.content_index), ): yield event elif isinstance(chunk, responses.ResponseReasoningTextDoneEvent): pass # content already accumulated via delta events elif isinstance(chunk, responses.ResponseOutputTextAnnotationAddedEvent): # Collect annotations if the setting is enabled if self._model_settings.get('openai_include_raw_annotations'): # `openai` 3.1 retyped `annotation` from `object` to a model union declared in the # event's own module, whose members are distinct classes from the identically # shaped ones `ResponseOutputText.annotations` uses. That distinction is invisible # in the payload but fatal to `responses_output_text_annotations_ta`, so normalize # to the wire dict both SDK shapes carry rather than serializing by type. annotation = chunk.annotation _annotations_by_item.setdefault(chunk.item_id, []).append( annotation.model_dump(mode='json') if isinstance(annotation, BaseModel) else annotation ) elif isinstance(chunk, responses.ResponseTextDeltaEvent): # Guard against delta=null from OpenAI-compatible gateways (e.g. Bifrost). if chunk.delta is not None: # pyright: ignore[reportUnnecessaryComparison] # Pop so the phase rides along with the `PartStartEvent` for the new text part # and isn't repeated on every subsequent delta. delta_provider_details: dict[str, Any] | None = None if (phase := _phase_by_item.pop(chunk.item_id, None)) is not None: delta_provider_details = {'phase': phase} for event in self._parts_manager.handle_text_delta( vendor_part_id=chunk.item_id, content=chunk.delta, id=chunk.item_id, provider_name=self.provider_name, provider_details=delta_provider_details, ): yield event elif isinstance(chunk, responses.ResponseTextDoneEvent): # Add annotations to provider_details if available provider_details: dict[str, Any] = {} annotations = _annotations_by_item.get(chunk.item_id) if annotations: provider_details['annotations'] = list(annotations) if chunk.logprobs: provider_details['logprobs'] = _map_logprobs(chunk.logprobs) if (phase := _phase_by_item.get(chunk.item_id)) is not None: provider_details['phase'] = phase if provider_details: for event in self._parts_manager.handle_text_delta( vendor_part_id=chunk.item_id, content='', provider_name=self.provider_name, provider_details=provider_details, ): yield event elif isinstance(chunk, responses.ResponseRefusalDeltaEvent): # Accumulate refusal text from deltas as a fallback in case the done event is missing. self._has_refusal = True self.finish_reason = 'content_filter' self._refusal_text += chunk.delta elif isinstance(chunk, responses.ResponseRefusalDoneEvent): # The done event contains the full refusal text, replacing any accumulated deltas. self._has_refusal = True self.finish_reason = 'content_filter' self._refusal_text = chunk.refusal elif isinstance(chunk, responses.ResponseWebSearchCallInProgressEvent): pass # there's nothing we need to do here elif isinstance(chunk, responses.ResponseWebSearchCallSearchingEvent): pass # there's nothing we need to do here elif isinstance(chunk, responses.ResponseWebSearchCallCompletedEvent): pass # there's nothing we need to do here elif isinstance(chunk, responses.ResponseAudioDeltaEvent): # pragma: lax no cover pass # there's nothing we need to do here elif isinstance(chunk, responses.ResponseCodeInterpreterCallCodeDeltaEvent): json_args_delta = to_json(chunk.delta).decode()[1:-1] # Drop the surrounding `"` maybe_event = self._parts_manager.handle_tool_call_delta( vendor_part_id=f'{chunk.item_id}-call', args=json_args_delta, ) if maybe_event is not None: # pragma: no branch yield maybe_event elif isinstance(chunk, responses.ResponseCodeInterpreterCallCodeDoneEvent): maybe_event = self._parts_manager.handle_tool_call_delta( vendor_part_id=f'{chunk.item_id}-call', args='"}', ) if maybe_event is not None: # pragma: no branch yield maybe_event elif isinstance(chunk, responses.ResponseCodeInterpreterCallCompletedEvent): pass # there's nothing we need to do here elif isinstance(chunk, responses.ResponseCodeInterpreterCallInProgressEvent): pass # there's nothing we need to do here elif isinstance(chunk, responses.ResponseCodeInterpreterCallInterpretingEvent): pass # there's nothing we need to do here elif isinstance(chunk, responses.ResponseImageGenCallCompletedEvent): # pragma: no cover pass # there's nothing we need to do here elif isinstance(chunk, responses.ResponseImageGenCallGeneratingEvent): pass # there's nothing we need to do here elif isinstance(chunk, responses.ResponseImageGenCallInProgressEvent): pass # there's nothing we need to do here elif isinstance(chunk, responses.ResponseImageGenCallPartialImageEvent): # Not present on the type, but present on the actual object. # See https://github.com/openai/openai-python/issues/2649 output_format = getattr(chunk, 'output_format', 'png') file_part = FilePart( content=BinaryImage( data=base64.b64decode(chunk.partial_image_b64), media_type=f'image/{output_format}', ), id=chunk.item_id, ) yield self._parts_manager.handle_part(vendor_part_id=f'{chunk.item_id}-file', part=file_part) elif isinstance(chunk, responses.ResponseMcpCallArgumentsDoneEvent): maybe_event = self._parts_manager.handle_tool_call_delta( vendor_part_id=f'{chunk.item_id}-call', args='}', ) if maybe_event is not None: # pragma: no branch yield maybe_event elif isinstance(chunk, responses.ResponseMcpCallArgumentsDeltaEvent): maybe_event = self._parts_manager.handle_tool_call_delta( vendor_part_id=f'{chunk.item_id}-call', args=chunk.delta, ) if maybe_event is not None: # pragma: no branch yield maybe_event elif isinstance(chunk, responses.ResponseMcpListToolsInProgressEvent): pass # there's nothing we need to do here elif isinstance(chunk, responses.ResponseMcpListToolsCompletedEvent): pass # there's nothing we need to do here elif isinstance(chunk, responses.ResponseMcpListToolsFailedEvent): # pragma: no cover pass # there's nothing we need to do here elif isinstance(chunk, responses.ResponseMcpCallInProgressEvent): pass # there's nothing we need to do here elif isinstance(chunk, responses.ResponseMcpCallFailedEvent): # pragma: no cover pass # there's nothing we need to do here elif isinstance(chunk, responses.ResponseMcpCallCompletedEvent): pass # there's nothing we need to do here elif isinstance(chunk, responses.ResponseFileSearchCallCompletedEvent): pass # there's nothing we need to do here elif isinstance(chunk, responses.ResponseFileSearchCallSearchingEvent): pass # there's nothing we need to do here elif isinstance(chunk, responses.ResponseFileSearchCallInProgressEvent): pass # there's nothing we need to do here else: # pragma: no cover warnings.warn( f'Handling of this event type is not yet implemented. Please report on our GitHub: {chunk}', UserWarning, ) if self._refusal_text: self.provider_details = {**(self.provider_details or {}), 'refusal': self._refusal_text} # This is used to resume suspended background streams with `starting_after`. if self.state == 'suspended' and self._last_sequence_number is not None: self.provider_details = { **(self.provider_details or {}), 'last_sequence_number': self._last_sequence_number, } def _store_conversation_id(self, response: responses.Response) -> None: if response.conversation: self.provider_details = {**(self.provider_details or {}), 'conversation_id': response.conversation.id} def _map_usage(self, response: responses.Response) -> usage.RequestUsage: return _map_usage(response, self._provider_name, self._provider_url, self.model_name) @property def model_name(self) -> OpenAIModelName: """Get the model name of the response.""" return self._model_name @property def provider_name(self) -> str: """Get the provider name.""" return self._provider_name @property def provider_url(self) -> str: """Get the provider base URL.""" return self._provider_url @property def timestamp(self) -> datetime: """Get the timestamp of the response.""" return self._timestamp @dataclass(init=False) class OpenAICompaction(AbstractCapability[AgentDepsT]): """Compaction capability for OpenAI Responses API. Automatically compacts conversation history to keep long-running agent runs within manageable context limits. Two modes are supported, selected by the `stateless` flag: - **Stateful mode** (default, `stateless=False`): configures [OpenAI's server-side auto-compaction](https://developers.openai.com/api/docs/guides/compaction) via the `context_management` field on the regular `/responses` request. The server triggers compaction when input tokens cross a threshold, and the compacted item is returned alongside the normal response. On subsequent requests, only that item and the content after it are sent. Compatible with [`openai_previous_response_id='auto'`][pydantic_ai.models.openai.OpenAIResponsesModelSettings.openai_previous_response_id] and server-side conversation state. Configurable with `token_threshold` (`compact_threshold` on the API). If omitted, OpenAI picks a server-side default. - **Stateless mode** (`stateless=True`): calls the stateless `/responses/compact` endpoint from a `before_model_request` hook when your trigger condition is met. Use this in [ZDR](https://openai.com/enterprise-privacy/) environments where OpenAI must not retain conversation data, when you set [`openai_store=False`][pydantic_ai.models.openai.OpenAIResponsesModelSettings.openai_store], or when you need explicit out-of-band control over when compaction runs. Requires either `message_count_threshold` or a custom `trigger` callable. If `stateless` is not set, it is inferred from which parameters you provide: passing any stateless-only parameter (`message_count_threshold` or `trigger`) implies `stateless=True`; otherwise stateful mode is used. Example usage: ```python {test="skip"} from pydantic_ai import Agent from pydantic_ai.models.openai import OpenAICompaction # Stateful mode with OpenAI's server-side default threshold: agent = Agent( 'openai-responses:gpt-5.2', capabilities=[OpenAICompaction()], ) # Stateful mode with a custom token threshold: agent = Agent( 'openai-responses:gpt-5.2', capabilities=[OpenAICompaction(token_threshold=100_000)], ) # Stateless mode for ZDR environments or explicit control: agent = Agent( 'openai-responses:gpt-5.2', capabilities=[OpenAICompaction(message_count_threshold=20)], ) ``` """ def __init__( self, *, stateless: bool | None = None, token_threshold: int | None = None, message_count_threshold: int | None = None, trigger: Callable[[list[ModelMessage]], bool] | None = None, ) -> None: """Initialize the OpenAI compaction capability. Args: stateless: Select the compaction mode explicitly. If `None` (the default), the mode is inferred from the other parameters: passing any stateless-only parameter (`message_count_threshold` or `trigger`) implies `stateless=True`; otherwise stateful mode is used. token_threshold: Stateful-mode only. Input token threshold at which OpenAI's server-side compaction is triggered. Corresponds to `compact_threshold` in the `context_management` API field. If `None`, OpenAI picks a server-side default. message_count_threshold: Stateless-mode only. Compact when the message count exceeds this threshold. trigger: Stateless-mode only. Custom callable that decides whether to compact based on the current messages. Takes precedence over `message_count_threshold`. """ has_stateless_only = message_count_threshold is not None or trigger is not None has_stateful_only = token_threshold is not None if stateless is None: stateless = has_stateless_only if stateless: if has_stateful_only: raise UserError( '`token_threshold` is only valid for stateful compaction (`stateless=False`). ' 'For stateless `/compact` endpoint compaction, use `message_count_threshold` or `trigger`.' ) if not has_stateless_only: raise UserError( '`stateless=True` requires `message_count_threshold` or `trigger` ' 'to determine when to invoke the `/compact` endpoint.' ) else: if has_stateless_only: raise UserError( '`message_count_threshold` and `trigger` are only valid for stateless compaction ' '(`stateless=True`). For stateful server-side compaction, use `token_threshold` ' '(or omit it to use the OpenAI-managed default).' ) self.stateless = stateless self.token_threshold = token_threshold self.message_count_threshold = message_count_threshold self.trigger = trigger def get_model_settings(self) -> Callable[[RunContext[AgentDepsT]], ModelSettings] | None: if self.stateless: return None edit: ContextManagement = {'type': 'compaction'} if self.token_threshold is not None: edit['compact_threshold'] = self.token_threshold def resolve(ctx: RunContext[AgentDepsT]) -> ModelSettings: # If the user already set `openai_context_management` on their model settings, # defer to it entirely — we don't want to end up with two conflicting `compaction` # entries, since OpenAI's context_management list only meaningfully supports one. if ctx.model_settings: existing = cast(dict[str, Any], ctx.model_settings).get('openai_context_management') if existing: return cast(ModelSettings, {}) return cast(ModelSettings, {'openai_context_management': [edit]}) return resolve def _should_compact(self, messages: list[ModelMessage]) -> bool: if not self.stateless: return False if self.trigger is not None: return self.trigger(messages) if self.message_count_threshold is not None: return len(messages) > self.message_count_threshold return False # pragma: no cover async def before_model_request( self, ctx: RunContext[AgentDepsT], request_context: ModelRequestContext, ) -> ModelRequestContext: if not self._should_compact(request_context.messages): return request_context from .wrapper import WrapperModel model = request_context.model while isinstance(model, WrapperModel): model = model.wrapped if not isinstance(model, OpenAIResponsesModel): raise UserError( f'OpenAICompaction requires OpenAIResponsesModel, got {type(model).__name__}. ' f'Use the provider-specific compaction capability for your model.' ) # Need at least 2 messages (history + current request) to compact if len(request_context.messages) < 2: # pragma: no cover return request_context # Compact all messages except the last (current) request compact_ctx = ModelRequestContext( model=request_context.model, messages=request_context.messages[:-1], model_settings=request_context.model_settings, model_request_parameters=request_context.model_request_parameters, ) compacted_response = await request_context.model.compact_messages(compact_ctx) # Replace message history with compaction + last request request_context.messages = [compacted_response, request_context.messages[-1]] return request_context @classmethod def get_serialization_name(cls) -> str | None: return 'OpenAICompaction' def _make_raw_content_updater(delta: str, index: int) -> Callable[[dict[str, Any] | None], dict[str, Any]]: """Create a callback that updates `provider_details['raw_content']`. This is used for streaming raw CoT from gpt-oss models. The callback pattern keeps `raw_content` logic in OpenAI code while the parts manager stays provider-agnostic. """ def update_provider_details(existing: dict[str, Any] | None) -> dict[str, Any]: details = {**(existing or {})} raw_list: list[str] = list(details.get('raw_content', [])) while len(raw_list) <= index: raw_list.append('') raw_list[index] += delta details['raw_content'] = raw_list return details return update_provider_details # Convert logprobs to a serializable format def _map_logprobs( logprobs: list[chat_completion_token_logprob.ChatCompletionTokenLogprob] | list[responses.response_output_text.Logprob] | list[responses.response_text_done_event.Logprob], ) -> list[dict[str, Any]]: return [ { 'token': lp.token, 'bytes': lp.bytes if not isinstance(lp, responses.response_text_done_event.Logprob) else None, 'logprob': lp.logprob, 'top_logprobs': [ { 'token': tlp.token, 'bytes': tlp.bytes if not isinstance(tlp, responses.response_text_done_event.LogprobTopLogprob) else None, 'logprob': tlp.logprob, } for tlp in (lp.top_logprobs or []) ], } for lp in logprobs ] def _support_tool_forcing( model_name: str, openai_profile: OpenAIModelProfile, model_settings: OpenAIChatModelSettings | OpenAIResponsesModelSettings, model_request_parameters: ModelRequestParameters, ) -> bool: """Check if the model supports forced tool use, raising UserError if explicitly requested but unsupported.""" if not openai_profile.get('openai_supports_tool_choice_required', True): return _reject_tool_forcing( model_name, model_settings, model_request_parameters, 'This model does not support forcing tool use.', ) if not openai_profile.get('openai_supports_forced_tool_choice_with_thinking', True) and _reasoning_active( openai_profile, model_settings, model_request_parameters ): return _reject_tool_forcing( model_name, model_settings, model_request_parameters, 'This model does not support forcing tool use while thinking is enabled. ' "Disable thinking with `thinking=False` or `openai_reasoning_effort='none'`, " "or use `tool_choice='auto'`.", ) return True def _reject_tool_forcing( model_name: str, model_settings: OpenAIChatModelSettings | OpenAIResponsesModelSettings, model_request_parameters: ModelRequestParameters, reason: str, ) -> bool: """Fall back to unforced tool choice, unless the user asked for forcing explicitly.""" explicit_choice = model_settings.get('tool_choice') # `resolve_tool_choice` maps `ToolOrOutput` to required mode when direct output isn't allowed, # so that shape requests forcing just as explicitly as `'required'` or a tool list. explicit_forcing = ( explicit_choice == 'required' or isinstance(explicit_choice, list) or ( isinstance(explicit_choice, ToolOrOutput) and not (model_request_parameters.allow_text_output or model_request_parameters.allow_image_output) ) ) if explicit_forcing: raise UserError(f'tool_choice={explicit_choice!r} is not supported by model {model_name!r}. {reason}') return False def _map_compaction_item( item: ResponseCompactionItem, system: str, *, standing_prompt_planted: bool = False ) -> CompactionPart: """Convert an OpenAI `ResponseCompactionItem` to a `CompactionPart`. `standing_prompt_planted` stamps the part as minted by our own `responses.compact` call, whose input window explicitly plants the standing prompt — the provenance that lets the trim rely on the compaction item's retention instead of re-sending the standing prompt (see `_trim_messages_before_compaction`). Compaction items arriving in ordinary responses are left unstamped: their window may itself have relied on an earlier item's retention, and retention is only reliable for a single hop. """ provider_details = item.model_dump() if standing_prompt_planted: provider_details[STANDING_PROMPT_PLANTED_KEY] = True return CompactionPart( content=None, id=item.id, provider_name=system, provider_details=provider_details, ) def _map_usage( response: chat.ChatCompletion | ChatCompletionChunk | responses.Response | responses.CompactedResponse, provider: str, provider_url: str, model: str, ) -> usage.RequestUsage: response_usage = response.usage if response_usage is None: return usage.RequestUsage() usage_data = response_usage.model_dump(exclude_none=True) details = { k: v for k, v in usage_data.items() if k not in {'prompt_tokens', 'completion_tokens', 'input_tokens', 'output_tokens', 'total_tokens'} if isinstance(v, int) } response_data = dict(model=model, usage=usage_data) if isinstance(response_usage, responses.ResponseUsage): api_flavor = 'responses' input_tokens_details = usage_data.get('input_tokens_details') if getattr(response_usage, 'output_tokens_details', None) is not None: details['reasoning_tokens'] = getattr(response_usage.output_tokens_details, 'reasoning_tokens', 0) else: details['reasoning_tokens'] = 0 else: api_flavor = 'chat' input_tokens_details = usage_data.get('prompt_tokens_details') if response_usage.completion_tokens_details is not None: details.update(response_usage.completion_tokens_details.model_dump(exclude_none=True)) request_usage = usage.RequestUsage.extract( response_data, provider=provider, provider_url=provider_url, provider_fallback='openai', api_flavor=api_flavor, details=details, ) # genai-prices maps OpenAI's nested `cache_write_tokens` on the `openai` extractors as of # https://github.com/pydantic/genai-prices/pull/463 (in 0.1.4), but not every OpenAI-compatible # provider's extractor does — Azure's still omits it — so lift it manually here. # TODO: Remove this block once those remaining extractors map it. Check `prices/providers/azure.yml`. if _is_str_dict(input_tokens_details): cache_write_tokens = input_tokens_details.get('cache_write_tokens') if isinstance(cache_write_tokens, int): request_usage.cache_write_tokens = cache_write_tokens return request_usage def _map_provider_details( choice: chat_completion_chunk.Choice | chat_completion.Choice, ) -> dict[str, Any] | None: provider_details: dict[str, Any] = {} # Add logprobs to provider_details if available if choice.logprobs is not None and choice.logprobs.content: provider_details['logprobs'] = _map_logprobs(choice.logprobs.content) if raw_finish_reason := choice.finish_reason: provider_details['finish_reason'] = raw_finish_reason return provider_details or None def _group_settled_portable_function_calls( messages: list[ModelMessage], message_index: int, message: ModelResponse, *, client_tool_search_active: bool, ) -> Sequence[ModelResponsePart]: """Reorder one assistant turn's settled function calls to the end of their segment. Only for endpoints whose profile clears `openai_responses_supports_interleaved_function_calls`. Such an endpoint folds each call into the assistant message next to it, so an assistant item sitting between two calls splits them into separate messages, each carrying an unanswered call. Item IDs are deliberately *not* consulted: an endpoint that merges items this way derives an item's position from the surrounding sequence rather than its identity, so an ID pins nothing (verified against DeepSeek: the grouped order is accepted with reasoning, message and `function_call` IDs all present on the wire, while the interleaved order is rejected with them). Reordering is skipped when the turn carries a native or compaction item the provider owns, or when any of its calls is still unanswered — an unanswered call is rejected in either order, so grouping cannot rescue it. "Answered" is deliberately wire-local and stricter than `_agent_graph._dangling_tool_calls_by_response`: only the requests between this response and the next one count, and repeated call IDs are counted rather than shadowed. A model must not import the graph layer, and the two answer different questions, so they are allowed to disagree — but only in the direction where this one declines to reorder. Keep any change on that side. """ parts = message.parts if any(isinstance(part, (NativeToolCallPart, NativeToolReturnPart, CompactionPart)) for part in parts): return parts unsettled_call_counts: dict[str, int] = {} for part in parts: if isinstance(part, ToolCallPart) and not ( client_tool_search_active and part.tool_name == TOOL_SEARCH_FUNCTION_TOOL_NAME ): unsettled_call_counts[part.tool_call_id] = unsettled_call_counts.get(part.tool_call_id, 0) + 1 for following_message_index in range(message_index + 1, len(messages)): following_message = messages[following_message_index] if isinstance(following_message, ModelResponse): break for part in following_message.parts: if ( ( isinstance(part, ToolReturnPart) and not (client_tool_search_active and part.tool_name == TOOL_SEARCH_FUNCTION_TOOL_NAME) ) or ( isinstance(part, RetryPromptPart) and part.tool_name is not None and not (client_tool_search_active and part.tool_name == TOOL_SEARCH_FUNCTION_TOOL_NAME) ) ) and (count := unsettled_call_counts.get(part.tool_call_id)): if count == 1: del unsettled_call_counts[part.tool_call_id] else: unsettled_call_counts[part.tool_call_id] = count - 1 if unsettled_call_counts: return parts grouped_parts: list[ModelResponsePart] = [] segment: list[ModelResponsePart] = [] def flush_segment() -> None: grouped_parts.extend(part for part in segment if not isinstance(part, ToolCallPart)) grouped_parts.extend(part for part in segment if isinstance(part, ToolCallPart)) segment.clear() for part in parts: if ( client_tool_search_active and isinstance(part, ToolCallPart) and part.tool_name == TOOL_SEARCH_FUNCTION_TOOL_NAME ): flush_segment() grouped_parts.append(part) else: segment.append(part) flush_segment() return grouped_parts def _split_combined_tool_call_id(combined_id: str) -> tuple[str, str | None]: # When reasoning, the Responses API requires the `ResponseFunctionToolCall` to be returned with both the `call_id` and `id` fields. # Before our `ToolCallPart` gained the `id` field alongside `tool_call_id` field, we combined the two fields into a single string stored on `tool_call_id`. if '|' in combined_id: call_id, id = combined_id.split('|', 1) return call_id, id else: return combined_id, None def _response_tool_call_id(tool_call_id: str, response_id: str | None) -> str: """Qualify a Responses tool-call ID with its response ID, or return it unchanged. `response_id` is the provider response ID when the model's tool-call IDs are response-scoped (so the ID must be made history-wide unique), or `None` to leave the call ID as-is. """ return f'{response_id}:{tool_call_id}' if response_id is not None else tool_call_id def _provider_response_tool_call_id(tool_call_id: str, response_id: str | None) -> str: """Restore a response-scoped provider tool-call ID from its normalized form.""" if response_id is None: return tool_call_id return tool_call_id.removeprefix(f'{response_id}:') def _map_code_interpreter_tool_call( item: responses.ResponseCodeInterpreterToolCall, provider_name: str ) -> tuple[NativeToolCallPart, NativeToolReturnPart, list[FilePart]]: result: dict[str, Any] = { 'status': item.status, } file_parts: list[FilePart] = [] logs: list[str] = [] if item.outputs: for output in item.outputs: if isinstance(output, responses.response_code_interpreter_tool_call.OutputImage): file_parts.append( FilePart( content=BinaryImage.from_data_uri(output.url), id=item.id, ) ) elif isinstance(output, responses.response_code_interpreter_tool_call.OutputLogs): logs.append(output.logs) else: assert_never(output) if logs: result['logs'] = logs call_part = NativeToolCallPart( tool_name=CodeExecutionTool.kind, tool_call_id=item.id, args={ 'container_id': item.container_id, 'code': item.code or '', }, provider_name=provider_name, ) call_part.otel_metadata = {'code_arg_name': 'code', 'code_arg_language': 'python'} return ( call_part, NativeToolReturnPart( tool_name=CodeExecutionTool.kind, tool_call_id=item.id, content=result, provider_name=provider_name, ), file_parts, ) def _tool_search_namespace_for_synthesis( tool_name: str, model_request_parameters: ModelRequestParameters ) -> str | None: """Return the synthetic OpenAI namespace for a cross-provider replay, or `None`. OpenAI-origin calls round-trip `provider_details['namespace']`. Non-OpenAI history lacks that field, but OpenAI rejects a replayed call to a tool that isn't in the default namespace when the call doesn't say so (`Missing namespace for function_call '...'. It does not exist in the default namespace.`). For the flat deferred function tools this adapter emits, OpenAI-generated calls use `namespace == tool_name` — verified by live probe against a capability owning multiple deferred tools. What decides is the tool's placement on *this* request's wire, not what kind of tool it is: a name declared through an `additional_tools` item, or occupying a `tools` entry with its schema withheld behind `defer_loading`, lives outside the default namespace, while a plain `tools` entry is default-namespace even if stored history happens to mention its name — tagging that one would be exactly the mismatch the error above complains about, from the other direction. """ if tool_name not in model_request_parameters.revealed_tool_names: return None for tool in model_request_parameters.function_tools: if tool.name == tool_name and model_request_parameters.visibility_of(tool.name) in ( 'deferred', 'via_history', ): return tool_name return None def _has_tool_search(model_request_parameters: ModelRequestParameters) -> bool: """Whether the current run carries any `ToolSearchTool` builtin. Used to gate two related behaviors: * Filter the local `search_tools` function tool out of the wire `tools[]` (it's represented by the builtin instead). * Promote local-shape `ToolSearch*Part` history into native `tool_search_call` / `tool_search_output` items (`execution='client'`), unlocking previously-discovered `defer_loading=True` tools without forcing the model to re-search. Both behaviors apply regardless of the active strategy (default native, named native, or custom callable) — the gate is "tool search is active in some form". """ return any(isinstance(t, ToolSearchTool) for t in model_request_parameters.native_tools) def _find_search_tool_definition( model_request_parameters: ModelRequestParameters, ) -> ToolDefinition | None: """Locate the local `search_tools` function-tool definition in the current request. In custom-callable tool search mode, `ToolSearchToolset` leaves its `search_tools` function tool in `function_tools` (no `unless_native`), so we look it up by name. """ return next( (t for t in model_request_parameters.function_tools if t.name == TOOL_SEARCH_FUNCTION_TOOL_NAME), None, ) def _normalize_tool_search_args(raw: Any) -> ToolSearchArgs: """Translate an OpenAI `tool_search_call.arguments` payload into `ToolSearchArgs`. OpenAI's wire shape varies by execution mode: * **Server-executed `tool_search`**: the result carries the picked tool paths under `paths`. Fold those into the canonical `queries` slot for cross-provider history. * **Client-executed `tool_search` (`execution='client'`)**: arguments mirror the schema we registered for the local `search_tools` function tool — already `{"queries": list[str]}` — so pass through unchanged. Empty `arguments={}` (the streaming-mid first-event case) returns `{'queries': []}`. Any other shape we don't recognize raises `UnexpectedModelBehavior` — schema drift in the OpenAI SDK should surface loudly at the parse boundary, not silently degrade. """ if isinstance(raw, dict): if not raw: return {'queries': []} queries = raw.get('queries') # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType] if isinstance(queries, list): return {'queries': [q for q in queries if isinstance(q, str)]} # pyright: ignore[reportUnknownVariableType] paths = raw.get('paths') # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType] if isinstance(paths, list): return {'queries': [p for p in paths if isinstance(p, str)]} # pyright: ignore[reportUnknownVariableType] raise UnexpectedModelBehavior(f'Unrecognized tool_search arguments shape: {raw!r}') def _tool_search_output_call_ids(response: responses.Response) -> dict[str, str]: """Pair hosted tool-search outputs with calls, using response order when IDs are absent.""" pending_call_ids = deque( item.id for item in response.output if isinstance(item, responses.ResponseToolSearchCall) and item.execution == 'server' and item.call_id is None ) server_call_ids = { item.call_id or item.id for item in response.output if isinstance(item, responses.ResponseToolSearchCall) and item.execution == 'server' } output_call_ids: dict[str, str] = {} for item in response.output: if not isinstance(item, responses.ResponseToolSearchOutputItem) or item.execution != 'server': continue if item.call_id is not None and item.call_id not in server_call_ids: continue if (call_id := _match_tool_search_output_to_call(item.call_id, pending_call_ids)) is not None: output_call_ids[item.id] = call_id return output_call_ids def _match_tool_search_output_to_call(output_call_id: str | None, pending_call_ids: deque[str]) -> str | None: """Match an output to a call and remove that call from the pending queue.""" if output_call_id is not None: # The provider's explicit association is authoritative. if output_call_id in pending_call_ids: pending_call_ids.remove(output_call_id) return output_call_id if pending_call_ids: # Anonymous hosted-search outputs follow their calls in FIFO order. return pending_call_ids.popleft() return None def _map_tool_search_call(item: ResponseToolSearchCall, provider_name: str) -> NativeToolSearchCallPart: """Map an OpenAI server-executed tool-search call without fabricating its output.""" call_id = item.call_id or item.id return NativeToolSearchCallPart( provider_name=provider_name, args=_normalize_tool_search_args(item.arguments), tool_call_id=call_id, id=item.id, provider_details={'call_id': item.call_id, 'execution': item.execution, 'status': item.status}, ) def _tool_search_replay_details( part: NativeToolCallPart | NativeToolSearchReturnPart, ) -> tuple[str | None, Literal['in_progress', 'completed', 'incomplete'], dict[str, Any]]: """Read provider-native tool-search identity and status with backward-compatible defaults.""" details = part.provider_details or {} call_id = details.get('call_id', part.tool_call_id) if not isinstance(call_id, str | None): call_id = part.tool_call_id status = details.get('status') if status not in ('in_progress', 'completed', 'incomplete'): status = 'completed' return call_id, status, details def _lacks_tool_search_output_identity(part: NativeToolSearchReturnPart) -> bool: """Identify return parts that did not come from a preserved `tool_search_output` item. Parts built from a real output item always carry its `id` in `provider_details` (see `_build_tool_search_return_part`). Parts without it are pre-fix history (only `status` was stashed) or metadata-stripped round-trips. For those, replay falls back to sending the call only, the shape pre-fix code sent and the only one proven against the live API for such histories; fabricating an output item risks colliding with server-side state the call already references. """ output_id = (part.provider_details or {}).get('id') return not (isinstance(output_id, str) and output_id) class _MapToolDefinition(Protocol): def __call__(self, f: ToolDefinition, *, visibility: ToolVisibility) -> responses.FunctionToolParam: ... def _build_tool_search_output_param( part: ToolSearchReturnPart | NativeToolSearchReturnPart, call_id: str | None, execution: Literal['server', 'client'], status: Literal['in_progress', 'completed', 'incomplete'], model_request_parameters: ModelRequestParameters, map_tool_definition: _MapToolDefinition, ) -> ResponseToolSearchOutputItemParamParam: """Build a `tool_search_output` replay param from normalized discovery state. Looks up each discovered tool name in `function_tools` and emits a `FunctionToolParam` so OpenAI sees the same `Tool` definitions it originally loaded in the prior turn. The shape is `ResponseToolSearchOutputItemParamParam`. Reads the typed [`ToolSearchReturnContent`][pydantic_ai.messages.ToolSearchReturnContent] off of `part.content`; the tool-return value is the contract. Uses the same `map_tool_definition` the model used at initial send-time, so `strict` honors the user's preference and the model profile rather than a replay-only override. """ discovered = [match['name'] for match in part.discovered_tools] tool_defs_by_name = {t.name: t for t in model_request_parameters.function_tools} tool_params: list[responses.tool_param.ToolParam] = [ cast( 'responses.tool_param.ToolParam', map_tool_definition(tool_def, visibility=model_request_parameters.visibility_of(name)), ) for name in discovered if (tool_def := tool_defs_by_name.get(name)) is not None ] return { 'type': 'tool_search_output', 'execution': execution, 'tools': tool_params, 'call_id': call_id, 'status': status, } def _map_client_tool_search_call(item: ResponseToolSearchCall, provider_name: str) -> ToolSearchCallPart: """Map a client-executed OpenAI `tool_search_call` into a typed `ToolSearchCallPart`. With `ToolSearchToolParam(execution='client')`, OpenAI still emits the call wrapped as a `tool_search_call` item but leaves execution to us: the standard agent-graph tool-execution path runs the local `search_tools` function and produces the matching `ToolSearchReturnPart`. OpenAI's wire shape for `tool_search_call.arguments` mirrors the schema we registered with the builtin — `{"queries": list[str]}` matching the cross-provider [`ToolSearchArgs`][pydantic_ai.messages.ToolSearchArgs] — so we forward it unchanged. Streaming-mid item events carry `arguments={}` (empty), which validates fine as `{"queries": []}` per `_normalize_tool_search_args`. Emits the typed [`ToolSearchCallPart`][pydantic_ai.messages.ToolSearchCallPart] subclass directly — the wire shape is distinct enough that the adapter can identify this as a tool-search call without going through the framework's post-hoc `_narrow_tool_call_parts` pass. """ call_id = item.call_id or item.id args = _normalize_tool_search_args(item.arguments) return ToolSearchCallPart( tool_name=TOOL_SEARCH_FUNCTION_TOOL_NAME, args=args, tool_call_id=call_id, id=item.id, provider_name=provider_name, tool_kind='tool-search', ) def _build_tool_search_return_part( call_id: str, output_item: responses.ResponseToolSearchOutputItem, provider_name: str, ) -> NativeToolSearchReturnPart: """Build the typed return part for an actual OpenAI tool-search output. Writes the cross-provider [`ToolSearchReturnContent`][pydantic_ai.messages.ToolSearchReturnContent] to `content` (carrying only the matched tool names — the full [`ToolDefinition`][pydantic_ai.tools.ToolDefinition] is injected on the next request via defer-loading, so description is redundant here) and stashes the provider item identity and state on `provider_details` for replay. """ matches: list[ToolSearchMatch] = [] # `output_item.tools` is a union of OpenAI Responses tool variants; only # function tools carry a stable `name` field. Other variants (file_search, # image generation, etc.) can't appear here in practice but aren't # statically excluded from the union, so we filter by type. for t in output_item.tools: if isinstance(t, responses.FunctionTool): matches.append({'name': t.name}) return NativeToolSearchReturnPart( provider_name=provider_name, content={'discovered_tools': matches}, tool_call_id=call_id, provider_details={ 'id': output_item.id, 'call_id': output_item.call_id, 'execution': output_item.execution, 'status': output_item.status, }, ) def _map_web_search_tool_call( item: responses.ResponseFunctionWebSearch, provider_name: str ) -> tuple[NativeToolCallPart, NativeToolReturnPart]: args: dict[str, Any] | None = None result = { 'status': item.status, } if action := item.action: # We need to exclude None values because of https://github.com/pydantic/pydantic-ai/issues/3653 args = action.model_dump(mode='json', exclude_none=True) # To prevent `Unknown parameter: 'input[2].action.sources'` for `ActionSearch` if sources := args.pop('sources', None): result['sources'] = sources return ( NativeToolCallPart( tool_name=WebSearchTool.kind, tool_call_id=item.id, args=args, provider_name=provider_name, id=item.id, ), NativeToolReturnPart( tool_name=WebSearchTool.kind, tool_call_id=item.id, content=result, provider_name=provider_name, ), ) def _map_file_search_tool_call( item: responses.ResponseFileSearchToolCall, provider_name: str, ) -> tuple[NativeToolCallPart, NativeToolReturnPart]: args = {'queries': item.queries} result: dict[str, Any] = { 'status': item.status, } if item.results is not None: result['results'] = [r.model_dump(mode='json') for r in item.results] return ( NativeToolCallPart( tool_name=FileSearchTool.kind, tool_call_id=item.id, args=args, provider_name=provider_name, id=item.id, ), NativeToolReturnPart( tool_name=FileSearchTool.kind, tool_call_id=item.id, content=result, provider_name=provider_name, ), ) def _map_image_generation_tool_call( item: responses.response_output_item.ImageGenerationCall, provider_name: str ) -> tuple[NativeToolCallPart, NativeToolReturnPart, FilePart | None]: result = { 'status': item.status, } # Not present on the type, but present on the actual object. # See https://github.com/openai/openai-python/issues/2649 if background := getattr(item, 'background', None): result['background'] = background if quality := getattr(item, 'quality', None): result['quality'] = quality if size := getattr(item, 'size', None): result['size'] = size if revised_prompt := getattr(item, 'revised_prompt', None): result['revised_prompt'] = revised_prompt output_format = getattr(item, 'output_format', 'png') file_part: FilePart | None = None if item.result: file_part = FilePart( content=BinaryImage( data=base64.b64decode(item.result), media_type=f'image/{output_format}', ), id=item.id, ) # For some reason, the streaming API leaves `status` as `generating` even though generation has completed. result['status'] = 'completed' return ( NativeToolCallPart( tool_name=ImageGenerationTool.kind, tool_call_id=item.id, provider_name=provider_name, ), NativeToolReturnPart( tool_name=ImageGenerationTool.kind, tool_call_id=item.id, content=result, provider_name=provider_name, ), file_part, ) def _map_mcp_list_tools( item: responses.response_output_item.McpListTools, provider_name: str ) -> tuple[NativeToolCallPart, NativeToolReturnPart]: tool_name = ':'.join([MCPServerTool.kind, item.server_label]) return ( NativeToolCallPart( tool_name=tool_name, tool_call_id=item.id, provider_name=provider_name, args={'action': 'list_tools'}, ), NativeToolReturnPart( tool_name=tool_name, tool_call_id=item.id, content=item.model_dump(mode='json', include={'tools', 'error'}), provider_name=provider_name, ), ) def _map_mcp_call( item: responses.response_output_item.McpCall, provider_name: str ) -> tuple[NativeToolCallPart, NativeToolReturnPart]: tool_name = ':'.join([MCPServerTool.kind, item.server_label]) return ( NativeToolCallPart( tool_name=tool_name, tool_call_id=item.id, args={ 'action': 'call_tool', 'tool_name': item.name, 'tool_args': json.loads(item.arguments) if item.arguments else {}, }, provider_name=provider_name, ), NativeToolReturnPart( tool_name=tool_name, tool_call_id=item.id, # Dumped rather than read off the item like `output` alone would allow: `openai` 3.1 retyped # `McpCall.error` from `str` to a model union, so reading the attribute puts an SDK model # into a message part that then can't be serialized with the message history. `warnings=False` # because pre-3.1 the wire's error object lands in that `str`-typed field unconverted. content=item.model_dump(mode='json', include={'output', 'error'}, warnings=False), provider_name=provider_name, ), )