Agent Server¶

AgentServer is the reference way to run any Tulip Agent as a service — drop in an Agent (a refund bot, a deploy operator, an access-grant approver, or a SOC triage agent), and get a FastAPI app with /invoke, /stream, and per-thread management out of the box. It re-emits the same event stream the Python API exposes as Server-Sent Events, gated by optional bearer-token auth.

Single shared key — single principal

AgentServer takes one api_key. Every authenticated request presents that same token, so all callers hash to the same principal and share one thread namespace. The bearer check (a 401 on /invoke, /stream and /threads/{tid}) is what stops a caller without the key; the principal prefix on thread IDs only keeps thread IDs from colliding between deployments with different keys that share a checkpointer — it does not give you per-analyst or per-tenant isolation behind a shared key. For real multi-tenant isolation, run one keyed AgentServer instance per tenant.

from tulip.server import AgentServer

server = AgentServer(
    agent=refund_agent,
    title="Refund desk",
    api_key="…",                       # bearer-token auth
)

if __name__ == "__main__":
    server.run(host="0.0.0.0", port=8080)

When to use it¶

Situation	Use AgentServer?
Exposing an agent to a web/mobile UI (a support console, an ops dashboard, a SOC console)	yes — SSE plus per-thread persistence is what you want
Internal one-off task, single Python script	no — call `agent.run_sync(...)` directly
Embedding an agent in your own FastAPI service	possible, but consider importing `AgentServer.app` and mounting it under your existing app
Scaling out across many workers, one thread resumable on any of them	yes, with an `S3Backend` (or another shared checkpointer) so every worker sees the same thread state

Getting started¶

1. Wrap the refund bot¶

from tulip.agent import Agent
from tulip.core.termination import ToolCalled, ConfidenceMet, MaxIterations
from tulip.memory.backends.file import FileCheckpointer
from tulip.server import AgentServer

agent = Agent(
    model="openai:gpt-4o-mini",
    tools=[lookup_order, check_refund_policy, issue_refund],
    system_prompt="You are a refund agent. Cite order evidence; escalate without it.",
    reflexion=True,
    checkpointer=FileCheckpointer(base_dir="./cases"),
    termination=(ToolCalled("issue_refund") & ConfidenceMet(0.9)) | MaxIterations(8),
)

server = AgentServer(agent=agent, api_key="…")
server.run(host="0.0.0.0", port=8080)

The SOC variant is one swap: tools=security_toolset(siem=True, edr=True, threat_intel=True) (from tulip_security, in the tulip-agents-security package) and a ToolCalled("isolate_host") terminal.

2. Call `/invoke` (one-shot)¶

curl -sS -X POST http://localhost:8080/invoke \
  -H "Authorization: Bearer $TULIP_SERVER_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"prompt": "Refund the duplicate charge on ord-4821.", "thread_id": "ord-4821"}'

Returns the full AgentResult JSON in one response. Use this for scheduled jobs, workflow-engine steps, and anything that doesn't render incrementally.

3. Call `/stream` (Server-Sent Events)¶

/stream is a POST endpoint: the prompt comes from the JSON body and the token from the Authorization header. The browser EventSource API only issues GET requests, so it can't drive this route — use fetch() + a ReadableStream reader and parse the data: frames yourself. Each frame is data: {json}\n\n; the JSON carries a type field (think, tool_start, tool_complete, done, error), and the stream ends with a literal data: [DONE] frame. The route does not emit named SSE events.

const res = await fetch("/stream", {
  method: "POST",
  headers: {
    "Content-Type": "application/json",
    "Authorization": "Bearer " + token,
  },
  body: JSON.stringify({ prompt: "Refund the duplicate charge on ord-4821.", thread_id: "ord-4821" }),
});

