PydanticAI Adapter

Kitaru's PydanticAI adapter lets you reuse an existing PydanticAI agent with Kitaru durability.

Use it when you want:

• an explicit outer @checkpoint replay boundary
• child-event visibility for agent model calls and tool calls
• optional human input from adapter tools via flow-level kitaru.wait(...)
• run-level summary metadata (pydantic_ai_run_summaries)

1) Install optional adapter dependency

uv sync --extra local --extra pydantic-ai

2) Wrap an agent

from kitaru import flow, checkpoint
from pydantic_ai import Agent
from pydantic_ai.models.test import TestModel
from kitaru.adapters.pydantic_ai import CapturePolicy, KitaruAgent

researcher = KitaruAgent(
    Agent(TestModel(), name="researcher"),
    capture=CapturePolicy(
        tool_capture="full",
        tool_capture_overrides={
            "quick_check": "metadata",
            "noop": None,  # None fully opts this tool out of tracking
        },
    ),
)

@checkpoint(type="llm_call")
def run_research(topic: str) -> str:
    return researcher.run_sync(f"Research {topic}").output

@flow
def research_flow(topic: str) -> str:
    return run_research(topic)

The outer checkpoint remains the replay boundary. Adapter-internal model/tool calls are tracked as child events under that checkpoint.

3) Tool capture modes

Capture policy is observability-only (it does not change tool execution):

• "full" (default): metadata + args/result artifacts + timings
• "metadata": metadata + timings, no args/result artifacts
• None: no adapter child event or artifacts for that tool (HITL interception still works)

Set a global default via CapturePolicy.tool_capture and override individual tools via CapturePolicy.tool_capture_overrides.

4) Add adapter-level HITL tools (optional)

from kitaru.adapters.pydantic_ai import hitl_tool

@hitl_tool(question="Approve publish?", schema=bool)
def approve_publish(summary: str) -> bool:
    # Body is skipped in HITL mode.
    return False

When the agent invokes this tool, the adapter translates it to a flow-level kitaru.wait(...) under the hood. PydanticAI's native requires_approval=True flag and ApprovalRequired / CallDeferred exceptions are also auto-bridged to kitaru.wait(...) without needing the @hitl_tool decorator.

Runtime behavior and guardrails

• No nested checkpoint replay boundaries are introduced by adapter internals.
• Adapter child events stay child events; they do not become standalone durable calls.
• run() / run_sync() at flow scope (outside checkpoints) use one synthetic checkpoint so tracking still works. Tune it via KitaruAgent(..., turn_checkpoint_config={"runtime": "isolated"}).
• run() / run_sync() outside any flow auto-opens a local flow (in-process only; on remote stacks, wrap explicitly with @kitaru.flow).
• Streamed model requests record a transcript artifact (*_stream_transcript) for replay/inspection.

Example in this repository

uv sync --extra local --extra pydantic-ai
uv run examples/pydantic_ai_agent/pydantic_ai_adapter.py

The example prints:

• execution ID
• final result
• child-event count captured under checkpoint metadata
• run summary count for wrapped agent runs

For the broader catalog, see Examples.