Test your endpoint locally

import asyncio
import json
import os
import uuid
from datetime import datetime, timezone

from fastapi import Depends, FastAPI, Header, HTTPException
from fastapi.responses import JSONResponse, StreamingResponse
from pydantic import BaseModel, Field

API_KEY = os.environ.get("AGENT_API_KEY", "")
STEPS = [("fetch", "Fetch invoice"), ("summarize", "Summarize")]


def now() -> str:
    return datetime.now(timezone.utc).isoformat()


def check_key(api_key: str | None = Header(default=None, alias="api-key")) -> None:
    if API_KEY and api_key != API_KEY:
        raise HTTPException(status_code=401, detail="unauthorized")


app = FastAPI(dependencies=[Depends(check_key)])
runs: dict[str, dict] = {}


class StartRun(BaseModel):
    orchestratorWorkflowId: str | None = None
    versionId: str | None = None
    inputs: dict[str, str] = Field(default_factory=dict)
    idempotencyKey: str | None = None


def emit(run: dict, event_type: str, step_id: str | None = None, fragment: dict | None = None) -> None:
    run["events"].append({
        "eventType": event_type,
        "orchestratorRunId": run["id"],
        "stepId": step_id,
        "payloadFragment": json.dumps(fragment) if fragment is not None else None,
        "at": now(),
    })


def set_step(run: dict, step_id: str, status: str, **fields) -> None:
    step = next(s for s in run["steps"] if s["stepId"] == step_id)
    step["status"] = status
    step.update(fields)
    emit(run, "step.status", step_id, {"status": status})


async def work(run: dict) -> None:
    try:
        invoice = run["inputs"].get("invoiceId", "unknown")
        set_step(run, "fetch", "running", startedAt=now())
        await asyncio.sleep(1)  # your real work goes here
        set_step(run, "fetch", "completed", completedAt=now(), delivered=f"Fetched {invoice}")

        set_step(run, "summarize", "running", startedAt=now())
        await asyncio.sleep(1)
        set_step(run, "summarize", "completed", completedAt=now(), delivered="Summary text",
                 tokenCount=1200, inputTokens=1000, outputTokens=200, model="<model-id>")
        run["status"] = "completed"
        emit(run, "run.completed", fragment={"summary": "Summary text"})
    except asyncio.CancelledError:
        run["status"] = "cancelled"
        emit(run, "run.status", fragment={"status": "cancelled"})
        raise
    except Exception as exc:
        run["status"] = "failed"
        run["failureReason"] = f"{type(exc).__name__}: {exc}"
        emit(run, "run.status", fragment={"status": "failed", "reason": run["failureReason"]})
    finally:
        run["finishedAt"] = now()


def run_dto(run: dict) -> dict:
    return {"orchestratorRunId": run["id"], "orchestratorWorkflowId": run["workflowId"],
            "status": run["status"], "startedAt": run["startedAt"], "finishedAt": run["finishedAt"]}


def get_run(run_id: str) -> dict:
    if run_id not in runs:
        raise HTTPException(status_code=404, detail="run not found")
    return runs[run_id]


@app.get("/health")
async def health():
    return {"status": "healthy"}


@app.post("/runs")
async def start_run(body: StartRun):
    run_id = "run_" + uuid.uuid4().hex
    run = runs[run_id] = {
        "id": run_id, "workflowId": body.orchestratorWorkflowId or "invoice-summarizer",
        "status": "running", "startedAt": now(), "finishedAt": None, "inputs": body.inputs,
        "failureReason": None, "events": [],
        "steps": [{"stepId": sid, "name": name, "status": "pending", "tokenCount": 0}
                  for sid, name in STEPS],
    }
    run["task"] = asyncio.create_task(work(run))
    return run_dto(run)


@app.post("/runs/{run_id}/cancel")
async def cancel_run(run_id: str):
    run = get_run(run_id)
    if run["status"] in ("running", "human_intervention"):
        run["task"].cancel()
        await asyncio.sleep(0)
    return run_dto(run)


@app.get("/runs/{run_id}/status")
async def run_status(run_id: str):
    run = get_run(run_id)
    started = [s for s in run["steps"] if s["status"] != "pending"]
    body = {"orchestratorRunId": run_id, "status": run["status"], "currentStep": len(started),
            "totalSteps": len(run["steps"]),
            "steps": [{k: v for k, v in s.items() if k != "delivered" or s["status"] == "waiting"}
                      for s in run["steps"]]}
    if run["failureReason"]:
        body["failureReason"] = run["failureReason"]
    return body


@app.get("/runs/{run_id}/tokens")
async def run_tokens(run_id: str):
    run = get_run(run_id)
    total = sum(s.get("tokenCount", 0) for s in run["steps"])
    return {"orchestratorRunId": run_id, "agentTokens": total, "hostTokens": 0, "totalTokens": total,
            "inputTokens": sum(s.get("inputTokens", 0) for s in run["steps"]),
            "outputTokens": sum(s.get("outputTokens", 0) for s in run["steps"]),
            "costUsd": 0}


@app.get("/runs/{run_id}/steps/{step_id}/content")
async def step_content(run_id: str, step_id: str):
    step = next((s for s in get_run(run_id)["steps"] if s["stepId"] == step_id), None)
    if step is None:
        return JSONResponse(status_code=404, content={"error": "step not found"})
    return {"stepId": step_id, "action": None, "delivered": step.get("delivered"),
            "reasoning": None, "activityLog": []}


@app.get("/runs/{run_id}/events")
async def run_events(run_id: str):
    run = get_run(run_id)

    async def stream():
        sent = 0
        while True:
            while sent < len(run["events"]):
                yield f"data: {json.dumps(run['events'][sent])}\n\n"
                sent += 1
            if run["finishedAt"] and sent >= len(run["events"]):
                return
            await asyncio.sleep(0.2)

    return StreamingResponse(stream(), media_type="text/event-stream")

AGENT_API_KEY=dev-key uvicorn app:app --port 8080

H='api-key: dev-key'
curl -s -H "$H" localhost:8080/health                 # registration check
curl -s -o /dev/null -w '%{http_code}\n' localhost:8080/health   # 401 without the key

RUN=$(curl -s -H "$H" -H 'Content-Type: application/json' \
  -d '{"orchestratorWorkflowId":"invoice-summarizer","versionId":null,"inputs":{"invoiceId":"INV-1042"},"idempotencyKey":null}' \
  localhost:8080/runs | python3 -c 'import sys, json; print(json.load(sys.stdin)["orchestratorRunId"])')

curl -s -N -H "$H" localhost:8080/runs/$RUN/events    # live view, closes when the run ends
curl -s -H "$H" localhost:8080/runs/$RUN/status       # the poll
curl -s -H "$H" localhost:8080/runs/$RUN/tokens       # the poll
curl -s -H "$H" localhost:8080/runs/$RUN/steps/summarize/content