AI-written adapter code forwarded unvalidated client-supplied UploadedFile references to providers, enabling crafted clients to access server-side files.

+from pydantic_ai.messages import UploadedFile from pydantic_ai.models import Model, ModelRequestParameters, StreamedResponse, download_item from pydantic_ai.providers import Provider, infer_provider from pydantic_ai.providers.bedrock import BedrockModelProfile, remove_bedrock_geo_prefix@@ -63,10 +64,8 @@         ConverseStreamOutputTypeDef,         ConverseStreamResponseTypeDef,         CountTokensRequestTypeDef,-        DocumentBlockTypeDef,         DocumentSourceTypeDef,         GuardrailConfigurationTypeDef,-        ImageBlockTypeDef,         InferenceConfigurationTypeDef,         MessageUnionTypeDef,         PerformanceConfigurationTypeDef,@@ -81,10 +80,32 @@         ToolSpecificationTypeDef,         ToolTypeDef,         ToolUseBlockOutputTypeDef,-        VideoBlockTypeDef,     )  +_SUPPORTED_IMAGE_FORMATS = ('jpeg', 'png', 'gif', 'webp')+_SUPPORTED_VIDEO_FORMATS = ('mkv', 'mov', 'mp4', 'webm', 'flv', 'mpeg', 'mpg', 'wmv', 'three_gp')

++ LatestBedrockModelNames = Literal[     'amazon.titan-tg1-large',     'amazon.titan-text-lite-v1',@@ -169,6 +190,15 @@ }  

