# The amc_sdk Python package

Agent Mission Control developer docs > Bring your own agent

> Serve AMC's whole run contract from one Python run function: steps, tokens, cost, renames, assignees and questions for people.

**Before you start:** Python 3.10 or later, and the `amc_sdk` package that ships with AMC.

`amc_sdk` is AMC's library for exposing an endpoint AMC can drive. You write a run function and a small config; `build_app` returns a FastAPI app that serves every endpoint, with the run store, run and step state, typed events, camelCase DTOs and the SSE loop built in.

> **Note:** `amc_sdk` is not Anthropic's Claude Agent SDK (`claude-agent-sdk`). That library builds an agent; this one exposes an agent to AMC. They compose: build with one, expose with the other.

## Install

`amc_sdk` ships with AMC and isn't on PyPI. Install it from the copy your AMC provider gives you. It depends on `fastapi` and `pydantic`; the `serve` extra adds `uvicorn`.

```bash
pip install "./amc_sdk[serve]"
```

## A complete agent

`app.py`
```python
import os

from fastapi import HTTPException, Request

from amc_sdk import Agent, AgentType, HealthReport, InputParam, RunContext, StepDef, build_app

API_KEY = os.environ.get("AGENT_API_KEY", "")


async def verify_caller(request: Request) -> None:
    if API_KEY and request.headers.get("api-key") != API_KEY:
        raise HTTPException(status_code=401, detail="unauthorized")


def health() -> HealthReport:
    if not os.environ.get("MODEL_API_KEY"):
        return HealthReport(ready=False, detail="MODEL_API_KEY is not set")
    return HealthReport(ready=True)


async def summarize(invoice_id: str) -> tuple[str, dict]:
    # Call your model here. Return the text and the usage it reported.
    return f"Summary of {invoice_id}", {"input": 4200, "output": 610, "cached": 3000}


agent = Agent(
    workflow_id="invoice-summarizer",
    name="Invoice Summarizer",
    description="Summarizes one invoice.",
    type="workflow",
    steps=[
        StepDef("fetch", "Fetch invoice"),
        StepDef("summarize", "Summarize"),
    ],
    agent_types=[AgentType("invoice-summarizer", "Invoice Summarizer", ["text"])],
    input_parameters=[
        InputParam("invoiceId", type="string", required=True, description="Invoice number"),
        InputParam("ownerEmail", type="string", role="assignee", description="Who approves it"),
    ],
    health=health,
)


@agent.run
async def run(ctx: RunContext) -> None:
    invoice_id = ctx.inputs["invoiceId"]
    ctx.run_renamed(f"invoice-{invoice_id}")
    if ctx.inputs.get("ownerEmail"):
        ctx.set_assignee(ctx.inputs["ownerEmail"])

    ctx.step_started("fetch")
    ctx.tool_activity("fetch", tool="get_invoice", text=f'{{"id": "{invoice_id}"}}', result='{"total": 1200}')
    ctx.step_completed("fetch", token_count=0, delivered=f"Fetched {invoice_id}")

    ctx.step_started("summarize", agent_name="Summarizer")
    ctx.thinking("summarize", "Grouping line items by cost center")
    summary, usage = await summarize(invoice_id)
    ctx.step_completed(
        "summarize",
        token_count=usage["input"] + usage["output"],
        delivered=summary,
        input_tokens=usage["input"],
        output_tokens=usage["output"],
        cached_input_tokens=usage["cached"],
        model="<model-id>",
    )
    ctx.set_output({"invoiceId": invoice_id, "summary": summary})


app = build_app(agent, verify_caller=verify_caller, max_concurrency=4)
```

```bash
AGENT_API_KEY=dev-key MODEL_API_KEY=<your-model-key> uvicorn app:app --host 0.0.0.0 --port 8080
```

## The run function

The handler is `async def (ctx: RunContext) -> None`, set with `@agent.run`. The SDK runs it in the background for each `POST /runs` and owns the run-level status:

