from __future__ import annotations as _annotations import base64 import dataclasses import hashlib import html import mimetypes import os import warnings from abc import ABC, abstractmethod from collections.abc import Callable, Collection, Mapping, Sequence from copy import deepcopy from dataclasses import KW_ONLY, dataclass, field, replace from datetime import datetime from mimetypes import MimeTypes from os import PathLike from pathlib import Path from typing import ( TYPE_CHECKING, Annotated, Any, ClassVar, Generic, Literal, TypeAlias, TypeGuard, cast, get_args, overload, ) from urllib.parse import urlparse import pydantic import pydantic_core from genai_prices import types as genai_types from pydantic.alias_generators import to_snake from pydantic.dataclasses import dataclass as pydantic_dataclass from typing_extensions import TypeAliasType, TypeVar, assert_never from pydantic_ai._genai_prices import calculate_price_for_usage from . import _otel_messages, _utils from ._event_registry import ( RESERVED_EVENT_TAGS as _RESERVED_EVENT_TAGS, EventRegistry as _EventRegistry, event_family_schema as _event_family_schema, guard_post_init as _guard_post_init, inherited_namespace as _inherited_namespace, inject_tag_field as _inject_tag_field, is_redefinition as _is_redefinition, keeps_canonical_registration as _keeps_canonical_registration, shadowed_envelope_fields as _shadowed_envelope_fields, undecorated_field_base as _undecorated_field_base, ) from ._instrumentation import redact_binary_content, serialize_any from ._utils import generate_tool_call_id as _generate_tool_call_id, now_utc as _now_utc from .exceptions import ModelRetry, UnexpectedModelBehavior, UserError from .usage import RequestUsage if TYPE_CHECKING: from .models.instrumented import InstrumentationSettings # Key used to wrap malformed tool-call arguments so they can still be round-tripped # through a model API without crashing. The specific string 'INVALID_JSON' is the # value recommended by the Anthropic docs for this situation: # https://docs.anthropic.com/en/docs/build-with-claude/tool-use/overview#handling-tool-use-errors INVALID_JSON_KEY = 'INVALID_JSON' _mime_types = MimeTypes() # Replicate what is being done in `mimetypes.init()` _mime_types.read_windows_registry() for file in mimetypes.knownfiles: if os.path.isfile(file): _mime_types.read(file) # pragma: lax no cover # TODO check for added mimetypes in Python 3.11 when dropping support for Python 3.10: # https://github.com/python/cpython/blob/3.11/Lib/mimetypes.py # Document types _mime_types.add_type('application/rtf', '.rtf') _mime_types.add_type('application/vnd.openxmlformats-officedocument.spreadsheetml.sheet', '.xlsx') _mime_types.add_type('application/vnd.openxmlformats-officedocument.wordprocessingml.document', '.docx') _mime_types.add_type('text/markdown', '.mdx') _mime_types.add_type('text/markdown', '.md') _mime_types.add_type('text/x-asciidoc', '.asciidoc') # Image types _mime_types.add_type('image/webp', '.webp') # Video types _mime_types.add_type('video/3gpp', '.three_gp') _mime_types.add_type('video/x-matroska', '.mkv') _mime_types.add_type('video/x-ms-wmv', '.wmv') _mime_types.add_type('video/x-flv', '.flv') # Audio types # NOTE: aac is platform specific (linux: audio/x-aac, macos: audio/aac) but x-aac is deprecated https://mimetype.io/audio/aac _mime_types.add_type('audio/aac', '.aac') _mime_types.add_type('audio/aiff', '.aiff') _mime_types.add_type('audio/flac', '.flac') _mime_types.add_type('audio/ogg', '.oga') _mime_types.add_type('audio/wav', '.wav') # Text/data file types not recognized by default mimetypes # YAML: RFC 9512 (https://www.rfc-editor.org/rfc/rfc9512.html) _mime_types.add_type('application/yaml', '.yaml') _mime_types.add_type('application/yaml', '.yml') # TOML: RFC 9519 (https://www.rfc-editor.org/rfc/rfc9519.html) _mime_types.add_type('application/toml', '.toml') # XML is recognized as `text/xml` on some systems, but it needs to be `application/xml` per RFC 7303 (https://www.rfc-editor.org/rfc/rfc7303.html) _mime_types.add_type('application/xml', '.xml') AudioMediaType: TypeAlias = Literal['audio/wav', 'audio/mpeg', 'audio/ogg', 'audio/flac', 'audio/aiff', 'audio/aac'] ImageMediaType: TypeAlias = Literal['image/jpeg', 'image/png', 'image/gif', 'image/webp'] DocumentMediaType: TypeAlias = Literal[ 'application/pdf', 'text/plain', 'text/csv', 'application/vnd.openxmlformats-officedocument.wordprocessingml.document', 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet', 'text/html', 'text/markdown', 'application/msword', 'application/vnd.ms-excel', ] VideoMediaType: TypeAlias = Literal[ 'video/x-matroska', 'video/quicktime', 'video/mp4', 'video/webm', 'video/x-flv', 'video/mpeg', 'video/x-ms-wmv', 'video/3gpp', ] AudioFormat: TypeAlias = Literal['wav', 'mp3', 'oga', 'flac', 'aiff', 'aac'] ImageFormat: TypeAlias = Literal['jpeg', 'png', 'gif', 'webp'] DocumentFormat: TypeAlias = Literal['csv', 'doc', 'docx', 'html', 'md', 'pdf', 'txt', 'xls', 'xlsx'] VideoFormat: TypeAlias = Literal['mkv', 'mov', 'mp4', 'webm', 'flv', 'mpeg', 'mpg', 'wmv', 'three_gp'] FinishReason: TypeAlias = Literal[ 'stop', 'length', 'content_filter', 'tool_call', 'error', ] """Reason the model finished generating the response. Mostly normalized to OpenTelemetry semantic convention values. Whether the agent should automatically continue is determined by `ModelResponse.state`, not by this field. """ ModelResponseState: TypeAlias = Literal['complete', 'incomplete', 'suspended', 'interrupted'] """Lifecycle state of a model response. - `'complete'`: the response has been fully received from the model. - `'incomplete'`: the response is still being streamed and may receive more parts. Yielded by [`AgentStream.response`][pydantic_ai.result.AgentStream.response] and [`StreamedRunResult.stream_response`][pydantic_ai.result.StreamedRunResult.stream_response] while iteration is in flight. - `'suspended'`: the model paused mid-turn and expects a continuation request. Used by Anthropic `pause_turn` and OpenAI background mode. Pydantic AI issues these continuations transparently for both `agent.run` and `agent.run_stream`, merging every segment into a single completed [`ModelResponse`][pydantic_ai.messages.ModelResponse], so a finished turn in the message history is never left in this state. - `'interrupted'`: generation was explicitly stopped before the model finished. Set when a streamed response is cancelled via [`StreamedResponse.cancel()`][pydantic_ai.models.StreamedResponse.cancel], and when a realtime turn is cut off by a barge-in or [`RealtimeSession.interrupt()`][pydantic_ai.realtime.RealtimeSession.interrupt] — in which case the cut-off point is recorded on the last [`SpeechPart.interrupted_at_ms`][pydantic_ai.messages.SpeechPart.interrupted_at_ms]. """ ModelRequestState: TypeAlias = Literal['complete', 'interrupted'] """Lifecycle state of a model request.""" ForceDownloadMode: TypeAlias = bool | Literal['allow-local'] """Type for the force_download parameter on FileUrl subclasses. - `False`: The URL is sent directly to providers that support it. For providers that don't, the file is downloaded with SSRF protection (blocks private IPs and cloud metadata). - `True`: The file is always downloaded with SSRF protection (blocks private IPs and cloud metadata). - `'allow-local'`: The file is always downloaded, allowing private IPs but still blocking cloud metadata. """ def _serialize_provider_details_delta( value: dict[str, Any] | Callable[[dict[str, Any] | None], dict[str, Any]] | None, ) -> dict[str, Any] | None: # A callable `provider_details` is a transient merge callback used while chaining deltas; it cannot be # JSON-serialized, so it is emitted as `null`. Once the delta is applied to a `ThinkingPart` the callback is # resolved to a concrete dict, which serializes normally. This is scoped to JSON mode (`when_used='json'`) so # Python-mode `model_dump()` keeps the callback intact for in-memory round-trips. if callable(value): return None return value ProviderDetailsDelta: TypeAlias = Annotated[ dict[str, Any] | Callable[[dict[str, Any] | None], dict[str, Any]] | None, pydantic.PlainSerializer(_serialize_provider_details_delta, return_type=dict[str, Any] | None, when_used='json'), ] """Type for provider_details input: can be a static dict, a callback to update existing details, or None.""" @dataclass(repr=False) class SystemPromptPart: """A system prompt, generally written by the application developer. This gives the model context and guidance on how to respond. """ content: str """The content of the prompt.""" _: KW_ONLY timestamp: datetime = field(default_factory=_now_utc) """The timestamp of the prompt.""" dynamic_ref: str | None = None """The ref of the dynamic system prompt function that generated this part. Only set if system prompt is dynamic, see [`system_prompt`][pydantic_ai.agent.Agent.system_prompt] for more information. """ part_kind: Literal['system-prompt'] = 'system-prompt' """Part type identifier, this is available on all parts as a discriminator.""" def otel_message_parts(self, settings: InstrumentationSettings) -> list[_otel_messages.MessagePart]: return [_otel_messages.TextPart(type='text', **{'content': self.content} if settings.include_content else {})] __repr__ = _utils.dataclasses_no_defaults_repr def _multi_modal_content_identifier(identifier: str | bytes) -> str: """Generate stable identifier for multi-modal content to help LLM in finding a specific file in tool call responses.""" if isinstance(identifier, str): identifier = identifier.encode('utf-8') return hashlib.sha1(identifier, usedforsecurity=False).hexdigest()[:6] @pydantic_dataclass(repr=False, config=pydantic.ConfigDict(validate_by_name=True)) class FileUrl(ABC): """Abstract base class for any URL-based file.""" url: str """The URL of the file.""" _: KW_ONLY force_download: ForceDownloadMode = False """Controls whether the file is downloaded and how SSRF protection is applied: * If `False`, the URL is sent directly to providers that support it. For providers that don't, the file is downloaded with SSRF protection (blocks private IPs and cloud metadata). * If `True`, the file is always downloaded with SSRF protection (blocks private IPs and cloud metadata). * If `'allow-local'`, the file is always downloaded, allowing private IPs but still blocking cloud metadata. """ vendor_metadata: dict[str, Any] | None = None """Vendor-specific metadata for the file. Supported by: - `GoogleModel`: `VideoUrl.vendor_metadata` is used as `video_metadata`: https://ai.google.dev/gemini-api/docs/video-understanding#customize-video-processing, and `vendor_metadata['media_resolution']` is forwarded as the per-Part `media_resolution` field for any file type: https://ai.google.dev/gemini-api/docs/media-resolution - `OpenAIChatModel`, `OpenAIResponsesModel`: `ImageUrl.vendor_metadata['detail']` is used as `detail` setting for images - `XaiModel`: `ImageUrl.vendor_metadata['detail']` is used as `detail` setting for images - `GroqModel`: `ImageUrl.vendor_metadata['detail']` is used as `detail` setting for images - `MistralModel`: `ImageUrl.vendor_metadata['detail']` is used as `detail` setting for images """ _media_type: Annotated[str | None, pydantic.Field(alias='media_type', default=None, exclude=True)] = field( compare=False, default=None ) _identifier: Annotated[str | None, pydantic.Field(alias='identifier', default=None, exclude=True)] = field( compare=False, default=None ) # `pydantic_dataclass` replaces `__init__` so this method is never used. # The signature is kept so that pyright/IDE hints recognize the `media_type` and `identifier` aliases. def __init__( self, url: str, *, media_type: str | None = None, identifier: str | None = None, force_download: ForceDownloadMode = False, vendor_metadata: dict[str, Any] | None = None, # Required for inline-snapshot which expects all dataclass `__init__` methods to take all field names as kwargs. _media_type: str | None = None, _identifier: str | None = None, ) -> None: ... # pragma: no cover @pydantic.computed_field @property def media_type(self) -> str: """Return the media type of the file, based on the URL or the provided `media_type`.""" return self._media_type or self._infer_media_type() @pydantic.computed_field @property def identifier(self) -> str: """The identifier of the file, such as a unique ID. This identifier can be provided to the model in a message to allow it to refer to this file in a tool call argument, and the tool can look up the file in question by iterating over the message history and finding the matching `FileUrl`. This identifier is only automatically passed to the model when the `FileUrl` is returned by a tool. If you're passing the `FileUrl` as a user message, it's up to you to include a separate text part with the identifier, e.g. "This is file :" preceding the `FileUrl`. It's also included in inline-text delimiters for providers that require inlining text documents, so the model can distinguish multiple files. """ return self._identifier or _multi_modal_content_identifier(self.url) @abstractmethod def _infer_media_type(self) -> str: """Infer the media type of the file based on the URL.""" raise NotImplementedError @property @abstractmethod def format(self) -> str: """The file format.""" raise NotImplementedError __repr__ = _utils.dataclasses_no_defaults_repr @pydantic_dataclass(repr=False, config=pydantic.ConfigDict(validate_by_name=True)) class VideoUrl(FileUrl): """A URL to a video.""" url: str """The URL of the video.""" _: KW_ONLY kind: Literal['video-url'] = 'video-url' """Type identifier, this is available on all parts as a discriminator.""" # `pydantic_dataclass` replaces `__init__` so this method is never used. # The signature is kept so that pyright/IDE hints recognize the aliases for the `_media_type` and `_identifier` fields. def __init__( self, url: str, *, media_type: str | None = None, identifier: str | None = None, force_download: ForceDownloadMode = False, vendor_metadata: dict[str, Any] | None = None, kind: Literal['video-url'] = 'video-url', # Required for inline-snapshot which expects all dataclass `__init__` methods to take all field names as kwargs. _media_type: str | None = None, _identifier: str | None = None, ) -> None: ... # pragma: no cover def _infer_media_type(self) -> str: """Return the media type of the video, based on the url.""" # Assume that YouTube videos are mp4 because there would be no extension # to infer from. This should not be a problem, as Gemini disregards media # type for YouTube URLs. if self.is_youtube: return 'video/mp4' mime_type, _ = _mime_types.guess_type(self.url) if mime_type is None: raise ValueError( f'Could not infer media type from video URL: {self.url}. Explicitly provide a `media_type` instead.' ) return mime_type @property def is_youtube(self) -> bool: """True if the URL is on a YouTube host that models can resolve directly. This is a specific set of hosts rather than every YouTube-owned domain, so `music.youtube.com` is deliberately not one of them. """ # Exact hosts, not a `.youtube.com` suffix match: Google rejects # `music.youtube.com` as a `file_uri` with 400 INVALID_ARGUMENT, on the Gemini API # and on Vertex alike, so a suffix match would hand it a URL it cannot resolve. # Membership is also read by `download_item`, so a host added here stops being # downloadable on every other provider too — verify both before extending this. return urlparse(self.url).hostname in ('youtu.be', 'youtube.com', 'www.youtube.com', 'm.youtube.com') @property def format(self) -> VideoFormat: """The file format of the video. The choice of supported formats were based on the Bedrock Converse API. Other APIs don't require to use a format. """ return _video_format_lookup[self.media_type] @pydantic_dataclass(repr=False, config=pydantic.ConfigDict(validate_by_name=True)) class AudioUrl(FileUrl): """A URL to an audio file.""" url: str """The URL of the audio file.""" _: KW_ONLY kind: Literal['audio-url'] = 'audio-url' """Type identifier, this is available on all parts as a discriminator.""" # `pydantic_dataclass` replaces `__init__` so this method is never used. # The signature is kept so that pyright/IDE hints recognize the aliases for the `_media_type` and `_identifier` fields. def __init__( self, url: str, *, media_type: str | None = None, identifier: str | None = None, force_download: ForceDownloadMode = False, vendor_metadata: dict[str, Any] | None = None, kind: Literal['audio-url'] = 'audio-url', # Required for inline-snapshot which expects all dataclass `__init__` methods to take all field names as kwargs. _media_type: str | None = None, _identifier: str | None = None, ) -> None: ... # pragma: no cover def _infer_media_type(self) -> str: """Return the media type of the audio file, based on the url. References: - Gemini: https://ai.google.dev/gemini-api/docs/audio#supported-formats """ mime_type, _ = _mime_types.guess_type(self.url) if mime_type is None: raise ValueError( f'Could not infer media type from audio URL: {self.url}. Explicitly provide a `media_type` instead.' ) return mime_type @property def format(self) -> AudioFormat: """The file format of the audio file.""" return _audio_format_lookup[self.media_type] @pydantic_dataclass(repr=False, config=pydantic.ConfigDict(validate_by_name=True)) class ImageUrl(FileUrl): """A URL to an image.""" url: str """The URL of the image.""" _: KW_ONLY kind: Literal['image-url'] = 'image-url' """Type identifier, this is available on all parts as a discriminator.""" # `pydantic_dataclass` replaces `__init__` so this method is never used. # The signature is kept so that pyright/IDE hints recognize the aliases for the `_media_type` and `_identifier` fields. def __init__( self, url: str, *, media_type: str | None = None, identifier: str | None = None, force_download: ForceDownloadMode = False, vendor_metadata: dict[str, Any] | None = None, kind: Literal['image-url'] = 'image-url', # Required for inline-snapshot which expects all dataclass `__init__` methods to take all field names as kwargs. _media_type: str | None = None, _identifier: str | None = None, ) -> None: ... # pragma: no cover def _infer_media_type(self) -> str: """Return the media type of the image, based on the url.""" mime_type, _ = _mime_types.guess_type(self.url) if mime_type is None: raise ValueError( f'Could not infer media type from image URL: {self.url}. Explicitly provide a `media_type` instead.' ) return mime_type @property def format(self) -> ImageFormat: """The file format of the image. The choice of supported formats were based on the Bedrock Converse API. Other APIs don't require to use a format. """ return _image_format_lookup[self.media_type] @pydantic_dataclass(repr=False, config=pydantic.ConfigDict(validate_by_name=True)) class DocumentUrl(FileUrl): """The URL of the document.""" url: str """The URL of the document.""" _: KW_ONLY kind: Literal['document-url'] = 'document-url' """Type identifier, this is available on all parts as a discriminator.""" # `pydantic_dataclass` replaces `__init__` so this method is never used. # The signature is kept so that pyright/IDE hints recognize the aliases for the `_media_type` and `_identifier` fields. def __init__( self, url: str, *, media_type: str | None = None, identifier: str | None = None, force_download: ForceDownloadMode = False, vendor_metadata: dict[str, Any] | None = None, kind: Literal['document-url'] = 'document-url', # Required for inline-snapshot which expects all dataclass `__init__` methods to take all field names as kwargs. _media_type: str | None = None, _identifier: str | None = None, ) -> None: ... # pragma: no cover def _infer_media_type(self) -> str: """Return the media type of the document, based on the url.""" mime_type, _ = _mime_types.guess_type(self.url) if mime_type is None: raise ValueError( f'Could not infer media type from document URL: {self.url}. Explicitly provide a `media_type` instead.' ) return mime_type @property def format(self) -> DocumentFormat: """The file format of the document. The choice of supported formats were based on the Bedrock Converse API. Other APIs don't require to use a format. """ media_type = self.media_type try: return _document_format_lookup[media_type] except KeyError as e: raise ValueError(f'Unknown document media type: {media_type}') from e @dataclass(repr=False) class TextContent: """String content that is tagged with additional metadata. This is useful for including metadata that can be accessed programmatically by the application, but is not sent to the LLM. """ content: str """The content that is sent to the LLM.""" _: KW_ONLY metadata: Any = None """Additional data that can be accessed programmatically by the application but is not sent to the LLM. `ModelMessagesTypeAdapter` preserves this field, but as application-only data it is not guaranteed to survive a round-trip through the UI adapters; see [Storing and loading messages](../message-history.md#storing-and-loading-messages-to-json). """ kind: Literal['text-content'] = 'text-content' """Type identifier, this is available on all parts as a discriminator.""" __repr__ = _utils.dataclasses_no_defaults_repr @pydantic_dataclass( repr=False, config=pydantic.ConfigDict( ser_json_bytes='base64', val_json_bytes='base64', ), ) class BinaryContent: """Binary content, e.g. an audio or image file.""" data: bytes """The binary file data. Use `.base64` to get the base64-encoded string. """ _: KW_ONLY media_type: AudioMediaType | ImageMediaType | DocumentMediaType | str """The media type of the binary data.""" vendor_metadata: dict[str, Any] | None = None """Vendor-specific metadata for the file. Supported by: - `GoogleModel`: `BinaryContent.vendor_metadata` is used as `video_metadata`: https://ai.google.dev/gemini-api/docs/video-understanding#customize-video-processing, and `BinaryContent.vendor_metadata['media_resolution']` is forwarded as the per-Part `media_resolution` field: https://ai.google.dev/gemini-api/docs/media-resolution - `OpenAIChatModel`, `OpenAIResponsesModel`: `BinaryContent.vendor_metadata['detail']` is used as `detail` setting for images - `XaiModel`: `BinaryContent.vendor_metadata['detail']` is used as `detail` setting for images - `GroqModel`: `BinaryContent.vendor_metadata['detail']` is used as `detail` setting for images - `MistralModel`: `BinaryContent.vendor_metadata['detail']` is used as `detail` setting for images """ _identifier: Annotated[str | None, pydantic.Field(alias='identifier', default=None, exclude=True)] = field( compare=False, default=None ) kind: Literal['binary'] = 'binary' """Type identifier, this is available on all parts as a discriminator.""" # `pydantic_dataclass` replaces `__init__` so this method is never used. # The signature is kept so that pyright/IDE hints recognize the `identifier` alias for the `_identifier` field. def __init__( self, data: bytes, *, media_type: AudioMediaType | ImageMediaType | DocumentMediaType | str, identifier: str | None = None, vendor_metadata: dict[str, Any] | None = None, kind: Literal['binary'] = 'binary', # Required for inline-snapshot which expects all dataclass `__init__` methods to take all field names as kwargs. _identifier: str | None = None, ) -> None: ... # pragma: no cover @staticmethod def narrow_type(bc: BinaryContent) -> BinaryContent | BinaryImage: """Narrow the type of the `BinaryContent` to `BinaryImage` if it's an image.""" if bc.is_image: return BinaryImage( data=bc.data, media_type=bc.media_type, identifier=bc.identifier, vendor_metadata=bc.vendor_metadata, ) else: return bc @classmethod def from_data_uri(cls, data_uri: str) -> BinaryContent: """Create a `BinaryContent` from a data URI.""" prefix = 'data:' if not data_uri.startswith(prefix): raise ValueError('Data URI must start with "data:"') body = data_uri[len(prefix) :] if ';base64,' not in body: raise ValueError('Data URI must be base64-encoded (expected ";base64," marker)') media_type, data = body.split(';base64,', 1) return cls.narrow_type(cls(data=base64.b64decode(data), media_type=media_type)) @classmethod def from_path(cls, path: PathLike[str]) -> BinaryContent: """Create a `BinaryContent` from a path. Defaults to 'application/octet-stream' if the media type cannot be inferred. Raises: FileNotFoundError: if the file does not exist. PermissionError: if the file cannot be read. """ path = Path(path) if not path.exists(): raise FileNotFoundError(f'File not found: {path}') media_type, _ = _mime_types.guess_type(path) if media_type is None: media_type = 'application/octet-stream' return cls.narrow_type(cls(data=path.read_bytes(), media_type=media_type)) @pydantic.computed_field @property def identifier(self) -> str: """Identifier for the binary content, such as a unique ID. This identifier can be provided to the model in a message to allow it to refer to this file in a tool call argument, and the tool can look up the file in question by iterating over the message history and finding the matching `BinaryContent`. This identifier is only automatically passed to the model when the `BinaryContent` is returned by a tool. If you're passing the `BinaryContent` as a user message, it's up to you to include a separate text part with the identifier, e.g. "This is file :" preceding the `BinaryContent`. It's also included in inline-text delimiters for providers that require inlining text documents, so the model can distinguish multiple files. """ return self._identifier or _multi_modal_content_identifier(self.data) @property def data_uri(self) -> str: """Convert the `BinaryContent` to a data URI.""" return f'data:{self.media_type};base64,{self.base64}' @property def base64(self) -> str: """Return the binary data as a base64-encoded string. Default encoding is UTF-8.""" return base64.b64encode(self.data).decode() @property def is_audio(self) -> bool: """Return `True` if the media type is an audio type.""" return self.media_type.startswith('audio/') @property def is_image(self) -> bool: """Return `True` if the media type is an image type.""" return self.media_type.startswith('image/') @property def is_video(self) -> bool: """Return `True` if the media type is a video type.""" return self.media_type.startswith('video/') @property def is_document(self) -> bool: """Return `True` if the media type is a document type.""" return self.media_type in _document_format_lookup @property def format(self) -> str: """The file format of the binary content.""" try: if self.is_audio: return _audio_format_lookup[self.media_type] elif self.is_image: return _image_format_lookup[self.media_type] elif self.is_video: return _video_format_lookup[self.media_type] else: return _document_format_lookup[self.media_type] except KeyError as e: raise ValueError(f'Unknown media type: {self.media_type}') from e __repr__ = _utils.dataclasses_no_defaults_repr @pydantic_dataclass( repr=False, config=pydantic.ConfigDict( ser_json_bytes='base64', val_json_bytes='base64', ), ) class BinaryImage(BinaryContent): """Binary content that's guaranteed to be an image.""" # `pydantic_dataclass` replaces `__init__` so this method is never used. # The signature is kept so that pyright/IDE hints recognize the `identifier` alias for the `_identifier` field. def __init__( self, data: bytes, *, media_type: ImageMediaType | str, identifier: str | None = None, vendor_metadata: dict[str, Any] | None = None, kind: Literal['binary'] = 'binary', # Required for inline-snapshot which expects all dataclass `__init__` methods to take all field names as kwargs. _identifier: str | None = None, ) -> None: ... # pragma: no cover def __post_init__(self): if not self.is_image: raise ValueError('`BinaryImage` must have a media type that starts with "image/"') @pydantic_dataclass( repr=False, config=pydantic.ConfigDict( ser_json_bytes='base64', val_json_bytes='base64', ), ) class BinaryAudio(BinaryContent): """Binary content that's guaranteed to be audio.""" # `pydantic_dataclass` replaces `__init__` so this method is never used. # The signature is kept so that pyright/IDE hints recognize the `identifier` alias for the `_identifier` field. def __init__( self, data: bytes, *, media_type: AudioMediaType | str, identifier: str | None = None, vendor_metadata: dict[str, Any] | None = None, kind: Literal['binary'] = 'binary', # Required for inline-snapshot which expects all dataclass `__init__` methods to take all field names as kwargs. _identifier: str | None = None, ) -> None: ... # pragma: no cover def __post_init__(self): if not self.is_audio: raise ValueError('`BinaryAudio` must have a media type that starts with "audio/"') @dataclass class CachePoint: """A cache point marker for prompt caching. Can be inserted into UserPromptPart.content to mark cache boundaries. Models that don't support caching will filter these out. Supported by: - Anthropic - Amazon Bedrock (Converse API) - OpenAI (GPT-5.6 models) - OpenRouter (Anthropic and Gemini models via `OpenRouterModel`, plus OpenAI GPT-5.6 models when using `OpenAIChatModel` or `OpenAIResponsesModel` with `OpenRouterProvider`) """ kind: Literal['cache-point'] = 'cache-point' """Type identifier, this is available on all parts as a discriminator.""" ttl: Literal['5m', '1h'] = '5m' """The cache time-to-live, either "5m" (5 minutes) or "1h" (1 hour). Supported by: * Anthropic — see https://docs.claude.com/en/docs/build-with-claude/prompt-caching#1-hour-cache-duration for more information. * Amazon Bedrock (Converse API) — see https://docs.aws.amazon.com/bedrock/latest/userguide/prompt-caching.html for more information. * OpenAI ignores this per-marker value and uses the request-wide `openai_prompt_cache_options['ttl']` setting instead. * OpenRouter with Anthropic models (automatically omitted for Gemini models, which do not support explicit TTL). """ UploadedFileProviderName: TypeAlias = Literal[ 'anthropic', 'openai', 'google', 'google-cloud', 'google-gla', 'google-vertex', 'bedrock', 'xai', ] """Provider names supported by [`UploadedFile`][pydantic_ai.messages.UploadedFile]. The `'google-gla'` and `'google-vertex'` values are retained for backward compatibility with message history captured before the v2 provider rename — current code emits `'google'` and `'google-cloud'` respectively. """ @pydantic_dataclass(repr=False, config=pydantic.ConfigDict(validate_by_name=True)) class UploadedFile: """A reference to a file uploaded to a provider's file storage by ID. This allows referencing files that have been uploaded via provider-specific file APIs rather than providing the file content directly. Supported by: - [`AnthropicModel`][pydantic_ai.models.anthropic.AnthropicModel] - [`OpenAIChatModel`][pydantic_ai.models.openai.OpenAIChatModel] - [`OpenAIResponsesModel`][pydantic_ai.models.openai.OpenAIResponsesModel] - [`BedrockConverseModel`][pydantic_ai.models.bedrock.BedrockConverseModel] - [`GoogleModel`][pydantic_ai.models.google.GoogleModel] (Gemini API: [Files API](https://ai.google.dev/gemini-api/docs/files) URIs, Google Cloud: GCS `gs://` URIs) - [`XaiModel`][pydantic_ai.models.xai.XaiModel] """ file_id: str """The provider-specific file identifier. For most providers, this is the file ID returned by the provider's upload API. For GoogleModel (Google Cloud), this must be a GCS URI (`gs://bucket/path`). For GoogleModel (Gemini API), this must be a Google Files API URI (`https://generativelanguage.googleapis.com/...`). For BedrockConverseModel, this must be an S3 URI (`s3://bucket/key`). """ provider_name: UploadedFileProviderName """The provider this file belongs to. This is required because file IDs are not portable across providers, and using a file ID with the wrong provider will always result in an error. Tip: Use `model.system` to get the provider name dynamically. """ _: KW_ONLY vendor_metadata: dict[str, Any] | None = None """Vendor-specific metadata for the file. The expected shape of this dictionary depends on the provider: Supported by: - `GoogleModel`: used as `video_metadata` for video files, and `UploadedFile.vendor_metadata['media_resolution']` is forwarded as the per-Part `media_resolution` field: https://ai.google.dev/gemini-api/docs/media-resolution - `OpenAIResponsesModel`: `UploadedFile.vendor_metadata['detail']` is used as `detail` setting for image files """ _media_type: Annotated[str | None, pydantic.Field(alias='media_type', default=None, exclude=True)] = field( compare=False, default=None ) _identifier: Annotated[str | None, pydantic.Field(alias='identifier', default=None, exclude=True)] = field( compare=False, default=None ) kind: Literal['uploaded-file'] = 'uploaded-file' """Type identifier, this is available on all parts as a discriminator.""" # `pydantic_dataclass` replaces `__init__` so this method is never used. # The signature is kept so that pyright/IDE hints recognize the `media_type` and `identifier` aliases. def __init__( self, file_id: str, provider_name: UploadedFileProviderName, *, media_type: str | None = None, vendor_metadata: dict[str, Any] | None = None, identifier: str | None = None, kind: Literal['uploaded-file'] = 'uploaded-file', # Required for inline-snapshot which expects all dataclass `__init__` methods to take all field names as kwargs. _media_type: str | None = None, _identifier: str | None = None, ) -> None: ... # pragma: no cover @pydantic.computed_field @property def media_type(self) -> str: """Return the media type of the file, inferred from `file_id` if not explicitly provided. Note: Inference relies on the file extension in `file_id`. For opaque file IDs (e.g., `'file-abc123'`), the media type will default to `'application/octet-stream'`. Inference relies on Python's `mimetypes` module, whose results may vary across platforms. Required by some providers (e.g., Bedrock) for certain file types. """ if self._media_type is not None: return self._media_type parsed = urlparse(self.file_id) mime_type, _ = _mime_types.guess_type(parsed.path) return mime_type or 'application/octet-stream' @pydantic.computed_field @property def identifier(self) -> str: """The identifier of the file, such as a unique ID. This identifier can be provided to the model in a message to allow it to refer to this file in a tool call argument, and the tool can look up the file in question by iterating over the message history and finding the matching `UploadedFile`. This identifier is only automatically passed to the model when the `UploadedFile` is returned by a tool. If you're passing the `UploadedFile` as a user message, it's up to you to include a separate text part with the identifier, e.g. "This is file :" preceding the `UploadedFile`. """ return self._identifier or _multi_modal_content_identifier(self.file_id) @property def format(self) -> str: """A general-purpose media-type-to-format mapping. Maps media types to format strings (e.g. `'image/png'` -> `'png'`). Covers image, video, audio, and document types. Currently used by Bedrock, which requires explicit format strings. """ media_type = self.media_type try: if media_type.startswith('image/'): return _image_format_lookup[media_type] elif media_type.startswith('video/'): return _video_format_lookup[media_type] elif media_type.startswith('audio/'): return _audio_format_lookup[media_type] else: return _document_format_lookup[media_type] except KeyError as e: raise ValueError(f'Unknown media type: {media_type}') from e __repr__ = _utils.dataclasses_no_defaults_repr MultiModalContent = Annotated[ ImageUrl | AudioUrl | DocumentUrl | VideoUrl | Annotated[BinaryContent, pydantic.AfterValidator(BinaryContent.narrow_type)] | UploadedFile, pydantic.Discriminator('kind'), ] """Union of all multi-modal content types with a discriminator for Pydantic validation.""" # Explicit tuple for readability; validated against MultiModalContent in tests MULTI_MODAL_CONTENT_TYPES: tuple[type, ...] = (ImageUrl, AudioUrl, DocumentUrl, VideoUrl, BinaryContent, UploadedFile) _FILE_URL_KINDS: tuple[str, ...] = (ImageUrl.kind, AudioUrl.kind, DocumentUrl.kind, VideoUrl.kind) """The `kind` values of the `FileUrl` subclasses: the multi-modal items whose media type is inferred from the URL when they were given none, and so the only ones a tool return has to spell a `media_type` out for (see `_RequireUrlMediaType`).""" def is_multi_modal_content(obj: Any) -> TypeGuard[MultiModalContent]: """Check if obj is a MultiModalContent type, enabling type narrowing.""" return isinstance(obj, MULTI_MODAL_CONTENT_TYPES) UserContent: TypeAlias = str | TextContent | MultiModalContent | CachePoint """A single item of user prompt content: a string, a typed text or multi-modal content part, or a [`CachePoint`][pydantic_ai.messages.CachePoint] marker.""" # Explicit tuple for readability; validated against `UserContent` in tests _USER_CONTENT_TYPES: tuple[type, ...] = (str, TextContent, *MULTI_MODAL_CONTENT_TYPES, CachePoint) _NOT_USER_CONTENT = ( 'Serialize the value yourself before passing it, e.g. with Pydantic (`pydantic_core.to_json()`) ' 'or `pydantic_ai.format_as_xml()`.' ) _ToolReturnValueT = TypeVar('_ToolReturnValueT', default=Any) """Type variable for the return value type in `ToolReturn[T]`. When `ToolReturn` is used without a type parameter (bare `ToolReturn`), this defaults to `Any`, meaning no return schema is generated. When specified (e.g. `ToolReturn[User]`), the return schema is generated from the inner type. """ @dataclass(repr=False) class ToolReturn(Generic[_ToolReturnValueT]): """A structured tool return that separates the tool result from additional content sent to the model. Can be parameterized with a type to enable return schema generation: - `ToolReturn[User]` — generates a return schema for `User` - `ToolReturn` (bare) — no return schema generated """ return_value: ToolReturnContent """The return value to be used in the tool response.""" _: KW_ONLY content: str | Sequence[UserContent] | None = None """Content sent to the model as a separate `UserPromptPart`. Use this when you want content to appear outside the tool result message. For multimodal content that should be sent natively in the tool result, return it directly from the tool function or include it in `return_value`. """ metadata: Any = None """Additional data accessible by the application but not sent to the LLM.""" tools: list[str] | None = None """Names of deferred tools made available by this tool call. The names are recorded verbatim in message history in a sibling [`ToolAvailabilityDeltaPart`][pydantic_ai.messages.ToolAvailabilityDeltaPart], then filtered against the currently served tool definitions at render time. A name that matches no deferred tool, such as a typo or an always-visible tool, is a silent no-op by design. """ kind: Literal['tool-return'] = 'tool-return' __repr__ = _utils.dataclasses_no_defaults_repr _document_format_lookup: dict[str, DocumentFormat] = { 'application/pdf': 'pdf', 'text/plain': 'txt', 'text/csv': 'csv', 'application/vnd.openxmlformats-officedocument.wordprocessingml.document': 'docx', 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet': 'xlsx', 'text/html': 'html', 'text/markdown': 'md', 'application/msword': 'doc', 'application/vnd.ms-excel': 'xls', } _audio_format_lookup: dict[str, AudioFormat] = { 'audio/mpeg': 'mp3', 'audio/wav': 'wav', 'audio/flac': 'flac', 'audio/ogg': 'oga', 'audio/aiff': 'aiff', 'audio/aac': 'aac', } _image_format_lookup: dict[str, ImageFormat] = { 'image/jpeg': 'jpeg', 'image/png': 'png', 'image/gif': 'gif', 'image/webp': 'webp', } _video_format_lookup: dict[str, VideoFormat] = { 'video/x-matroska': 'mkv', 'video/quicktime': 'mov', 'video/mp4': 'mp4', 'video/webm': 'webm', 'video/x-flv': 'flv', 'video/mpeg': 'mpeg', 'video/x-ms-wmv': 'wmv', 'video/3gpp': 'three_gp', } _kind_to_modality_lookup: dict[str, Literal['image', 'audio', 'video']] = { 'image-url': 'image', 'audio-url': 'audio', 'video-url': 'video', } def _infer_modality_from_media_type(media_type: str) -> Literal['image', 'audio', 'video'] | None: """Infer modality from media type for OTel GenAI semantic conventions.""" if media_type.startswith('image/'): return 'image' elif media_type.startswith('audio/'): return 'audio' elif media_type.startswith('video/'): return 'video' return None def _convert_binary_to_otel_part( media_type: str, base64_content: Callable[[], str], settings: InstrumentationSettings ) -> _otel_messages.BlobPart | _otel_messages.BinaryDataPart: """Convert binary content to OTel message part based on version.""" if settings.version >= 4: blob_part = _otel_messages.BlobPart(type='blob', mime_type=media_type) modality = _infer_modality_from_media_type(media_type) if modality is not None: blob_part['modality'] = modality if settings.include_content and settings.include_binary_content: blob_part['content'] = base64_content() return blob_part else: converted_part = _otel_messages.BinaryDataPart(type='binary', media_type=media_type) if settings.include_content and settings.include_binary_content: converted_part['content'] = base64_content() return converted_part @dataclass(repr=False) class UserPromptPart: """A user prompt, generally written by the end user. Content comes from the `user_prompt` parameter of [`Agent.run`][pydantic_ai.agent.AbstractAgent.run], [`Agent.run_sync`][pydantic_ai.agent.AbstractAgent.run_sync], and [`Agent.run_stream`][pydantic_ai.agent.AbstractAgent.run_stream]. """ content: str | Sequence[UserContent] """The content of the prompt.""" _: KW_ONLY timestamp: datetime = field(default_factory=_now_utc) """The timestamp of the prompt.""" part_kind: Literal['user-prompt'] = 'user-prompt' """Part type identifier, this is available on all parts as a discriminator.""" def __post_init__(self) -> None: # Every model's message mapper walks this content and hands each item to an exhaustive match. What # is not `UserContent` gets there as a bare `AssertionError: Expected code to be unreachable`, and # what is iterable but not a sequence -- a `dict`, most of all -- is walked as its keys, silently # sending them to the model as the prompt. Both are caught here, where the value comes in, rather # than once per mapper. `ValueError`, not `UserError`: `__post_init__` also runs when Pydantic # deserializes message history, where a `ValueError` becomes a `ValidationError` with location info. content = self.content if isinstance(content, str): return # `bytes` is a `Sequence` of `int`, so it passes the check below and fails on its first item with a # message about an `int` the caller never wrote. if not isinstance(content, Sequence) or isinstance(content, bytes | bytearray): raise ValueError( '`UserPromptPart.content` must be a `str` or a sequence of `UserContent` items, ' f'got `{type(content).__name__}`. {_NOT_USER_CONTENT}' ) for index, item in enumerate(content): if not isinstance(item, _USER_CONTENT_TYPES): raise ValueError( f'`UserPromptPart.content[{index}]` must be a `UserContent` item, ' f'got `{type(item).__name__}`. {_NOT_USER_CONTENT}' ) def otel_message_parts(self, settings: InstrumentationSettings) -> list[_otel_messages.MessagePart]: parts: list[_otel_messages.MessagePart] = [] content: Sequence[UserContent] = [self.content] if isinstance(self.content, str) else self.content for part in content: if isinstance(part, str | TextContent): content_str = part if isinstance(part, str) else part.content parts.append( _otel_messages.TextPart( type='text', **({'content': content_str} if settings.include_content else {}) ) ) elif isinstance(part, ImageUrl | AudioUrl | DocumentUrl | VideoUrl): if settings.version >= 4: uri_part = _otel_messages.UriPart(type='uri') modality = _kind_to_modality_lookup.get(part.kind) if modality is not None: uri_part['modality'] = modality try: # don't fail the whole message if media type can't be inferred for some reason, just omit it uri_part['mime_type'] = part.media_type except ValueError: pass if settings.include_content: uri_part['uri'] = part.url parts.append(uri_part) else: parts.append( _otel_messages.MediaUrlPart( type=part.kind, **{'url': part.url} if settings.include_content else {}, ) ) elif isinstance(part, BinaryContent): parts.append(_convert_binary_to_otel_part(part.media_type, lambda p=part: p.base64, settings)) elif isinstance(part, UploadedFile): # UploadedFile references provider-hosted files by file_id (OTel GenAI spec FilePart) # Infer modality from media_type - OTel spec defines: image, video, audio (or any string) category = part.media_type.split('/', 1)[0] if category in ('image', 'audio', 'video'): modality = category else: modality = 'document' # default for PDFs, text, etc. file_part = _otel_messages.FilePart(type='file', modality=modality, mime_type=part.media_type) if settings.include_content: file_part['file_id'] = part.file_id parts.append(file_part) elif isinstance(part, CachePoint): # CachePoint is a marker, not actual content - skip it for otel pass else: parts.append({'type': part.kind}) # pragma: no cover return parts __repr__ = _utils.dataclasses_no_defaults_repr RETURN_VALUE_KEY = 'return_value' """Key used to wrap non-dict tool return values in `model_response_object()`.""" tool_return_ta: pydantic.TypeAdapter[Any] = pydantic.TypeAdapter( Any, config=pydantic.ConfigDict(defer_build=True, ser_json_bytes='base64', val_json_bytes='base64') ) class _StrPassthrough: """The `str` arm of `ToolReturnContent`, matched entirely in Rust in both validation modes. Strings dominate the node count of a typical structured tool return, and every node of one crosses this union, so `str` is checked before the container arms. It is not a micro-optimisation: without it a string falls through all four remaining arms, measured at ~14x slower in `validate_python`. That figure is Rust-side arm-walking, so no frame count can pin it; what guards the arm against deletion is the `dump_json` leg of `test_tool_return_content_json_paths_make_no_per_node_python_calls`. `is_instance_schema` leaves `str` subclasses (a `StrEnum` returned by a tool, say) as they are, where pydantic's `str` validator would coerce them to a plain `str`; it can't run against JSON, where a strict `str` schema is exact anyway. Not `pydantic.InstanceOf[str]`, which builds the same validator but also attaches a wrap serializer — reintroducing a Python call per string node on the dump path. """ @classmethod def __get_pydantic_core_schema__( cls, _source_type: Any, _handler: pydantic.GetCoreSchemaHandler ) -> pydantic_core.CoreSchema: return pydantic_core.core_schema.json_or_python_schema( json_schema=pydantic_core.core_schema.str_schema(strict=True), python_schema=pydantic_core.core_schema.is_instance_schema(str), ) class _RequireUrlMediaType: """The `MultiModalContent` arm of `ToolReturnContent`, with an explicit `media_type` required of its URL items. A tool return is arbitrary user data, so this arm has to separate a multimodal item we serialized from a mapping a tool happened to build. For the four [`FileUrl`][pydantic_ai.messages.FileUrl] kinds, `media_type` draws that line, because those are the items whose media type the URL alone cannot always supply: `FileUrl.media_type` infers one from the URL when it was given none, and a URL with no usable extension raises `Could not infer media type` — on the *dump*, not on the load that built the object, so a history that had loaded cleanly could no longer be saved ([issue #4190](https://github.com/pydantic/pydantic-ai/issues/4190)). An item reconstructed here brings its own media type and never reaches that inference, and a URL mapping without one was never dumped by us: it stays a plain `Mapping` and reaches the caller with the keys its tool put in it. Nothing is required of the other two kinds, which cannot fail that way and so keep rehydrating from the fields they declare: `media_type` is a required field on `BinaryContent`, and `UploadedFile.media_type` falls back to `application/octet-stream` instead of raising. The requirement is a *non-empty* string. `FileUrl` infers whenever `_media_type` is falsy, so `''` would reconstruct an item that raises on dump after all, and no dump of ours writes one. The check is chained onto each URL choice of the tagged union rather than written as a validator, because any Python callable on this union is called once per node of the decoded payload — the cost [issue #7472](https://github.com/pydantic/pydantic-ai/issues/7472) was about. Chained inside the union it costs nothing measurable: the discriminator has already read `kind` in Rust, so only a mapping claiming one of the four URL kinds pays for it. The same check chained ahead of the union runs on every mapping node instead, measured at 1.29x on a dict-heavy payload. In python mode the check also admits an instance of ours, which reaches the choice as itself rather than as a mapping, carrying whatever media type it was built with. Making `_media_type` a required field on a copy of each dataclass schema would say the same thing with no wrapper at all, and does not work: two core schemas for one dataclass do not reliably build two validators, and the copy's requirement is dropped outright when no pydantic plugin is installed. `handler` hands back the tagged union `UserContent` also uses, so the copy is what keeps the requirement off a user prompt, which still accepts a file whose media type is inferred. The asserts guard the two shapes the surgery reads: that the union is still discriminated on a literal tag, and that each URL tag carries a schema of its own rather than a string aliasing another tag's. """ @classmethod def __get_pydantic_core_schema__( cls, source_type: Any, handler: pydantic.GetCoreSchemaHandler ) -> pydantic_core.CoreSchema: schema = deepcopy(handler(source_type)) assert schema['type'] == 'tagged-union', schema['type'] for kind in _FILE_URL_KINDS: choice = schema['choices'][kind] assert isinstance(choice, dict), choice schema['choices'][kind] = pydantic_core.core_schema.chain_schema([cls._names_a_media_type(), choice]) return schema @staticmethod def _names_a_media_type() -> pydantic_core.CoreSchema: mapping_naming_its_media_type = pydantic_core.core_schema.typed_dict_schema( { 'media_type': pydantic_core.core_schema.typed_dict_field( pydantic_core.core_schema.str_schema(min_length=1) ) }, extra_behavior='allow', ) return pydantic_core.core_schema.json_or_python_schema( json_schema=mapping_naming_its_media_type, # An instance is already one of ours and reaches the arm as itself, not as a mapping. python_schema=pydantic_core.core_schema.union_schema( [ mapping_naming_its_media_type, pydantic_core.core_schema.is_instance_schema(FileUrl), ], mode='left_to_right', ), ) if TYPE_CHECKING: # Simpler type for static analysis - recursive TypeAliasType with Any produces spurious Unknown types ToolReturnContent: TypeAlias = MultiModalContent | Sequence[Any] | Mapping[str, Any] | Any else: # Recursive type for runtime Pydantic validation - enables automatic reconstruction of # BinaryContent/FileUrl objects nested inside dicts/lists during deserialization. # # `left_to_right` is required because smart-union resolution otherwise picks `Mapping`/`Any` # over the inner-discriminated `MultiModalContent` branch in python mode, leaving multimodal # leaves as plain dicts. It also keeps arm selection in Rust: a callable `pydantic.Discriminator` # is invoked once per node of the decoded payload, making validation O(JSON nodes) Python calls. # # Falling through arm by arm is what keeps a user dict a plain mapping: `{'kind': 'binary', # 'label': 'foo'}` merely reuses one of our `kind` values, fails `MultiModalContent` — whose members # each require the fields they declare, and whose URL members additionally require a `media_type` # here — and lands on `Mapping`. The `Any` arm catches everything else — scalars, `bytes`, non-str # mapping keys — unchanged. ToolReturnContent = TypeAliasType( 'ToolReturnContent', Annotated[ Annotated[str, _StrPassthrough] | Annotated[MultiModalContent, _RequireUrlMediaType] | Mapping[str, 'ToolReturnContent'] | Sequence['ToolReturnContent'] | Any, pydantic.Field(union_mode='left_to_right'), ], ) tool_return_content_ta: pydantic.TypeAdapter[ToolReturnContent] = pydantic.TypeAdapter( ToolReturnContent, config=pydantic.ConfigDict(defer_build=True) ) """TypeAdapter for `ToolReturnContent` — used by UI adapters to rehydrate multimodal items (`BinaryContent`, `ImageUrl`, etc.) from raw JSON/dict payloads carried in wire-protocol fields typed as `Any` (e.g. Vercel's `ToolOutputAvailablePart.output`).""" ToolPartKind: TypeAlias = Literal['tool-search', 'capability-load'] """Discriminator value for the typed call/return-part subclass associated with a tool. Set on [`BaseToolCallPart.tool_kind`][pydantic_ai.messages.BaseToolCallPart.tool_kind], [`BaseToolReturnPart.tool_kind`][pydantic_ai.messages.BaseToolReturnPart.tool_kind], and [`ToolDefinition.tool_kind`][pydantic_ai.tools.ToolDefinition.tool_kind]. Extended as new typed-part families (e.g. web search) gain dedicated subclasses. Distinct from [`ToolKind`][pydantic_ai.tools.ToolKind] (invocation semantics — `'function'`, `'output'`, `'external'`, `'unapproved'`). """ _TOOL_PART_KINDS: tuple[ToolPartKind, ...] = get_args(ToolPartKind) def parse_tool_kind(value: str) -> ToolPartKind | None: """Return `value` if it's a known [`ToolPartKind`][pydantic_ai.messages.ToolPartKind], else `None`. UI adapters call this at the wire boundary to validate an untrusted client-supplied `tool_kind` string before setting it on a part, so an unknown value degrades to `None` rather than asserting a bogus discriminator. """ return next((kind for kind in _TOOL_PART_KINDS if kind == value), None) INTERRUPTED_TOOL_RETURN_CONTENT = 'The tool call was interrupted before a result was produced.' """Placeholder content for a tool call that was interrupted before producing a result (e.g. by run cancellation). Shared between the agent graph's history repair and the UI adapters' stream closeout so both synthesize the same `outcome='interrupted'` return.""" def _tool_result_provenance_tags(tool_name: str, tool_call_id: str, file_identifier: str) -> tuple[str, str]: """The open and close tags framing one tool-produced file that has to travel on the user channel. A provider whose tool result channel is text-only can only deliver a tool's multimodal output on the user channel — the same one the end user's own uploads travel on — which leaves the model unable to tell a tool attachment from something the person it is talking to attached. The tags name the call the media came from, so tool output, which may be attacker-influenced, is no longer presented at user trust level. `Model.prepare_messages` frames a `SystemPromptPart` a provider can't send natively as `...` for the same reason. Each file is framed on its own so that `file_id` carries the same identifier the tool result text cross-references as "See file .", which one set of tags around a whole call's files could not name. They identify rather than prove. This is prompt text like any other, so a tool can emit the closing tag and a user can type the opening one; the attribute values are escaped, and a provider that accepts media in its tool result channel sends it there and never renders these at all. """ return ( f'', '', ) @dataclass(repr=False) class BaseToolReturnPart: """Base class for tool return parts.""" tool_name: str """The name of the tool that was called.""" content: ToolReturnContent """The tool return content, which may include multimodal files.""" tool_call_id: str = field(default_factory=_generate_tool_call_id) """The tool call identifier, this is used by some models including OpenAI. In case the tool call id is not provided by the model, Pydantic AI will generate a random one. """ _: KW_ONLY tool_kind: ToolPartKind | None = None """Discriminator for the typed subclass of this part (e.g. `'tool-search'`). `None` for any part without a typed subclass — including all user-defined tools and all native tools without a dedicated typed call/return shape. Subclasses that pin this to a [`ToolPartKind`][pydantic_ai.messages.ToolPartKind] literal: * [`ToolSearchCallPart`][pydantic_ai.messages.ToolSearchCallPart] / [`ToolSearchReturnPart`][pydantic_ai.messages.ToolSearchReturnPart] — `'tool-search'` * [`NativeToolSearchCallPart`][pydantic_ai.messages.NativeToolSearchCallPart] / [`NativeToolSearchReturnPart`][pydantic_ai.messages.NativeToolSearchReturnPart] — `'tool-search'` """ metadata: Any = None """Additional data accessible by the application but not sent to the LLM.""" timestamp: datetime = field(default_factory=_now_utc) """The timestamp, when the tool returned.""" outcome: Literal['success', 'failed', 'denied', 'interrupted'] = 'success' """The outcome of the tool call. - `'success'`: The tool executed successfully. - `'failed'`: The tool call failed — the tool raised an error during execution (the common case), or an args validator or tool hook reported a failure via [`ToolFailed`][pydantic_ai.exceptions.ToolFailed]. - `'denied'`: The tool call was denied — either by the approval mechanism or by a [`HandleDeferredToolCalls`][pydantic_ai.capabilities.HandleDeferredToolCalls] handler returning [`ToolDenied`][pydantic_ai.tools.ToolDenied]. - `'interrupted'`: The tool call did not produce a result because the run was interrupted (e.g. a cancelled stream or a crash mid-execution); synthesized during message-history repair. Only `'failed'` is mapped to a provider's native error channel (e.g. Anthropic `is_error`, Bedrock `status='error'`). A denial is a deliberate policy decision rather than a runtime error, while an interruption means no result was produced. Both are sent as ordinary results; their content tells the model what happened without suggesting a transient tool failure. """ def _split_content(self) -> tuple[list[Any], list[MultiModalContent], bool]: """Split content into non-file and file parts. Returns: A 3-tuple of (`data_parts`, `file_parts`, `was_list`) where `was_list` indicates whether the original content was a list. """ if is_multi_modal_content(self.content): return [], [self.content], False elif isinstance(self.content, list): non_files: list[Any] = [] files: list[MultiModalContent] = [] # ToolReturnContent uses a recursive TypeAliasType at runtime (for Pydantic validation) # but a simpler union at type-check time, so pyright can't infer `p`'s type from the list. for p in self.content: # pyright: ignore[reportUnknownVariableType, reportUnknownMemberType] if is_multi_modal_content(p): files.append(p) else: non_files.append(p) return non_files, files, True return [self.content], [], False def _unwrap_data(self) -> tuple[Any, list[MultiModalContent]]: """Split content and unwrap single-item data lists. Returns the unwrapped data value (or None if empty) and the file parts. Single-item lists are unwrapped when content was scalar or when files were filtered out. """ data, files, was_list = self._split_content() if not data: return None, files # Unwrap single-item data: either content was originally scalar (!was_list) # or extracting files reduced a multi-item list to one element. if len(data) == 1 and (not was_list or bool(files)): return data[0], files return data, files @property def files(self) -> list[MultiModalContent]: """The multimodal file parts from `content` (`ImageUrl`, `AudioUrl`, `DocumentUrl`, `VideoUrl`, `BinaryContent`).""" _, files, _ = self._split_content() return files @overload def content_items(self, *, mode: Literal['raw'] = 'raw') -> list[ToolReturnContent]: ... @overload def content_items(self, *, mode: Literal['str'], wrap_if_error: bool = True) -> list[str | MultiModalContent]: ... @overload def content_items( self, *, mode: Literal['jsonable'], wrap_if_error: bool = True ) -> list[Any | MultiModalContent]: ... def content_items( self, *, mode: Literal['raw', 'str', 'jsonable'] = 'raw', wrap_if_error: bool = True ) -> list[ToolReturnContent] | list[str | MultiModalContent] | list[Any | MultiModalContent]: """Return content as a flat list for iteration, with optional serialization. Args: mode: Controls serialization of non-file items: - `'raw'`: No serialization. Returns items as-is. - `'str'`: Non-file items are serialized to strings via `tool_return_ta`. File items (`MultiModalContent`) pass through unchanged. - `'jsonable'`: Non-file items are serialized to JSON-compatible Python objects via `tool_return_ta`. File items pass through unchanged. wrap_if_error: Whether to wrap failed tool returns in an `{"error": ...}` object (ignored in `'raw'` mode). When `True` (the default), a failed return's non-file data collapses into a single wrapped error item so providers without a native error channel still see the failure explicitly; files pass through unchanged. Set this to `False` when the provider has a native error channel (e.g. Anthropic `is_error`) and should receive the content unwrapped. """ items: list[ToolReturnContent] if isinstance(self.content, list): items = self.content # pyright: ignore[reportUnknownVariableType,reportUnknownMemberType] else: items = [self.content] if mode == 'raw': return items if wrap_if_error and self.outcome == 'failed': wrapped = self.model_response_str() if mode == 'str' else self.model_response_object() return [wrapped, *self.files] result: list[str | MultiModalContent] | list[Any | MultiModalContent] = [] for item in items: if is_multi_modal_content(item): result.append(item) elif isinstance(item, str): result.append(item) elif mode == 'str': result.append(tool_return_ta.dump_json(item, by_alias=True).decode()) else: result.append(tool_return_ta.dump_python(item, mode='json', by_alias=True)) return result def model_response_str(self, *, wrap_if_error: bool = True) -> str: """Return a string representation of the data content for the model. This excludes multimodal files - use `.files` to get those separately. Args: wrap_if_error: Whether to wrap failed tool returns in an `{"error": ...}` object. Set this to `False` when the provider has a native error channel. """ value, _ = self._unwrap_data() if value is None: response = '' elif isinstance(value, str): response = value else: response = tool_return_ta.dump_json(value, by_alias=True).decode() if wrap_if_error and self.outcome == 'failed': return tool_return_ta.dump_json({'error': response}).decode() return response def model_response_object(self, *, wrap_if_error: bool = True) -> dict[str, Any]: """Return a dictionary representation of the data content, wrapping non-dict types appropriately. This excludes multimodal files - use `.files` to get those separately. Gemini supports JSON dict return values, but no other JSON types, hence we wrap anything else in a dict. Args: wrap_if_error: Whether to wrap failed tool returns in an `{"error": ...}` object. Set this to `False` when the provider has a native error channel. """ if wrap_if_error and self.outcome == 'failed': return {'error': self.model_response_str(wrap_if_error=False)} value, _ = self._unwrap_data() if value is None: return {} json_content = tool_return_ta.dump_python(value, mode='json', by_alias=True) if _utils.is_str_dict(json_content): return json_content else: return {RETURN_VALUE_KEY: json_content} def structured_content(self) -> dict[str, Any] | list[Any] | None: """Return `content` as structured JSON data (a `dict` or `list`), or `None` if it has none. A JSON string is parsed; already-structured content is returned as-is; a plain/non-JSON string, scalar, or multimodal content yields `None` (there is no structured payload). A read-side companion to [`files`][pydantic_ai.messages.BaseToolReturnPart.files] and [`model_response_object`][pydantic_ai.messages.BaseToolReturnPart.model_response_object]; some UI wire formats (e.g. AG-UI) transmit tool results as JSON strings, so [`narrow_type`][pydantic_ai.messages.ToolReturnPart.narrow_type] uses it to recover the structured payload a typed return subclass expects. """ content = self.content if isinstance(content, str): try: content = pydantic_core.from_json(content) except ValueError: return None if isinstance(content, dict): return cast('dict[str, Any]', content) if isinstance(content, list): return cast('list[Any]', content) return None def model_response_str_and_user_content(self, *, wrap_if_error: bool = True) -> tuple[str, list[UserContent]]: """Build a text-only tool result with multimodal files extracted for a trailing user message. For providers whose tool result API only accepts text. Multimodal files are referenced by identifier in the tool result text ('See file {id}.') and included in full in the returned file content list. Each file is framed by `_tool_result_provenance_tags` so the model can tell it from the user's own uploads, which travel on the same channel. Args: wrap_if_error: Whether to wrap failed tool returns in an `{"error": ...}` object. Set this to `False` when the provider has a native error channel. """ _, files, was_list = self._split_content() if not files: return self.model_response_str(wrap_if_error=wrap_if_error), [] tool_content_parts: list[str] = [] file_content: list[UserContent] = [] for item in self.content_items(mode='str', wrap_if_error=False): if is_multi_modal_content(item): tool_content_parts.append(f'See file {item.identifier}.') open_tag, close_tag = _tool_result_provenance_tags(self.tool_name, self.tool_call_id, item.identifier) file_content.extend([open_tag, item, close_tag]) elif isinstance(item, str): # pragma: no branch tool_content_parts.append(item) if wrap_if_error and self.outcome == 'failed': error = {'error': self.model_response_str(wrap_if_error=False)} file_references = [f'See file {file.identifier}.' for file in files] return tool_return_ta.dump_json([error, *file_references]).decode(), file_content if was_list: return tool_return_ta.dump_json(tool_content_parts).decode(), file_content # Safe: when was_list is False, content is either scalar data (→ str item) or a single # MultiModalContent (→ 'See file ...' placeholder), so tool_content_parts always has one entry. return tool_content_parts[0], file_content def otel_message_parts(self, settings: InstrumentationSettings) -> list[_otel_messages.MessagePart]: part = _otel_messages.ToolCallResponsePart( type='tool_call_response', id=self.tool_call_id, name=self.tool_name, ) if settings.include_content and self.content is not None: part['result'] = serialize_any(redact_binary_content(self.content, settings)) return [part] def has_content(self) -> bool: """Return `True` if the tool return has content.""" return self.content is not None # pragma: no cover __repr__ = _utils.dataclasses_no_defaults_repr @dataclass(repr=False) class ToolReturnPart(BaseToolReturnPart): """A tool return message, this encodes the result of running a tool.""" _: KW_ONLY part_kind: Literal['tool-return'] = 'tool-return' """Part type identifier, this is available on all parts as a discriminator.""" @staticmethod def narrow_type(part: ToolReturnPart, *, tool_kind: ToolPartKind | None = None) -> ToolReturnPart: """Promote a base `ToolReturnPart` to its typed subclass when its `tool_kind` is registered. Best-effort: returns the part unchanged when the `tool_kind` (kwarg or on the part) resolves to no registered subclass, and strips an unsubstantiated `tool_kind` when the part's data doesn't validate against that subclass — keeping it on a base part would break a [`ModelMessagesTypeAdapter`][pydantic_ai.messages.ModelMessagesTypeAdapter] round-trip. For direct construction; Pydantic deserialization promotes automatically via the discriminated union. """ return _narrow_return(part, _TOOL_RETURN_NARROWERS, tool_kind) @dataclass(repr=False) class NativeToolReturnPart(BaseToolReturnPart): """A tool return message from a native tool. For native tools with a stable cross-provider shape (currently `tool_search`), a `NativeToolReturnPart` may be promoted to a typed subclass like [`NativeToolSearchReturnPart`][pydantic_ai.messages.NativeToolSearchReturnPart] with a narrowed `content` `TypedDict`. See `NativeToolCallPart` for the pattern. """ _: KW_ONLY provider_name: str | None = None """The name of the provider that generated the response. Required to be set when `provider_details` is set. """ provider_details: dict[str, Any] | None = None """Additional data returned by the provider that can't be mapped to standard fields. This is used for data that is required to be sent back to APIs, as well as data users may want to access programmatically. When this field is set, `provider_name` is required to identify the provider that generated this data.""" part_kind: Literal['builtin-tool-return'] = 'builtin-tool-return' """Part type identifier, this is available on all parts as a discriminator.""" @staticmethod def narrow_type(part: NativeToolReturnPart, *, tool_kind: ToolPartKind | None = None) -> NativeToolReturnPart: """Promote a base `NativeToolReturnPart` to its typed subclass when its `tool_kind` is registered. Best-effort: returns the part unchanged when the `tool_kind` (kwarg or on the part) resolves to no registered subclass, and strips an unsubstantiated `tool_kind` when the part's data doesn't validate against that subclass — keeping it on a base part would break a [`ModelMessagesTypeAdapter`][pydantic_ai.messages.ModelMessagesTypeAdapter] round-trip. For direct construction; Pydantic deserialization promotes automatically via the discriminated union. """ return _narrow_return(part, _NATIVE_RETURN_NARROWERS, tool_kind) error_details_ta = pydantic.TypeAdapter(list[pydantic_core.ErrorDetails], config=pydantic.ConfigDict(defer_build=True)) @dataclass(repr=False) class RetryPromptPart: """A message back to a model asking it to try again. This can be sent for a number of reasons: * Pydantic validation of tool arguments failed, here content is derived from a Pydantic [`ValidationError`][pydantic_core.ValidationError] * a tool raised a [`ModelRetry`][pydantic_ai.exceptions.ModelRetry] exception * no tool was found for the tool name * the model returned plain text when a structured response was expected * Pydantic validation of a structured response failed, here content is derived from a Pydantic [`ValidationError`][pydantic_core.ValidationError] * an output validator raised a [`ModelRetry`][pydantic_ai.exceptions.ModelRetry] exception """ content: list[pydantic_core.ErrorDetails] | str """Details of why and how the model should retry. If the retry was triggered by a [`ValidationError`][pydantic_core.ValidationError], this will be a list of error details. """ _: KW_ONLY tool_name: str | None = None """The name of the tool that was called, if any.""" tool_call_id: str = field(default_factory=_generate_tool_call_id) """The tool call identifier, this is used by some models including OpenAI. In case the tool call id is not provided by the model, Pydantic AI will generate a random one. """ timestamp: datetime = field(default_factory=_now_utc) """The timestamp, when the retry was triggered.""" part_kind: Literal['retry-prompt'] = 'retry-prompt' """Part type identifier, this is available on all parts as a discriminator.""" @classmethod def from_error( cls, error: pydantic_core.ValidationError | ModelRetry, *, tool_name: str | None = None, tool_call_id: str | None = None, ) -> RetryPromptPart: """Build the retry prompt for a failed tool call or output validation. This is the exact message the model receives when the error is handled by the agent loop, so anything else presenting the failure (e.g. instrumentation spans) must build it the same way. """ content = ( error.errors(include_url=False, include_context=False) if isinstance(error, pydantic_core.ValidationError) else error.message ) part = cls(content=content, tool_name=tool_name) if tool_call_id: part.tool_call_id = tool_call_id return part def model_response(self) -> str: """Return a string message describing why the retry is requested.""" if isinstance(self.content, str): if self.tool_name is None: description = f'Validation feedback:\n{self.content}' else: description = self.content else: # For NativeOutput retries (no `tool_name`) the generated JSON is already in the model's # context, so top-level errors' `input` just duplicates it. Tool-call retries keep `input` # so the model sees what arguments it sent alongside the error. if self.tool_name is None: exclude = { i: {'ctx', 'input'} if len(e.get('loc', ())) <= 1 else {'ctx'} for i, e in enumerate(self.content) } else: exclude = {'__all__': {'ctx'}} json_errors = error_details_ta.dump_json(self.content, exclude=exclude, indent=2) plural = isinstance(self.content, list) and len(self.content) != 1 description = ( f'{len(self.content)} validation error{"s" if plural else ""}:\n```json\n{json_errors.decode()}\n```' ) return f'{description}\n\nFix the errors and try again.' def otel_message_parts(self, settings: InstrumentationSettings) -> list[_otel_messages.MessagePart]: if self.tool_name is None: return [ _otel_messages.TextPart( type='text', **({'content': self.model_response()} if settings.include_content else {}) ) ] else: part = _otel_messages.ToolCallResponsePart( type='tool_call_response', id=self.tool_call_id, name=self.tool_name, ) if settings.include_content: part['result'] = self.model_response() return [part] __repr__ = _utils.dataclasses_no_defaults_repr # `ModelRequestPart` is defined further down (after the typed `ToolSearchReturnPart` # subclass) so it can include the local `search_tools` return as a discriminated-union # member. The forward reference inside `ModelRequest.parts` works because of # `from __future__ import annotations` at the top of this module. @dataclass(frozen=True, repr=False) class AgentInstructionSource: """The agent's own instructions. There is exactly one agent in scope, so it carries no id. """ def __str__(self) -> str: return 'agent' __repr__ = _utils.dataclasses_no_defaults_repr @dataclass(frozen=True, repr=False) class ToolsetInstructionSource: """A toolset with an `id`.""" id: str def __str__(self) -> str: return f'toolset:{self.id}' __repr__ = _utils.dataclasses_no_defaults_repr @dataclass(frozen=True, repr=False) class CapabilityInstructionSource: """A capability with an `id`.""" id: str def __str__(self) -> str: return f'capability:{self.id}' __repr__ = _utils.dataclasses_no_defaults_repr InstructionSource: TypeAlias = AgentInstructionSource | ToolsetInstructionSource | CapabilityInstructionSource """The source that authored an instruction part.""" @dataclass(frozen=True, repr=False) class InstructionId: """The key an instruction part is addressed by: who contributed it, and which of their parts it is.""" source: InstructionSource """Who contributed the part: the agent, a toolset with an `id`, or a capability with an `id`.""" _: KW_ONLY name: str | None = None """Which of that source's parts this is, or `None` to address everything the source contributes.""" def __str__(self) -> str: return f'{self.source}:{self.name}' if self.name is not None else str(self.source) __repr__ = _utils.dataclasses_no_defaults_repr def _deserialize_instruction_id(value: str | InstructionId | None) -> InstructionId | None: """Rebuild a key from the string it renders to. A namespace this version doesn't know can only come from a newer one, whose keys it has no way to address anyway, so it reads as unaddressable rather than failing the message it arrived on. """ if not isinstance(value, str): return value if value == 'agent': return InstructionId(AgentInstructionSource()) if value.startswith('agent:'): return InstructionId(AgentInstructionSource(), name=value.removeprefix('agent:')) for namespace, source_type in ( ('toolset', ToolsetInstructionSource), ('capability', CapabilityInstructionSource), ): prefix = f'{namespace}:' if value.startswith(prefix): source_id, separator, name = value.removeprefix(prefix).partition(':') return InstructionId(source_type(source_id), name=name if separator else None) return None def _serialize_instruction_id(value: InstructionId | None) -> str | None: return str(value) if value is not None else None SerializedInstructionId: TypeAlias = Annotated[ InstructionId | None, pydantic.BeforeValidator(_deserialize_instruction_id), pydantic.PlainSerializer(_serialize_instruction_id, return_type=str | None, when_used='json'), ] """An [`InstructionId`][pydantic_ai.messages.InstructionId] that persists as the string it renders to.""" @dataclass(repr=False) class InstructionPart: """A single instruction part with metadata about its origin. Instructions are composed of one or more parts, each of which can be static (from a literal string) or dynamic (from a function, template, or toolset). This distinction allows model implementations to make intelligent caching decisions — e.g. Anthropic's prompt caching can cache the static prefix while leaving dynamic instructions uncached. """ content: str """The text content of this instruction part.""" _: KW_ONLY dynamic: bool = False """Whether this instruction came from a dynamic source (function, template, or toolset). Static instructions (`dynamic=False`) come from literal strings passed to `Agent(instructions=...)`. Dynamic instructions (`dynamic=True`) come from `@agent.instructions` functions, `TemplateStr`, or toolset `get_instructions()` methods. """ name: str | None = None """What the author calls this part, relative to whatever contributes it. Name a part relative to what you own — `'limits'`, not `'toolset:weather:limits'` — and the source you contribute through qualifies it into an [`id`][pydantic_ai.messages.InstructionPart.id]. A name cannot contain `:`, which delimits the segments of an id, and cannot be `'agent'`, which is the key of the agent's own instructions. Naming a part whose source has no identity of its own leaves `id` as `None`: there is no source key to qualify the name against, so it says what the part is without making it addressable. """ id: SerializedInstructionId = None """The stable key this part is addressed by, or `None` if nothing addresses it. The framework issues this while collecting instructions, from the author's [`name`][pydantic_ai.messages.InstructionPart.name] and the identity of the source that contributed the part — declare a `name` rather than setting this yourself. A consumer reading [`ModelRequestParameters.instruction_parts`][pydantic_ai.models.ModelRequestParameters.instruction_parts] can persist configuration against an id, because it survives the reordering and rewording that a part's position and text do not. [`InstructionId.source`][pydantic_ai.messages.InstructionId.source] is who contributed the part: - [`AgentInstructionSource`][pydantic_ai.messages.AgentInstructionSource] — the agent's literal instructions - [`ToolsetInstructionSource`][pydantic_ai.messages.ToolsetInstructionSource] — a toolset with an `id` - [`CapabilityInstructionSource`][pydantic_ai.messages.CapabilityInstructionSource] — a capability with an `id` An id renders and serializes as its segments joined by `:` — `'agent'`, `'toolset:weather'`, or `'capability:budget:remaining'`. """ part_kind: Literal['instruction'] = 'instruction' """Part type identifier, used as a discriminator for deserialization.""" @staticmethod def join(parts: Sequence[InstructionPart]) -> str | None: """Join instruction parts into a single string, separated by double newlines.""" return '\n\n'.join(p.content for p in parts).strip() or None @staticmethod def sorted(parts: Sequence[InstructionPart]) -> list[InstructionPart]: """Sort instruction parts with static (`dynamic=False`) before dynamic, preserving relative order.""" return sorted(parts, key=lambda p: p.dynamic) __repr__ = _utils.dataclasses_no_defaults_repr @dataclass(repr=False, kw_only=True) class ToolAvailabilityDeltaPart: """Records that the set of tools available to the model changed at this point. Additions only. Withdrawing a tool is not supported yet, because no provider can be told about one without also invalidating the prompt cache this part exists to protect: Anthropic rejects a reference to a tool the request doesn't declare, so a withdrawn tool has to leave the `tools` array, and that is itself the invalidation. The name says *availability* rather than *addition* so removals can join once they can be done cache-safely — see https://github.com/pydantic/pydantic-ai/issues/6985. """ tools_added: Annotated[ list[str], pydantic.Field(validation_alias=pydantic.AliasChoices('tools_added', 'added')) ] = field(default_factory=lambda: []) """Names of tools this point in history reveals. A reveal is what the model has been *shown*; whether the tool is callable is the broader availability question, which for a capability-owned tool also asks whether its owning capability is loaded. """ tool_call_id: str | None = None """The tool call associated with the change, if any.""" part_kind: Literal['tool-availability-delta'] = 'tool-availability-delta' """Part type identifier, this is available on all parts as a discriminator.""" def otel_message_parts(self, settings: InstrumentationSettings) -> list[_otel_messages.MessagePart]: """Render the change as trace content. Tool names are recorded regardless of `include_content`: they aren't user content, they're already visible in the request's tool definitions, and a run where the model suddenly can call something is unreadable without them. """ changes = ', '.join(f'+{name}' for name in self.tools_added) return [_otel_messages.TextPart(type='text', content=f'Tool availability changed: {changes}')] __repr__ = _utils.dataclasses_no_defaults_repr @dataclass(repr=False) class ModelRequest: """A request generated by Pydantic AI and sent to a model, e.g. a message from the Pydantic AI app to the model.""" parts: Sequence[ModelRequestPart] """The parts of the user message.""" _: KW_ONLY # Default is None for backwards compatibility with old serialized messages that don't have this field. # Using a default_factory would incorrectly fill in the current time for deserialized historical messages. timestamp: datetime | None = None """The timestamp when the request was sent to the model.""" instructions: str | None = None """The instructions string for this request, rendered from structured instruction parts.""" kind: Literal['request'] = 'request' """Message type identifier, this is available on all parts as a discriminator.""" run_id: str | None = None """The unique identifier of the agent run in which this message originated.""" conversation_id: str | None = None """The unique identifier of the conversation this message belongs to. A conversation spans potentially multiple agent runs that share message history. Emitted as the `gen_ai.conversation.id` OpenTelemetry span attribute on the agent run. """ metadata: dict[str, Any] | None = None """Additional data that can be accessed programmatically by the application but is not sent to the LLM.""" state: ModelRequestState = 'complete' """Lifecycle state of the request. Set to `'interrupted'` when the request was being assembled (e.g. collecting tool returns) and the run was abnormally terminated by an exception or cancellation before the request was sent to the model. In that case `parts` holds only the tool returns that were collected, and is empty if none were. Appears in [`capture_run_messages`][pydantic_ai.capture_run_messages] output so consumers can detect partial state. """ def __post_init__(self) -> None: for part in self.parts: if isinstance(part, SpeechPart) and part.speaker != 'user': # `ValueError`, not `UserError`: `__post_init__` also runs when Pydantic deserializes # message history, where a `ValueError` becomes a `ValidationError` with location info. raise ValueError( f"`SpeechPart` in `ModelRequest.parts` must have `speaker='user'`, got {part.speaker!r}" ) @classmethod def user_text_prompt(cls, user_prompt: str, *, instructions: str | None = None) -> ModelRequest: """Create a `ModelRequest` with a single user prompt as text.""" return cls(parts=[UserPromptPart(user_prompt)], instructions=instructions) __repr__ = _utils.dataclasses_no_defaults_repr @dataclass(repr=False) class TextPart: """A plain text response from a model.""" content: str """The text content of the response.""" _: KW_ONLY id: str | None = None """An optional identifier of the text part. When this field is set, `provider_name` is required to identify the provider that generated this data. """ provider_name: str | None = None """The name of the provider that generated the response. Required to be set when `provider_details` or `id` is set. """ provider_details: dict[str, Any] | None = None """Additional data returned by the provider that can't be mapped to standard fields. This is used for data that is required to be sent back to APIs, as well as data users may want to access programmatically. When this field is set, `provider_name` is required to identify the provider that generated this data. """ part_kind: Literal['text'] = 'text' """Part type identifier, this is available on all parts as a discriminator.""" def has_content(self) -> bool: """Return `True` if the text content is non-empty.""" return bool(self.content) __repr__ = _utils.dataclasses_no_defaults_repr @dataclass(repr=False) class ThinkingPart: """A thinking response from a model.""" content: str """The thinking content of the response.""" _: KW_ONLY id: str | None = None """The identifier of the thinking part. When this field is set, `provider_name` is required to identify the provider that generated this data. """ signature: str | None = None """The signature of the thinking. Supported by: * Anthropic (corresponds to the `signature` field) * Bedrock (corresponds to the `signature` field) * Google (corresponds to the `thought_signature` field) * OpenAI (corresponds to the `encrypted_content` field) When this field is set, `provider_name` is required to identify the provider that generated this data. """ provider_name: str | None = None """The name of the provider that generated the response. Signatures are only sent back to the same provider. Required to be set when `provider_details`, `id` or `signature` is set. """ provider_details: dict[str, Any] | None = None """Additional data returned by the provider that can't be mapped to standard fields. This is used for data that is required to be sent back to APIs, as well as data users may want to access programmatically. When this field is set, `provider_name` is required to identify the provider that generated this data. """ part_kind: Literal['thinking'] = 'thinking' """Part type identifier, this is available on all parts as a discriminator.""" def has_content(self) -> bool: """Return `True` if the thinking content is non-empty.""" return bool(self.content) __repr__ = _utils.dataclasses_no_defaults_repr STANDING_PROMPT_PLANTED_KEY = 'pydantic_ai_standing_prompt_planted' """`CompactionPart.provider_details` key stamped on compaction items minted by our own compact call, whose input window explicitly planted the standing prompt. Provenance for `_trim_messages_before_compaction`'s `standing_prompt_retained` fast path: only a stamped item is trusted to retain the standing prompt; anything else — an externally supplied or spliced history, or an item produced by a provider-initiated compaction of an ordinary request window — gets the standing prompt re-inserted.""" @dataclass(repr=False) class CompactionPart: """A compaction part that summarizes previous conversation history. Compaction parts contain an opaque or readable summary of prior messages, produced by provider-specific compaction mechanisms. They must be round-tripped back to the same provider in subsequent requests. For Anthropic, `content` contains a readable text summary. For OpenAI, `content` is `None` and the encrypted data is stored in `provider_details`. """ content: str | None = None """The compaction summary text, if available. For Anthropic: a readable text summary of compacted messages. For OpenAI: `None` (the compacted content is encrypted and stored in `provider_details`). """ _: KW_ONLY id: str | None = None """The identifier of the compaction part. When this field is set, `provider_name` is required to identify the provider that generated this data. """ provider_name: str | None = None """The name of the provider that generated the compaction. Compaction data is only sent back to the same provider. Required to be set when `provider_details` or `id` is set. """ provider_details: dict[str, Any] | None = None """Additional data returned by the provider that can't be mapped to standard fields. For OpenAI: contains `encrypted_content` and other fields from `ResponseCompactionItem`. When this field is set, `provider_name` is required to identify the provider that generated this data. """ part_kind: Literal['compaction'] = 'compaction' """Part type identifier, this is available on all parts as a discriminator.""" def has_content(self) -> bool: """Return `True` if the compaction content is non-empty.""" return bool(self.content) __repr__ = _utils.dataclasses_no_defaults_repr @dataclass(repr=False) class FilePart: """A file response from a model.""" content: Annotated[BinaryContent, pydantic.AfterValidator(BinaryContent.narrow_type)] """The file content of the response.""" _: KW_ONLY id: str | None = None """The identifier of the file part. When this field is set, `provider_name` is required to identify the provider that generated this data. """ provider_name: str | None = None """The name of the provider that generated the response. Required to be set when `provider_details` or `id` is set. """ provider_details: dict[str, Any] | None = None """Additional data returned by the provider that can't be mapped to standard fields. This is used for data that is required to be sent back to APIs, as well as data users may want to access programmatically. When this field is set, `provider_name` is required to identify the provider that generated this data. """ part_kind: Literal['file'] = 'file' """Part type identifier, this is available on all parts as a discriminator.""" def has_content(self) -> bool: """Return `True` if the file content is non-empty.""" return bool(self.content.data) __repr__ = _utils.dataclasses_no_defaults_repr @dataclass(repr=False, kw_only=True) class SpeechPart: """Spoken audio exchanged during a realtime session, paired with its transcript. This part is a member of both [`ModelRequestPart`][pydantic_ai.messages.ModelRequestPart] and [`ModelResponsePart`][pydantic_ai.messages.ModelResponsePart], distinguished by `speaker`: in `ModelRequest.parts` the speaker is always `'user'`; in `ModelResponse.parts` it is always `'assistant'`. This invariant is enforced at runtime when a message is constructed. Standard (non-realtime) models can't consume this part directly; when history containing it is used in an agent run, [`Model.prepare_messages`][pydantic_ai.models.Model.prepare_messages] converts user-speaker parts to [`UserPromptPart`][pydantic_ai.messages.UserPromptPart]s and assistant-speaker parts to [`TextPart`][pydantic_ai.messages.TextPart]s. """ speaker: Literal['user', 'assistant'] """Whether the audio was spoken by the end user or by the model.""" transcript: str | None = None """The transcript of the audio. `None` if transcription was unavailable.""" audio: BinaryContent | None = None """The audio data, if retained. Audio is only retained when the realtime session is configured to do so (see the `audio_retention` setting), so this is usually `None`. """ interrupted_at_ms: int | None = None """The offset into this part's audio where playback was interrupted, in milliseconds. `None` when the part was not interrupted. It may also be `None` for an interrupted turn when the provider reported the interruption without an offset. This is relative to this part's audio, not wall-clock or session-relative time. """ id: str | None = None """The provider item ID, used to correlate the part with provider-side conversation items.""" provider_name: str | None = None """The name of the provider that generated or transcribed the audio. Required to be set when `provider_details` or `id` is set. """ provider_details: dict[str, Any] | None = None """Additional data returned by the provider that can't be mapped to standard fields. This is used for data that is required to be sent back to APIs, as well as data users may want to access programmatically. When this field is set, `provider_name` is required to identify the provider that generated this data. """ part_kind: Literal['speech'] = 'speech' """Part type identifier, this is available on all parts as a discriminator.""" @property def content(self) -> str: """The transcript, or an empty string if transcription was unavailable. Mirrors [`TextPart.content`][pydantic_ai.messages.TextPart.content] so code that renders message parts generically can treat spoken content like text. """ return self.transcript or '' def otel_message_parts(self, settings: InstrumentationSettings) -> list[_otel_messages.MessagePart]: parts: list[_otel_messages.MessagePart] = [] if self.transcript is not None: parts.append( _otel_messages.TextPart( type='text', **({'content': self.transcript} if settings.include_content else {}) ) ) if (audio := self.audio) is not None: parts.append(_convert_binary_to_otel_part(audio.media_type, lambda: audio.base64, settings)) return parts def has_content(self) -> bool: """Return `True` if the part has a transcript or retained audio.""" return bool(self.transcript) or self.audio is not None __repr__ = _utils.dataclasses_no_defaults_repr @dataclass(repr=False) class BaseToolCallPart: """A tool call from a model.""" tool_name: str """The name of the tool to call.""" args: str | dict[str, Any] | None = None """The arguments to pass to the tool. This is stored either as a JSON string or a Python dictionary depending on how data was received. """ tool_call_id: str = field(default_factory=_generate_tool_call_id) """The tool call identifier, this is used by some models including OpenAI. In case the tool call id is not provided by the model, Pydantic AI will generate a random one. """ _: KW_ONLY tool_kind: ToolPartKind | None = None """Discriminator for the typed subclass of this part (e.g. `'tool-search'`). See [`BaseToolReturnPart.tool_kind`][pydantic_ai.messages.BaseToolReturnPart.tool_kind] for the full semantics. """ id: str | None = None """An optional identifier of the tool call part, separate from the tool call ID. This is used by some APIs like OpenAI Responses. When this field is set, `provider_name` is required to identify the provider that generated this data. """ provider_name: str | None = None """The name of the provider that generated the response. Native tool calls are only sent back to the same provider. Required to be set when `provider_details` or `id` is set. """ provider_details: dict[str, Any] | None = None """Additional data returned by the provider that can't be mapped to standard fields. This is used for data that is required to be sent back to APIs, as well as data users may want to access programmatically. When this field is set, `provider_name` is required to identify the provider that generated this data. """ def __post_init__(self) -> None: # Per-instance attribute populated by the instrumentation layer from # `ToolDefinition.metadata` to drive OTel rendering hints (e.g. syntax highlighting). # Declared here rather than as a dataclass field so it stays out of `__init__`, # equality, repr, Pydantic JSON schema, and serialization. self.otel_metadata: _otel_messages.ToolCallPartOtelMetadata | None = None def args_as_dict(self, *, raise_if_invalid: bool = False) -> dict[str, Any]: """Return the arguments as a Python dictionary. This is just for convenience with models that require dicts as input. Args: raise_if_invalid: If `True`, a `ValueError` or `AssertionError` caused by malformed or non-object JSON in `args` will be re-raised. When `False` (the default), such JSON is handled gracefully by returning `{'INVALID_JSON': ''}` so that the value can still be sent to a model API (e.g. during a retry flow) without crashing. """ if not self.args: return {} if isinstance(self.args, dict): return self.args try: args = pydantic_core.from_json(self.args) assert isinstance(args, dict), 'args should be a dict' return cast(dict[str, Any], args) except (ValueError, AssertionError): if raise_if_invalid: raise return {INVALID_JSON_KEY: self.args} def args_as_json_str(self) -> str: """Return the arguments as a JSON string. This is just for convenience with models that require JSON strings as input. JSON that's malformed or doesn't represent an object is handled gracefully by returning `'{"INVALID_JSON":""}'`, matching [`args_as_dict`][pydantic_ai.messages.BaseToolCallPart.args_as_dict], so that the value can still be sent to a model API (e.g. during a retry flow) instead of being rejected by one that requires an object. Because of that, this is not the way to render args that are still streaming in: a partial fragment that only becomes valid JSON once the following deltas are concatenated would be degraded to the wrapper. Emit those verbatim instead, as the UI event streams do. """ if not self.args: return '{}' if isinstance(self.args, str): try: if isinstance(pydantic_core.from_json(self.args), dict): # Returned verbatim rather than re-serialized, as the exact bytes the model produced # (key order, whitespace) matter for prompt caching. return self.args except ValueError: pass return pydantic_core.to_json(self.args_as_dict()).decode() def has_content(self) -> bool: """Return `True` if the tool call has content.""" return self.args not in ('', {}, None) __repr__ = _utils.dataclasses_no_defaults_repr @dataclass(repr=False) class ToolCallPart(BaseToolCallPart): """A tool call from a model.""" _: KW_ONLY part_kind: Literal['tool-call'] = 'tool-call' """Part type identifier, this is available on all parts as a discriminator. Note that this is different from `ToolCallPartDelta.part_delta_kind`.""" @staticmethod def narrow_type(part: ToolCallPart, *, tool_kind: ToolPartKind | None = None) -> ToolCallPart: """Promote a base `ToolCallPart` to its typed subclass when its `tool_kind` is registered. Best-effort: returns the part unchanged when the `tool_kind` (kwarg or on the part) resolves to no registered subclass, and strips an unsubstantiated `tool_kind` when the part's data doesn't validate against that subclass — keeping it on a base part would break a [`ModelMessagesTypeAdapter`][pydantic_ai.messages.ModelMessagesTypeAdapter] round-trip. For direct construction; Pydantic deserialization promotes automatically via the discriminated union. """ return _narrow_call(part, _TOOL_CALL_NARROWERS, tool_kind) @dataclass(repr=False) class NativeToolCallPart(BaseToolCallPart): """A tool call to a native tool. For native tools with a stable cross-provider shape (currently `tool_search`), this base class can be promoted to a typed subclass with a narrowed `args` `TypedDict`. See [`NativeToolSearchCallPart`][pydantic_ai.messages.NativeToolSearchCallPart] for the canonical example. Adding a typed subclass for a future native tool (see `pydantic_ai._tool_search` for a worked example): 1. Add a sibling `pydantic_ai/_.py` module that defines the cross-provider `TypedDict`s, the `NativeToolCallPart` / `NativeToolReturnPart` subclasses, and registers their narrowers into `_NATIVE_CALL_NARROWERS` / `_NATIVE_RETURN_NARROWERS` keyed by `tool_kind`. Subclass overrides `tool_kind: Literal['']` to match the emitting [`AbstractNativeTool.kind`][pydantic_ai.native_tools.AbstractNativeTool.kind], and shadows `args` / `content` with a narrower type. 2. Late-import the new module from this file (alongside the existing tool-search import) so registration runs whenever `pydantic_ai.messages` is imported. 3. Add the subclass to `ModelResponsePart`'s discriminated union and to `_model_response_part_discriminator` so Pydantic deserialization auto-promotes on `model_validate` / `model_validate_json`. Dispatch is by `tool_kind`, not `tool_name`. This protects users whose tools happen to share a name with one of ours from accidentally getting their parts promoted (and failing shape validation against the typed `args`/`content`). The `provider_details` field carries genuinely non-portable provider extras (e.g. Anthropic's `strategy: 'bm25' | 'regex'` for tool search). Promote a field to a typed slot in `args` / `content` only when at least two of OpenAI, Anthropic, and Google support it (cf. [issue #3885](https://github.com/pydantic/pydantic-ai/issues/3885)). MCP server tools land here with `tool_kind='mcp_server'` (label stays in `tool_name='mcp_server: