Pydantic AI

ActionCapability turns registered actions into typed Pydantic AI tools. The model can propose a command and see a safe preview, but it cannot create financial authority or bypass the runtime.

python -m pip install "threvo-actions[pydantic-ai]==0.1.3"

The integration is tested against pydantic-ai-slim==2.33.0. It installs no provider SDK. The complete example below uses FunctionModel, so it runs offline without an API key.

Run the example

uv run --extra pydantic-ai python -m examples.docs.pydantic_ai_agent

Output:

The refund was authoritatively verified.
executor calls: 1

Complete agent

This file imports the quickstart action, which contains the host ports and runtime. Both files are executable in the repository.

"""A complete Pydantic AI agent with a confirm-first refund tool.

Run with:

    uv run --extra pydantic-ai python -m examples.docs.pydantic_ai_agent

The FunctionModel keeps this example offline and deterministic. Replace it with
your provider model when integrating the same capability in an application.
"""

from __future__ import annotations

import asyncio
from dataclasses import dataclass

from pydantic_ai import Agent, DeferredToolRequests
from pydantic_ai.messages import ModelMessage, ModelResponse, TextPart, ToolCallPart
from pydantic_ai.models import override_allow_model_requests
from pydantic_ai.models.function import AgentInfo, FunctionModel

from examples.docs.quickstart import AGENT, CONSUMER, REQUESTER, TENANT, Demo, build_demo
from threvo_actions.integrations.pydantic_ai import (
    ActionAgentContext,
    ActionCapability,
    ActionToolBinding,
    DeferredActionRequest,
)


@dataclass(frozen=True)
class AgentDependencies:
    tenant_reference: str
    demo: Demo


def action_context(deps: AgentDependencies) -> ActionAgentContext:
    # In a web application, build this from the authenticated server session.
    return ActionAgentContext(
        tenant_reference=deps.tenant_reference,
        requesting_principal=REQUESTER,
        proposing_agent=AGENT,
        evidence_consumer=CONSUMER,
    )


