PydanticAI adapter

For PydanticAI. Returns pydantic_ai.Tool instances ready for Agent(tools=…). Each tool wraps a governed callable that runs the policy pipeline before invoking your registered function.

Install

pip install kitelogik pydantic-ai

pydantic-ai is not a hard dependency — only imported when you call pydantic_tools().

Setup

from kitelogik import OPAClient, PolicyGate, SessionContext
from kitelogik.adapters.pydantic_ai import PydanticAIAdapter

gate = PolicyGate(opa_client=OPAClient())
context = SessionContext(
    session_id="sess_001",
    user_role="analyst",
    session_scopes=["read_customer"],
)

adapter = PydanticAIAdapter(gate=gate, context=context)

Register tools

register(name, fn, description="", action=None) — chainable.

async def get_customer(customer_id: str) -> str:
    """Look up a customer record by ID."""
    return f"Customer {customer_id}: Acme Corp"

adapter.register("get_customer", get_customer, description="Look up a customer")

The function's signature is preserved via functools.wraps, so PydanticAI's parameter-type validation works against the underlying function's annotations.

Pass governed tools to a PydanticAI Agent

from pydantic_ai import Agent

agent = Agent(
    "openai:gpt-4o",
    tools=adapter.pydantic_tools(),    # ← list of governed pydantic_ai.Tool
)

result = await agent.run("Look up customer cust_001")

Tools are constructed with takes_ctx=False — context flows through the adapter's SessionContext rather than via PydanticAI's own RunContext. If you need the framework's RunContext for non-governance purposes, build the Tool yourself and call into a governed callable inside the body.

What happens on a deny

Denied calls return a JSON string {"blocked": true, "reason": "..."} from the tool — the agent loop continues; the model sees the structured denial.

To customise the message, pass deny_message= to the adapter constructor.

Constructor parameters

PydanticAIAdapter extends BaseGovernedAdapter:

PydanticAIAdapter(
    gate: PolicyGate,
    context: SessionContext,
    sanitize: bool = True,
    deny_message: str | None = None,
)

Param	Default	Purpose
sanitize	True	Run prompt-injection sanitiser on string return values
deny_message	"Action blocked by governance policy."	Surfaced to the model on deny

Source

kitelogik/adapters/pydantic_ai.py on GitHub.