Agent Server

This notebook deploys an on-call incident triage copilot as an HTTP API. AgentServer wraps any Tulip Agent in a FastAPI app: synchronous invoke, streaming SSE, and persisted incident threads. Auth is a single shared bearer API key (constructor arg or TULIP_SERVER_API_KEY); the presented key derives the checkpoint namespace: thread ids are stored as <principal>:<tid>, where <principal> is the first 12 hex chars of the SHA-256 of the key (anon when no key is configured). That keeps thread ids from colliding between deployments with different keys that share one checkpointer; when a key is configured, a caller that does not present it gets a 401 from the bearer check before the route runs, so no thread is read. It is not per-engineer isolation: the server holds one key, so everyone who can call it shares one principal and one thread namespace. For per-tenant isolation, run one keyed AgentServer instance per tenant — see Agent Server.

Endpoints:

• POST /invoke — synchronous invocation.
• POST /stream — SSE streaming.
• GET /threads/{tid} — load a persisted incident thread.
• DELETE /threads/{tid} — drop a persisted incident thread.
• GET /health — health check.

When to use AgentServer vs A2AServer:

• AgentServer: first-party HTTP API. Persisted threads, single shared-key bearer auth, key-namespaced thread checkpoints. Use when Tulip is the system of record and clients are yours (a PagerDuty webhook or a deploy dashboard).
• A2AServer: cross-framework interop with the A2A message spec. Use when another A2A-speaking agent runtime needs to call your Tulip agent.

Run it:

# Smoke test against a TestClient (no live server, no live model; still needs REDIS_URL):
TULIP_MODEL_PROVIDER=mock python examples/notebook_68_agent_server.py

# Boot a real uvicorn server on http://127.0.0.1:8000:
TULIP_NOTEBOOK_BOOT=1 python examples/notebook_68_agent_server.py

Prerequisites:

• pip install fastapi uvicorn and the Redis extra (pip install "tulip-agents[redis]") — the checkpointer's Redis client is imported lazily, so without it the first checkpoint read or write raises ImportError.
• A Redis instance with REDIS_URL set. Both run modes require it: the notebook checks the variable before it chooses between the TestClient smoke test and the live uvicorn boot, prints what's missing, and exits. The Redis-backed checkpointer is what makes GET /threads/{tid} survive a restart.

Source

The listing starts after the module docstring; the prose above replaces it. One printed hint near the end of example_server() says two engineers with different bearer tokens see different incidents for the same thread_id. That holds only across two keyed AgentServer instances sharing one checkpointer: a single keyed server rejects any other token with a 401 on its invoke, stream and thread routes.

import os
import sys

from config import get_model

from tulip.agent import Agent, AgentConfig
from tulip.memory.backends import RedisBackend, StorageBackendAdapter
from tulip.server import AgentServer


_REQUIRED_ENV = ("REDIS_URL",)


def _missing_env() -> list[str]:
    return [name for name in _REQUIRED_ENV if not os.environ.get(name)]


def _build_checkpointer():
    # The key namespace is ``prefix``; RedisBackend takes **kwargs, so a wrong
    # name is swallowed silently and every notebook shares one keyspace.
    backend = RedisBackend(
        url=os.environ["REDIS_URL"],
        prefix="tulip_notebook_68:",
    )
    # A backend is raw key/value storage; StorageBackendAdapter is what gives it
    # the Checkpointer contract the agent loop expects.
    return StorageBackendAdapter(backend)


_TRIAGE_PROMPT = (
    "You are an on-call incident triage copilot for an SRE team. Given an "
    "alert summary, say whether it warrants paging the on-call engineer and "
    "why. Be concise."
)


# Smoke test the server with FastAPI's TestClient. No port is bound.