++ def _insert_cache_point_before_trailing_documents(     content: list[Any],     *,@@ -834,8 +864,8 @@ def _get_last_user_message_content(messages: list[MessageUnionTypeDef]) -> list[             return None         return content -    @staticmethod     async def _map_user_prompt(  # noqa: C901+        self,         part: UserPromptPart,         document_count: Iterator[int],         supports_prompt_caching: bool,@@ -849,62 +879,64 @@ async def _map_user_prompt(  # noqa: C901                     content.append({'text': item})                 elif isinstance(item, BinaryContent):                     format = item.format+                    source: DocumentSourceTypeDef = {'bytes': item.data}                     if item.is_document:-                        name = f'Document {next(document_count)}'-                        assert format in ('pdf', 'txt', 'csv', 'doc', 'docx', 'xls', 'xlsx', 'html', 'md')-                        content.append({'document': {'name': name, 'format': format, 'source': {'bytes': item.data}}})+                        content.append(_make_document_block(f'Document {next(document_count)}', format, source))                     elif item.is_image:

+@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]+    - [`XaiModel`][pydantic_ai.models.xai.XaiModel]+    """+

+    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+    """++    _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.

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

@@ -235,16 +237,29 @@ def load_messages(cls, messages: Sequence[UIMessage]) -> list[ModelMessage]:  #                         try:                             file = BinaryContent.from_data_uri(part.url)                         except ValueError:-                            media_type_prefix = part.media_type.split('/', 1)[0]-                            match media_type_prefix:-                                case 'image':-                                    file = ImageUrl(url=part.url, media_type=part.media_type)-                                case 'video':-                                    file = VideoUrl(url=part.url, media_type=part.media_type)-                                case 'audio':-                                    file = AudioUrl(url=part.url, media_type=part.media_type)-                                case _:-                                    file = DocumentUrl(url=part.url, media_type=part.media_type)+                            # Check provider_metadata for UploadedFile data+                            provider_meta = load_provider_metadata(part.provider_metadata)+                            uploaded_file_id = provider_meta.get('file_id')+                            uploaded_file_provider = provider_meta.get('provider_name')

+                                )+                            else:+                                media_type_prefix = part.media_type.split('/', 1)[0]+                                match media_type_prefix:+                                    case 'image':+                                        file = ImageUrl(url=part.url, media_type=part.media_type)+                                    case 'video':+                                        file = VideoUrl(url=part.url, media_type=part.media_type)+                                    case 'audio':+                                        file = AudioUrl(url=part.url, media_type=part.media_type)+                                    case _:+                                        file = DocumentUrl(url=part.url, media_type=part.media_type)                         user_prompt_content.append(file)                     else:  # pragma: no cover                         raise ValueError(f'Unsupported user message part type: {type(part)}')@@ -680,6 +695,17 @@ def _convert_user_prompt_part(part: UserPromptPart) -> list[UIMessagePart]:                 ui_parts.append(FileUIPart(url=item.data_uri, media_type=item.media_type))             elif isinstance(item, ImageUrl | AudioUrl | VideoUrl | DocumentUrl):                 ui_parts.append(FileUIPart(url=item.url, media_type=item.media_type))+            elif isinstance(item, UploadedFile):+                # Store uploaded file info in provider_metadata for round-trip support+                provider_metadata = dump_provider_metadata(+                    file_id=item.file_id,+                    provider_name=item.provider_name,+                    vendor_metadata=item.vendor_metadata,+                    identifier=item._identifier,  # pyright: ignore[reportPrivateUsage]+                )+                ui_parts.append(+                    FileUIPart(url=item.file_id, media_type=item.media_type, provider_metadata=provider_metadata)+                )             elif isinstance(item, CachePoint):                 # CachePoint is metadata for prompt caching, skip for UI conversion                 pass

++    Set to `True` to keep client-submitted uploaded files after auditing your frontend. Some+    adapters (e.g. AG-UI) additionally use this flag to round-trip agent-generated files and+    uploaded files through their protocol-specific message representation; see the adapter for+    details.+    """+     @classmethod     async def from_request(         cls,@@ -218,6 +240,7 @@ async def from_request(         manage_system_prompt: Literal['server', 'client'] = 'server',         allowed_file_url_schemes: frozenset[str] = frozenset({'http', 'https'}),         allowed_file_url_force_download: frozenset[ForceDownloadMode] = frozenset(),+        preserve_file_data: bool = False,         **kwargs: Any,     ) -> Self:         """Create an adapter from a request.@@ -232,6 +255,7 @@ async def from_request(             manage_system_prompt=manage_system_prompt,             allowed_file_url_schemes=allowed_file_url_schemes,             allowed_file_url_force_download=allowed_file_url_force_download,+            preserve_file_data=preserve_file_data,             **kwargs,         ) @@ -323,6 +347,12 @@ def sanitize_messages(

         - [`ToolCallPart`][pydantic_ai.messages.ToolCallPart] and           [`NativeToolCallPart`][pydantic_ai.messages.NativeToolCallPart] entries at           the end of the history that don't have a matching entry in@@ -342,6 +372,7 @@ def sanitize_messages(         stripped_system_prompt = False         disallowed_url_schemes: set[str] = set()         reset_force_download_values: set[ForceDownloadMode] = set()+        dropped_uploaded_file_providers: set[str] = set()         dangling_tool_call_names: list[str] = []         last_index = len(messages) - 1 @@ -353,6 +384,7 @@ def sanitize_messages(                     strip_system_prompt=strip_system_prompt,                     disallowed_schemes=disallowed_url_schemes,                     reset_force_download_values=reset_force_download_values,+                    dropped_uploaded_file_providers=dropped_uploaded_file_providers,                 )                 stripped_system_prompt = stripped_system_prompt or request_stripped_system_prompt                 if new_request_parts:@@ -366,6 +398,7 @@ def sanitize_messages(                     dangling_names=dangling_tool_call_names if index == last_index else None,                     disallowed_schemes=disallowed_url_schemes,                     reset_force_download_values=reset_force_download_values,+                    dropped_uploaded_file_providers=dropped_uploaded_file_providers,                 )                 if new_response_parts:                     sanitized.append(replace(message, parts=new_response_parts))@@ -408,6 +441,18 @@ def sanitize_messages(

             allowed_file_url_force_download=allowed_file_url_force_download,+            preserve_file_data=preserve_file_data,             **kwargs,         ) @@ -191,6 +193,7 @@ async def dispatch_request(         manage_system_prompt: Literal['server', 'client'] = 'server',         allowed_file_url_schemes: frozenset[str] = frozenset({'http', 'https'}),         allowed_file_url_force_download: frozenset[ForceDownloadMode] = frozenset(),

             allowed_file_url_force_download=allowed_file_url_force_download,+            preserve_file_data=preserve_file_data,             **kwargs,         ) 

+    (the default), they are dropped. If your AG-UI frontend uses activities, be aware that+    `pydantic_ai_*` activity types are reserved for internal round-trip use and should be+    ignored by frontend activity handlers.+    """

-    preserve_file_data: bool = False-    """Whether to preserve agent-generated files and uploaded files in AG-UI message conversion.--    When `True`, agent-generated files and uploaded files are stored as-    [activity messages](https://docs.ag-ui.com/concepts/activities) during `dump_messages`-    and restored during `load_messages`, enabling full round-trip fidelity.-    When `False` (default), they are silently dropped.--    If your AG-UI frontend uses activities, be aware that `pydantic_ai_*` activity types-    are reserved for internal round-trip use and should be ignored by frontend activity handlers.-    """-     @classmethod     def build_run_input(cls, body: bytes) -> RunAgentInput:         """Build an AG-UI run input object from the request body."""