NewIntroducing Integrations
All examples
Pydantic AI
Updated
8 min read
by Anlyon Team

Pydantic AI tool approval: require a human decision before an agent sends email

Pydantic AI's requires_approval ends the run with DeferredToolRequests until your code answers. This example uses deferred tools differently: the tool invokes an Anlyon action, defers while an approver reviews the exact email, and resumes the conversation with the graded receipt. Anlyon holds the Resend key and a shared limit counts every recipient.

pydantic aipythonapprovalsemailtool callinghuman in the loop

To require approval for a Pydantic AI tool, mark it requires_approval=True or raise ApprovalRequired, and the run ends with DeferredToolRequests until your code answers. That decision is made by your code, in your process. To put the decision and the credential outside the agent, have the tool invoke an Anlyon action that requires approval and defer with CallDeferred while an approver reviews the exact request. When they decide, Anlyon makes the call and you hand its receipt back to the model.

This example gives a Pydantic AI support agent the ability to email customers. Four things hold without relying on the model:

  • Every email waits for an approver, who sees the recipients, the subject and the body.
  • The agent never holds the Resend key. It is in the Anlyon vault and attached at dispatch.
  • A shared limit caps the day's email, counted per recipient across to, cc and bcc, for every agent, session and key in the environment.
  • The sender is fixed by the action. The model can choose who to write to and what to say. It cannot change who the email is from.

Written against pydantic-ai 2.57 and anlyon 0.6 (Python).

TL;DR

Pydantic AI requires_approvalAnlyon action, deferred with CallDeferred
Where the decision is madeYour code builds DeferredToolResultsAn approver in Anlyon, from the console or the API
Who can decideWhoever your code trusts to answerAn approver in your workspace, never the key that asked
Where the Resend key livesWherever your tool reads itThe Anlyon vault, attached at dispatch
What runs after approvalYour tool function, with the approved or overridden argsThe exact request that was reviewed
Limit across agents and sessionsNot part of the flagA shared count of emails, reserved before dispatch
What the model gets backYour tool's return valueThe invocation's receipt grade, in plain words

How does Pydantic AI's own tool approval work?

As of 2026-10-18, Pydantic AI's deferred tools come in two kinds: tools that require approval, and tools that are executed externally. A tool that always needs approval takes requires_approval=True. One that needs it only for some arguments raises ApprovalRequired unless ctx.tool_call_approved is already true:

from pydantic_ai import Agent, ApprovalRequired, DeferredToolRequests, DeferredToolResults, RunContext, ToolDenied

files_agent = Agent("openai:gpt-5-mini", output_type=[str, DeferredToolRequests])


@files_agent.tool_plain(requires_approval=True)
def delete_file(path: str) -> str:
    return f"File {path!r} deleted"


@files_agent.tool
def update_file(ctx: RunContext, path: str, content: str) -> str:
    if path == ".env" and not ctx.tool_call_approved:
        raise ApprovalRequired(metadata={"reason": "protected"})
    return f"File {path!r} updated"


result = files_agent.run_sync("Delete __init__.py and clear .env")
requests = result.output  # DeferredToolRequests: the calls waiting for a decision

results = DeferredToolResults()
for call in requests.approvals:
    results.approvals[call.tool_call_id] = (
        True if call.tool_name == "update_file" else ToolDenied("Deleting files is not allowed")
    )

result = files_agent.run_sync(message_history=result.all_messages(), deferred_tool_results=results)

A HandleDeferredToolCalls capability can answer the same requests inline, without ending the run, when the approver is in the same process.

The Pydantic AI docs are direct about the limit of this. As of 2026-10-18 they say that approval is not an authorization boundary against an untrusted client: when an agent is served over a UI adapter, the client submits the approval with the rest of the request, so a client that can reach the endpoint can approve a tool call of its own making. The docs say to authenticate the endpoint and enforce authorization for sensitive actions inside the tool function itself. And whatever the decision, the tool runs in your process with whatever key it reads.

How does the Anlyon version fit together?

