PydanticAI — Class Deep Dives Vol. 6

import { Aside } from ‘@astrojs/starlight/components’;

Ten class groups from the pydantic_ai 1.104.0 source covering the structured tool-return API, tagged text content with app-only metadata, file/image message parts, the four direct-API functions for raw model calls, the ModelRequestParameters wire-format object, the complete exception hierarchy for model-provider errors, the HITL approval result pair, the JSON-schema transformer base class, Temporal durable-execution integration, and the internal ToolsetTool definition wrapper.

1. `ToolReturn`

Module: pydantic_ai.messages
Import: from pydantic_ai import ToolReturn or from pydantic_ai.messages import ToolReturn

ToolReturn separates three things that often need to travel independently:

Field	Sent to LLM?	Purpose
`return_value`	Yes — as the tool-result message	The primary structured answer (used for return schema)
`content`	Yes — as a `UserPromptPart` after the result	Supplementary multimodal data to surface in the next turn
`metadata`	No	App-only data accessible via `ctx.partial_output`/logging

Class signature

@dataclass(repr=False)
class ToolReturn(Generic[_ToolReturnValueT]):
    return_value: ToolReturnContent          # str | BinaryContent | Sequence[UserContent] | ...
    content: str | Sequence[UserContent] | None = None
    metadata: Any = None
    kind: Literal['tool-return'] = 'tool-return'

ToolReturn is generic: ToolReturn[User] generates a return schema for User, while bare ToolReturn (or ToolReturn[Any]) generates no return schema.

Basic usage — add metadata without changing the model’s view

from dataclasses import dataclass
from pydantic_ai import Agent
from pydantic_ai.messages import ToolReturn
from pydantic_ai.tools import RunContext

@dataclass
class Deps:
    audit_log: list

agent = Agent('openai:gpt-4o', deps_type=Deps)

@agent.tool
async def lookup_customer(ctx: RunContext[Deps], customer_id: str) -> ToolReturn:
    record = {'id': customer_id, 'name': 'Alice', 'tier': 'gold'}
    ctx.deps.audit_log.append({'action': 'lookup', 'id': customer_id})
    return ToolReturn(
        return_value=f"Customer {customer_id}: Alice (gold tier)",
        metadata={'db_latency_ms': 12, 'cache_hit': False},
    )

Typed return — generate a return schema

from pydantic import BaseModel
from pydantic_ai import Agent
from pydantic_ai.messages import ToolReturn
from pydantic_ai.tools import RunContext

class WeatherReport(BaseModel):
    temperature_c: float
    condition: str
    humidity_pct: int

agent = Agent('openai:gpt-4o')

@agent.tool
async def get_weather(ctx: RunContext[None], city: str) -> ToolReturn[WeatherReport]:
    """Fetch current weather. The model receives a validated WeatherReport schema."""
    report = WeatherReport(temperature_c=22.5, condition='sunny', humidity_pct=45)
    return ToolReturn(return_value=report)

`content` field — inject extra multimodal data into the next turn

Use content when you want to push an image, audio clip, or follow-up text into the conversation outside the tool-result message.

from pydantic_ai.messages import ToolReturn
from pydantic_ai import ImageUrl

@agent.tool
async def capture_screenshot(ctx, url: str) -> ToolReturn:
    screenshot_url = f"https://screenshots.example.com/{url}"
    return ToolReturn(
        return_value="Screenshot captured.",
        # This appears as a UserPromptPart in the NEXT model request,
        # not inside the tool result — useful for vision workflows.
        content=[
            "Here is the screenshot for your reference:",
            ImageUrl(url=screenshot_url),
        ],
    )

Accessing metadata in a hook

from pydantic_ai.capabilities import Hooks

hooks = Hooks()

@hooks.after_tool_execute
async def log_tool_metadata(ctx, tool_call_part, result_or_error):
    partial = ctx.partial_output
    if partial and hasattr(partial, 'metadata') and partial.metadata:
        print(f"[{ctx.tool_name}] metadata={partial.metadata}")

agent = Agent('openai:gpt-4o', capabilities=[hooks])

2. `TextContent`

Module: pydantic_ai.messages
Import: from pydantic_ai.messages import TextContent

TextContent is a string that carries app-only metadata alongside the text that the LLM actually sees. The metadata field is never serialised to the model request — it is purely for application-side consumption (logging, UI rendering, post-processing).

Class signature

@dataclass(repr=False)
class TextContent:
    content: str          # sent to the LLM
    metadata: Any = None  # NOT sent to the LLM
    kind: Literal['text-content'] = 'text-content'

TextContent is a valid UserContent item (part of the UserContent union), so it can appear anywhere a plain str would appear in a user prompt.

Use case — annotate retrieved passages for the UI without polluting the context

import asyncio
from pydantic_ai import Agent
from pydantic_ai.messages import TextContent

agent = Agent('openai:gpt-4o', system_prompt='Answer questions using the provided context.')

async def ask_with_sources(question: str, passages: list[dict]) -> str:
    # Annotate each passage with its source URL — the LLM only sees the text
    tagged_passages = [
        TextContent(
            content=p['text'],
            metadata={'source_url': p['url'], 'score': p['score']},
        )
        for p in passages
    ]
    result = await agent.run(
        [*tagged_passages, question],  # TextContent items inline with the question
    )
    return result.output

passages = [
    {'text': 'Pydantic AI was released in 2024.', 'url': 'https://docs.ai/', 'score': 0.97},
    {'text': 'It supports OpenAI, Anthropic, and Gemini.', 'url': 'https://docs.ai/models', 'score': 0.88},
]
answer = asyncio.run(ask_with_sources('When was Pydantic AI released?', passages))
print(answer)

