Durable Execution Patterns

Three separate reliability concerns

"Make this production-ready" usually bundles together three distinct problems. Pydantic AI gives each one its own, separate mechanism, rather than one big "be reliable" flag. Retries for the model's own mistakes are covered in Lesson 8, ModelRetry from an output validator, and available per-tool too, via retries= on @agent.tool. Fallback for a whole provider failing is covered in Lesson 22, FallbackModel. Idempotency for tools with real side effects is not something Pydantic AI does for you, this lesson's actual subject, because only you know what "processing the same request twice" means for your system.

Per-tool retries

A tool can have its own retry budget, separate from the agent's overall retries.

@agent.tool(retries=3)
def flaky_lookup(ctx: RunContext[None], query: str) -> str:
    ...
    raise ModelRetry("Lookup timed out, try a narrower query.")

Each ModelRetry here counts against this tool's 3, not the agent's global limit, so one unreliable tool doesn't eat into every other tool's retry budget.

Idempotency: the pattern this lesson actually builds

If a tool call has a real side effect, charging a card, sending an email, and a retry, from the model reformulating its approach, from your own infrastructure retrying a failed request, causes the same tool to be called again with the same arguments, you don't want the side effect to happen twice. The standard fix is an idempotency key: a stable identifier for "this logical operation," checked against a record of operations already completed, before the side effect runs.

from dataclasses import dataclass, field

from dotenv import load_dotenv
from pydantic_ai import Agent, RunContext

load_dotenv()


@dataclass
class PaymentDeps:
    processed: set[str] = field(default_factory=set)


agent = Agent(
    "google:gemini-3.5-flash-lite",
    deps_type=PaymentDeps,
    system_prompt=(
        "You process payments. Always call charge_card with a stable "
        "idempotency_key derived from the order id, e.g. 'order-123'."
    ),
)


@agent.tool
def charge_card(
    ctx: RunContext[PaymentDeps], idempotency_key: str, amount: float
) -> str:
    """Charge a card for an amount, safe to call more than once.

    Args:
        idempotency_key: A stable identifier for this exact charge.
        amount: The amount in dollars to charge.
    """
    if idempotency_key in ctx.deps.processed:
        return f"Already processed '{idempotency_key}', not charging again."
    ctx.deps.processed.add(idempotency_key)
    return f"Charged ${amount:.2f} (key={idempotency_key})."


def main() -> None:
    deps = PaymentDeps()

    print("First charge for order-123:")
    result = agent.run_sync("Charge $42.00 for order-123.", deps=deps)
    print(" ", result.output)

    print("\nRetrying the exact same request (simulating a client retry):")
    result = agent.run_sync("Charge $42.00 for order-123.", deps=deps)
    print(" ", result.output)

    print(f"\nOperations actually processed: {deps.processed}")

In a real system, processed would be a database table or a Redis set, something that survives past one Python process, not an in-memory set. The shape of the check, look up the key, skip if seen, record it before or while doing the real work, is what matters, and it's identical whether the storage behind it is a set() or Postgres.