The other kind of deferred tool, executed externally, is the one that matches an approval held outside your process:

Pydantic AI run (holds an Anlyon key: invoke, read)
  -> tool send_customer_email
    -> anlyon.actions.invoke("send-support-email", input, operation_key=...)
       pending approval: the tool raises CallDeferred(metadata={"invocation_id": ...})
  <- the run ends with DeferredToolRequests. You store the conversation.
       ... an approver reviews the exact email in the Anlyon console ...
       on approval, Anlyon reserves the recipients against the daily limit,
       attaches the vaulted Resend key and sends
  -> later: read the invocation, answer the deferred call with its result,
     and resume the stored conversation

1. One-time setup, with an operator key

Run this from a deploy script, not from the agent. It stores the Resend key, defines the action on the resend.email_send adapter, and sets a daily limit in emails.

# setup.py
import os
from anlyon import Client

ops = Client(api_key=os.environ["ANLYON_OPERATOR_KEY"])

# A full access key: a sending access key cannot read the email back.
ops.secrets.put("RESEND_KEY", value=os.environ["RESEND_API_KEY"], allowed_hosts=["api.resend.com"])

ops.actions.create(
    name="send-support-email",
    description="Send one email to a customer from the support address.",
    requires_approval=True,
    adapter={
        "type": "resend.email_send",
        "config": {"domain": "mail.example.com", "fromAddresses": ["support@mail.example.com"]},
        "credentialSecret": "RESEND_KEY",
    },
)

# 200 recipients a day, across every agent, session and key in this environment.
ops.impact_limits.create(
    name="support-email-per-day",
    dimension="emails",
    period="day",
    amount=200,
    action_name="send-support-email",
)

fromAddresses is why the sender is fixed: the adapter refuses any from that is not on the list. Creating a limit needs impact-limits:write, which no key gets by default, so it belongs to the operator key and not to the agent's.

2. The agent and the tool

# support_agent.py
from dataclasses import dataclass

from anlyon import AsyncClient
from pydantic_ai import Agent, CallDeferred, DeferredToolRequests, RunContext


@dataclass
class Deps:
    anlyon: AsyncClient


support_agent = Agent(
    "openai:gpt-5-mini",
    deps_type=Deps,
    output_type=[str, DeferredToolRequests],
    instructions="You answer support tickets. Reply to the customer with send_customer_email. Report the outcome exactly as the tool describes it.",
)

REPLY = {
    "confirmed": "Sent. Resend's own record returned the email on read-back.",
    "acknowledged": "Resend accepted the email. Nothing has read it back yet. Do not send it again.",
    "unknown": "No usable response from Resend. Anlyon does not send it again, and neither should you.",
    "failed": "Resend rejected the email. Tell the user it was not sent.",
    "refused": "Anlyon refused the email before sending it, for example because the daily limit is used up.",
    "denied": "An approver denied the email. Do not send it again.",
    "pending": "The email is not finished yet. Do not send it again.",
}


def describe(invocation: dict) -> str:
    return REPLY[invocation.get("grade") or "pending"]


@support_agent.tool
async def send_customer_email(ctx: RunContext[Deps], ticket_id: str, to: str, subject: str, body: str) -> str:
    """Send one plain-text email to a customer. Every email waits for an approver before it is sent."""
    result = await ctx.deps.anlyon.actions.invoke(
        "send-support-email",
        {"from": "Support <support@mail.example.com>", "to": [to], "subject": subject, "text": body},
        operation_key=f"email:{ticket_id}",
    )
    invocation = result.data
    if invocation["pendingApproval"]:
        raise CallDeferred(metadata={"invocation_id": invocation["id"]})
    return describe(invocation)

The operation key ties the send to the ticket. A model that calls the tool twice for the same ticket gets the existing effect back, not a second review or a second email. If a ticket can get several replies, add your own message id to the key.

3. Run, store, and resume

# tickets.py
import os

from anlyon import AsyncClient
from pydantic_ai import ConversationTypeAdapter, DeferredToolRequests

from support_agent import Deps, describe, support_agent


