Orchestration and Integrations
Read this file when the user wants multi-agent coordination, graphs, direct model calls, A2A, durable execution, embeddings, evals, or third-party integrations.
Coordinate Multiple Agents
Use agent delegation when one agent should call another and return the result.
from pydantic_ai import Agent, RunContext
parent = Agent('openai:gpt-5.2')
researcher = Agent('openai:gpt-5.2', output_type=str)
@parent.tool
async def research(ctx: RunContext[None], topic: str) -> str:
result = await researcher.run(f'Research: {topic}', usage=ctx.usage)
return result.outputGood 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.
from dataclasses import dataclass
from pydantic_graph import BaseNode, End, Graph, GraphRunContext
@dataclass
class FirstNode(BaseNode[None, None, int]):
value: int
async def run(self, ctx: GraphRunContext) -> 'SecondNode | End[int]':
if self.value >= 5:
return End(self.value)
return SecondNode(self.value + 1)
@dataclass
class SecondNode(BaseNode):
value: int
async def run(self, ctx: GraphRunContext) -> FirstNode:
return FirstNode(self.value)
graph = Graph(nodes=[FirstNode, SecondNode])
result = graph.run_sync(FirstNode(0))Call the Model Without Using an Agent
Use the direct API when the user wants a single model request without agent orchestration.
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.
Expose Agents as HTTP Servers (A2A)
Use agent.to_a2a() when the agent should be exposed as an ASGI app that speaks the A2A protocol.
from pydantic_ai import Agent
agent = Agent('openai:gpt-5.2')
app = agent.to_a2a()Use Durable Execution
Use the durable execution integrations when the run must survive crashes, retries, or long-lived workflows.
Temporal entry points:
TemporalAgentPydanticAIWorkflowPydanticAIPlugin
There are parallel integrations for DBOS and Prefect.
Use Embeddings for RAG
Use Embedder(...) for query/document embeddings when the user is building retrieval or semantic search.
from pydantic_ai import Embedder
embedder = Embedder('openai:text-embedding-3-small')Use LangChain or ACI.dev Tools
Third-party integrations to reach for:
tool_from_langchainLangChainToolsettool_from_aciACIToolset
Use these when the user explicitly wants those ecosystems 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:
CaseDataset- evaluator classes from
pydantic_evals.evaluators
Build Custom Toolsets, Models, or Agents
Extensibility entry points:
AbstractToolset/WrapperToolsetModel/WrapperModelAbstractAgent/WrapperAgentAbstractCapability
Reach for these only when the built-in primitives are genuinely insufficient.