Building an E-Commerce Cart Agent with Structured Tool Outputs

Shopping cart tools are where agent integrations usually rot. The add_to_cart tool returns a sentence like “Added 2 widgets to your cart, total is now $49.98”, the model paraphrases it, and three turns later the agent confidently reports a subtotal it invented. When the tool then fails - out of stock, invalid coupon - the error arrives as prose too, and the model’s recovery is whatever it can infer from the prose.

The fix is not a better prompt. It is giving every cart tool a return type: a Pydantic model that is validated before it ever reaches the model’s context. In Pydantic AI, a function tool’s signature and docstring become the schema the model sees, arguments are validated before your code runs, and if a tool’s result does not validate, the framework hands the error back to the model as a retry prompt instead of silently feeding it garbage.

This post builds a complete cart domain agent in Pydantic AI: add_item, update_quantity, remove_item, get_pricing, and prepare_checkout, each returning a validated model, with deliberate handling for the errors a model can recover from, structured partial failures at checkout time, and a TestModel-based test suite that runs with no API key.

The schema is the contract

Start with the data, not the tools. Every tool in this agent returns a subclass of one base result type, so “did this operation succeed, and why not” is always answerable in code:

from enum import StrEnum
from typing import Literal

from pydantic import BaseModel, Field


class Outcome(StrEnum):
    OK = 'ok'
    OUT_OF_STOCK = 'out_of_stock'
    INVALID_QUANTITY = 'invalid_quantity'
    UNKNOWN_SKU = 'unknown_sku'
    INVALID_COUPON = 'invalid_coupon'
    PAYMENT_REVIEW = 'payment_review'


class CartToolResult(BaseModel):
    """Base result for every cart tool: an outcome code plus a human summary."""

    outcome: Outcome
    summary: str = Field(description='One sentence the shopper would understand')


class CartLine(BaseModel):
    sku: str
    title: str
    quantity: int = Field(ge=1)
    unit_price_cents: int = Field(ge=0)
    line_total_cents: int = Field(ge=0)


class CartSnapshot(CartToolResult):
    """The current state of the cart after the operation."""

    lines: list[CartLine]
    subtotal_cents: int = Field(ge=0)
    currency: str = 'USD'

Two design decisions do most of the work here:

Failure is data, not an exception. An out-of-stock item is a normal commerce event, not a crash. Encoding it as outcome=OUT_OF_STOCK inside a valid result means the model always gets a parseable object and a clear instruction for what to tell the shopper - no exception text to garble, no empty context.

Money is integers. unit_price_cents: int = Field(ge=0) makes it structurally impossible for the model’s downstream reasoning to round a float into a wrong total, and ge=0 rejects a negative price before any business rule runs. Pydantic builds the JSON Schema from these annotations, so the constraints double as documentation in the tool schema the model reads.

Pydantic’s validation core is Rust-fast and the same engine underneath FastAPI and the major LLM SDKs - you are not adding a bespoke parsing layer, you are using the ecosystem’s default.

The cart store and the tools

The agent’s dependencies carry the store. A plain dataclass injected via deps_type keeps the tools testable and the store swappable:

from dataclasses import dataclass, field

from pydantic_ai import Agent, RunContext, ModelRetry


@dataclass
class CartStore:
    lines: dict[str, CartLine] = field(default_factory=dict)
    coupon: str | None = None

    def subtotal(self) -> int:
        return sum(l.line_total_cents for l in self.lines.values())


@dataclass
class CartDeps:
    store: CartStore
    catalog: dict[str, tuple[str, int]]  # sku -> (title, unit_price_cents)

Now the five tools. Each has a real docstring (it becomes the model-visible tool description), typed parameters (they become the input schema, with parameter descriptions extracted from the docstring), and a validated return type:

def _snapshot(store: CartStore, outcome: Outcome, summary: str) -> CartSnapshot:
    return CartSnapshot(
        outcome=outcome,
        summary=summary,
        lines=list(store.lines.values()),
        subtotal_cents=store.subtotal(),
    )


agent = Agent(
    'openai:gpt-6.1-sol',
    deps_type=CartDeps,
    instructions=(
        'You are a shopping cart assistant. Always call a tool before '
        'quoting a price or a subtotal; never compute totals yourself. '
        'If a tool returns a non-ok outcome, tell the shopper what happened '
        'and offer the next step.'
    ),
)


