Building a backend API for an AI agent in Python

Why an agent is different from a normal endpoint

A typical API request is short, does one thing and either succeeds or fails. An agent request can run for a long time, make many model and tool calls, change data in other systems, pause for a person and fail halfway through. Treating it as a slow function behind a single POST route works for a demo, but it breaks when connections drop or users retry.

This guide sets out concepts that apply in any Python web framework. It focuses on the agent-specific parts. For the broader decisions around an LLM feature, see adding an LLM feature to a full stack app.

Model the API around runs

The most useful design step is to make the unit of agent work into a resource of its own. A small set of concepts covers most products:

• Conversation or thread. A long-lived container for related messages, owned by a user.
• Message. One user or assistant turn.
• Run. One execution of the agent, with its own status, inputs, limits, outcome and usage.
• Step or event. A record of something that happened during a run, such as a model call or tool result.

Separating a run from its conversation means that a client can ask "what is happening?" at any point, reconnect after a dropped connection, cancel the work or retry after a failure without confusing the history. An illustrative set of routes might look like this:

Give runs an explicit state machine

A run should always be in one known state, and only certain transitions should be allowed. This turns vague situations into ones your code and UI can handle:

# Illustrative only. Allowed run state transitions.
TRANSITIONS = {
    "queued": {"running", "cancelled"},
    "running": {"waiting_for_approval", "succeeded", "failed", "cancelled"},
    "waiting_for_approval": {"running", "cancelled"},
    "succeeded": set(),
    "failed": set(),
    "cancelled": set(),
}

def change_status(run, new_status):
    if new_status not in TRANSITIONS[run.status]:
        raise ValueError(f"{run.status} cannot become {new_status}")
    run.status = new_status

In a real system, the update should be atomic in the database, for example a conditional update on the current status, so that two workers cannot both make a transition. The waiting_for_approval state is what makes human-in-the-loop approval workable, because a pause may last far longer than any HTTP request.

Decide where the work runs

You have three broad options:

• Inside the request. The handler runs the agent and returns the answer. This is simple and fine for quick, safe tasks, but it ties the work to one connection and one process.
• In-process background tasks. The handler responds at once and the agent continues on the same server. This is easy to set up, but work can be lost when the process restarts, so it suits only tasks that are safe to lose or that are recovered by some other mechanism.
• A durable queue and workers. The handler records the run and enqueues it, and a separate worker carries it out. There are more parts to operate, but work survives restarts, and web and worker capacity can scale separately.

For long or consequential agents, a queue is usually the safer choice. Most queues redeliver a job if a worker dies, which means that a run can execute more than once, so the steps must be written to tolerate repetition. Saving a checkpoint after each step lets a new worker resume instead of starting over, an idea explored in building an orchestrator with LangGraph.

Agent workloads spend most of their time waiting on network calls, so asynchronous Python can be a natural fit. If you use it, make sure that blocking libraries do not run directly on the event loop, where they would stall every other request in that process.

If you are using FastAPI, serving an LLM app with FastAPI covers the framework-specific side, including project structure and avoiding a blocked event loop.

Make requests safe to repeat

Clients retry and users double-click. Without protection, each repeat could start another expensive run, or send a second email. Accept an idempotency key from the client on the route that starts a run, enforce uniqueness per user in the database and return the original run if the key has been seen before:

# Illustrative only. Framework-neutral service function.
def start_run(user, conversation_id, text, idempotency_key):
    conversation = load_conversation(conversation_id)
    if conversation is None or conversation.owner_id != user.id:
        raise NotFound()

    existing = find_run(user.id, idempotency_key)
    if existing:
        return existing

    # Create the message and run in one transaction. A unique
    # constraint on (user_id, idempotency_key) rejects a racing
    # duplicate; on that error, load and return the existing run.
    run = create_message_and_run(
        conversation, text, user.id, idempotency_key
    )
    enqueue_run(run.id)
    return run

