Serving over HTTP

loomflow.serve.create_app(agent) wraps an Agent in a plain ASGI-3 application — no FastAPI or Starlette dependency. Because it speaks the raw ASGI protocol, it runs under any ASGI server (uvicorn, hypercorn, daphne) and mounts inside an existing FastAPI/Starlette app.

# my_service.py
from loomflow import Agent
from loomflow.serve import create_app
 
agent = Agent("You are a support assistant.", model="gpt-4.1-mini")
app = create_app(agent)

pip install 'loomflow[serve]'          # adds uvicorn
uvicorn my_service:app --port 8000

The app holds one shared Agent — multi-tenant by design. Each request awaits agent.run(...) with the user_id from the request body as the tenancy partition. No worker pool, no long-lived state: serverless-friendly.

Routes

POST /run

curl -s localhost:8000/run -d '{
  "prompt": "My card was charged twice.",
  "user_id": "alice",
  "session_id": "sess-42"
}'

{
  "output": "…",
  "session_id": "sess-42",
  "usage": {
    "input_tokens": 412, "cached_input_tokens": 0,
    "cache_write_tokens": 0, "output_tokens": 96,
    "total_tokens": 508, "cost_usd": 0.00031, "turns": 1
  },
  "interrupted": false,
  "interruption_reason": null
}

POST /stream — Server-Sent Events

Same body as /run; the response is one SSE frame per agent Event (the event name is the event kind), terminated by an event: done frame carrying the final result payload. A client disconnect cancels the underlying run — the stream generator is closed, which triggers the agent’s break-to-cancel contract, so abandoned tabs don’t keep burning tokens.

POST /resume

Resumes a prior session — from_checkpoint selects a durable checkpoint ("latest" or a checkpoint id) when the agent has checkpointing enabled.

Error shapes

Errors are structured JSON, never tracebacks: missing/empty prompt → 400, bodies over 1 MB → 413, unknown route → 404, wrong method → 405 (with an Allow header), agent exception → 500 {"error": "<ExceptionType>", "message": "..."}.

Multi-tenant posture: require_user_id

app = create_app(agent, require_user_id=True)

Rejects any /run / /stream / /resume request whose body lacks a user_id with a 422 — the deployment posture where anonymous runs must not share the default partition.

Mounting inside FastAPI

Any ASGI app mounts, so composing with an existing service is one line:

from fastapi import FastAPI
from loomflow.serve import create_app
 
api = FastAPI()
api.mount("/agent", create_app(agent))   # POST /agent/run, /agent/stream, ...

The loom CLI

Installing loomflow adds a loom console script (stdlib argparse, no click/typer dependency):

# One-shot prompt through an Agent (default model: echo — no API key)
loom run "summarise this" --model openai:gpt-4o
loom run "hello" --stream --user-id alice --session-id s1 --tone concise
 
# Serve an Agent over HTTP (needs the serve extra for uvicorn)
loom serve my_service:agent --host 0.0.0.0 --port 8000
 
# Run a JSONL dataset through loomflow.eval; exit 1 on unmet thresholds
loom eval cases.jsonl --agent my_service:agent --threshold exact_match=0.9
 
# Print the version
loom version

loom eval prints a JSON report and exits non-zero when a case errored or a threshold is unmet — see Evals for the full harness (richer metrics, LLM judge, online sampling).

Exposing the agent to other agents

HTTP serving is for your own clients. To expose the agent to other frameworks’ agents over a standard protocol, serve it via A2A instead — same pure-ASGI posture, plus a discovery card and JSON-RPC methods.