Designing a ReAct Loop from Scratch in FastAPI or Express
Expose a framework-free ReAct loop as an HTTP API: accept a user goal, run a bounded model-and-tool cycle, return the final answer plus stop metadata. This page is a recipe for FastAPI (Python) with an Express (TypeScript) twin for teams on Node.
Summary
Build a POST /v1/agent/runs endpoint that owns max turns, tool dispatch, and structured stop reasons. Keep tools pure, the model client swappable, and the HTTP layer free of business side effects.
Recipe
- Define a run request schema:
input, optionalthread_id, budgets (max_turns,timeout_s). - Define a run response schema:
output,stop_reason,turns,run_id, optional trace summary. - Implement pure tools + a name→callable dispatcher with JSON Schema specs for the model.
- Implement
run_react_loop(messages, tools, budgets) -> RunResultwith no HTTP imports. - Wrap the loop in FastAPI or Express; map timeouts and cancels to stop reasons.
- Add request ids, structured logs, and per-tool timeouts.
- Deny unknown tools; return tool errors as observations, not 500s, when recoverable.
- Add authn at the edge; pass a reduced auth context into tools that need it.
Working Example
Python (FastAPI)
# pip install fastapi uvicorn openai pydantic
import json
import os
import time
import uuid
from typing import Any, Callable, Literal
from fastapi import FastAPI, Header
from openai import OpenAI
from pydantic import BaseModel, Field
client = OpenAI(
api_key=os.environ["OPENAI_API_KEY"],
base_url=os.environ.get("OPENAI_BASE_URL") or None,
)
MODEL = os.environ.get("AGENT_MODEL", "gpt-4.1-mini") # verify at build
def get_weather(city: str) -> dict[str, Any]:
return {"city": city, "cond": "demo-sunny", "c": 22}
DISPATCH: dict[str, Callable[..., Any]] = {"get_weather": get_weather}
TOOLS_SPEC = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Demo weather for a city.",
"parameters": {
"type": "object",
"properties": {"city": {"type": "string"}},
"required": ["city"],
},
},
}
]
StopReason = Literal["final_answer", "max_turns", "timeout", "model_error"]
class RunRequest(BaseModel):
input: str
max_turns: int = Field(default=5, ge=1, le=20)
timeout_s: float = Field(default=30.0, ge=1, le=120)
class RunResponse(BaseModel):
run_id: str
output: str | None
stop_reason: StopReason
turns: int
ok: bool
def run_react_loop(user_text: str, max_turns: int, deadline: float) -> RunResponse:
run_id = str(uuid.uuid4())
messages: list[dict[str, Any]] = [
{"role": "system", "content": "Use tools when helpful. Be concise."},
{"role": "user", "content": user_text},
]
try:
for turn in range(1, max_turns + 1):
if time.monotonic() > deadline:
return RunResponse(
run_id=run_id, output=None, stop_reason="timeout",
turns=turn - 1, ok=False,
)
resp = client.chat.completions.create(
model=MODEL, messages=messages, tools=TOOLS_SPEC, tool_choice="auto",
)
msg = resp.choices[0].message
assistant: dict[str, Any] = {"role": "assistant", "content": msg.content or ""}
if msg.tool_calls:
assistant["tool_calls"] = [
{
"id": tc.id,
"type": "function",
"function": {
"name": tc.function.name,
"arguments": tc.function.arguments,
},
}
for tc in msg.tool_calls
]
messages.append(assistant)
if not msg.tool_calls:
return RunResponse(
run_id=run_id, output=msg.content or "",
stop_reason="final_answer", turns=turn, ok=True,
)
for tc in msg.tool_calls:
name = tc.function.name
try:
args = json.loads(tc.function.arguments or "{}")
fn = DISPATCH[name]
content = json.dumps(fn(**args))
except Exception as exc:
content = json.dumps({"error": str(exc), "tool": name})
messages.append(
{"role": "tool", "tool_call_id": tc.id, "content": content}
)
return RunResponse(
run_id=run_id, output=None, stop_reason="max_turns",
turns=max_turns, ok=False,
)
except Exception:
return RunResponse(
run_id=run_id, output=None, stop_reason="model_error", turns=0, ok=False,
)
app = FastAPI()
@app.post("/v1/agent/runs", response_model=RunResponse)
def create_run(
body: RunRequest,
x_request_id: str | None = Header(default=None),
) -> RunResponse:
# x_request_id: log/correlate in production
deadline = time.monotonic() + body.timeout_s
return run_react_loop(body.input, body.max_turns, deadline)TypeScript (Express sketch)
// npm i express openai zod
import express from "express";
import OpenAI from "openai";
import { z } from "zod";
const client = new OpenAI({
apiKey: process.env.OPENAI_API_KEY,
baseURL: process.env.OPENAI_BASE_URL,
});
const MODEL = process.env.AGENT_MODEL ?? "gpt-4.1-mini"; // verify at build
const RunRequest = z.object({
input: z.string().min(1),
max_turns: z.number().int().min(1).max(20).default(5),
timeout_s: z.number().min(1).max(120).default(30),
});
const tools = [
{
type: "function" as const,
function: {
name: "get_weather",
description: "Demo weather for a city.",
parameters: {
type: "object",
properties: { city: { type: "string" } },
required: ["city"],
},
},
},
];
const dispatch: Record<string, (args: any) => unknown> = {
get_weather: ({ city }) => ({ city, cond: "demo-sunny", c: 22 }),
};
const app = express();
app.use(express.json());
app.post("/v1/agent/runs", async (req, res) => {
const parsed = RunRequest.safeParse(req.body);
if (!parsed.success) return res.status(400).json(parsed.error.flatten());
const { input, max_turns, timeout_s } = parsed.data;
const deadline = Date.now() + timeout_s * 1000;
const run_id = crypto.randomUUID();
const messages: OpenAI.Chat.ChatCompletionMessageParam[] = [
{ role: "system", content: "Use tools when helpful. Be concise." },
{ role: "user", content: input },
];
try {
for (let turn = 1; turn <= max_turns; turn++) {
if (Date.now() > deadline) {
return res.json({
run_id, output: null, stop_reason: "timeout", turns: turn - 1, ok: false,
});
}
const resp = await client.chat.completions.create({
model: MODEL, messages, tools, tool_choice: "auto",
});
const msg = resp.choices[0].message;
messages.push(msg);
if (!msg.tool_calls?.length) {
return res.json({
run_id, output: msg.content ?? "", stop_reason: "final_answer",
turns: turn, ok: true,
});
}
for (const tc of msg.tool_calls) {
const name = tc.function.name;
let content: string;
try {
const args = JSON.parse(tc.function.arguments || "{}");
content = JSON.stringify(dispatch[name](args));
} catch (e) {
content = JSON.stringify({ error: String(e), tool: name });
}
messages.push({ role: "tool", tool_call_id: tc.id, content });
}
}
return res.json({
run_id, output: null, stop_reason: "max_turns", turns: max_turns, ok: false,
});
} catch {
return res.json({
run_id, output: null, stop_reason: "model_error", turns: 0, ok: false,
});
}
});- Loop logic is independent of the web framework; HTTP only validates and returns.
- Express and FastAPI differ in validation libraries, not in ReAct semantics.
- Prefer async workers for long runs; keep the request path for short interactive agents.
Deep Dive
Why API-first custom loops
Putting the loop behind HTTP forces contracts:
- Stable request/response schemas for web and mobile clients
- Auth at the edge instead of inside tool code
- Horizontal scale via normal service patterns
- Easier dual implementation (Python worker + TS BFF) sharing the same JSON shapes
Frameworks can also sit behind HTTP. The custom path is justified when the loop stays small and you want zero graph runtime in-process.
Separation of concerns
| Layer | Owns | Must not own |
|---|---|---|
| HTTP | Auth, validation, status codes | Tool side effects |
| Loop | Turns, tool dispatch, stop reasons | SQL schemas, UI markup |
| Tools | Domain actions | Model SDK types |
| Model client | Provider I/O | Product authorization |
If Express handlers call SQL and the model in one function, you will not be able to eval or reuse the loop.
Streaming
Interactive UIs often need token or event streams:
- SSE from FastAPI/Express with events:
assistant.delta,tool.started,tool.finished,run.completed. - Keep the same stop reasons on the final event.
- Do not stream raw secrets from tool args.
Vercel AI SDK and similar clients can consume streams while your custom runtime remains the source of truth for tools.
Concurrency and idempotency
- Use
run_idand optionalIdempotency-Keyfor clients that retry POSTs. - Do not assume one process; if tools are not idempotent, document that clearly.
- For multi-instance breakers and rate limits, shared Redis state beats in-process counters.
Auth context into tools
Pass an explicit context object:
@dataclass
class AuthContext:
user_id: str
roles: frozenset[str]
tenant_id: strTools that mutate state take ctx as a host-injected argument, not as model-supplied JSON.
Never let the model choose user_id for privileged actions.
When FastAPI vs Express
| Choose | When |
|---|---|
| FastAPI | Python ML/data stack, Pydantic-heavy schemas, shared Python tools |
| Express (or Fastify) | TS monorepo, Node ops fluency, edge/BFF colocation |
| Either + queue worker | Runs longer than HTTP timeouts |
The ReAct design is portable; language is a team fit choice.
Gotchas
- Blocking the event loop with sync HTTP tools in async FastAPI without a threadpool.
- No deadline checks inside the loop - only a reverse proxy timeout that drops the client while the agent keeps spending.
- Returning 500 on tool errors instead of packing errors into tool messages for model recovery.
- Putting API keys in tool results or logs.
- Allowing the model to pass tenant or user ids that override auth context.
- Giant tool catalogs on every request; filter by route or role.
- Assuming Chat Completions tool wire format forever - verify shapes when switching to Responses API or other providers.
- Skipping max body size and prompt size limits on public endpoints.
Alternatives
| Approach | When | Trade-off |
|---|---|---|
| Custom FastAPI/Express loop (this page) | Simple agents, full control | You own reliability features |
| OpenAI Agents SDK as library in the service | Want handoffs/guardrails fast on OpenAI | Provider coupling |
| LangGraph / MS Agent Framework behind HTTP | Durable graphs, complex HITL | Heavier dependency |
| Serverless one-shot function | Rare short runs | Cold starts; harder long loops |
| Queue + worker | Minutes-long jobs | Async UX complexity |
FAQs
Should the loop live in the web process or a worker?
Short interactive runs can live in the web process with tight timeouts. Anything that regularly exceeds ~30-60s belongs in a worker with a run status API.
How do I test the endpoint without burning tokens?
Unit-test run_react_loop with a fake model client that returns scripted tool_calls. Integration-test the HTTP schema with that fake. Reserve live-model tests for a small CI job.
Can I support multiple models per request?
Yes via a model field validated against an allowlist. Keep routing policy in your code, not in free-form client strings without checks.
How does this relate to conceptual ReAct?
This is ReAct as a control loop on an API. Thoughts may be latent in tool choice rather than verbose "Thought:" prose. See the architectures section for the pattern itself.
Do I need WebSockets?
Not for most agents. SSE or chunked HTTP streaming is enough for tokens and tool events. WebSockets help when clients push mid-run cancels or multimodal streams.
Where do plugins fit?
Register tools at app startup from a plugin registry; the loop only sees TOOLS_SPEC and DISPATCH. See the plugin architecture article in this section.
How do I cancel a run?
For in-request loops, abort when the client disconnects if your stack signals it. For workers, store a cancel flag checked each turn.
Is Express production-ready for agents?
Yes if you add the same budgets, auth, and observability you would in Python. Language is not the reliability bottleneck; missing stop conditions are.
Should I expose tool traces to end users?
Usually no raw traces. Offer user-safe progress events and keep full traces for operators with redaction.
When should I stop custom and adopt a framework?
When you need durable resume, complex joins, or you are reinventing a graph engine. Use the maintenance cheatsheet in this section.
Related
- Custom Agent Runtimes Basics - loop and dispatcher without HTTP
- Plugin Architecture for Tools in a Custom Runtime - registration system
- Building an Eval Harness for a Custom Runtime - score the API behavior
- Circuit Breakers and Observability Without a Third-Party Framework - harden the host
- ReAct: Reasoning and Acting in an Interleaved Loop - pattern foundations
- Why Some Teams Build a Custom Agent Runtime Instead of Adopting a Framework - motivation
Stack versions: Pins from the category manifest (verify at build): OpenRouter (~315+ models, July 2026 pricing/fees); LangGraph 1.0+; CrewAI 1.14+; Microsoft Agent Framework 1.0; Vercel AI SDK 6; Pydantic AI (latest); LlamaIndex (latest); OpenAI Agents SDK (latest + MCP); MCP (Linux Foundation governance); A2A (HTTP+SSE+JSON-RPC 2.0); Solana
@solana/web3.js+@solana/spl-token.