const reader = res.body.getReader();
const decoder = new TextDecoder();
let buf = "";
for (;;) {
  const { value, done } = await reader.read();
  if (done) break;
  buf += decoder.decode(value, { stream: true });
  const frames = buf.split("\n\n");
  buf = frames.pop();                               // keep the partial frame
  for (const frame of frames) {
    const line = frame.replace(/^data: /, "");
    if (line === "[DONE]") continue;
    const evt = JSON.parse(line);
    if (evt.type === "tool_start") {
      status.innerText = `running ${evt.tool}`;     // lookup_order, issue_refund…
    } else if (evt.type === "done") {
      output.innerText = evt.message;
    }
  }
}

The data: payload is the JSON projection of each agent event (type plus its fields), derived from the same async for event in agent.run(...) stream the Python API exposes.

Serving a graph¶

AgentServer only needs the agent contract — an async run(prompt, *, thread_id=None, metadata=None) that yields events. A compiled StateGraph doesn't speak that directly, so wrap it in GraphRunnable:

from tulip.multiagent.graph import StateGraph, START, END
from tulip.server import AgentServer, GraphRunnable


async def draft(state):
    return {"answer": f"Draft reply for: {state['prompt']}"}


graph = StateGraph()
graph.add_node("draft", draft)
graph.add_edge(START, "draft")
graph.add_edge("draft", END)

server = AgentServer(
    agent=GraphRunnable(graph.compile(), input_key="prompt", output_key="answer"),
    api_key="…",
)
server.run(host="0.0.0.0", port=8080)

The request's prompt lands in the graph's initial state under input_key. Each streamed graph event goes out on /stream as a think frame, and the final state's output_key value becomes the message of the done frame and of the /invoke response (with no output_key, the whole final state is stringified). thread_id and metadata are accepted but not passed to the graph, so the case persistence flow and the /threads/{tid} routes don't apply: those routes read agent.config.checkpointer, and GraphRunnable has no config, so they fail with a 500 rather than the no-checkpointer 404. The same adapter slots into tulip.a2a.A2AServer.

Endpoints¶

Path	Method	Body	Returns
`/invoke`	POST	`{"prompt": "...", "thread_id": "..."}`	`InvokeResponse` JSON (`message`, `success`, `stop_reason`, `iterations`, `tool_calls`, `duration_ms`)
`/stream`	POST	same	`text/event-stream` SSE of `data: {json}` frames
`/health`	GET	—	liveness probe (200 OK)
`/threads/{tid}`	GET	—	thread state (requires checkpointer)
`/threads/{tid}`	DELETE	—	drop a thread

There is no /resume route on the reference AgentServer — see Human-in-the-loop below for how the in-process interrupt/resume flow actually works.

/docs, /redoc, and /openapi.json are only mounted when debug=True in your settings — production deployments don't expose schema by default.

Auth and thread scoping¶

Bearer token (opt-in). Auth is off by default. Pass api_key="..." to the constructor or set TULIP_SERVER_API_KEY to turn it on; then every request must carry Authorization: Bearer <token>, constant-time compared with hmac.compare_digest.
Refuses to start off-loopback. With no api_key (argument or TULIP_SERVER_API_KEY) and no allow_unauthenticated=True, server.run(host=...) raises RuntimeError for any non-loopback host — it does not fall back to 127.0.0.1. run() itself defaults to host="127.0.0.1", so a bare server.run() still starts for local work. The check lives in AgentServer.run() only: serve server.app under your own uvicorn or gunicorn and it is bypassed (the app just logs a warning), so set the key yourself or terminate auth upstream.
Single-principal thread namespacing. The principal is derived server-side from the presented key — sha256(token)[:12], or anon when unauthenticated. Because the server holds one shared api_key, every authenticated caller resolves to the same principal, and thread IDs are prefixed with it (<principal>:<tid>). A caller without the key never reaches a thread — the bearer check rejects it with a 401 first. The prefix keeps thread IDs from colliding between deployments with different keys that share one checkpointer — it does not separate two analysts sharing the same key. For per-tenant isolation, run one keyed AgentServer per tenant.

server = AgentServer(
    agent=agent,
    api_key=os.environ["TULIP_SERVER_API_KEY"],
)

For unauthenticated dev:

server = AgentServer(agent=agent, allow_unauthenticated=True)
server.run(host="127.0.0.1", port=8080)   # never 0.0.0.0