def example_server():
    """Create an incident-triage copilot server with health, invoke, and stream endpoints."""
    print("=== On-Call Incident Triage Copilot Server ===\n")

    model = get_model()

    agent = Agent(
        config=AgentConfig(
            system_prompt=_TRIAGE_PROMPT,
            max_iterations=5,
            model=model,
            # Redis checkpointer so /threads/{id} survives restarts —
            # an incident outlives any one server process.
            checkpointer=_build_checkpointer(),
        )
    )

    server = AgentServer(
        agent=agent,
        title="On-Call Incident Triage Copilot API",
        description="An incident-triage copilot exposed as an HTTP API",
    )

    from fastapi.testclient import TestClient

    client = TestClient(server.app)

    r = client.get("/health")
    print(f"GET /health: {r.json()}")

    # Explicit thread_id (one per incident) so we can read it back through GET /threads.
    r = client.post(
        "/invoke",
        json={
            "prompt": (
                "Alert: checkout-api p99 latency at 4.2s (SLO 800ms) for 6 "
                "minutes after the v3.8.1 rollout. Page on-call?"
            ),
            "thread_id": "inc-4711",
        },
    )
    data = r.json()
    print(f"POST /invoke: {data['message']} (success={data['success']})")

    r = client.post("/stream", json={"prompt": "Name 3 common signs of a bad deploy."})
    print(f"POST /stream: status={r.status_code}")

    r = client.get("/threads/inc-4711")
    if r.status_code == 200:
        thread = r.json()
        print(
            f"GET /threads/inc-4711: iteration={thread['iteration']}, "
            f"messages={len(thread['messages'])}"
        )
    else:
        print(f"GET /threads/inc-4711: status={r.status_code}")

    r = client.get("/threads/inc-never-opened")
    print(f"GET /threads/inc-never-opened: status={r.status_code}")

    # DELETE is idempotent — a second call returns deleted=False.
    r = client.delete("/threads/inc-4711")
    print(f"DELETE /threads/inc-4711: {r.json()}")

    print("\nTo run as a real server, set TULIP_NOTEBOOK_BOOT=1 and run this")
    print("file directly. Example session:")
    print("  TULIP_NOTEBOOK_BOOT=1 TULIP_MODEL_PROVIDER=openai \\")
    print("      python examples/notebook_68_agent_server.py")
    print("  curl -s -X POST http://127.0.0.1:8000/invoke \\")
    print("       -H 'Content-Type: application/json' \\")
    print(
        '       -d \'{"prompt":"Page on-call? CrashLoopBackOff on payments-worker, 12 restarts."}\''
    )
    print("\nWith api_key= set, every /threads call is principal-scoped:")
    print("  AgentServer(agent=agent, api_key='secret')")
    print("  # Two on-call engineers with different bearer tokens see different")
    print("  # incidents for the same client-supplied thread_id.")
    return server


def boot_live_server() -> None:
    """Bind a live uvicorn instance.

    Gated behind TULIP_NOTEBOOK_BOOT=1 so the integration runner that
    imports every notebook doesn't hang on a blocking server.
    """
    model = get_model()
    agent = Agent(
        config=AgentConfig(
            system_prompt=_TRIAGE_PROMPT,
            max_iterations=5,
            model=model,
            checkpointer=_build_checkpointer(),
        )
    )
    server = AgentServer(
        agent=agent,
        title="On-Call Incident Triage Copilot API",
        description="An incident-triage copilot exposed as an HTTP API",
    )
    print("Booting AgentServer on http://127.0.0.1:8000 — Ctrl-C to stop.")
    print("Try: curl -X POST http://127.0.0.1:8000/invoke \\")
    print("          -H 'Content-Type: application/json' \\")
    print(
        '          -d \'{"prompt":"Page on-call? CrashLoopBackOff on payments-worker, 12 restarts."}\''
    )
    server.run(host="127.0.0.1", port=8000)


if __name__ == "__main__":
    missing = _missing_env()
    if missing:
        print("\n--- Notebook 68: Agent Server (Redis-backed incident threads) ---")
        print(
            "Required environment variables not set; skipping the live "
            "demo so this file still runs cleanly in CI.\n"
        )
        for name in missing:
            print(f"  - {name}")
        print("\nStart a Redis instance, set REDIS_URL, and re-run.")
        sys.exit(0)
    if os.getenv("TULIP_NOTEBOOK_BOOT") == "1":
        boot_live_server()
    else:
        example_server()