- **Returns normally:** the run is `completed`, and `run.completed` carries `ctx.output`.
- **Raises:** the run is `failed`. Every step still pending, running or waiting is marked failed with the error text (`ExceptionType: message`), which AMC shows as the failure reason.
- **Cancelled** (`POST /runs/{id}/cancel`): the run is `cancelled`, and unfinished steps are marked failed with `cancelled`.
- **Returns normally after `ctx.mark_partial(reason)`:** the run is still `completed`, and `/status` publishes `outcome: "partial"` with `reason` as `failureReason`. AMC shows **Completed with issues**.

`ctx.inputs` is the flat string map from `POST /runs`. The SDK doesn't parse or validate it; AMC has already validated it against your registered contract.

## RunContext

| Call | What it does |
| --- | --- |
| `step_started(step_id, agent_name=None)` | Marks the step `running` and stamps `startedAt`. |
| `step_completed(step_id, token_count, ...)` | Marks it `completed`. Keyword arguments: `delivered`, `input_tokens`, `output_tokens`, `cached_input_tokens`, `model`, `model_usage`, `agent_id`. |
| `step_failed(step_id, detail=None)` | Marks one step `failed` with `detail` as its reason, while the run continues. |
| `thinking(step_id, text)` | Adds a `thinking` row to the step's activity log. |
| `tool_activity(step_id, tool, text="", result=None)` | Adds a `tool` row with the tool name, its input and its result. |
| `activity(step_id, text, type="tool", tool=None, result=None)` | Adds any activity row. Identical rows aren't repeated. |
| `token_delta(step_id, text)` | Emits a `token.delta` event. |
| `await ask_human(step_id, question)` | Parks the step on a question and returns the answer posted to `/runs/{id}/messages`. See the limitation below. |
| `set_assignee(email)` | Declares who the run is about. Published as `assigneeRef` on the next poll. `None` does nothing. |
| `run_renamed(name)` | Renames this run. Published as `runName` and as a `run.renamed` event. Blank does nothing. |
| `mark_partial(reason)` | Declares that the run finished with issues; `reason` says what wasn't done. It doesn't end the run. Once the handler returns, `/status` publishes `outcome: "partial"` and `failureReason`. A blank `reason` raises `ValueError`. The reason is trimmed and capped at 2000 characters, and the last call wins. Raising still fails the run, and on a failed or cancelled run the call is ignored. AMC never derives a partial outcome from failed steps. |
| `set_cost(cost_usd)` | The run's own cost. AMC prefers it over its own pricing. |
| `set_step_tokens(step_id, token_count, ...)` | Sets a step's tokens directly, for the failure path. |
| `set_output(dict)` | The run's output, sent with `run.completed`. |
| `emit(event_type, step_id=None, fragment=None)` | Appends a raw event. Rarely needed. |

## Report spend on the failure path

A run that raises before `step_completed` reports zero tokens unless you record them first:

```python
    try:
        summary, usage = await summarize(invoice_id)
    except Exception:
        ctx.set_step_tokens("summarize", token_count=spent_in + spent_out,
                            input_tokens=spent_in, output_tokens=spent_out,
                            model="<model-id>", cached_input_tokens=spent_cached)
        ctx.set_cost(spent_usd)
        raise  # the run fails, with the error as its reason
```

## Config objects

| Object | Fields |
| --- | --- |
| `Agent` | `workflow_id`, `name`, `description`, `type` (`"workflow"` by default), `steps`, `version` (`"1"`), `agent_types`, `input_parameters`, `run_metadata`, `health`. |
| `StepDef` | `step_id`, `name`, `description=None`. One per step, in order. `step_id` is what AMC keys the step by. |
| `InputParam` | `name`, `type="string"`, `required=False`, `format`, `description`, `default`, `enum_values`, `role`. |
| `AgentType` | `key`, `display_name`, `capabilities`. |
| `HealthReport` | `ready=True`, `detail=None`, `status=None`. |