Human-in-the-loop¶

Reads (lookup_order, check_refund_policy, get_deploy_status) can auto-run. Writes that change the world — issue_refund, deploy_service, isolating a host — should pause for a human first. The SDK primitive for this is the in-process interrupt() / ask_user() call: a tool calls it, the agent yields an InterruptEvent and pauses, and the caller continues with agent.resume(response) (see Interrupts).

from tulip.core.interrupt import interrupt

@tool
def request_human_approval(reason: str, action: str) -> str:
    """Pause the run until a human approves the pending action."""
    return interrupt({"reason": reason, "action": action})

AgentServer does not expose this over HTTP yet

There is no PendingApproval exception in the SDK and no /resume endpoint on the reference AgentServer. The interrupt/resume flow above runs in-process. Over /stream, an unexpected tool exception is caught and returned as a sanitized {"type": "error", "error": "internal error", "correlation_id": ...} frame (details land in the server log under that id) — it does not park a resumable case. To drive human-in-the-loop over HTTP today, handle InterruptEvent in your own wrapper around Agent.run / Agent.resume and add your own resume route.

Case persistence¶

If the underlying Agent has a checkpointer, the server honours thread_id in the request body for cross-request continuity. Same key + same thread_id → same thread, same accumulated state.

# Day 1
curl -X POST .../invoke -d '{"prompt":"Open a case for the duplicate charge on ord-4821", "thread_id":"ord-4821"}'
# Day 2 — same thread_id, the case continues
curl -X POST .../invoke -d '{"prompt":"What did we establish so far?", "thread_id":"ord-4821"}'

For multi-worker deployments, swap the checkpointer to one the workers share so a thread written on one worker can be loaded on any other — S3Backend(bucket=..., prefix=...) is the zero-friction path (it implements BaseCheckpointer directly, so pass it straight to the agent); the redis_checkpointer(...) and postgresql_checkpointer(...) factories work too. See Checkpointers.

Deployment¶

The server is plain FastAPI — deploy it however you deploy FastAPI.

Target	Path
Kubernetes / container services	`docker build` and ship; gunicorn-uvicorn workers in front
serverless functions	Mangum-style adapter wrapping `server.app` (the FastAPI app is built on first access)
Compute / VM	`uvicorn myapp:app --workers 4 --port 8080`, where `myapp.py` sets `app = AgentServer(agent=agent, api_key=...).app` at module scope
Anywhere else FastAPI runs	…yes

Each of these serves server.app directly rather than calling server.run(), so the loopback check in AgentServer.run() never runs. Set api_key (or TULIP_SERVER_API_KEY) explicitly, or put an auth-terminating proxy in front and pass allow_unauthenticated=True.

Auth, rate-limiting, and request logging are FastAPI middleware concerns — Tulip does not own them. Add slowapi, prometheus-fastapi-instrumentator, or whatever your platform expects.

Common gotchas¶

Symptom	Likely cause
`RuntimeError: Refusing to bind AgentServer to '0.0.0.0'…`	No `api_key` and no `allow_unauthenticated=True`. Pick one.
Console SSE drops mid-run (~30s)	Reverse-proxy idle timeout. Bump `proxy_read_timeout` in nginx / `idle_timeout` on the LB, or have the agent send heartbeats every ~25s. A long-running tool call is the usual trigger.
Threads don't persist across restarts	`FileCheckpointer` writes to disk in the working directory — ephemeral container filesystems lose state on restart. Mount a volume or move to `S3Backend`.
`/threads/{tid}` 404s for the right tid	Thread IDs are prefixed with the principal — `<principal>:<tid>` is what's stored. The path you pass is your tid; the server prefixes. A thread written under a different key (the key was rotated, or another deployment sharing the checkpointer wrote it), or written in-process by `agent.run(..., thread_id=...)` without the prefix, won't be found. A request with a wrong or missing key gets a 401, not a 404.

Source and notebook¶

notebook_68_agent_server.py — runnable wrapper plus a curl client.
tulip.server — AgentServer, GraphRunnable, InvokeRequest, InvokeResponse.