AI Agents & Loop Engineering

Part IV / OpenAPI & AI Agents

Why an AI Agent Belongs in a Backend Chapter

An LLM call is a function that turns text into text. An AI agent is a backend workflow that gives the model a controlled way to inspect state, select an action, call a tool, observe the result, and decide what to do next.

That distinction matters. A prompt can answer a question. An agent can complete a bounded task such as: inspect an order, check inventory, ask for approval, reserve stock, and return a result. The model supplies probabilistic decisions; the backend supplies identity, permissions, durable state, tools, limits, and auditability.

The useful mental model is not “the model runs the business.” It is:

The backend owns the state machine. The model proposes the next transition.

A production agent therefore has two contracts:

1. The OpenAI model contract: messages, tool schemas, structured output, model limits, and refusal behavior.
2. The backend contract: authenticated input, authorized tools, idempotency, timeouts, retries, persistence, and a stable API response.

The OpenAI specification describes how the model receives messages and tools. It does not grant the model access to your database or authorize a payment. Your backend remains the policy enforcement point.

The Smallest Useful Agent

We will build a support agent that can look up an order and return a grounded answer. The model may choose the get_order tool; Python and LangGraph execute the tool and feed its result back to the model.

Install the dependencies in a service virtual environment:

pip install langgraph langchain-openai pydantic
export OPENAI_API_KEY="..."

The tool is a typed backend function, not an arbitrary Python function exposed to the model. Its description and schema are sent to the model, while the implementation keeps the real authorization and data access rules.

from typing import Annotated, Literal
from typing_extensions import TypedDict

from langchain_core.messages import AnyMessage, HumanMessage, SystemMessage, ToolMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages


class AgentState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    user_id: str
    tenant_id: str
    steps: int


@tool
def get_order(order_id: str) -> dict:
    """Read an order visible to the authenticated tenant."""
    # Production code must query through a repository that applies tenant scope.
    return {"order_id": order_id, "status": "shipped", "eta": "2026-09-21"}


model = ChatOpenAI(model="gpt-4.1-mini", temperature=0)
model_with_tools = model.bind_tools([get_order])


def call_model(state: AgentState) -> dict:
    response = model_with_tools.invoke([
        SystemMessage(content=(
            "You are a support agent. Use tools for facts. "
            "Never invent order status. Keep the answer concise."
        )),
        *state["messages"],
    ])
    return {"messages": [response], "steps": state["steps"] + 1}


def run_tools(state: AgentState) -> dict:
    last = state["messages"][-1]
    tool_messages = []
    for call in last.tool_calls:
        if call["name"] == "get_order":
            result = get_order.invoke(call["args"])
            tool_messages.append(ToolMessage(
                content=str(result),
                tool_call_id=call["id"],
            ))
    return {"messages": tool_messages}


def route_after_model(state: AgentState) -> Literal["tools", "done"]:
    last = state["messages"][-1]
    if getattr(last, "tool_calls", None) and state["steps"] < 4:
        return "tools"
    return "done"


graph = StateGraph(AgentState)
graph.add_node("model", call_model)
graph.add_node("tools", run_tools)
graph.add_edge(START, "model")
graph.add_conditional_edges(
    "model",
    route_after_model,
    {"tools": "tools", "done": END},
)
graph.add_edge("tools", "model")
agent = graph.compile()


result = agent.invoke({
    "messages": [HumanMessage(content="Where is order A-42?")],
    "user_id": "user-7",
    "tenant_id": "tenant-3",
    "steps": 0,
})
print(result["messages"][-1].content)

The same loop in Go with an OpenAI-compatible API

The Go version uses the wire contract directly: POST /chat/completions, a messages array, and a tools array containing JSON Schema. That makes the example portable across OpenAI-compatible gateways and providers. The complete HTTP server, including POST /agent-runs, is in 26.AI-Agents-and-Loop-Engineering/code/go/agent_server.go; its API contract is in code/openapi.yaml.

type ChatMessage struct {
    Role       string     `json:"role"`
    Content    string     `json:"content,omitempty"`
    ToolCallID string     `json:"tool_call_id,omitempty"`
    ToolCalls  []ToolCall `json:"tool_calls,omitempty"`
}

type ToolCall struct {
    ID       string       `json:"id"`
    Type     string       `json:"type"`
    Function ToolFunction `json:"function"`
}

type ToolFunction struct {
    Name      string `json:"name"`
    Arguments string `json:"arguments"`
}

type CompletionRequest struct {
    Model    string        `json:"model"`
    Messages []ChatMessage `json:"messages"`
    Tools    []Tool        `json:"tools,omitempty"`
}

func (a *Agent) Run(ctx context.Context, prompt string) (string, error) {
    messages := []ChatMessage{
        {Role: "system", Content: "Use tools for facts; never invent order status."},
        {Role: "user", Content: prompt},
    }

    for step := 0; step < 4; step++ {
        response, err := a.complete(ctx, messages)
        if err != nil {
            return "", err
        }
        assistant := response.Choices[0].Message
        messages = append(messages, assistant)
        if len(assistant.ToolCalls) == 0 {
            return assistant.Content, nil
        }

        for _, call := range assistant.ToolCalls {
            result, err := a.executeAllowlistedTool(call)
            if err != nil {
                result = map[string]any{"error": err.Error()}
            }
            encoded, _ := json.Marshal(result)
            messages = append(messages, ChatMessage{
                Role: "tool", ToolCallID: call.ID, Content: string(encoded),
            })
        }
    }
    return "", errors.New("agent step limit exceeded")
}