def offline_model() -> FunctionModel:
    calls = 0

    def respond(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse:
        nonlocal calls
        del messages
        calls += 1
        if calls == 1:
            return ModelResponse(
                parts=[
                    ToolCallPart(
                        "refund",
                        {"order_reference": "ORD-42"},
                        tool_call_id="refund-call:1",
                    )
                ]
            )
        if [tool.name for tool in info.function_tools] != ["refund"]:
            raise RuntimeError("the refund tool was not registered")
        return ModelResponse(parts=[TextPart("The refund was authoritatively verified.")])

    return FunctionModel(respond)


async def main() -> None:
    demo = build_demo()

    async def establish_authority(
        request: DeferredActionRequest,
        *,
        deps: AgentDependencies,
    ) -> bool:
        # A real handler authenticates the confirmer and applies separation of
        # duties before recording evidence. Framework approval alone is not enough.
        await deps.demo.approve(request.proposal_reference)
        return True

    refund = ActionToolBinding(
        definition=demo.action,
        context_resolver=action_context,
        name="refund",
        description="Prepare a refund and show a safe preview before execution.",
    )
    actions = ActionCapability[AgentDependencies](
        runtime=demo.runtime,
        bindings=[refund],
        inline_authority_handler=establish_authority,
    )
    agent = Agent(
        offline_model(),
        deps_type=AgentDependencies,
        output_type=[str, DeferredToolRequests],
        capabilities=[actions],
    )

    with override_allow_model_requests(False):
        result = await agent.run(
            "Refund order ORD-42",
            deps=AgentDependencies(tenant_reference=TENANT, demo=demo),
        )

    print(result.output)
    print(f"executor calls: {demo.host.executor_calls}")


if __name__ == "__main__":
    asyncio.run(main())

What the integration does

1. ActionToolBinding exposes only RefundCommand as the model-visible schema. Tenant, requester, evidence consumer, authority, private state, executor, and verifier are not tool arguments.
2. The first tool call prepares a proposal and raises Pydantic AI's deferred approval request with safe metadata.
3. The host authority handler authenticates and authorizes the confirmer, then records bound AuthorityEvidence in the action runtime.
4. Pydantic AI resumes the tool call. The capability resolves trusted context again and asks the runtime to execute the stored proposal.
5. If execution is immediately due for verification, the capability performs one reconciliation attempt and returns a display-safe ActionToolResult.

The agent is ordinary Pydantic AI:

agent = Agent(
    "openai:gpt-5.2",
    deps_type=AgentDependencies,
    output_type=[str, DeferredToolRequests],
    capabilities=[actions],
)

Replace the offline FunctionModel with your provider and set its credentials as Pydantic AI documents. The action control flow stays the same.

Deferred approval across requests

Inline authority is convenient for a trusted server flow, but a real approval often spans HTTP requests or users. In that case:

The repository includes a second complete program for this flow:

uv run --extra pydantic-ai python -m examples.docs.pydantic_ai_deferred

It prints the same verified result as the inline example, but execution pauses until the host records authority. The flow is:

1. run the agent and persist prepared.all_messages() according to your chat retention policy;
2. render the proposal using trusted server data;
3. record bound authority after the authenticated decision;
4. call actions.build_continuation_results(...); and
5. invoke agent.run(...) with the prior messages and deferred results.

"""A Pydantic AI approval that pauses and resumes across requests.

Run with:

    uv run --extra pydantic-ai python -m examples.docs.pydantic_ai_deferred

The FunctionModel keeps this example offline and deterministic. In production,
persist the returned message history and record authority in an authenticated
approval endpoint before resuming the agent.
"""

from __future__ import annotations

import asyncio

from pydantic_ai import Agent, DeferredToolRequests
from pydantic_ai.models import override_allow_model_requests

from examples.docs.pydantic_ai_agent import (
    AgentDependencies,
    action_context,
    offline_model,
)
from examples.docs.quickstart import TENANT, build_demo
from threvo_actions.integrations.pydantic_ai import ActionCapability, ActionToolBinding


async def main() -> None:
    demo = build_demo()
    dependencies = AgentDependencies(tenant_reference=TENANT, demo=demo)
    actions = ActionCapability[AgentDependencies](
        runtime=demo.runtime,
        bindings=[
            ActionToolBinding(
                definition=demo.action,
                context_resolver=action_context,
                name="refund",
                description="Prepare a refund and show a safe preview before execution.",
            )
        ],
    )
    agent = Agent(
        offline_model(),
        deps_type=AgentDependencies,
        output_type=[str, DeferredToolRequests],
        capabilities=[actions],
    )

    with override_allow_model_requests(False):
        prepared = await agent.run("Refund order ORD-42", deps=dependencies)
    if not isinstance(prepared.output, DeferredToolRequests):
        raise RuntimeError("the action did not request authority")

    call_id = "refund-call:1"
    metadata = prepared.output.metadata.get(call_id)
    if not isinstance(metadata, dict):
        raise RuntimeError("the authority request has no continuation metadata")
    proposal_reference = metadata.get("proposal_reference")
    if not isinstance(proposal_reference, str):
        raise RuntimeError("the authority request has no proposal reference")

    # This line belongs in an authenticated approval endpoint. The host must
    # authorize the confirmer and enforce separation of duties before recording.
    await demo.approve(proposal_reference)

    continuation = actions.build_continuation_results(
        prepared.output,
        decisions={call_id: True},
    )
    with override_allow_model_requests(False):
        completed = await agent.run(
            "Continue after server authority was recorded",
            deps=dependencies,
            message_history=prepared.all_messages(),
            deferred_tool_results=continuation,
        )

    print(completed.output)
    print(f"executor calls: {demo.host.executor_calls}")


if __name__ == "__main__":
    asyncio.run(main())

ToolApproved only permits framework continuation. Returning True from an inline handler without calling record_authority() still leaves the proposal authority_pending and the executor is not called.

Trust rules

• Build ActionAgentContext from authenticated dependencies, never model arguments.
• Treat deferred metadata and message history as untrusted routing input.
• Do not use ToolApproved.override_args to change a prepared action.
• Render previews and results from the stored proposal, not model prose.
• Treat only verified as authoritative completion.
• Schedule later runtime.reconcile() calls when the capability returns verification_pending.

See the Pydantic AI API reference for every integration type.