# Test your endpoint locally

Agent Mission Control developer docs > Bring your own agent

> A complete minimal endpoint, and the calls to make against it in the order AMC makes them.

**Before you start:** Python 3.10 or later, and `pip install fastapi uvicorn`.

This service implements the whole run contract in about 150 lines, with an in-memory store and two steps. Use it to see the wire format working, or as the starting shape for a service in any language.

`app.py`
```python
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")
```

To add questions for people, add the `ask_human` helper and the `/messages` route from [Ask a person mid-run](https://www.agentmissioncontrol.ai/developers/byo-human-in-the-loop/).

## Drive it the way AMC does

`Terminal 1`
```bash
AGENT_API_KEY=dev-key uvicorn app:app --port 8080
```

`Terminal 2`
```bash
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
```

## Check before you register

- `/health` returns `2xx` with the right credential and `401` without it.
- `POST /runs` returns in well under a second, with status `running`.
- `/status` shows every step from the start, and ends `completed` with the last step's tokens in.
- Every step that used a model reports `inputTokens`, `outputTokens` and `model`.
- A forced failure ends `failed` with a readable `failureReason`.
- Cancel works while the run is running and while it is waiting on a person.
- Finished runs stay readable at `/status` and `/steps/{stepId}/content`.

## Register a test endpoint

- AMC must be able to reach the endpoint from its own network. For an AMC in the cloud, expose your local port over an https tunnel or deploy to a test host.
- Use **API key** authentication. Key mode accepts `http` endpoints; managed identity needs `https` and an approved host.
- Choose **Flight Director / custom**, check **This resource uses the AMC SDK** (or enter a path segment such as `invoice-summarizer`), and enter the endpoint and key. See [Register your agent](https://www.agentmissioncontrol.ai/developers/register-your-agent/).

## Related

- [Bring your own agent](https://www.agentmissioncontrol.ai/developers/byo-overview/)
- [The amc_sdk Python package](https://www.agentmissioncontrol.ai/developers/byo-amc-sdk/)
- [Errors and timeouts](https://www.agentmissioncontrol.ai/developers/byo-errors-and-timeouts/)

---
Source: https://www.agentmissioncontrol.ai/developers/byo-test-locally/ · Last updated 2026-10-21 · Full index: https://www.agentmissioncontrol.ai/llms.txt