Use case — attach render hints for a streaming UI

from pydantic_ai.messages import TextContent

# A streaming UI can inspect TextContent.metadata to choose how to display each chunk
def build_user_turn(chunks: list[dict]) -> list[TextContent]:
    return [
        TextContent(
            content=chunk['text'],
            metadata={
                'chunk_id': chunk['id'],
                'render_as': 'code' if chunk.get('is_code') else 'prose',
            }
        )
        for chunk in chunks
    ]

3. `FilePart` + `BinaryImage`

Module: pydantic_ai.messages
Imports: from pydantic_ai.messages import FilePart, BinaryImage

FilePart is a model-response part — it appears in ModelResponse.parts when the model returns a file (e.g. a generated image, a document, or a rendered PDF). BinaryImage is a specialised subclass of BinaryContent that enforces an image/* media type at construction.

`FilePart` signature

@dataclass(repr=False)
class FilePart:
    content: BinaryContent         # validated to be a BinaryImage via AfterValidator
    id: str | None = None          # provider-assigned file identifier
    provider_name: str | None = None   # required when id or provider_details is set
    provider_details: dict[str, Any] | None = None  # provider-specific round-trip data
    part_kind: Literal['file'] = 'file'

    def has_content(self) -> bool: ...

provider_details holds data that must be round-tripped back to the API on the next call (e.g. OpenAI’s file IDs for DALL-E outputs). provider_name must be set whenever either id or provider_details is populated.

`BinaryImage` signature

@pydantic_dataclass(config=ConfigDict(ser_json_bytes='base64', val_json_bytes='base64'))
class BinaryImage(BinaryContent):
    # Same fields as BinaryContent: data, media_type, identifier, vendor_metadata
    def __post_init__(self):
        if not self.is_image:
            raise ValueError('`BinaryImage` must have a media type that starts with "image/"')

BinaryImage inherits BinaryContent.is_image (a @property that checks media_type.startswith('image/')). The Pydantic dataclass config serialises data: bytes as base64 for JSON persistence.

Extracting generated images from a model response

from pydantic_ai.messages import FilePart, ModelResponse

def extract_images(response: ModelResponse) -> list[FilePart]:
    return [part for part in response.parts if isinstance(part, FilePart)]

# In a tool or hook:
@hooks.after_model_request
async def save_generated_images(ctx, response: ModelResponse, usage):
    for file_part in extract_images(response):
        if file_part.has_content():
            img_bytes: bytes = file_part.content.data
            media: str = file_part.content.media_type   # e.g. 'image/png'
            # persist or forward img_bytes …

Round-tripping provider file IDs

Some providers (e.g. OpenAI) return a file ID that must be echoed back on the next call. FilePart.provider_details carries this opaque payload:

from pydantic_ai.messages import FilePart

def round_trip_file_parts(
    file_parts: list[FilePart],
) -> list[dict]:
    """Convert FileParts to the wire format expected by the provider on the next turn."""
    result = []
    for fp in file_parts:
        if fp.provider_details and fp.provider_name == 'openai':
            result.append({
                'type': 'image_file',
                'image_file': {'file_id': fp.id, **fp.provider_details},
            })
    return result

Constructing a `BinaryImage` manually

from pydantic_ai.messages import BinaryImage

with open('photo.jpg', 'rb') as f:
    img_data = f.read()

img = BinaryImage(data=img_data, media_type='image/jpeg')
assert img.is_image        # True
assert img.media_type == 'image/jpeg'

# JSON round-trip — bytes are base64-encoded
import json
from pydantic import TypeAdapter
ta = TypeAdapter(BinaryImage)
serialised = ta.dump_json(img)
restored = ta.validate_json(serialised)
assert restored.data == img_data

4. Direct API — `model_request` / `model_request_sync` / `model_request_stream` / `model_request_stream_sync`

Module: pydantic_ai.direct
Imports:

from pydantic_ai.direct import (
    model_request,
    model_request_sync,
    model_request_stream,
    model_request_stream_sync,
)

The Direct API makes raw requests to a model without an Agent. The only abstractions are model-string resolution, optional OpenTelemetry instrumentation, and an instruction_parts fix-up for models that read that field instead of ModelRequest.instructions. There is no tool dispatch, no output validation, no retries, and no capability pipeline.

The four functions at a glance

Function	Async?	Streaming?	Returns
`model_request`	Yes	No	`ModelResponse`
`model_request_sync`	No	No	`ModelResponse`
`model_request_stream`	Yes (context manager)	Yes	`AsyncContextManager[StreamedResponse]`
`model_request_stream_sync`	No (context manager)	Yes	`StreamedResponseSync` (context manager)

`model_request` — single async request

import asyncio
from pydantic_ai import ModelRequest
from pydantic_ai.direct import model_request

async def main():
    response = await model_request(
        'anthropic:claude-haiku-4-5',
        [ModelRequest.user_text_prompt('What is the capital of France?')],
    )
    # response is a ModelResponse
    text = next(p.content for p in response.parts if hasattr(p, 'content'))
    print(text)          # 'The capital of France is Paris.'
    print(response.usage)  # RequestUsage(input_tokens=..., output_tokens=...)

asyncio.run(main())

`model_request_sync` — synchronous wrapper

Identical to model_request but blocks the calling thread. Cannot be used inside an active event loop (raises RuntimeError). Useful for scripts, notebooks, and CLI tools.

from pydantic_ai import ModelRequest
from pydantic_ai.direct import model_request_sync

response = model_request_sync(
    'openai:gpt-4o-mini',
    [ModelRequest.user_text_prompt('Summarise the Zen of Python in one sentence.')],
)
print(response.parts[0].content)

`model_request_stream` — async streaming

import asyncio
from pydantic_ai import ModelRequest
from pydantic_ai.direct import model_request_stream
from pydantic_ai.messages import PartDeltaEvent, TextPartDelta

async def stream_translation():
    msgs = [ModelRequest.user_text_prompt('Translate "Hello, world!" to Spanish.')]
    async with model_request_stream('openai:gpt-4o', msgs) as stream:
        async for event in stream:
            if isinstance(event, PartDeltaEvent) and isinstance(event.delta, TextPartDelta):
                print(event.delta.content_delta, end='', flush=True)
    print()  # newline

asyncio.run(stream_translation())

`model_request_stream_sync` — synchronous streaming via background thread

model_request_stream_sync runs the async stream in a background threading.Thread and exposes a synchronous iterator. It must be used as a context manager:

from pydantic_ai import ModelRequest
from pydantic_ai.direct import model_request_stream_sync
from pydantic_ai.messages import PartDeltaEvent, TextPartDelta

msgs = [ModelRequest.user_text_prompt('Write a haiku about Python.')]
with model_request_stream_sync('openai:gpt-4o', msgs) as stream:
    for event in stream:
        if isinstance(event, PartDeltaEvent) and isinstance(event.delta, TextPartDelta):
            print(event.delta.content_delta, end='', flush=True)
    # After iteration, access metadata:
    print(f"\nModel: {stream.model_name} | Timestamp: {stream.timestamp}")

Passing `ModelRequestParameters`

All four functions accept an optional model_request_parameters argument for advanced control over tools, output mode, and native tools:

from pydantic_ai import ModelRequest
from pydantic_ai.direct import model_request
from pydantic_ai.models import ModelRequestParameters
from pydantic_ai.tools import ToolDefinition

async def call_with_tools():
    tool = ToolDefinition(
        name='get_time',
        description='Returns the current UTC time.',
        parameters_json_schema={'type': 'object', 'properties': {}},
    )
    response = await model_request(
        'openai:gpt-4o',
        [ModelRequest.user_text_prompt('What time is it?')],
        model_request_parameters=ModelRequestParameters(function_tools=[tool]),
    )
    print(response.parts)

With `ModelSettings` and instrumentation

from pydantic_ai import ModelRequest, ModelSettings
from pydantic_ai.direct import model_request

response = await model_request(
    'anthropic:claude-opus-4-8',
    [ModelRequest.user_text_prompt('Explain quantum entanglement.')],
    model_settings=ModelSettings(max_tokens=256, temperature=0.3),
    instrument=True,   # enable OpenTelemetry tracing
)

5. `ModelRequestParameters`

Module: pydantic_ai.models
Import: from pydantic_ai.models import ModelRequestParameters

ModelRequestParameters is the wire-format object that the agent passes to Model.request() / Model.request_stream(). It captures everything a model provider needs about the shape of the response and the tools available.

Class signature

@dataclass(repr=False, kw_only=True)
class ModelRequestParameters:
    function_tools: list[ToolDefinition] = []
    native_tools: list[AbstractNativeTool] = []          # replaces deprecated `builtin_tools`
    output_mode: OutputMode = 'text'                     # 'text' | 'tool' | 'native' | 'prompted' | 'auto'
    output_object: OutputObjectDefinition | None = None
    output_tools: list[ToolDefinition] = []
    prompted_output_template: str | Literal[False] | None = None
    allow_text_output: bool = True
    allow_image_output: bool = False
    instruction_parts: list[InstructionPart] | None = None
    thinking: ThinkingLevel | None = None                # None = model default

    @cached_property
    def tool_defs(self) -> dict[str, ToolDefinition]: ...

    def with_default_output_mode(self, mode: StructuredOutputMode) -> ModelRequestParameters: ...

Field reference

Field	Default	Description
`function_tools`	`[]`	Regular `@agent.tool` tools sent to the model
`native_tools`	`[]`	Built-in / native tools (WebSearch, Thinking, etc.)
`output_mode`	`'text'`	How structured output is requested from the model
`output_object`	`None`	Schema for the structured output object
`output_tools`	`[]`	Special tools used to deliver structured outputs
`allow_text_output`	`True`	Whether free-text responses are accepted
`allow_image_output`	`False`	Whether image responses are accepted
`instruction_parts`	`None`	Structured system-prompt parts (static vs dynamic)
`thinking`	`None`	Thinking effort level (`None` = model default)

`instruction_parts` — static vs dynamic instructions

instruction_parts is a list of InstructionPart objects. Each part carries a dynamic flag:

dynamic=False — literal string passed to Agent(instructions='...'). Anthropic/Bedrock can place a cache boundary after the last static part.
dynamic=True — produced by @agent.instructions functions, TemplateStr, or toolset.get_instructions(). Must not be cached.

from pydantic_ai.models import ModelRequestParameters
from pydantic_ai.messages import InstructionPart

mrp = ModelRequestParameters(
    instruction_parts=[
        InstructionPart(content='You are a helpful coding assistant.', dynamic=False),
        InstructionPart(content=f'Today is {date.today()}.', dynamic=True),
    ]
)

`with_default_output_mode` — resolve `'auto'` mode atomically

When output_mode='auto', the model (not the user) chooses between 'tool', 'native', and 'prompted'. with_default_output_mode resolves this atomically and keeps allow_text_output in sync:

mrp = ModelRequestParameters(output_mode='auto')
resolved = mrp.with_default_output_mode('tool')
# resolved.output_mode == 'tool', resolved.allow_text_output == False

resolved = mrp.with_default_output_mode('native')
# resolved.output_mode == 'native', resolved.allow_text_output == True

`tool_defs` — merged tool look-up dictionary

tool_defs is a @cached_property combining function_tools and output_tools into a single dict keyed by name. It is used internally to dispatch tool calls back from the model:

from pydantic_ai.models import ModelRequestParameters
from pydantic_ai.tools import ToolDefinition

t1 = ToolDefinition(name='search', description='Web search', parameters_json_schema={'type': 'object'})
t2 = ToolDefinition(name='calculator', description='Maths', parameters_json_schema={'type': 'object'})
mrp = ModelRequestParameters(function_tools=[t1, t2])
print(list(mrp.tool_defs.keys()))  # ['search', 'calculator']

6. Error hierarchy — `ModelAPIError` / `ModelHTTPError` / `ContentFilterError` / `IncompleteToolCall`

Module: pydantic_ai.exceptions
Import: from pydantic_ai.exceptions import ModelHTTPError, ContentFilterError, IncompleteToolCall
(all are also importable from pydantic_ai directly)

Hierarchy

AgentRunError (RuntimeError)
├── UsageLimitExceeded
├── ConcurrencyLimitExceeded
├── UnexpectedModelBehavior
│   ├── ContentFilterError       ← content moderation triggered → empty response
│   └── IncompleteToolCall       ← token limit hit mid tool-call
└── ModelAPIError
    └── ModelHTTPError           ← 4xx / 5xx HTTP status codes

`ModelAPIError`

Raised when a provider API request fails for any reason (network timeout, authentication failure, quota exceeded). Adds model_name to the base AgentRunError:

@dataclass
class ModelAPIError(AgentRunError):
    model_name: str    # e.g. 'gpt-4o-mini'
    message: str       # from AgentRunError

`ModelHTTPError`

Raised for 4xx/5xx responses. Adds status_code and body:

@dataclass
class ModelHTTPError(ModelAPIError):
    status_code: int          # e.g. 429
    model_name: str
    body: object | None       # parsed JSON or raw string

`ContentFilterError`

Raised when the model’s content filter is triggered and the response is empty. Subclass of UnexpectedModelBehavior, so it carries an optional body string for diagnostics:

from pydantic_ai.exceptions import ContentFilterError

# ContentFilterError has no extra fields — the parent's body and message are sufficient

`IncompleteToolCall`

Raised when the model stops mid-stream while emitting a tool call (usually due to hitting the token limit). The tool call is therefore malformed and cannot be executed:

from pydantic_ai.exceptions import IncompleteToolCall

Production error-handling pattern

import asyncio
import logging
from pydantic_ai import Agent
from pydantic_ai.exceptions import (
    ModelHTTPError,
    ContentFilterError,
    IncompleteToolCall,
    UsageLimitExceeded,
    UnexpectedModelBehavior,
)

log = logging.getLogger(__name__)
agent = Agent('openai:gpt-4o')

async def safe_run(prompt: str, max_retries: int = 3) -> str | None:
    for attempt in range(max_retries):
        try:
            result = await agent.run(prompt)
            return result.output
        except ContentFilterError as e:
            log.warning('Content filter triggered: %s — returning None', e.message)
            return None
        except IncompleteToolCall as e:
            log.warning('Tool call truncated (attempt %d/%d): %s', attempt + 1, max_retries, e.message)
            if attempt == max_retries - 1:
                raise
        except ModelHTTPError as e:
            if e.status_code == 429:
                # Rate-limited: exponential back-off
                wait = 2 ** attempt
                log.info('Rate limited (status 429), waiting %ds', wait)
                await asyncio.sleep(wait)
            elif e.status_code >= 500:
                log.error('Provider server error %d, retrying', e.status_code)
            else:
                raise  # 4xx other than 429 → re-raise
        except UsageLimitExceeded:
            log.error('Token budget exhausted')
            raise
        except UnexpectedModelBehavior as e:
            log.error('Unexpected model behaviour: %s\nbody: %s', e.message, e.body)
            raise
    return None

result = asyncio.run(safe_run('Tell me a joke.'))

Catching by status code

from pydantic_ai.exceptions import ModelHTTPError

try:
    result = await agent.run(user_input)
except ModelHTTPError as e:
    match e.status_code:
        case 401:
            raise ValueError('Invalid API key') from e
        case 429:
            raise RuntimeError('Rate limit hit — try again later') from e
        case 503:
            raise RuntimeError('Provider temporarily unavailable') from e
        case _:
            raise

7. `ToolApproved` + `ToolDenied`

Module: pydantic_ai.tools
Imports: from pydantic_ai.tools import ToolApproved, ToolDenied
(also: from pydantic_ai import ToolApproved, ToolDenied)

ToolApproved and ToolDenied are the two possible outcomes of a human-in-the-loop (HITL) approval step. They are returned by the approval callback passed to ApprovalRequiredToolset (or raised as ApprovalRequired inside a tool to defer to the caller).

Class signatures

@dataclass(kw_only=True)
class ToolApproved:
    override_args: dict[str, Any] | None = None   # optionally replace the model's args
    kind: Literal['tool-approved'] = 'tool-approved'

@dataclass
class ToolDenied:
    message: str = 'The tool call was denied.'
    kind: Literal['tool-denied'] = 'tool-denied'

How they fit into the HITL workflow

Agent calls tool → tool raises ApprovalRequired(metadata={...})
  ↓
DeferredToolRequests delivered to your application
  ↓
Human reviews: return ToolApproved(override_args={...}) or ToolDenied('Reason')
  ↓
Agent resumes with approved args (or gets the denial message)

Using `ApprovalRequiredToolset` with the approval pair

import asyncio
from pydantic_ai import Agent
from pydantic_ai.tools import ToolApproved, ToolDenied
from pydantic_ai.toolsets import ApprovalRequiredToolset, FunctionToolset

async def send_email(to: str, subject: str, body: str) -> str:
    print(f'Sending email to {to}: {subject}')
    return f'Email sent to {to}'

async def human_approval(tool_name: str, args: dict) -> ToolApproved | ToolDenied:
    """Simulate a human reviewer."""
    print(f'REVIEW REQUEST: {tool_name}({args})')
    if args.get('to', '').endswith('@external.com'):
        return ToolDenied(message='Cannot send emails to external domains.')
    # Optionally sanitise the subject before approving
    sanitised = {**args, 'subject': args['subject'].replace('URGENT', '')}
    return ToolApproved(override_args=sanitised)

toolset = ApprovalRequiredToolset(
    FunctionToolset([send_email]),
    approval_callback=human_approval,
)
agent = Agent('openai:gpt-4o', toolsets=[toolset])

async def main():
    result = await agent.run(
        'Send an URGENT welcome email to alice@internal.com with body "Hello!"'
    )
    print(result.output)

asyncio.run(main())

`ToolApproved.override_args` — modify arguments before execution

override_args lets the reviewer replace the model’s chosen arguments. If None, the original model-generated arguments are used unchanged. If set, the dict must match the tool’s parameter schema exactly (they are re-validated before execution):

# Approve but cap the dollar amount
return ToolApproved(override_args={**original_args, 'amount_usd': min(original_args['amount_usd'], 500)})

`ToolDenied.message` — sent back to the model

The message string is returned to the model as a ToolReturnPart (the tool result). The model can then decide to ask the user for clarification, choose a different tool, or stop:

return ToolDenied(message='Payment above $500 requires manager sign-off. Please request a smaller amount.')

Programmatic approval — batch review

from pydantic_ai import Agent
from pydantic_ai.tools import ToolApproved, ToolDenied
from pydantic_ai import DeferredToolRequests

BLOCKED_TOOLS = {'delete_file', 'drop_table'}

async def policy_approval(tool_name: str, args: dict) -> ToolApproved | ToolDenied:
    if tool_name in BLOCKED_TOOLS:
        return ToolDenied(message=f'Tool {tool_name!r} is not permitted in this environment.')
    if len(str(args)) > 4096:
        return ToolDenied(message='Tool arguments exceed the 4KB policy limit.')
    return ToolApproved()

8. `JsonSchemaTransformer` + `InlineDefsJsonSchemaTransformer`

Module: pydantic_ai._json_schema
Import: from pydantic_ai import JsonSchemaTransformer, InlineDefsJsonSchemaTransformer

JsonSchemaTransformer is the abstract base class for walking and transforming a JSON schema tree. It is called during Model.prepare_request() to convert Pydantic-generated schemas into the format required by each provider.

Abstract interface

@dataclass(init=False)
class JsonSchemaTransformer(ABC):
    schema: JsonSchema
    strict: bool | None             # enforce strict compatibility
    is_strict_compatible: bool      # whether schema passes strict mode
    prefer_inlined_defs: bool       # inline $defs rather than keep $ref
    defs: dict[str, JsonSchema]     # copy of schema.$defs

    @abstractmethod
    def transform(self, schema: JsonSchema) -> JsonSchema:
        """Apply provider-specific transformations to a single schema node."""
        ...

    def walk(self) -> JsonSchema:
        """Entry point — walks the full schema, calling transform() at each node."""
        ...

How `walk()` works

Pops $defs from a deep-copy of the schema.
Recursively handles object, array, and union (anyOf/oneOf) sub-schemas.
Calls transform(schema) on every node after structural handling.
Re-attaches $defs (unless prefer_inlined_defs=True, in which case refs are inlined).

`InlineDefsJsonSchemaTransformer`

The concrete subclass that inlines all $ref definitions. The transform method is a no-op — the inlining happens in walk() via prefer_inlined_defs=True:

class InlineDefsJsonSchemaTransformer(JsonSchemaTransformer):
    def __init__(self, schema: JsonSchema, *, strict: bool | None = None):
        super().__init__(schema, strict=strict, prefer_inlined_defs=True)

    def transform(self, schema: JsonSchema) -> JsonSchema:
        return schema   # no-op — just inline $defs

Writing a custom `JsonSchemaTransformer`

from pydantic_ai import JsonSchemaTransformer
from typing import Any

class RemoveDescriptionsTransformer(JsonSchemaTransformer):
    """Strip 'description' fields — useful when provider charges per-token on schemas."""

    def transform(self, schema: dict[str, Any]) -> dict[str, Any]:
        schema.pop('description', None)
        schema.pop('title', None)
        return schema

# Apply it:
original_schema = {
    'type': 'object',
    'title': 'UserQuery',
    'description': 'A user query',
    'properties': {
        'text': {'type': 'string', 'description': 'The query text'},
    },
}
transformer = RemoveDescriptionsTransformer(original_schema)
clean_schema = transformer.walk()
# {'type': 'object', 'properties': {'text': {'type': 'string'}}}
print(clean_schema)

Enforcing provider-specific rules

from pydantic_ai import JsonSchemaTransformer
from typing import Any

class OpenAIStrictTransformer(JsonSchemaTransformer):
    """Convert nullable unions to OpenAI strict-mode compatible format."""

    def transform(self, schema: dict[str, Any]) -> dict[str, Any]:
        # OpenAI strict mode: no 'default' or 'minimum'/'maximum' on required fields
        schema.pop('default', None)
        schema.pop('minimum', None)
        schema.pop('maximum', None)
        schema.pop('minLength', None)
        schema.pop('maxLength', None)
        return schema

    def walk(self) -> dict[str, Any]:
        result = super().walk()
        # Mark strict after a clean walk
        result['additionalProperties'] = False
        return result

Using a transformer with the Direct API

from pydantic_ai import InlineDefsJsonSchemaTransformer
from pydantic_ai.tools import ToolDefinition

raw_schema = {
    '$defs': {
        'Address': {
            'type': 'object',
            'properties': {
                'street': {'type': 'string'},
                'city': {'type': 'string'},
            }
        }
    },
    'type': 'object',
    'properties': {
        'name': {'type': 'string'},
        'address': {'$ref': '#/$defs/Address'},
    }
}

inlined = InlineDefsJsonSchemaTransformer(raw_schema).walk()
# Address is inlined — no more $defs or $ref
tool = ToolDefinition(
    name='create_user',
    description='Create a new user',
    parameters_json_schema=inlined,
)

9. `TemporalAgent`

Module: pydantic_ai.durable_exec.temporal
Import: from pydantic_ai.durable_exec.temporal import TemporalAgent
Extra: pip install "pydantic-ai[temporal]"

TemporalAgent wraps any Agent (or WrapperAgent) to run inside a Temporal workflow, making every model call and tool execution a durable Temporal activity. The original agent continues to work normally outside Temporal — wrapping is non-destructive.

Constructor parameters

Parameter	Default	Description
`wrapped`	—	The agent to wrap
`name`	`None` → agent.name	Unique prefix for Temporal activity names (required)
`models`	`None`	Mapping of named `Model` instances available at runtime
`provider_factory`	`None`	Callback `(provider_name, ctx) → Model` for runtime model strings
`event_stream_handler`	`None`	Custom event handler (replaces the wrapped agent’s handler)
`activity_config`	60 s timeout	Base `ActivityConfig` for all activities
`model_activity_config`	`{}`	Per-model activity config (merged with base)
`toolset_activity_config`	`{}`	Per-toolset activity config (keyed by toolset `id`)
`tool_activity_config`	`{}`	Per-tool activity config (`{toolset_id: {tool_name: config \| False}}`)
`run_context_type`	`TemporalRunContext`	Subclass for serialising `RunContext` across activity boundaries
`temporalize_toolset_func`	built-in	Custom function to wrap non-standard toolsets for Temporal

UserError and PydanticUserError are automatically marked non-retryable in the base RetryPolicy — these represent programming mistakes that retrying cannot fix.

Minimal example — durable research agent

# pip install "pydantic-ai[temporal]" temporalio
import asyncio
from pydantic_ai import Agent
from pydantic_ai.durable_exec.temporal import TemporalAgent
from pydantic_ai.durable_exec.temporal._workflow import PydanticAIWorkflow
from temporalio import workflow
from temporalio.client import Client
from temporalio.worker import Worker

base_agent = Agent('openai:gpt-4o', name='research-agent')

@base_agent.tool_plain
async def web_search(query: str) -> str:
    """Search the web for information."""
    return f'Results for: {query}'  # replace with real search

# Wrap the agent — base_agent still works normally
temporal_agent = TemporalAgent(base_agent)

@workflow.defn
class ResearchWorkflow(PydanticAIWorkflow):
    __pydantic_ai_agents__ = [temporal_agent]

    @workflow.run
    async def run(self, topic: str) -> str:
        result = await temporal_agent.run(f'Research the topic: {topic}')
        return result.output

async def main():
    client = await Client.connect('localhost:7233')
    async with Worker(
        client,
        task_queue='research-queue',
        workflows=[ResearchWorkflow],
        activities=[
            *temporal_agent.temporal_activities,   # auto-registered activities
        ],
    ):
        result = await client.execute_workflow(
            ResearchWorkflow.run,
            'quantum computing',
            id='research-1',
            task_queue='research-queue',
        )
        print(result)

Per-toolset and per-tool activity config

from datetime import timedelta
from temporalio.workflow import ActivityConfig

temporal_agent = TemporalAgent(
    base_agent,
    activity_config=ActivityConfig(start_to_close_timeout=timedelta(seconds=60)),
    toolset_activity_config={
        'slow-toolset-id': ActivityConfig(start_to_close_timeout=timedelta(minutes=10)),
    },
    tool_activity_config={
        'my-toolset-id': {
            'fast_cache_lookup': False,    # skip activity — function has no I/O
            'send_email': ActivityConfig(start_to_close_timeout=timedelta(seconds=30)),
        }
    },
)

Custom `run_context_type` — expose extra fields to activities

By default, TemporalRunContext only serialises a subset of RunContext fields. To expose custom fields (e.g. metadata) across activity boundaries, subclass it:

from pydantic_ai.durable_exec.temporal import TemporalRunContext

class MyRunContext(TemporalRunContext):
    @classmethod
    def serialize_run_context(cls, ctx) -> dict:
        base = super().serialize_run_context(ctx)
        base['metadata'] = ctx.metadata   # make metadata available in activities
        return base

temporal_agent = TemporalAgent(base_agent, run_context_type=MyRunContext)

Registering multiple models

When the workflow needs to switch between models at runtime, register them by name:

from pydantic_ai.models.openai import OpenAIModel
from pydantic_ai.models.anthropic import AnthropicModel

temporal_agent = TemporalAgent(
    base_agent,
    models={
        'fast': OpenAIModel('gpt-4o-mini'),
        'smart': AnthropicModel('claude-opus-4-8'),
    },
)

# At runtime, inside the workflow:
result = await temporal_agent.run(prompt, model='smart')

10. `ToolsetTool`

Module: pydantic_ai.toolsets.abstract
Import: from pydantic_ai.toolsets.abstract import ToolsetTool

ToolsetTool is the internal envelope that wraps a ToolDefinition once a toolset has been asked for its tools. It is the object the agent works with at execution time — it carries not just the schema but also the retry budget, the args validator, and a reference to the parent toolset for error messages.

Most users never construct ToolsetTool directly. You encounter it in:

Custom AbstractToolset implementations (returned from get_tools)
before_tool_validate / after_tool_execute hooks where you can inspect it
Custom ToolManager subclasses

Class signature

@dataclass(kw_only=True)
class ToolsetTool(Generic[AgentDepsT]):
    toolset: AbstractToolset[AgentDepsT]       # parent toolset (for error messages)
    tool_def: ToolDefinition                    # name, description, parameters schema
    max_retries: int                            # retry budget for this tool
    args_validator: SchemaValidator | SchemaValidatorProt  # Pydantic-core validator
    args_validator_func: Callable[..., Any] | None = None  # post-schema custom validator

Accessing `ToolsetTool` in a hook

from pydantic_ai import Agent
from pydantic_ai.capabilities import Hooks

hooks = Hooks()

@hooks.before_tool_validate
async def log_tool_retry_budget(ctx, tool_call_part):
    # The hook receives ctx.tool (a ToolsetTool) if available via partial_output
    # More commonly: access via the agent's internal state
    print(f'Calling tool: {tool_call_part.tool_name}')

agent = Agent('openai:gpt-4o', capabilities=[hooks])

Implementing `AbstractToolset.get_tools` — returning `ToolsetTool`

When writing a custom toolset from scratch, your get_tools method must return list[ToolsetTool[AgentDepsT]]. Here’s a minimal example:

from pydantic_ai.toolsets.abstract import AbstractToolset, ToolsetTool
from pydantic_ai.tools import ToolDefinition
from pydantic_ai.tools import RunContext
from pydantic_core import SchemaValidator, core_schema
from typing import Any

class MathToolset(AbstractToolset):
    """A minimal toolset exposing a single `add` tool."""

    id = 'math-toolset'

    async def get_tools(self, ctx: RunContext) -> list[ToolsetTool]:
        schema = {
            'type': 'object',
            'properties': {
                'a': {'type': 'number'},
                'b': {'type': 'number'},
            },
            'required': ['a', 'b'],
        }
        validator = SchemaValidator(
            core_schema.typed_dict_schema({
                'a': core_schema.typed_dict_field(core_schema.float_schema()),
                'b': core_schema.typed_dict_field(core_schema.float_schema()),
            })
        )
        tool_def = ToolDefinition(
            name='add',
            description='Add two numbers.',
            parameters_json_schema=schema,
        )
        return [
            ToolsetTool(
                toolset=self,
                tool_def=tool_def,
                max_retries=3,
                args_validator=validator,
            )
        ]

    async def call_tool(
        self,
        tool_name: str,
        tool_args: dict[str, Any],
        ctx: RunContext,
        tool_def: ToolDefinition,
    ) -> Any:
        if tool_name == 'add':
            return tool_args['a'] + tool_args['b']
        raise ValueError(f'Unknown tool: {tool_name}')

`args_validator_func` — custom validation after schema validation

args_validator_func runs after the Pydantic-core schema validator but before execution. It receives the same typed parameters as the tool function plus RunContext, and should raise ModelRetry on failure:

from pydantic_ai.exceptions import ModelRetry
from pydantic_ai.toolsets.abstract import ToolsetTool
from pydantic_ai.tools import RunContext

async def validate_positive(ctx: RunContext, a: float, b: float) -> None:
    if a < 0 or b < 0:
        raise ModelRetry('Both values must be non-negative. Please try again.')

toolset_tool = ToolsetTool(
    toolset=my_toolset,
    tool_def=tool_def,
    max_retries=3,
    args_validator=validator,
    args_validator_func=validate_positive,   # extra validation layer
)

Summary table

Field	Type	When it matters
`toolset`	`AbstractToolset`	Error messages, toolset-level logging
`tool_def`	`ToolDefinition`	Name, description, JSON schema sent to model
`max_retries`	`int`	How many `ModelRetry` raises are tolerated
`args_validator`	`SchemaValidator`	Schema-validates model-supplied args before execution
`args_validator_func`	`Callable \| None`	Optional semantic validation after schema validation

Capstone — Direct API + error handling + custom schema

The following example combines several concepts from this volume: the Direct API for a raw model call, a custom JsonSchemaTransformer to strip descriptions, structured ToolReturn with metadata, and the production error-handling pattern.

import asyncio
from pydantic import BaseModel
from pydantic_ai import ModelRequest, ModelSettings
from pydantic_ai.direct import model_request
from pydantic_ai import JsonSchemaTransformer
from pydantic_ai.models import ModelRequestParameters
from pydantic_ai.tools import ToolDefinition
from pydantic_ai.exceptions import ModelHTTPError, ContentFilterError

class StripDescriptions(JsonSchemaTransformer):
    def transform(self, schema):
        schema.pop('description', None)
        schema.pop('title', None)
        return schema

raw_tool_schema = {
    'type': 'object',
    'title': 'WeatherParams',
    'description': 'Parameters for the weather tool',
    'properties': {
        'city': {'type': 'string', 'description': 'City name'},
    },
    'required': ['city'],
}
lean_schema = StripDescriptions(raw_tool_schema).walk()

weather_tool = ToolDefinition(
    name='get_weather',
    description='Get weather for a city',
    parameters_json_schema=lean_schema,
)

async def call_with_tool(city: str) -> str:
    try:
        response = await model_request(
            'openai:gpt-4o-mini',
            [ModelRequest.user_text_prompt(f"What's the weather in {city}?")],
            model_settings=ModelSettings(max_tokens=256),
            model_request_parameters=ModelRequestParameters(function_tools=[weather_tool]),
        )
        # For a real agent, you'd dispatch tool calls; here we just return the text
        for part in response.parts:
            if hasattr(part, 'content'):
                return part.content
        return 'No text response'
    except ContentFilterError:
        return 'Query blocked by content filter.'
    except ModelHTTPError as e:
        return f'API error {e.status_code}: {e.message}'

print(asyncio.run(call_with_tool('Paris')))

Continue to Class Deep Dives Vol. 7 → — AgentEventStream, ThinkingPart/ThinkingPartDelta, AudioUrl/VideoUrl/DocumentUrl, OutputContext, ModelRetry, RequestUsage, WebSearchTool/WebSearchUserLocation, MemoryTool, CodeExecutionTool, AbstractNativeTool.

PydanticAI — Class Deep Dives Vol. 6

1. ToolReturn

Class signature

Basic usage — add metadata without changing the model’s view

Typed return — generate a return schema

content field — inject extra multimodal data into the next turn

Accessing metadata in a hook

2. TextContent

Class signature

Use case — annotate retrieved passages for the UI without polluting the context

Use case — attach render hints for a streaming UI

3. FilePart + BinaryImage

FilePart signature

BinaryImage signature

Extracting generated images from a model response

Round-tripping provider file IDs

Constructing a BinaryImage manually

4. Direct API — model_request / model_request_sync / model_request_stream / model_request_stream_sync

The four functions at a glance

model_request — single async request

model_request_sync — synchronous wrapper

model_request_stream — async streaming

model_request_stream_sync — synchronous streaming via background thread

Passing ModelRequestParameters

With ModelSettings and instrumentation

5. ModelRequestParameters

Class signature

Field reference

instruction_parts — static vs dynamic instructions

with_default_output_mode — resolve 'auto' mode atomically

tool_defs — merged tool look-up dictionary

6. Error hierarchy — ModelAPIError / ModelHTTPError / ContentFilterError / IncompleteToolCall

Hierarchy

ModelAPIError

ModelHTTPError

ContentFilterError

IncompleteToolCall

Production error-handling pattern

Catching by status code

7. ToolApproved + ToolDenied

Class signatures

How they fit into the HITL workflow

Using ApprovalRequiredToolset with the approval pair

ToolApproved.override_args — modify arguments before execution

ToolDenied.message — sent back to the model

Programmatic approval — batch review

8. JsonSchemaTransformer + InlineDefsJsonSchemaTransformer

Abstract interface

How walk() works

InlineDefsJsonSchemaTransformer

Writing a custom JsonSchemaTransformer

Enforcing provider-specific rules

Using a transformer with the Direct API

9. TemporalAgent

Constructor parameters

Minimal example — durable research agent

Per-toolset and per-tool activity config

Custom run_context_type — expose extra fields to activities

Registering multiple models

10. ToolsetTool

Class signature

Accessing ToolsetTool in a hook

Implementing AbstractToolset.get_tools — returning ToolsetTool

args_validator_func — custom validation after schema validation

Summary table

Capstone — Direct API + error handling + custom schema

1. `ToolReturn`

`content` field — inject extra multimodal data into the next turn

2. `TextContent`

3. `FilePart` + `BinaryImage`

`FilePart` signature

`BinaryImage` signature

Constructing a `BinaryImage` manually

4. Direct API — `model_request` / `model_request_sync` / `model_request_stream` / `model_request_stream_sync`

`model_request` — single async request

`model_request_sync` — synchronous wrapper

`model_request_stream` — async streaming

`model_request_stream_sync` — synchronous streaming via background thread

Passing `ModelRequestParameters`

With `ModelSettings` and instrumentation

5. `ModelRequestParameters`

`instruction_parts` — static vs dynamic instructions

`with_default_output_mode` — resolve `'auto'` mode atomically

`tool_defs` — merged tool look-up dictionary

6. Error hierarchy — `ModelAPIError` / `ModelHTTPError` / `ContentFilterError` / `IncompleteToolCall`

`ModelAPIError`

`ModelHTTPError`

`ContentFilterError`

`IncompleteToolCall`

7. `ToolApproved` + `ToolDenied`

Using `ApprovalRequiredToolset` with the approval pair

`ToolApproved.override_args` — modify arguments before execution

`ToolDenied.message` — sent back to the model

8. `JsonSchemaTransformer` + `InlineDefsJsonSchemaTransformer`

How `walk()` works

`InlineDefsJsonSchemaTransformer`

Writing a custom `JsonSchemaTransformer`

9. `TemporalAgent`

Custom `run_context_type` — expose extra fields to activities

10. `ToolsetTool`

Accessing `ToolsetTool` in a hook

Implementing `AbstractToolset.get_tools` — returning `ToolsetTool`

`args_validator_func` — custom validation after schema validation