async def handle_ticket(ticket) -> str | bytes:
    async with AsyncClient(api_key=os.environ["ANLYON_AGENT_KEY"]) as anlyon:
        result = await support_agent.run(
            f"Ticket {ticket.id} from {ticket.email}:\n\n{ticket.body}",
            deps=Deps(anlyon),
        )
    if isinstance(result.output, DeferredToolRequests):
        # Waiting on an approver. Store the paused conversation, keyed by ticket.
        return ConversationTypeAdapter.dump_json(result.conversation)
    return result.output


async def resume_ticket(stored: bytes) -> str | None:
    conversation = ConversationTypeAdapter.validate_json(stored)
    requests = conversation.deferred_tool_requests
    async with AsyncClient(api_key=os.environ["ANLYON_AGENT_KEY"]) as anlyon:
        answers = {}
        for call in requests.calls:
            invocation_id = requests.metadata[call.tool_call_id]["invocation_id"]
            invocation = (await anlyon.actions.invocation(invocation_id)).data
            if invocation["status"] in ("pending_approval", "running"):
                return None  # not decided or not finished yet: try again later
            answers[call.tool_call_id] = describe(invocation)

        result = await support_agent.run(
            conversation=conversation,
            deferred_tool_results=requests.build_results(calls=answers),
            deps=Deps(anlyon),
        )
    return result.output

Call resume_ticket from a scheduled job, or from a handler for the action.succeeded and action.failed events (see Events and webhooks). Nothing is held open while the approver decides. The model sees the result as if the tool had returned it, and writes its final reply from that.

What does each outcome mean for the agent?

The tool and the resume step both turn the receipt grade into one instruction for the model. A denied approval grades denied. An approval that expired grades refused, and so does a send the daily limit cannot cover: nothing is sent. unknown means no usable response from Resend. Anlyon does not send it again. It looks for the send with a bounded search of recent sent email, and an operator can replay it under the same key inside Resend's 24-hour window. Past those bounds the effect stays unknown until an operator settles it. Every reply in REPLY tells the model not to send again, because a model that reads "error" will reasonably try.

Confirmed is Resend's record of the send, not inbox delivery. A sent email cannot be recalled, so the approval is the point to stop one.

Should you keep requires_approval as well?

Not on the same tool. A tool with requires_approval=True that then invokes an action requiring approval asks twice. Use Pydantic AI's approval for a tool whose effect stays inside your app, or to confirm with the user in a chat UI. Use the Anlyon action where the decision has to hold on the production call: bound to the exact request, made by an approver who is not the key that asked, and enforced on Anlyon's side while no other copy of the Resend key is in your process.

What this does not cover

Anlyon governs the actions routed through its hosted executor. A Pydantic AI tool that calls Resend with its own key is not governed by it. The limit is shared within one environment, not across environments. The adapter sends one email per effect, with no attachments, templates, scheduled sends or batch sends.

Free during Early Beta Access, no credit card. Start building →

Frequently asked questions

How do I require approval for a Pydantic AI tool call?

As of 2026-10-18, pass requires_approval=True to @agent.tool or @agent.tool_plain, or raise ApprovalRequired inside the tool when approval depends on the arguments. The run ends with a DeferredToolRequests output. Your code builds a DeferredToolResults with an approval or a ToolDenied for each call and starts a new run with it. A HandleDeferredToolCalls capability can resolve the same requests inline instead.

Why does the tool raise CallDeferred instead of using requires_approval?

Because the decision is made in Anlyon, not in your process. The tool submits the email to an Anlyon action that requires approval and defers with the invocation id. When the approver has decided, your code reads the invocation and returns its result to the model as the deferred tool result. Nobody has to approve twice.

Does a confirmed receipt mean the email reached the inbox?

No. For the resend.email_send adapter, confirmed means Resend's own record returned the email on read-back. Confirmed is Resend's record of the send, not inbox delivery.

Continue with a related task

Free tier, no credit card. One command if you use Claude or Cursor.

$ claude mcp add anlyon -- npx -y @anlyonhq/mcp-server