Support FileUrl.force_download in AnthropicModel and OpenAIResponsesModel (#3694)

Co-authored-by: Claude Sonnet 4.5 <noreply@anthropic.com>

Author: dsfaccini

docs/input.md

@@ -104,39 +104,34 @@ print(result.output)
 
 ## User-side download vs. direct file URL
 
-When you provide a URL using any of `ImageUrl`, `AudioUrl`, `VideoUrl` or `DocumentUrl`, Pydantic AI will typically send the URL directly to the model API so that the download happens on their side.
+When using one of `ImageUrl`, `AudioUrl`, `VideoUrl` or `DocumentUrl`, Pydantic AI will default to sending the URL to the model provider, so the file is downloaded on their side.
 
-Some model APIs do not support file URLs at all or for specific file types. In the following cases, Pydantic AI will download the file content and send it as part of the API request instead:
+Support for file URLs varies depending on type and provider:
 
-- [`OpenAIChatModel`][pydantic_ai.models.openai.OpenAIChatModel]: `AudioUrl` and `DocumentUrl`
-- [`OpenAIResponsesModel`][pydantic_ai.models.openai.OpenAIResponsesModel]: All URLs
-- [`AnthropicModel`][pydantic_ai.models.anthropic.AnthropicModel]: `DocumentUrl` with media type `text/plain`
-- [`GoogleModel`][pydantic_ai.models.google.GoogleModel] using GLA (Gemini Developer API): All URLs except YouTube video URLs and files uploaded to the [Files API](https://ai.google.dev/gemini-api/docs/files).
-- [`BedrockConverseModel`][pydantic_ai.models.bedrock.BedrockConverseModel]: All URLs except S3 URLs, specifically starting with `s3://`.
+| Model | Send URL directly | Download and send bytes | Unsupported |
+|-------|-------------------|-------------------------|-------------|
+| [`OpenAIChatModel`][pydantic_ai.models.openai.OpenAIChatModel] | `ImageUrl` | `AudioUrl`, `DocumentUrl` | `VideoUrl` |
+| [`OpenAIResponsesModel`][pydantic_ai.models.openai.OpenAIResponsesModel] | `ImageUrl`, `AudioUrl`, `DocumentUrl` | — | `VideoUrl` |
+| [`AnthropicModel`][pydantic_ai.models.anthropic.AnthropicModel] | `ImageUrl`, `DocumentUrl` (PDF) | `DocumentUrl` (`text/plain`) | `AudioUrl`, `VideoUrl` |
+| [`GoogleModel`][pydantic_ai.models.google.GoogleModel] (Vertex) | All URL types | — | — |
+| [`GoogleModel`][pydantic_ai.models.google.GoogleModel] (GLA) | [YouTube](models/google.md#document-image-audio-and-video-input), [Files API](models/google.md#document-image-audio-and-video-input) | All other URLs | — |
+| [`MistralModel`][pydantic_ai.models.mistral.MistralModel] | `ImageUrl`, `DocumentUrl` (PDF) | — | `AudioUrl`, `VideoUrl`, `DocumentUrl` (non-PDF) |
+| [`BedrockConverseModel`][pydantic_ai.models.bedrock.BedrockConverseModel] | S3 URLs (`s3://`) | `ImageUrl`, `DocumentUrl`, `VideoUrl` | `AudioUrl` |
 
-If the model API supports file URLs but may not be able to download a file because of crawling or access restrictions, you can instruct Pydantic AI to download the file content and send that instead of the URL by enabling the `force_download` flag on the URL object. For example, [`GoogleModel`][pydantic_ai.models.google.GoogleModel] on Vertex AI limits YouTube video URLs to one URL per request.
+A model API may be unable to download a file (e.g., because of crawling or access restrictions) even if it supports file URLs. For example, [`GoogleModel`][pydantic_ai.models.google.GoogleModel] on Vertex AI limits YouTube video URLs to one URL per request. In such cases, you can instruct Pydantic AI to download the file content locally and send that instead of the URL by setting `force_download` on the URL object:
 
-## Uploaded Files
-
-Some model providers like Google's Gemini API support [uploading files](https://ai.google.dev/gemini-api/docs/files). You can upload a file to the model API using the client you can get from the provider and use the resulting URL as input:
+```py {title="force_download.py" test="skip" lint="skip"}
+from pydantic_ai import ImageUrl, AudioUrl, VideoUrl, DocumentUrl
 
-```py {title="file_upload.py" test="skip"}
-from pydantic_ai import Agent, DocumentUrl
-from pydantic_ai.models.google import GoogleModel
-from pydantic_ai.providers.google import GoogleProvider
+ImageUrl(url='https://example.com/image.png', force_download=True)
+AudioUrl(url='https://example.com/audio.mp3', force_download=True)
+VideoUrl(url='https://example.com/video.mp4', force_download=True)
+DocumentUrl(url='https://example.com/doc.pdf', force_download=True)
+```
 
-provider = GoogleProvider()
-file = provider.client.files.upload(file='pydantic-ai-logo.png')
-assert file.uri is not None
+## Uploaded Files
 
-agent = Agent(GoogleModel('gemini-2.5-flash', provider=provider))
-result = agent.run_sync(
- [
- 'What company is this logo from?',
- DocumentUrl(url=file.uri, media_type=file.mime_type),
- ]
-)
-print(result.output)
-```
+Some model providers support passing URLs to files hosted on their platform:
 
-`BedrockConverseModel` supports `s3://<bucket-name>/<object-key>` URIs, provided that the assumed role has the `s3:GetObject` permission. An optional `bucketOwner` query parameter must be specified if the bucket is not owned by the account making the request. For example: `s3://my-bucket/my-file.png?bucketOwner=123456789012`.
+- [`GoogleModel`][pydantic_ai.models.google.GoogleModel] supports the [Files API](models/google.md#document-image-audio-and-video-input) for uploading and referencing files.
+- [`BedrockConverseModel`][pydantic_ai.models.bedrock.BedrockConverseModel] supports `s3://<bucket-name>/<object-key>` URIs, provided that the assumed role has the `s3:GetObject` permission. An optional `bucketOwner` query parameter must be specified if the bucket is not owned by the account making the request. For example: `s3://my-bucket/my-file.png?bucketOwner=123456789012`.

docs/models/google.md

@@ -199,7 +199,46 @@ agent = Agent(model)
 
 ## Document, Image, Audio, and Video Input
 
-`GoogleModel` supports multi-modal input, including documents, images, audio, and video. See the [input documentation](../input.md) for details and examples.
+`GoogleModel` supports multi-modal input, including documents, images, audio, and video.
+
+YouTube video URLs can be passed directly to Google models:
+
+```py {title="youtube_input.py" test="skip" lint="skip"}
+from pydantic_ai import Agent, VideoUrl
+from pydantic_ai.models.google import GoogleModel
+
+agent = Agent(GoogleModel('gemini-2.5-flash'))
+result = agent.run_sync(
+ [
+ 'What is this video about?',
+ VideoUrl(url='https://www.youtube.com/watch?v=dQw4w9WgXcQ'),
+ ]
+)
+print(result.output)
+```
+
+Files can be uploaded via the [Files API](https://ai.google.dev/gemini-api/docs/files) and passed as URLs:
+
+```py {title="file_upload.py" test="skip"}
+from pydantic_ai import Agent, DocumentUrl
+from pydantic_ai.models.google import GoogleModel
+from pydantic_ai.providers.google import GoogleProvider
+
+provider = GoogleProvider()
+file = provider.client.files.upload(file='pydantic-ai-logo.png')
+assert file.uri is not None
+
+agent = Agent(GoogleModel('gemini-2.5-flash', provider=provider))
+result = agent.run_sync(
+ [
+ 'What company is this logo from?',
+ DocumentUrl(url=file.uri, media_type=file.mime_type),
+ ]
+)
+print(result.output)
+```
+
+See the [input documentation](../input.md) for more details and examples.
 
 ## Model settings
 

pydantic_ai_slim/pydantic_ai/_mcp.py

@@ -91,7 +91,7 @@ def add_msg(
  'user',
  mcp_types.ImageContent(
  type='image',
- data=base64.b64encode(chunk.data).decode(),
+ data=chunk.base64,
  mimeType=chunk.media_type,
  ),
  )

pydantic_ai_slim/pydantic_ai/messages.py

@@ -474,7 +474,10 @@ class BinaryContent:
  """Binary content, e.g. an audio or image file."""
 
  data: bytes
- """The binary data."""
+ """The binary file data.
+
+ Use `.base64` to get the base64-encoded string.
+ """
 
  _: KW_ONLY
 
@@ -574,7 +577,12 @@ def identifier(self) -> str:
  @property
  def data_uri(self) -> str:
  """Convert the `BinaryContent` to a data URI."""
- return f'data:{self.media_type};base64,{base64.b64encode(self.data).decode()}'
+ 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:
@@ -776,7 +784,7 @@ def otel_message_parts(self, settings: InstrumentationSettings) -> list[_otel_me
  elif isinstance(part, BinaryContent):
  converted_part = _otel_messages.BinaryDataPart(type='binary', media_type=part.media_type)
  if settings.include_content and settings.include_binary_content:
- converted_part['content'] = base64.b64encode(part.data).decode()
+ converted_part['content'] = part.base64
  parts.append(converted_part)
  elif isinstance(part, CachePoint):
  # CachePoint is a marker, not actual content - skip it for otel
@@ -1396,7 +1404,7 @@ def new_event_body():
  'kind': 'binary',
  'media_type': part.content.media_type,
  **(
- {'binary_content': base64.b64encode(part.content.data).decode()}
+ {'binary_content': part.content.base64}
  if settings.include_content and settings.include_binary_content
  else {}
  ),
@@ -1430,7 +1438,7 @@ def otel_message_parts(self, settings: InstrumentationSettings) -> list[_otel_me
  elif isinstance(part, FilePart):
  converted_part = _otel_messages.BinaryDataPart(type='binary', media_type=part.content.media_type)
  if settings.include_content and settings.include_binary_content:
- converted_part['content'] = base64.b64encode(part.content.data).decode()
+ converted_part['content'] = part.content.base64
  parts.append(converted_part)
  elif isinstance(part, BaseToolCallPart):
  call_part = _otel_messages.ToolCallPart(type='tool_call', id=part.tool_call_id, name=part.tool_name)

pydantic_ai_slim/pydantic_ai/models/__init__.py

@@ -9,7 +9,7 @@
 import base64
 import warnings
 from abc import ABC, abstractmethod
-from collections.abc import AsyncIterator, Callable, Iterator
+from collections.abc import AsyncIterator, Callable, Iterator, Sequence
 from contextlib import asynccontextmanager, contextmanager
 from dataclasses import dataclass, field, replace
 from datetime import datetime
@@ -797,7 +797,7 @@ def base_url(self) -> str | None:
 
  @staticmethod
  def _get_instructions(
- messages: list[ModelMessage], model_request_parameters: ModelRequestParameters | None = None
+ messages: Sequence[ModelMessage], model_request_parameters: ModelRequestParameters | None = None
  ) -> str | None:
  """Get instructions from the first ModelRequest found when iterating messages in reverse.
 

pydantic_ai_slim/pydantic_ai/models/anthropic.py

@@ -71,7 +71,6 @@
  omit as OMIT,
  )
  from anthropic.types.beta import (
- BetaBase64PDFBlockParam,
  BetaBase64PDFSourceParam,
  BetaCacheControlEphemeralParam,
  BetaCitationsConfigParam,
@@ -105,6 +104,7 @@
  BetaRawMessageStreamEvent,
  BetaRedactedThinkingBlock,
  BetaRedactedThinkingBlockParam,
+ BetaRequestDocumentBlockParam,
  BetaRequestMCPServerToolConfigurationParam,
  BetaRequestMCPServerURLDefinitionParam,
  BetaServerToolUseBlock,
@@ -1047,6 +1047,31 @@ def _add_cache_control_to_last_param(
  # Add cache_control to the last param
  last_param['cache_control'] = self._build_cache_control(ttl)
 
+ @staticmethod
+ def _map_binary_data(data: bytes, media_type: str) -> BetaContentBlockParam:
+ # Anthropic SDK accepts file-like objects (IO[bytes]) and handles base64 encoding internally
+ if media_type.startswith('image/'):
+ return BetaImageBlockParam(
+ source={'data': io.BytesIO(data), 'media_type': media_type, 'type': 'base64'}, # type: ignore
+ type='image',
+ )
+ elif media_type == 'application/pdf':
+ return BetaRequestDocumentBlockParam(
+ source=BetaBase64PDFSourceParam(
+ data=io.BytesIO(data),
+ media_type='application/pdf',
+ type='base64',
+ ),
+ type='document',
+ )
+ elif media_type == 'text/plain':
+ return BetaRequestDocumentBlockParam(
+ source=BetaPlainTextSourceParam(data=data.decode('utf-8'), media_type=media_type, type='text'),
+ type='document',
+ )
+ else:
+ raise RuntimeError(f'Unsupported binary content media type for Anthropic: {media_type}')
+
  @staticmethod
  async def _map_user_prompt(
  part: UserPromptPart,
@@ -1062,30 +1087,25 @@ async def _map_user_prompt(
  elif isinstance(item, CachePoint):
  yield item
  elif isinstance(item, BinaryContent):
- if item.is_image:
- yield BetaImageBlockParam(
- source={'data': io.BytesIO(item.data), 'media_type': item.media_type, 'type': 'base64'}, # type: ignore
- type='image',
- )
- elif item.media_type == 'application/pdf':
- yield BetaBase64PDFBlockParam(
- source=BetaBase64PDFSourceParam(
- data=io.BytesIO(item.data),
- media_type='application/pdf',
- type='base64',
- ),
- type='document',
- )
- else:
- raise RuntimeError('Only images and PDFs are supported for binary content')
+ yield AnthropicModel._map_binary_data(item.data, item.media_type)
  elif isinstance(item, ImageUrl):
- yield BetaImageBlockParam(source={'type': 'url', 'url': item.url}, type='image')
+ if item.force_download:
+ downloaded = await download_item(item, data_format='bytes')
+ yield AnthropicModel._map_binary_data(downloaded['data'], item.media_type)
+ else:
+ yield BetaImageBlockParam(source={'type': 'url', 'url': item.url}, type='image')
  elif isinstance(item, DocumentUrl):
  if item.media_type == 'application/pdf':
- yield BetaBase64PDFBlockParam(source={'url': item.url, 'type': 'url'}, type='document')
+ if item.force_download:
+ downloaded = await download_item(item, data_format='bytes')
+ yield AnthropicModel._map_binary_data(downloaded['data'], item.media_type)
+ else:
+ yield BetaRequestDocumentBlockParam(
+ source={'url': item.url, 'type': 'url'}, type='document'
+ )
  elif item.media_type == 'text/plain':
  downloaded_item = await download_item(item, data_format='text')
- yield BetaBase64PDFBlockParam(
+ yield BetaRequestDocumentBlockParam(
  source=BetaPlainTextSourceParam(
  data=downloaded_item['data'], media_type=item.media_type, type='text'
  ),

pydantic_ai_slim/pydantic_ai/models/bedrock.py

@@ -2,7 +2,7 @@
 
 import functools
 import typing
-from collections.abc import AsyncIterator, Iterable, Iterator, Mapping
+from collections.abc import AsyncIterator, Iterable, Iterator, Mapping, Sequence
 from contextlib import asynccontextmanager
 from dataclasses import dataclass, field
 from datetime import datetime
@@ -545,7 +545,7 @@ def _map_tool_config(
 
  async def _map_messages( # noqa: C901
  self,
- messages: list[ModelMessage],
+ messages: Sequence[ModelMessage],
  model_request_parameters: ModelRequestParameters,
  model_settings: BedrockModelSettings | None,
  ) -> tuple[list[SystemContentBlockTypeDef], list[MessageUnionTypeDef]]:

pydantic_ai_slim/pydantic_ai/models/gemini.py

@@ -1,6 +1,5 @@
 from __future__ import annotations as _annotations
 
-import base64
 from collections.abc import AsyncIterator, Sequence
 from contextlib import asynccontextmanager
 from dataclasses import dataclass, field
@@ -377,9 +376,8 @@ async def _map_user_prompt(self, part: UserPromptPart) -> list[_GeminiPartUnion]
  if isinstance(item, str):
  content.append({'text': item})
  elif isinstance(item, BinaryContent):
- base64_encoded = base64.b64encode(item.data).decode('utf-8')
  content.append(
- _GeminiInlineDataPart(inline_data={'data': base64_encoded, 'mime_type': item.media_type})
+ _GeminiInlineDataPart(inline_data={'data': item.base64, 'mime_type': item.media_type})
  )
  elif isinstance(item, VideoUrl) and item.is_youtube:
  file_data = _GeminiFileDataPart(file_data={'file_uri': item.url, 'mime_type': item.media_type})