The important detail is that executeAllowlistedTool is ordinary Go code. It parses the model’s JSON arguments, checks identity and tenant scope, performs the repository operation, and returns a bounded result. The model never gets a database handle or a generic function runner.

The graph is explicit. model proposes either a final response or a tool call. tools executes only the calls that the backend recognizes. route_after_model decides whether to continue or finish, and the step limit prevents an accidental infinite loop.

Loop Engineering: Inspect, Plan, Act, Observe, Evaluate

A useful agent loop is an engineered control loop:

1. Inspect: load the request, trusted identity, prior state, available tools, and remaining budget.
2. Plan: ask the model for the next action or a final answer. Keep the plan bounded; do not require hidden chain-of-thought.
3. Act: validate and execute one allowlisted tool call.
4. Observe: append a structured tool result, not an unbounded log dump.
5. Evaluate: check whether the goal is complete, whether the result is valid, and whether another step is allowed.
6. Stop: return a result, request human approval, or return a typed failure.

This is loop engineering: the quality of an agent comes less from adding more autonomy and more from designing the loop’s state, transitions, stop conditions, and failure behavior.

What can go wrong in the loop?

• The model calls a tool repeatedly because the result is ambiguous.
• A tool returns a huge payload and consumes the context window.
• A retry repeats a payment, email, or mutation.
• A prompt injection inside retrieved content tries to change the agent’s instructions.
• The model declares success without evidence.
• A timeout leaves the run halfway through a multi-step task.

Design for each failure explicitly. Use a maximum step count and wall-clock deadline, summarize large observations, attach idempotency keys to mutations, treat retrieved text as untrusted data, and require a validator or evidence field before declaring success.

Tool Use: Turning Backend Capabilities into Safe Actions

A tool should have a narrow name, a typed input, a typed output, and a clear side-effect policy. Separate read tools from write tools. Reads can often run automatically; writes should be authorized by the backend and may require human approval.

from pydantic import BaseModel, Field


class RefundRequest(BaseModel):
    order_id: str = Field(pattern=r"^[A-Z]-[0-9]+$")
    reason: str = Field(min_length=5, max_length=300)
    idempotency_key: str


@tool
def request_refund(request: RefundRequest) -> dict:
    """Create a refund request; never directly issue money."""
    # The real service re-checks auth, tenant, order ownership and state here.
    # The idempotency key makes a retry safe.
    return {
        "status": "pending_approval",
        "order_id": request.order_id,
        "approval_required": True,
    }

The model can propose a refund, but it cannot bypass the approval state. This is a general pattern:

Do not let tool descriptions become your authorization system. Descriptions guide the model; backend code enforces the rules.

Sequential and Non-Sequential Agent Workflows

A sequential agent follows a predictable pipeline. It is easier to test and is preferable when every stage must happen in order.

from langgraph.graph import END, START, StateGraph
from typing_extensions import TypedDict


class ReportState(TypedDict):
    request: str
    facts: str
    draft: str
    answer: str


def retrieve(state: ReportState):
    return {"facts": "facts retrieved from the authorized knowledge service"}


def draft(state: ReportState):
    return {"draft": f"Draft based on {state['facts']} for {state['request']}"}


def verify(state: ReportState):
    return {"answer": state["draft"] + " [verified against retrieved facts]"}


workflow = StateGraph(ReportState)
workflow.add_node("retrieve", retrieve)
workflow.add_node("draft", draft)
workflow.add_node("verify", verify)
workflow.add_edge(START, "retrieve")
workflow.add_edge("retrieve", "draft")
workflow.add_edge("draft", "verify")
workflow.add_edge("verify", END)
sequential_agent = workflow.compile()

A non-sequential workflow branches when the state makes different paths useful. The model may classify the request, but the graph owns the routing. For example, a billing question can go to billing retrieval while a technical question goes to documentation retrieval; both paths can converge on a verifier.

from typing import Literal
from typing_extensions import TypedDict
from langgraph.graph import END, START, StateGraph


class RequestState(TypedDict):
    request: str
    category: str
    context: str
    answer: str


def classify(state: RequestState):
    category = "billing" if "invoice" in state["request"].lower() else "technical"
    return {"category": category}


def route(state: RequestState) -> Literal["billing", "technical"]:
    return state["category"] if state["category"] in {"billing", "technical"} else "technical"


def billing(state: RequestState):
    return {"context": "billing records retrieved through the billing service"}


def technical(state: RequestState):
    return {"context": "versioned technical documentation retrieved"}


def answer(state: RequestState):
    return {"answer": f"Answer using: {state['context']}"}


workflow = StateGraph(RequestState)
for name, fn in [("classify", classify), ("billing", billing),
                 ("technical", technical), ("answer", answer)]:
    workflow.add_node(name, fn)
