# Orchestration and Integrations

Read this file when the user wants multi-agent coordination, graphs, direct model calls, A2A, durable execution, embeddings, image generation, evals, or third-party integrations.

## Coordinate Multiple Agents

Use agent delegation when one agent should call another and return the result.

```python
from pydantic_ai import Agent, RunContext

parent = Agent('openai:gpt-5.2', name='parent_agent')
researcher = Agent('openai:gpt-5.2', name='researcher_agent', output_type=str)


@parent.tool
async def research(ctx: RunContext, topic: str) -> str:
    result = await researcher.run(f'Research: {topic}', usage=ctx.usage)
    return result.output
```

Delegating tools and output functions must be `async def` and use `await delegate.run(...)`; never call
`run_sync()` or `run_stream_sync()` inside them. The parent may still use `run_sync()` at the application boundary.

Good split:

- delegation via tools when the parent keeps control
- output functions or programmatic hand-off when control should move elsewhere

## Build Multi-Step Workflows with Graphs

Use `pydantic_graph` when the workflow is a state machine rather than a single agent loop. Compose graphs with `GraphBuilder` and typed step functions:

```python
from pydantic_graph import GraphBuilder, StepContext

g = GraphBuilder(input_type=int, output_type=int)


@g.step
async def increment(ctx: StepContext[None, None, int]) -> int:
    return ctx.inputs + 1


@g.step
async def double(ctx: StepContext[None, None, int]) -> int:
    return ctx.inputs * 2


g.add(
    g.edge_from(g.start_node).to(increment),
    g.edge_from(increment).to(double),
    g.edge_from(double).to(g.end_node),
)

graph = g.build()
result = graph.run_sync(inputs=3)
```

Use `await graph.run(inputs=...)` from async code.

## Call the Model Without Using an Agent

Use the direct API when the user wants a single model request without agent orchestration.

```python
from pydantic_ai import ModelRequest
from pydantic_ai.direct import model_request_sync

response = model_request_sync(
    'openai:gpt-5.2',
    [ModelRequest.user_text_prompt('Summarize this in one sentence.')],
)
```

Reach for this when there is no need for tools, retries, or agent loop state.

## Use Durable Execution

Use the durable execution integrations when the run must survive crashes, retries, or long-lived workflows.

Temporal entry points:

- `Agent(..., capabilities=[TemporalDurability(...)])`
- `PydanticAIWorkflow`
- `PydanticAIPlugin`
- `AgentPlugin`

`TemporalAgent`, `DBOSAgent`, and `PrefectAgent` are deprecated wrapper agents.

Pass every executing toolset that needs durable wrapping to the agent constructor. In particular, construct a `DynamicToolset` with an explicit `id` and pass it to `Agent(toolsets=[...])`; the `@agent.toolset` decorator registers after the engine's durable units were created. Toolsets that arrive later — via the decorator, `run(toolsets=...)`, `override(toolsets=...)`, or a per-run capability — are never wrapped for durable execution. Inside a workflow or flow, Temporal and Prefect reject runtime `MCPToolset` and `DynamicToolset` leaves, plus `FunctionToolset` leaves unless every async tool opts out of durable wrapping with `metadata={'temporal': False}` or `metadata={'prefect': False}` respectively; DBOS accepts a `FunctionToolset`, whose tools it runs inline either way, but rejects `MCPToolset` and `DynamicToolset`. A custom executing `AbstractToolset` leaf is not recognized by this guard and runs unwrapped, so do not add one at run time. The deprecated wrapper agents don't run this check — inside a workflow or flow they run the toolset list frozen at wrap time, so a toolset registered that late is silently left out. A toolset added at run time also cannot reuse a construction-time toolset's `id`.

Temporal and DBOS register durable units before their workers start, so attach capabilities at agent construction time. Passing `run(capabilities=[...])` inside one of their workflows raises a `UserError` unless the capability is the observer-only `Instrumentation`; broader support for observer-only capabilities is tracked in [#5477](https://github.com/pydantic/pydantic-ai/issues/5477), where users can share their use cases. Prefect creates tasks per call, so it has no such registration boundary and accepts a per-run capability that contributes no executing toolset; one that does is still rejected by the runtime-toolset guard.

A run-time `model=` inside a workflow must be a model-name string or an instance registered in the durability capability's `models=`. An unregistered `Model` instance raises a `UserError`: it can't be serialized into the activity/step/task, and rebuilding it from its `model_id` would build a different model. To build a specific instance inside the durable unit (e.g. per-user credentials from `deps`), pass a string and use a `ResolveModelId` capability.

## Handle MCP Tool Errors

Set `MCPToolset(tool_error_behavior=...)` according to the server error semantics:

- `'retry'` asks the model to correct and retry the call. This is the default.
- `'failed'` reports a completed failed result via `ToolFailed` without consuming retry budget.
- `'error'` propagates the underlying exception to application code.

Structured server error content is serialized as JSON for both `'retry'` and `'failed'`. Protocol and transport errors (as opposed to completed tool errors) stay retryable even under `'failed'`.

## Use Embeddings for RAG

Use `Embedder(...)` for query/document embeddings when the user is building retrieval or semantic search.

```python
from pydantic_ai import Embedder

embedder = Embedder('openai:text-embedding-3-small')
```

## Generate Images

Use `ImageGenerator(...)` when the application, rather than an agent, decides that an image should be created or edited.

```python
from pydantic_ai import ImageGenerator

generator = ImageGenerator('openai:gpt-image-2')
result = generator.generate_sync('A watercolor map of a floating city.')
image_bytes = result.image.data
```

Use `await generator.generate(...)` from async code. `result.image` is the first generated image as a `BinaryImage`; use `result.images` when you asked for several or need per-image metadata such as `revised_prompt`.

`ImageGenerationSettings` carries only the portable settings — `dimensions`, `aspect_ratio`, `extra_headers`, `extra_body`. Everything else, including image count, quality, output format, and background, is provider-prefixed on `OpenAIImageGenerationSettings`, `GoogleImageGenerationSettings`, or `XaiImageGenerationSettings`.

Pass reference images through `images=[...]` to edit or transform them. A provider content block raises `ContentFilterError` instead of returning an empty result, so a rejected prompt can be retried explicitly.

When the agent rather than the application should decide, use the `ImageGeneration` capability with `fallback_image_model='openai:gpt-image-2'` (or an `ImageGenerationModel`; an `ImageGenerator` goes on `local`), which calls the direct image model as a tool when the conversational model has no native image generation.

## Use LangChain Tools

Third-party integrations to reach for:

- `tool_from_langchain`
- `LangChainToolset`

Use these when the user explicitly wants the LangChain ecosystem instead of native Pydantic AI tools.

## Systematically Verify Agent Behavior with Evals

Use `pydantic_evals` when the user wants repeatable evaluation datasets and evaluators rather than ad hoc tests.

Common entry points:

- `Case`
- `Dataset`
- evaluator classes from `pydantic_evals.evaluators`

## Build Custom Toolsets, Models, or Agents

Extensibility entry points:

- `AbstractToolset` / `WrapperToolset`
- `Model` / `WrapperModel`
- `AbstractAgent` / `WrapperAgent`
- `AbstractCapability`

Reach for these only when the built-in primitives are genuinely insufficient.