A crash between the commit and the enqueue would strand a run, which is a known problem with writing to two systems. A common fix is a periodic job that re-enqueues old queued runs, or a transactional outbox. Also decide what happens when a second message arrives while a run is in progress on the same conversation. Rejecting it, queueing it or cancelling the first run are all defensible, but silently running both can corrupt the history.

Authentication and authorisation

Authenticate every route, then check ownership on every resource. Run IDs must never be enough on their own to read someone else's data. The more distinctive concern is what happens inside the run: the agent acts on behalf of a user, so tool execution should receive that user's identity from the server side and enforce their permissions, as described in tool calling for AI agents. A background worker has no request context, so identity has to be stored with the run and checked when it executes. Approval routes need their own check on who may approve.

Reporting progress and streaming

Write events to storage as the run proceeds and publish them to any connected client. Give each event an increasing sequence number, so that a client can reconnect and ask for everything after the last one it saw. Then the connection becomes a view onto the run, not the thing keeping it alive. Plain polling of the status route is a perfectly good starting point. For transports, event design and proxy issues, see streaming LLM responses to a web UI.

Cancellation, timeouts and limits

• Cancellation. Cancellation is a request, not an instant stop. Record it, and have the worker check before each step and each side effect. A tool call already under way may complete, so report what did happen.
• Timeouts at each level. Set them for model calls, for tools and for the run as a whole. Add a periodic sweep that finds runs stuck in running whose worker has stopped sending heartbeats.
• Step and usage budgets. Cap model turns, tool calls and tokens per run, and stop cleanly with a clear status when they are reached.
• Rate limits and quotas. Limit runs per user, not just requests per second, since a single request can create a lot of work. Limit concurrency too, so that one user cannot take up all the workers.
• Input limits. Validate message length and file size at the edge, before any model is involved.

Handling failures

Separate temporary faults, such as provider rate limits and timeouts, from permanent ones, such as invalid requests. Retry the first kind with exponential backoff and jitter, and fail the second kind at once. Record a safe, stable error code on the run for the client, and keep the details in internal logs. An API that distinguishes "the model service is busy" from "this run hit its step limit" is much easier to build a good interface on.

Model output should be validated like any other external input before it is stored, displayed or used to trigger something. See LLM guardrails and prompt injection defence for the wider picture.

Persistence and observability

Store conversations, messages, runs and events in ordinary database tables, with the agent's own state alongside them or in a dedicated checkpoint store. Deciding what gets carried between runs is covered in agent memory: short-term and long-term state.

Attach a trace ID to each run and pass it through the worker, model calls and tools. Record the prompt version, model name, token usage, step durations and outcome for every run. Agent content can be sensitive, so redact logs and set retention periods deliberately.

Testing the backend without a live model

Put the model behind a small interface of your own, then substitute a scripted fake in tests. This keeps tests fast and repeatable, and it makes rare situations easy to produce:

• A model that requests a tool, then fails on the following call.
• A worker that dies between two steps, followed by resumption.
• The same idempotency key sent twice at the same moment.
• A cancellation that arrives during a tool call.
• One user requesting another user's run, which should return not found.
• A run that reaches its step limit.

Keep a smaller set of tests against a real model to check for integration problems, and evaluate agent quality separately. For help building the service around an agent, including the interface that sits on top of it, see full stack development for AI products.

Frequently asked questions

Which Python framework is best for an AI agent API?

The concepts here matter more than the framework. Choose one your team knows and that fits your needs, including its support for asynchronous work and streaming if you need them. Check its current documentation for specifics.

Do I need a job queue for an AI agent?

Not always. A short, safe task can run inside the request. A durable queue becomes more useful when runs are long, have side effects or must survive restarts and dropped connections.

Should an agent API use streaming or polling?

Either can work, and many APIs support both. Polling a run's status is a simple starting point, and streaming gives a more responsive interface. Storing events with sequence numbers lets clients reconnect in either case.

How do I test an agent backend without calling a model?

Put the model behind your own small interface and use a scripted fake in tests. That makes failures, tool calls, cancellations and repeated requests repeatable, and a few separate tests can cover the real integration.