Building Production-Ready AI Agents With Python

A prototype agent is a while loop wrapped around a model call. A production agent is that same loop wrapped in schema validation, budget ceilings, retries, traces and a test suite that runs before every deploy. The loop is the small part.

I have shipped agents that screen CVs at volume, qualify B2B leads from natural-language queries, and drive branching voice call flows. The failures were never clever model failures. They were a tool called twice with the same order ID, a 429 retried instantly in a tight loop, a JSON blob that parsed fine and meant nothing. This post is the Python scaffolding that stops all three.

What makes an AI agent production-ready in Python?

An agent is production-ready when every failure mode has a defined behaviour: bad tool arguments are rejected before execution, transient errors are retried with backoff, runaway loops hit a ceiling, and every step is traceable after the fact. Model quality is the last variable I tune, not the first.

In my experience that right-hand column is roughly a week of engineering on top of a one-day prototype. Teams that skip it do not save the week, they spend it in production with users watching. If you want the wider delivery view around this, I wrote about it in shipping AI projects as a full-stack Python developer.

How do you define typed tools an agent cannot misuse?

Define the tool schema once as a Pydantic model, export it as JSON Schema for the model, and re-validate the arguments in your own code before touching the database. The model proposes; your validator decides.

from typing import Literal
from pydantic import BaseModel, Field

class RefundOrder(BaseModel):
    order_id: str = Field(pattern='^ORD-[0-9]{6}
#x27;)
    amount_cents: int = Field(gt=0, le=50_000)
    reason: Literal['damaged', 'late', 'wrong_item']

TOOLS = [{
    'type': 'function',
    'function': {
        'name': 'refund_order',
        'description': 'Refund one shipped order. Never call twice for the same order_id.',
        'parameters': RefundOrder.model_json_schema(),
        'strict': True,
    },
}]

def refund_order(raw: dict) -> dict:
    args = RefundOrder.model_validate(raw)   # raises before money moves
    return ledger.refund(
        args.order_id,
        args.amount_cents,
        idempotency_key=f'refund:{args.order_id}',
    )

Three details there matter more than the model you pick. The regex on order_id blocks hallucinated identifiers. The upper bound on amount_cents caps the blast radius of a bad call. The idempotency key makes a duplicate call a no-op instead of a second refund.

This is exactly the discipline behind AgentFlow, my visual AI agent builder: each block on the drag-and-drop canvas stands for one narrow, well-defined step, which is why non-engineers can wire pipelines without inventing new failure modes.

What does the agent loop actually look like?

The loop is call the model, check for tool calls, execute them, append the observations, repeat until the model answers or a limit trips. Everything interesting is in the limits.

import random, time

def run_agent(messages, tools, max_steps=8, max_cost_usd=0.50):
    spent = 0.0
    for step in range(max_steps):
        reply, cost = call_model(messages, tools)
        spent += cost
        if spent > max_cost_usd:
            return {'status': 'budget_exceeded', 'steps': step + 1}
        if not reply.tool_calls:
            return {'status': 'ok', 'output': reply.content, 'steps': step + 1}
        messages.append(reply)
        for call in reply.tool_calls:
            messages.append(dispatch(call))   # validates, executes, traces
    return {'status': 'step_ceiling', 'steps': max_steps}

def call_model(messages, tools, attempts=4):
    for i in range(attempts):
        try:
            return client.chat(messages=messages, tools=tools, timeout=30)
        except (RateLimitError, APITimeoutError, InternalServerError):
            if i == attempts - 1:
                raise
            time.sleep(min(2 ** i, 8) + random.random())

Note that budget_exceeded and step_ceiling are returned, not raised. A stuck agent should surface as a countable status in your dashboard, because the rate of those two statuses is the single best early warning that a prompt change went wrong.

"An unbounded agent loop is a credit card with no limit handed to a system that cannot explain itself."

How do you get structured outputs you can trust?

Use the provider's schema-constrained output mode, then validate the result again with your own Pydantic model. Schema adherence and semantic correctness are different problems, and only the first one is the vendor's job.

OpenAI's structured outputs guide describes strict mode as guaranteeing that responses conform to the JSON Schema you supply, which removes the parse-and-pray layer most early agent code is built on. Anthropic's tool use documentation takes the same shape: you hand Claude JSON Schema tool definitions and it returns typed calls against them. Both are worth wiring through your OpenAI API integration layer rather than reimplementing per project.

When layer three fails, do not retry blindly. Feed the validation error back as a tool result and give the agent exactly one repair attempt. If the repair fails, escalate to a human queue with the full trace attached.

How should retries and backoff work in production AI agents in Python?

Retry only transient failures, with exponential backoff plus random jitter, capped at three or four attempts. Retrying a validation error is not resilience, it is the same wrong request sent again at full price.

Jitter is not optional. Without it, every worker that got rate-limited in the same second retries in the same second, and you rebuild the spike you were trying to escape.

How do you trace and debug production AI agents in Python?

Emit one trace per task and one span per step, with the model, token counts, cost, latency, tool name and validation outcome attached. Without per-step traces, debugging an agent is guesswork with a paid API attached to it.

Store the prompt version hash. The first question after any regression is which prompt produced the bad run, and a hash answers it in one query. If your traces land in a warehouse for analysis, that is ordinary data engineering work, and it pays for itself the first time you diff two prompt versions on real traffic.

How do you build an eval harness before you ship?

Build a small labelled set from real inputs, score each case with a deterministic check where possible, and fail the build when the pass rate drops below your threshold. Fifty real cases beat five hundred synthetic ones.

That last step is what compounds. On the CV screening platform, the eval set was mostly grown from the cases that had already embarrassed us, which is why later prompt changes could be shipped in an afternoon instead of debated for a week.

How do you deploy and cost-control production AI agents in Python?

Run the loop inside a FastAPI service, push anything longer than a few seconds onto a queue, and put an idempotency key on every side effect. The API returns a task ID immediately; the worker owns the loop.

Cost structure is simple to reason about and easy to get wrong. You pay per input token and per output token, so cost per task equals tokens per step multiplied by steps per task multiplied by task volume, with cached prompt prefixes billed at a lower rate on the major providers. Rates change often, so check the current OpenAI and Anthropic pricing pages before you model anything.

The two levers that actually move that number are step count and context size, in that order. Cutting a loop from six steps to four saves more than any model swap, and trimming a retrieved-context blob usually saves more than shortening the system prompt. If you want this built and handed over rather than assembled in-house, that is the core of my full-stack Python development work, and it is the same architecture behind connecting the OpenAI API to a CRM.