workflow.add_edge(START, "classify")
workflow.add_conditional_edges(
    "classify", route, {"billing": "billing", "technical": "technical"}
)
workflow.add_edge("billing", "answer")
workflow.add_edge("technical", "answer")
workflow.add_edge("answer", END)
non_sequential_agent = workflow.compile()

Non-sequential does not mean uncontrolled. Every branch should have a known destination, a fallback, and a test. If the decision is deterministic, use ordinary code instead of paying for an LLM call.

Replacing a Series of LLM Calls with One Agent

A common first implementation is a fixed chain:

classify -> retrieve -> summarize -> write -> review

That chain is appropriate when every task needs every stage. It becomes wasteful when simple requests still pay for all five calls, or when one failed stage forces the entire chain to restart.

An agent can replace the chain when the next action depends on the current state:

inspect -> choose the smallest useful action -> observe -> evaluate -> repeat

For example, a support agent may:

• answer a general question with zero tools;
• call one order lookup for a tracking question;
• call an order lookup and then a policy lookup for a refund question;
• stop and request approval before creating a refund.

This reduces unnecessary calls, but it introduces model variability. The trade-off is not “agents are always better.” Use a fixed workflow when the sequence is known, an agent loop when the next step genuinely depends on evidence, and a hybrid when predictable stages contain an agentic subtask.

Using an Agent from a Backend API

An HTTP handler should not contain the whole loop. It should authenticate the caller, create a run, enqueue or invoke the agent service, and return a stable response. For work that can exceed the request timeout, return 202 Accepted and let the client poll or subscribe to run events.

from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel, Field

router = APIRouter()


class AgentRequest(BaseModel):
    prompt: str = Field(min_length=1, max_length=4000)
    idempotency_key: str = Field(min_length=8, max_length=100)


@router.post("/agent-runs", status_code=202)
async def create_agent_run(request: AgentRequest, user=Depends(current_user)):
    if not user.can_use_agent:
        raise HTTPException(status_code=403, detail="agent access is disabled")

    run = await run_repository.create_if_absent(
        user_id=user.id,
        tenant_id=user.tenant_id,
        idempotency_key=request.idempotency_key,
        prompt=request.prompt,
    )
    await queue.publish("agent.run.requested", {"run_id": run.id})
    return {"run_id": run.id, "status": run.status, "poll_url": f"/agent-runs/{run.id}"}

A worker can resume the graph from a checkpoint. Persist at least the run id, user and tenant ids, current state, status, step count, model usage, tool calls, errors, and timestamps. Do not persist secrets or raw sensitive prompts without a retention policy.

async def handle_agent_run(run_id: str) -> None:
    run = await run_repository.get(run_id)
    if run.status in {"succeeded", "failed", "cancelled"}:
        return

    await run_repository.mark_running(run_id)
    try:
        result = await asyncio.wait_for(
            agent.ainvoke({
                "messages": [HumanMessage(content=run.prompt)],
                "user_id": run.user_id,
                "tenant_id": run.tenant_id,
                "steps": run.steps,
            }, config={"configurable": {"thread_id": run.id}}),
            timeout=45,
        )
        await run_repository.mark_succeeded(run_id, result=result)
    except TimeoutError:
        await run_repository.mark_failed(run_id, code="agent_deadline_exceeded")
    except Exception:
        await run_repository.mark_failed(run_id, code="agent_execution_failed")
        raise

The queue gives the backend normal operational properties: retries, visibility timeouts, dead-letter handling, concurrency limits, and a place to cancel work. The agent is a worker workload, not a privileged replacement for the backend.

Production Checklist

Before shipping an agent, answer these questions with code and tests:

• Identity: Which authenticated user, tenant, and permissions are attached to every tool call?
• Scope: What is the maximum number of model calls, tool calls, tokens, and wall-clock time per run?
• State: Can a run resume after a worker crash without repeating a side effect?
• Idempotency: Does every mutation have a key and a deduplication record?
• Validation: Are model outputs parsed into typed schemas before use?
• Prompt injection: Is retrieved or user-provided text treated as data rather than instructions?
• Approval: Which actions require a human or a second service to approve them?
• Observability: Can you see latency, cost, tool errors, retries, refusal rate, and final outcome per run?
• Evaluation: Do fixtures test correct tool choice, wrong tool arguments, missing data, timeouts, and adversarial content?
• Fallback: What does the API return when the model, tool, queue, or checkpoint store is unavailable?

A good first release has fewer tools, smaller context, stricter budgets, and stronger traces than its demo. Add autonomy only when measurements show that the extra branch improves the task without increasing unacceptable risk.

The Chapter in One Breath

An AI agent is a controlled state machine around an LLM, not an LLM with unrestricted access to your system. LangGraph makes the state and transitions explicit: model nodes propose actions, tool nodes execute validated capabilities, and conditional edges decide whether the run continues or stops. Loop engineering makes the process dependable by bounding steps, deadlines, retries, side effects, approvals, and evidence. In a backend, expose agents behind ordinary authenticated APIs and queues, persist their runs, observe them like any other production workload, and keep authorization in deterministic code.