@agent.tool
def add_item(ctx: RunContext[CartDeps], sku: str, quantity: int) -> CartSnapshot:
    """Add a SKU to the cart, or increase its quantity.

    Args:
        sku: The catalog SKU, e.g. 'WIDGET-1'.
        quantity: How many to add; must be 1 or more.
    """
    entry = ctx.deps.catalog.get(sku)
    if entry is None:
        return _snapshot(ctx.deps.store, Outcome.UNKNOWN_SKU, f'No product with SKU {sku}.')
    if quantity < 1:
        raise ModelRetry('quantity must be at least 1; ask the shopper how many they want.')
    title, unit_price = entry
    existing = ctx.deps.store.lines.get(sku)
    new_qty = (existing.quantity if existing else 0) + quantity
    ctx.deps.store.lines[sku] = CartLine(
        sku=sku, title=title, quantity=new_qty,
        unit_price_cents=unit_price, line_total_cents=new_qty * unit_price,
    )
    return _snapshot(ctx.deps.store, Outcome.OK, f'Added {quantity} x {title}.')

update_quantity and remove_item follow the same shape - look up the line, return a typed failure or mutate the store, return the full snapshot. Returning the whole cart on every mutation is deliberate: the model never has to stitch together state across turns, and every tool result is independently truthful.

The pricing tool is where structured output pays for itself:

@agent.tool
def get_pricing(ctx: RunContext[CartDeps]) -> CartToolResult:
    """Return the current subtotal and any applied coupon."""
    store = ctx.deps.store
    if not store.lines:
        return CartToolResult(outcome=Outcome.OK, summary='The cart is empty.')
    if store.coupon == 'SAVE10':
        total = int(store.subtotal() * 0.9)
        return CartToolResult(
            outcome=Outcome.OK,
            summary=f'Subtotal {store.subtotal()} cents, coupon SAVE10 applied: {total} cents.',
        )
    return CartToolResult(
        outcome=Outcome.OK, summary=f'Subtotal {store.subtotal()} cents, no coupon applied.'
    )

Note what is absent: string concatenation of a total the model might misread, and no float arithmetic on money. The summary is prose for the model; the numbers are computed once, in code, and the model is instructed not to recompute them.

Checkout preparation with partial failure

prepare_checkout is the tool that justifies the whole design, because checkout is where partial failure is normal: the coupon can be invalid while the cart prices fine, an item can go out of stock while the rest is purchasable. Model that explicitly instead of collapsing to a single boolean:

class CheckoutBlocker(BaseModel):
    code: Literal['out_of_stock', 'invalid_coupon', 'payment_review', 'empty_cart']
    detail: str
    skus: list[str] = Field(default_factory=list)


class CheckoutPrep(CartToolResult):
    """Whether the cart is ready for checkout, and what blocks it if not."""

    blockers: list[CheckoutBlocker] = Field(default_factory=list)
    total_cents: int = Field(ge=0)
    estimated_tax_cents: int = Field(ge=0)


@agent.tool
def prepare_checkout(ctx: RunContext[CartDeps]) -> CheckoutPrep:
    """Validate the cart for checkout and return everything blocking it.

    Every blocker found is returned; a non-empty blockers list means the
    cart cannot be charged yet.
    """
    store = ctx.deps.store
    blockers: list[CheckoutBlocker] = []
    if not store.lines:
        blockers.append(CheckoutBlocker(code='empty_cart', detail='Cart has no items.'))
    oos = [s for s in store.lines if s in ctx.deps.out_of_stock]
    if oos:
        blockers.append(CheckoutBlocker(
            code='out_of_stock', detail='Items left the catalog.', skus=oos))
    if store.coupon is not None and store.coupon not in ctx.deps.valid_coupons:
        blockers.append(CheckoutBlocker(
            code='invalid_coupon', detail=f'Coupon {store.coupon} is not valid.'))
    subtotal = store.subtotal()
    total = subtotal if store.coupon == 'SAVE10' else subtotal
    return CheckoutPrep(
        outcome=Outcome.OK if not blockers else Outcome.PAYMENT_REVIEW,
        summary=('Ready for checkout.' if not blockers
                 else f'{len(blockers)} issue(s) must be resolved before checkout.'),
        blockers=blockers,
        total_cents=total,
        estimated_tax_cents=int(total * 0.0825),
    )

All the checks run, all the blockers come back, and the caller can branch on blockers in code while the model narrates it accurately. A single success: false would have forced the model to guess which of three independent problems it was looking at.

Errors the model can recover from

Pydantic AI distinguishes three failure channels, and choosing the right one is most of the craft:

Failure	Raise	What the model sees	Retry budget
Bad arguments the model can fix	ModelRetry('message')	A retry prompt with your message	Consumes the tool’s retries
Arguments failing schema validation	automatic	Generated retry prompt with validation details	Consumes the same budget
Terminal, definitive failure	ToolFailed('message')	A failed tool result to adapt to	Untouched

