Pydantic AI
View as MarkdownPydantic AI defines a typed agent with tools, dependencies, and output types. The Dex adapter runs a Pydantic AI agent inside a durable session. It is available for Python.
When to use it
Use it when your agent already uses Pydantic AI. Keep the agent, its tools, and its capabilities, and let Dex own the session. The adapter replaces Pydantic AI's own durable execution integrations.
How a session runs
Each RunModel Step runs the agent with a capability that marks every function tool as an external tool. Pydantic AI never runs an external tool, so the run ends right after the model asks for tools.
Each tool call runs in its own Step, in parallel: CallTool for a read or an idempotent write, CallSingleAttemptTool for any other write. A call that needs approval waits in AwaitToolApproval first. The tool Step runs the tool once, through the agent, so the agent's tool hooks run there. The next RunModel resumes the run with the recorded results.
A turn that looks up an order runs RunModel, CallTool, ResumeTurn, and RunModel.
Install
Install the adapter with its framework. Each adapter brings in the Dex Agents core.
pip install dex-agents-pydantic-ai "pydantic-ai-slim[openai]"
Example: agents/python/pydantic-ai/README.md
The example
The example is the order support agent that every framework page uses.
The Flow type is PydanticAIOrderSupport. The full example is under examples/python/agent-harnesses.
Agent and tools
The tools are plain Pydantic AI tools on the agent. The tool policies name each tool. The agent defers its model check, so a Client process needs no provider key.
def build_pydantic_ai_order_support_flow(
order_desk: OrderDesk, model: str | Model, slack_connection: slack.Connection | None
) -> AgentSessionFlow:
"""Builds the session Flow; ``model`` is a Pydantic AI model name, such as "openai:gpt-5.2"."""
# The model is resolved at the first run, so a Client process needs no provider key.
agent = Agent(model, instructions=ORDER_SUPPORT_INSTRUCTIONS, defer_model_check=True)
# The tools are closures, so they call the OrderDesk passed in here.
@agent.tool_plain
def look_up_order(order_id: str) -> str:
"""Look up one order by its order ID and return its items and status."""
return order_desk.describe_order(order_id)
@agent.tool_plain
def cancel_order(order_id: str, reason: str) -> str:
"""Cancel one order that has not shipped yet."""
return order_desk.cancel_order(order_id, reason)
return pydantic_ai_session_flow(
agent,
flow_type=PYDANTIC_AI_ORDER_SUPPORT_FLOW_TYPE,
tool_sources=build_slack_tool_sources(slack_connection),
tool_policies={
"look_up_order": ToolPolicy(ToolEffect.READ),
"cancel_order": ToolPolicy(ToolEffect.NON_IDEMPOTENT_WRITE),
},
)
Example: examples/python/agent-harnesses/agent_harness_examples/pydantic_ai_order_support.py
Worker
One Worker serves the session Flows of every harness in the example. Agent session Steps are async, so the Python Registry allows async handlers. Build the same session Flow in every process that starts sessions or calls their RPCs.
# Agent session Steps are async, so the Registry allows async handlers.
self.registry = Registry(
tuple(example.session_flow for example in self.examples), allow_async_handlers=True
)
config.blob_cache_dir.mkdir(parents=True, exist_ok=True)
self.blob_cache = open_blob_cache(BlobCacheConfig(str(config.blob_cache_dir), 1 << 30))
self.worker = AsyncWorker(
self.registry,
self.blob_cache,
WorkerOptions(
server_address=config.server_address,
pull=PullWorkerOptions(max_concurrent_tasks=MAX_CONCURRENT_TASKS),
),
)
self.client = AsyncClient(
self.registry, self.blob_cache, ClientOptions(server_address=config.server_address)
)
# ...
def _new_pydantic_ai_example(
config: AgentHarnessExamplesConfig, slack_connection: slack.Connection | None
) -> AgentHarnessExample:
order_desk = OrderDesk()
model = _pydantic_ai_model(config.pydantic_ai_model)
flow = build_pydantic_ai_order_support_flow(order_desk, model, slack_connection)
return AgentHarnessExample("pydantic-ai", flow, order_desk)
# ...
def _pydantic_ai_model(model: str) -> str | PydanticAIModel:
return build_pydantic_ai_demo_model() if is_demo_model(model) else model
Example: examples/python/agent-harnesses/agent_harness_examples/app.py
Session routes
Every harness serves these same routes. Session routes explains them and the session token.
@blueprint.post("/sessions")
async def start_session() -> Response:
# The server picks the session ID, so a caller cannot claim an existing session.
session_id = f"{harness_name}-order-support-{uuid.uuid4()}"
await sessions.start_session(session_id)
session_token = sign_session_id(app_state.config.session_key, session_id)
return jsonify(session_id=session_id, session_token=session_token)
@blueprint.post("/messages")
async def send_user_message() -> Response:
body, session_id = await session_request()
accepted = await sessions.send_user_message(session_id, _required_string(body, "text"))
return jsonify(accepted=accepted)
@blueprint.post("/interrupts/resolve")
async def resolve_interrupt() -> Response:
body, session_id = await session_request()
interrupt_id = _required_string(body, "interrupt_id")
if not isinstance(approved := body.get("approved"), bool):
raise BadRequest("approved must be a boolean")
rejection_message = body.get("rejection_message")
if rejection_message is not None and not isinstance(rejection_message, str):
raise BadRequest("rejection_message must be a string")
accepted = await sessions.resolve_interrupt(
session_id, interrupt_id, approved=approved, rejection_message=rejection_message
)
return jsonify(accepted=accepted)
@blueprint.get("/snapshot")
async def get_agent_snapshot() -> Response:
session_id = request.args.get("session_id", "")
if not session_id:
raise BadRequest("session_id is required")
authorize_session(session_id)
try:
snapshot = await read_agent_event_stream_snapshot(
app_state.client, example.session_flow, session_id
)
except FlowNotActiveOrNotFoundError as error:
raise NotFound("the agent session is missing or no longer retained") from error
return jsonify(snapshot.to_json())
Example: examples/python/agent-harnesses/agent_harness_examples/http_routes.py
Run it locally
Start Dex and the example
In a first terminal, render the session graph, then start Dex. Dex Web reads the graph when it starts, to show the session and its approval controls.
cd examples/python/agent-harnesses
uv sync --locked
mkdir -p definitions
uv run --frozen dex-agents-session-graph --flow-type PydanticAIOrderSupport \
--permission order-support.operate --out definitions/pydantic-ai.json
dexcli dev --flow-rendering-dir ./definitions
dexcli dev stays in the foreground. In a second terminal, start the example:
cd examples/python/agent-harnesses
uv run --frozen python -m agent_harness_examples
The app serves on 127.0.0.1:8081. Dex Web is at http://127.0.0.1:8802.
Without DEX_PYDANTIC_AI_MODEL, the agent uses mock/order-support, a local demo model that needs no key. It calls look_up_order for "look up order 1001" and cancel_order for "cancel order 1002".
| Variable | Default | Meaning |
|---|---|---|
| DEX_PYDANTIC_AI_MODEL | mock/order-support | A Pydantic AI model name, such as openai:gpt-5.2. |
| OPENAI_API_KEY | none | The provider key, for an OpenAI model. |
| DEX_AGENT_SESSION_KEY | a random key | The key that signs session tokens. |
Call the session
Start a session, ask the agent to cancel order 1002, and approve the call:
base=http://127.0.0.1:8081/products/agent-harnesses/pydantic-ai
session=$(curl -s -X POST "$base/sessions" -H 'Content-Type: application/json' -d '{}')
id=$(echo "$session" | jq -r .session_id)
token=$(echo "$session" | jq -r .session_token)
curl -s -X POST "$base/messages" -H 'Content-Type: application/json' \
-H "X-Agent-Session-Token: $token" -d "{\"session_id\": \"$id\", \"text\": \"Cancel order 1002\"}"
until interrupt=$(curl -s "$base/snapshot?session_id=$id" -H "X-Agent-Session-Token: $token" \
| jq -re '.snapshot.pending_interrupts[0].interrupt_id'); do sleep 1; done
curl -s -X POST "$base/interrupts/resolve" -H 'Content-Type: application/json' \
-H "X-Agent-Session-Token: $token" \
-d "{\"session_id\": \"$id\", \"interrupt_id\": \"$interrupt\", \"approved\": true}"
The loop reads the snapshot until the agent asks to run cancel_order. That call waits for approval, then runs once. Read the snapshot again to see the agent's answer.
Approve in Dex Web
Instead of the last command, you can approve the call in Dex Web. Follow the steps in Approve in Dex Web.
Connect Connectors
The agent can call Connector operations as tools, as Connectors as tools describes. The example adds the Slack tool slack_post_thread_reply, which replies in a Slack thread.
Offer the tool
Install the connectors extra of dex-agents and the Connector package, such as dex-connector-slack. The install commands are in Connector tools.
The Worker opens the Slack connection support-slack from the project configuration. connector_tool turns the operation into a tool, and ConnectorToolSource holds it. Without the connection, the agent gets no Slack tool.
def build_slack_tool_sources(connection: slack.Connection | None) -> list[ToolSource]:
"""Returns the Slack tool source, or no source when Slack is not configured."""
if connection is None:
return []
return [
ConnectorToolSource([
# The manifest does not ask for approval; this agent asks before every post.
connector_tool(
connection.post_thread_reply(), connection.reference, requires_approval=True
),
]),
]
# ...
async def open_slack_project_connection(
environment: Mapping[str, str] = os.environ,
) -> SlackProjectConnection | None:
# ...
# An incomplete or invalid configuration raises here, so the Worker fails at startup.
try:
project = await load_from_environment(environment)
except FileNotFoundError:
_log_slack_not_set_up("the connector configuration file does not exist")
return None
# ...
try:
connection = slack.new_project_connection(project, SLACK_CONNECTION_NAME)
except BaseException:
await project.close()
raise
return SlackProjectConnection(project, connection)
Example: examples/python/agent-harnesses/agent_harness_examples/slack_connector_tools.py
The Worker opens the connection once, at startup, and every agent uses it:
async def main() -> None:
config = AgentHarnessExamplesConfig.from_env()
slack_setup = await open_slack_project_connection()
try:
app_state = AgentHarnessExamplesApp(config, slack_setup.connection if slack_setup else None)
# ...
finally:
if slack_setup is not None:
await slack_setup.aclose()
Example: examples/python/agent-harnesses/agent_harness_examples/__main__.py
The agent passes the tool source next to its own tools:
return pydantic_ai_session_flow(
agent,
flow_type=PYDANTIC_AI_ORDER_SUPPORT_FLOW_TYPE,
tool_sources=build_slack_tool_sources(slack_connection),
# ...
)
Example: examples/python/agent-harnesses/agent_harness_examples/pydantic_ai_order_support.py
How a call runs
Each call of slack_post_thread_reply runs once, in its own CallSingleAttemptTool Step, after a person approves it. The Slack Connector tool explains why.
The adapter passes the Connector tools to each run as an ExternalToolset. Pydantic AI never runs an external tool, so a Connector tool call ends the run, and the call runs in its tool Step.
Configure the connection
Connections explains where the Worker finds the support-slack connection and what the Slack app needs.
To try it, start the example with DEX_CONNECTOR_CONFIG_FILE set. Then ask the agent to reply in a Slack thread, with the channel ID and the thread timestamp:
Reply in Slack thread C0123456789 1710000000.000100: Order 1002 ships on Thursday.
The post waits for approval, like cancel_order.
Limits
- Retries that Pydantic AI sends to the model itself make another model call in the same RunModel Step. Examples are invalid tool arguments and output that fails validation.
- Output functions run inside RunModel and are at least once. Put a write in a tool instead.
- Provider-run tools, such as WebSearchTool, repeat when RunModel reruns.
- A deferred tool returns text to the model. Images and files are lost.
- A tool that asks for approval only while it runs needs a ToolPolicy with requires_approval=True.
- UsageLimits and tool retry budgets apply per Step, not per turn.
The adapter README lists every limit. See Limits for long sessions for the server limits.