Skip to content

AxmeAI/pydantic-ai-human-approval-example

Repository files navigation

Pydantic AI Human Approval Example

Pydantic AI doesn't have human-in-the-loop. Add async approvals with reminders and timeout in 10 lines.

Pydantic AI gives you structured, type-safe AI agents. But when your agent drafts a customer response that needs manager sign-off before sending - Pydantic AI has no built-in way to pause, notify a human, wait hours, remind, and resume. AXME adds that.

Alpha - Built with AXME (AXP Intent Protocol). cloud.axme.ai - contact@axme.ai


The Problem

Pydantic AI is excellent at structured agent output. But production agents need human checkpoints:

Pydantic AI agent analyzes support ticket
  -> Drafts a response (structured, type-safe, great!)
  -> Now what? Who approves this before it goes to the customer?

Options:
  1. Print to console and wait? (agent blocks, session dies)
  2. Send email manually? (no tracking, no reminder, no timeout)
  3. Build a webhook + DB + polling system? (200+ lines of infrastructure)

Open issues confirm this gap:


The Solution

# Pydantic AI does the thinking
result = support_agent.run_sync(ticket_prompt)  # structured TicketAnalysis

# AXME handles the human gate
intent_id = client.send_intent({
    "intent_type": "intent.support.ticket_review.v1",
    "to_agent": "agent://myorg/production/support-analyst",
    "payload": {"ticket_id": "TICK-4821", "draft": result.data.draft_response},
})
approval = client.wait_for(intent_id)  # reminder in 5 min, timeout in 8h

Pydantic AI does the reasoning. AXME does the waiting.


Quick Start

pip install axme pydantic-ai
export AXME_API_KEY="your-key"   # Get one: axme login
export OPENAI_API_KEY="sk-..."   # For Pydantic AI LLM calls
from pydantic import BaseModel
from pydantic_ai import Agent
from axme import AxmeClient, AxmeClientConfig
import os

# 1. Pydantic AI: structured analysis
class TicketAnalysis(BaseModel):
    root_cause: str
    severity: str
    draft_response: str
    recommended_action: str

support_agent = Agent("openai:gpt-4o-mini", result_type=TicketAnalysis)

# 2. Run analysis
result = support_agent.run_sync("Ticket TICK-4821: API rate limit errors")
print(f"Analysis: {result.data.root_cause}")
print(f"Draft: {result.data.draft_response}")

# 3. AXME: human approval before sending
client = AxmeClient(AxmeClientConfig(api_key=os.environ["AXME_API_KEY"]))
intent_id = client.send_intent({
    "intent_type": "intent.support.ticket_review.v1",
    "to_agent": "agent://myorg/production/support-analyst",
    "payload": {
        "ticket_id": "TICK-4821",
        "analysis": result.data.model_dump(),
    },
})

# Agent suspends. Reminder in 5 min. Timeout in 8 hours.
approval = client.wait_for(intent_id)
print(f"Decision: {approval['status']}")

Before / After

Before: No HITL in Pydantic AI

# Pydantic AI generates the response...
result = support_agent.run_sync(ticket_prompt)

# ...but how does a human approve it?
print(result.data.draft_response)  # Print to console?
input("Press Enter to send...")     # Block the process?

# No reminders. No timeout. No audit trail.
# If the human walks away, the agent is stuck forever.

After: AXME Adds the Missing Piece

result = support_agent.run_sync(ticket_prompt)

intent_id = client.send_intent({
    "intent_type": "intent.support.ticket_review.v1",
    "to_agent": "agent://myorg/production/support-analyst",
    "payload": {"draft": result.data.draft_response},
})
approval = client.wait_for(intent_id)
# Reminder in 5 min. Escalation in 30 min. Timeout in 8h.

What Each Component Does

Component Role Framework
agent_pydantic.py Analyze ticket with structured output Pydantic AI
agent.py Simplified agent (no LLM, tests AXME flow) AXME SDK
initiator.py Send ticket intent, observe lifecycle AXME SDK
scenario.json Scenario definition (agent + human approval) AXME CLI

Pydantic AI does the thinking. AXME does the infrastructure.


How It Works

+-----------+  send_intent()   +----------------+  analyze    +-----------+
|           | ---------------> |                | ----------> |           |
| Initiator |                  |   AXME Cloud   |             | Pydantic  |
|           | <- wait_for() -- |   (platform)   | <- result   | AI Agent  |
|           |                  |                |             |           |
|           |                  |  WAITING for   |             +-----------+
|           |                  |  human approval|
|           |                  |                |  approve    +-----------+
|           |                  |                | <---------- |           |
|           | <- COMPLETED --- |  - remind 5m   |             |  Support  |
|           |                  |  - timeout 8h  |             |   Lead    |
+-----------+                  +----------------+             +-----------+

Run the Full Example

Prerequisites

curl -fsSL https://raw.githubusercontent.com/AxmeAI/axme-cli/main/install.sh | sh
axme login
pip install -r requirements.txt

Terminal 1 - submit the scenario

axme scenarios apply scenario.json

Terminal 2 - start the agent

# macOS
cat ~/Library/Application\ Support/axme/scenario-agents.json | grep -A2 pydantic-ai-support-demo

# Linux
cat ~/.config/axme/scenario-agents.json | grep -A2 pydantic-ai-support-demo
# Simplified agent (no LLM required)
AXME_API_KEY=<agent-key> python agent.py

# Full Pydantic AI agent (requires OPENAI_API_KEY)
AXME_API_KEY=<agent-key> OPENAI_API_KEY=sk-... python agent_pydantic.py

Terminal 1 - approve

axme tasks approve <intent_id>

Verify

axme intents get <intent_id>
# lifecycle_status: COMPLETED

Related


Built with AXME (AXP Intent Protocol).

About

Pydantic AI doesn't have human-in-the-loop. Add async approvals with reminders and timeout in 10 lines.

Topics

Resources

License

Stars

Watchers

Forks

Releases

No releases published

Packages

 
 
 

Contributors

Languages