Schema validation happens first: arguments are validated against the function signature, and on failure the framework generates a RetryPromptPart with the validation details and sends it back to the LLM so it can correct the call. ModelRetry is your way to inject a business rule into that same loop - “quantity must be at least 1; ask the shopper how many they want” - with the message reaching the model verbatim. The ToolFailed channel, added in newer Pydantic AI releases, is for calls that are complete but unsuccessful: the resource genuinely does not exist, or the upstream said no. The failed outcome is recorded in message history as a failed tool return, the model adapts rather than retries, and the per-tool retry counter is left alone - bounding repeated failures there is the job of run-level usage limits instead.

Retries are per tool, with a layered precedence: a per-tool @agent.tool(retries=N) beats a per-toolset limit, which beats the agent-wide Agent(retries={'tools': N}) default. The built-in default is 1, which is stingy for commerce tools - a model that guesses a SKU once and gets one correction before the run raises is a bad shopper experience. Setting retries={'tools': 3} on the agent, or per-tool on the ones with fuzzy inputs, is cheap insurance.

There is a fourth lever worth knowing: strict=True on a tool registration asks the provider to constrain generation so arguments conform to the schema in the first place. Pydantic AI translates it to each provider’s native enforcement - OpenAI strict tool definitions, Anthropic and Bedrock strict specs on opt-in, Gemini’s VALIDATED function-calling mode (enabled by default on Gemini 2.5 and newer). For structured arguments like a quantity: int parameter, strict mode rules out the “model emits the quantity as a string” class of failure before validation ever runs.

Testing without spending a cent

Pydantic AI’s testing story is the reason to pick it for a domain like this. The standard setup from the docs: TestModel or FunctionModel in place of the real model, Agent.override to swap it in, and ALLOW_MODEL_REQUESTS = False globally so an accidental live request fails loudly instead of quietly costing money.

import pytest
from pydantic_ai import models, capture_run_messages
from pydantic_ai.models.test import TestModel

models.ALLOW_MODEL_REQUESTS = False


@pytest.fixture
def store() -> CartStore:
    return CartStore()


@pytest.fixture
def deps(store: CartStore) -> CartDeps:
    return CartDeps(
        store=store,
        catalog={'WIDGET-1': ('Widget', 2499), 'GADGET-2': ('Gadget', 999)},
    )


async def test_add_item_returns_validated_snapshot(deps: CartDeps):
    with capture_run_messages() as messages:
        with agent.override(model=TestModel()):
            result = await agent.run('Add 2 WIDGET-1 to my cart', deps=deps)

    assert deps.store.lines['WIDGET-1'].quantity == 2
    assert deps.store.lines['WIDGET-1'].line_total_cents == 4998
    assert isinstance(result.output, CartSnapshot)

TestModel calls every registered tool and generates schema-valid arguments procedurally - no ML involved, just Python producing data that satisfies each tool’s JSON schema. That makes it perfect for asserting that your tools execute correctly and return valid models, but useless for asserting which arguments a real model would send. For that, FunctionModel replaces the LLM with your own function that receives the message list and returns a scripted response:

from pydantic_ai import ModelResponse, ToolCallPart
from pydantic_ai.models.function import AgentInfo, FunctionModel


def script_add_widget(messages, info: AgentInfo) -> ModelResponse:
    if len(messages) == 1:
        return ModelResponse(parts=[
            ToolCallPart('add_item', {'sku': 'WIDGET-1', 'quantity': 2})
        ])
    return ModelResponse(parts=[])


async def test_scripted_add_flow(deps: CartDeps):
    with agent.override(model=FunctionModel(script_add_widget)):
        await agent.run('I want two widgets', deps=deps)
    assert deps.store.subtotal() == 4998

capture_run_messages gives you the full ModelRequest/ModelResponse exchange - tool call parts, tool return parts, retry prompts - so a test can assert exactly which tool fired with which arguments. Together these cover the three things a cart agent must get right: tools mutate the store correctly, tool results validate, and retry paths surface recoverable errors instead of swallowing them.

Checklist

• Give every tool a Pydantic return model; never return prose the caller must parse.
• Model expected failures (out_of_stock, invalid_coupon) as outcome codes inside valid results, and reserve ToolFailed for terminal errors.
• Use ModelRetry with an actionable message for arguments the model can correct, and raise the default retry budget for fuzzy commerce tools.
• Keep money as integer cents with Field(ge=0); never float arithmetic the model can re-round.
• Return the full cart snapshot on every mutation so no turn depends on stitched-together state.
• Return all checkout blockers at once with a typed blockers list instead of a single failure flag.
• Set models.ALLOW_MODEL_REQUESTS = False and test with TestModel (execution and validation) plus FunctionModel (argument control).
• Consider strict=True on structured-argument tools to get provider-side schema enforcement.

Sources: Pydantic AI, Pydantic AI - Function Tools, Pydantic AI - Advanced Tool Features (retries and failures), Pydantic AI - Unit Testing, Pydantic Validation