`build_app` needs a handler and at least one `StepDef`. Every declared step is listed on `/status` from the start of each run, as `pending`.

## Health

With no `health` callable, `/health` returns `{"status": "healthy", "service": "<workflow_id>"}`. With one, `ready=False` returns `503` with `status: "not_configured"` and your `detail`. AMC's registration check treats any non-`2xx` as **Unreachable**, so check your own configuration (model keys, required files) here: a misconfigured agent can't be registered.

## build_app options

| Option | Default | Purpose |
| --- | --- | --- |
| `verify_caller` | Allow everyone | A FastAPI dependency run before every request. Raise `HTTPException` to reject. Supply one: the default is not a security control. |
| `on_startup` | None | `callable(agent)`, sync or async, run at startup. |
| `max_concurrency` | Unbounded | Cap on runs executing at once in this process. Extra runs wait their turn while reported as `running`. |
| `max_runs` | `200` | How many runs to keep in memory. The oldest finished runs are evicted first; running runs never are. |
| `sse_poll_seconds` | `0.1` | How often the SSE stream checks for new events. |
| `cors_allow_origins` | `["*"]` | CORS origins. |
| `title` | The agent's name | The OpenAPI title. |

## Limitations in 0.1.0

- **Questions don't open interventions yet.** `ask_human` reports the run as `waiting`, and AMC opens an intervention only for `human_intervention`. The run shows as **Intervention** with nothing to answer. Until this changes, an agent that must ask people should serve the contract directly, as in [Ask a person mid-run](https://www.agentmissioncontrol.ai/developers/byo-human-in-the-loop/).
- **The live view shows completion and renames only.** The SDK's step, token and failure events carry plain-text fragments, which AMC's live view can't read and drops. The poll still records everything, so the run page catches up within one poll.
- **Runs live in memory.** A restart loses them: AMC's polls and cancels for those runs then fail, and the runs stay live in AMC (see [Restarts](https://www.agentmissioncontrol.ai/developers/byo-errors-and-timeouts/)). Keep `max_runs` large enough that finished runs stay readable for as long as people review them.
- **Declarations stay on your side.** The SDK publishes `input_parameters` and `steps` at `GET /workflows/{id}`, but AMC doesn't read them. Enter the same input contract and declared steps when you register.

## Test with FastAPI's TestClient

`test_app.py`
```python
import time

from fastapi.testclient import TestClient

from app import app


def wait_for(client, run_id, status):
    for _ in range(100):
        body = client.get(f"/runs/{run_id}/status").json()
        if body["status"] == status:
            return body
        time.sleep(0.02)
    raise AssertionError(body)


def test_run_completes():
    with TestClient(app) as client:
        run = client.post("/runs", json={"inputs": {"invoiceId": "INV-1042"}}).json()
        status = wait_for(client, run["orchestratorRunId"], "completed")
        assert status["runName"] == "invoice-INV-1042"
        tokens = client.get(f"/runs/{run['orchestratorRunId']}/tokens").json()
        assert tokens["inputTokens"] == 4200


def test_missing_input_fails_the_run():
    with TestClient(app) as client:
        run = client.post("/runs", json={"inputs": {}}).json()
        status = wait_for(client, run["orchestratorRunId"], "failed")
        assert "KeyError" in status["steps"][0]["failureReason"]
```

Use `TestClient` as a context manager. Without it, the client's event loop closes between requests and cancels the background run.

## Related

- [Bring your own agent](https://www.agentmissioncontrol.ai/developers/byo-overview/)
- [Ask a person mid-run](https://www.agentmissioncontrol.ai/developers/byo-human-in-the-loop/)
- [Report tokens and cost](https://www.agentmissioncontrol.ai/developers/byo-tokens-and-cost/)
- [Test your endpoint locally](https://www.agentmissioncontrol.ai/developers/byo-test-locally/)

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