# Serve the agent to AMC with amc_sdk

Agent Mission Control developer docs > Claude skill, self-hosted

> Swap the plain-HTTP host for `app.py` on `amc_sdk.build_app`, map SDK events to AMC steps, and report tokens and cost AMC can price.

**Before you start:** A generated agent core, and the `amc_sdk` package that ships with AMC.

The AMC variant keeps the `claude-agent-sdk` core and replaces the host. `app.py` describes the agent to `amc_sdk`, drives the core inside a run handler, and reports through a `RunContext`. `amc_sdk.build_app()` serves the rest of the AMC contract. This page covers only what is specific to a Claude skill; the general contract is in [The integration contract](https://www.agentmissioncontrol.ai/developers/integration-contract/).

> **Note:** `amc_sdk` is AMC's library for exposing an endpoint AMC can drive. It is unrelated to Anthropic's `claude-agent-sdk`, which runs the skill. The AMC variant uses both: `claude-agent-sdk` inside, `amc_sdk` as the HTTP surface.

## What build_app serves

| Endpoint | Purpose |
| --- | --- |
| `GET /health` | Readiness. AMC probes it when you register. |
| `GET /workflows`, `GET /workflows/{id}` | Definition discovery, built from your `Agent` config |
| `GET /agent-types` | Agent type entries |
| `POST /runs` | Starts a run. Body: `{"inputs": {...}}`, a flat string map. Returns `orchestratorRunId` and `status`. |
| `POST /runs/{id}/cancel` | Best-effort cancel |
| `POST /runs/{id}/messages` | A human's answer to a paused step |
| `GET /runs/{id}/status` | Per-step status, tokens, and timestamps |
| `GET /runs/{id}/events` | SSE stream of run events |
| `GET /runs/{id}/content`, `GET /runs/{id}/steps/{stepId}/content` | Produced content |
| `GET /runs/{id}/tokens` | Token usage for ROI and budgets |

The run store, step state, event recorder, camelCase DTOs, and SSE loop all live in `amc_sdk`. Your `app.py` keeps only the `Agent(...)` definition, input parsing, the `@agent.run` handler, and a health probe.

## Files

- Keep `claude_skills_agent.py` as generated.
- Skip `http_agent_server.py`, `claude_skills_agent_executor.py`, and the HTTP parts of `__main__.py`.
- Add `app.py`.

## Map SDK events to the run

| claude-agent-sdk event | RunContext call |
| --- | --- |
| Run starts | `ctx.step_started(STEP_ID, agent_name=...)` |
| `ThinkingBlock` | `ctx.thinking(STEP_ID, block.thinking)`: a `thinking` detail row |
| `ToolUseBlock` plus its `ToolResultBlock` | `ctx.tool_activity(STEP_ID, tool=block.name, text=<args>, result=<result>)`: one `tool` row with name, input, and result |
| `TextBlock` | `ctx.token_delta(STEP_ID, text)` |
| `ResultMessage` (success) | `ctx.step_completed(STEP_ID, token_count=i+o, delivered=<final text>, input_tokens=i, output_tokens=o, model=<model id>, cached_input_tokens=<cache read>, model_usage=<optional>)` |
| Final output | `ctx.set_output({...})` and `ctx.set_cost(total_cost_usd)` |

`ToolResultBlock`s arrive on a `UserMessage`, not on the `AssistantMessage` that made the call. Buffer each `ToolUseBlock` by `block.id`, then emit one `ctx.tool_activity(...)` when the block whose `tool_use_id` matches lands. AMC renders these rows as collapsible sub-steps under the step's `delivered` text. Keep the capture fail-safe so it never breaks a run, and gate it behind an `EMIT_STEP_ACTIVITY_LOG` env flag so you can turn it off without a redeploy.

## Example app.py

The template's core yields display strings for the plain-HTTP surface. For this variant, factor the options-building part of `run_conversation_stream()` into a method (called `build_options()` here) so `app.py` can run `query()` and see raw SDK messages. Put the buffering in the core's query loop, as the skill recommends, if you prefer; the routing to `ctx` belongs in `app.py`.

`app.py`
```python
import os

import uvicorn
from amc_sdk import Agent, AgentType, HealthReport, InputParam, RunContext, StepDef, build_app
from claude_agent_sdk import query

from claude_skills_agent import (
    DEFAULT_MODEL, SKILL_NAME, SKILL_PARENT_DIR, ReportGeneratorSkillsAgent,
)

STEP_ID = "main"
EMIT_ACTIVITY = os.getenv("EMIT_STEP_ACTIVITY_LOG", "1") == "1"
core = ReportGeneratorSkillsAgent()


def _health() -> HealthReport:
    has_creds = bool(os.getenv("ANTHROPIC_API_KEY"))
    skill_md = os.path.join(SKILL_PARENT_DIR, ".claude", "skills", SKILL_NAME, "SKILL.md")
    if has_creds and os.path.isfile(skill_md):
        return HealthReport(ready=True, detail="Fully configured")
    return HealthReport(ready=False, detail="Missing Claude credentials or SKILL.md")


amc_agent = Agent(
    workflow_id="report-generator",        # catalog key, globally unique
    name="Report Generator",
    description="Builds a weekly report with the report-generator skill.",
    type="workflow",
    steps=[StepDef(STEP_ID, "Generate report")],
    agent_types=[AgentType("report-generator", "Report Generator", ["text", "tool-use"])],
    input_parameters=[InputParam("topic", required=False)],
    health=_health,
)


def _read_model_usage(msg) -> dict[str, dict[str, int]]:
    """{model: {input, output, cached}}; input includes cache reads and writes."""
    out: dict[str, dict[str, int]] = {}
    for model, e in (getattr(msg, "model_usage", None) or {}).items():
        if not isinstance(e, dict):
            continue
        cached = int(e.get("cacheReadInputTokens") or 0)
        out[model] = {
            "input": int(e.get("inputTokens") or 0) + cached
                     + int(e.get("cacheCreationInputTokens") or 0),
            "output": int(e.get("outputTokens") or 0),
            "cached": cached,
        }
    return out


@amc_agent.run
async def _run(ctx: RunContext) -> None:
    await core.create_agent()
    prompt = core._build_prompt(ctx.inputs.get("topic", ""))
    pending: dict[str, tuple[str, str]] = {}   # tool_use_id -> (name, input)
    text: list[str] = []
    usage: dict[str, dict[str, int]] = {}
    cost = None

    ctx.step_started(STEP_ID, agent_name="Report Generator")
    try:
        async for msg in query(prompt=prompt, options=core.build_options()):
            kind = type(msg).__name__
            if kind == "ResultMessage":
                usage = _read_model_usage(msg)
                cost = getattr(msg, "total_cost_usd", None)
                continue
            content = getattr(msg, "content", None)
            if not isinstance(content, list):
                continue
            for block in content:
                b = type(block).__name__
                if b == "TextBlock" and block.text.strip():
                    text.append(block.text.strip())
                    ctx.token_delta(STEP_ID, block.text)
                elif not EMIT_ACTIVITY:
                    continue
                elif b == "ThinkingBlock":
                    ctx.thinking(STEP_ID, block.thinking)
                elif b == "ToolUseBlock":
                    pending[block.id] = (block.name, str(block.input))
                elif b == "ToolResultBlock" and block.tool_use_id in pending:
                    name, args = pending.pop(block.tool_use_id)
                    ctx.tool_activity(STEP_ID, tool=name, text=args, result=str(block.content))
    except Exception:
        i, o, cached = _totals(usage)
        ctx.set_step_tokens(STEP_ID, token_count=i + o, input_tokens=i, output_tokens=o,
                            model=DEFAULT_MODEL, cached_input_tokens=cached)
        ctx.set_cost(cost)
        raise                                # run failed, error text becomes the reason

    i, o, cached = _totals(usage)
    final = "\n\n".join(text)
    ctx.step_completed(
        STEP_ID, token_count=i + o, delivered=final,
        input_tokens=i, output_tokens=o, model=DEFAULT_MODEL, cached_input_tokens=cached,
        model_usage=usage if len(usage) > 1 else None,   # display-only breakdown
    )
    ctx.set_output({"result": final})
    ctx.set_cost(cost)


def _totals(usage: dict[str, dict[str, int]]) -> tuple[int, int, int]:
    return (sum(m["input"] for m in usage.values()),
            sum(m["output"] for m in usage.values()),
            sum(m["cached"] for m in usage.values()))


app = build_app(amc_agent)

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8080)
```

> **Important:** The `Agent` field for the input contract is `input_parameters`. Some skill notes write `input_params`, which `Agent` does not accept.

## Two rules that carry the money

- **Report `input_tokens` and `output_tokens` separately**, not just `token_count`. AMC prices prompt and completion tokens at different per-model rates.
- **Pass `model` and `cached_input_tokens`.** Read them off `ResultMessage.model_usage`: input is uncached plus `cacheReadInputTokens` plus `cacheCreationInputTokens`; cached is `cacheReadInputTokens`. A step with no `model` is priced at AMC's default rate. That is wrong money, with no error.

When Task sub-agents run on a different model from the orchestrator, one step spans several models. Pass the per-model projection as `model_usage={model: {"input", "output", "cached"}}` and AMC shows it as the per-model table on the run detail. Here it is display-only: AMC prefers the exact `ctx.set_cost(total_cost_usd)` for the invoice over any per-model recompute. Omit it for a single-model run.

## The failure path

Raising from the `@agent.run` handler is how a run fails. `amc_sdk` marks the run `failed`, and the error text becomes the failed step's content. AMC records it as the run's failure reason and uses it for the suggested cause it shows the operator. Before re-raising, call `ctx.set_step_tokens(...)` and `ctx.set_cost(...)`, or a failed run is priced at zero. `ctx.step_failed(STEP_ID, detail)` marks a step failed while the run continues.

## Optional: completed with issues

When a step failed but the run still produced a usable result, call `ctx.mark_partial("Slack post failed; the report was saved but nobody was notified")` and return normally. The run stays `completed`, and `/status` publishes `outcome: "partial"` with the reason as `failureReason`, so AMC shows **Completed with issues**. The reason must say what wasn't done: a blank one raises `ValueError`. It is trimmed and capped at 2000 characters, and the last call wins. The call doesn't end the run. 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, so a `ctx.step_failed` without `mark_partial` leaves a clean completed run.

## Optional: rename the run

Call `ctx.run_renamed("report-generator-weekly")` once the run's subject is known. It sets the run's display name (published as `runName` on `/status`) and emits a `run.renamed` event so a watching browser updates in place. It is sticky (a blank name never clears a set one), idempotent, and has no effect once the run is terminal. It renames only this run.

## Optional: ask a human

`answer = await ctx.ask_human(STEP_ID, "Which region should the report cover?")` pauses the step and the run as `waiting`. AMC opens an intervention, and when someone answers it posts to `/runs/{id}/messages` and the call returns their text. See [The integration contract](https://www.agentmissioncontrol.ai/developers/integration-contract/) for how AMC routes the question.

## Packaging

`amc_sdk` ships with AMC and is not on PyPI. Install it from your copy (`pip install -e <path-to>/amc_sdk`); it brings `fastapi` and `pydantic`. Build the image from a context that includes the `amc_sdk/` folder.

`Dockerfile (AMC variant changes)`
```dockerfile
COPY amc_sdk/ ./amc_sdk/
RUN pip install ./amc_sdk
EXPOSE 8080
CMD ["python", "app.py"]
```

`pyproject.toml` names the agent's modules under `[tool.setuptools] py-modules` (`__main__`, `claude_skills_agent`, `claude_skills_agent_executor`, `http_agent_server`). Keep that list in step with the files you actually keep.

There is no Foundry deploy step. AMC reaches the container at the endpoint you register. See [Deploy and register in AMC](https://www.agentmissioncontrol.ai/developers/self-hosted-deploy-and-register/).

## Inbound auth

`build_app(agent, verify_caller=...)` takes a FastAPI dependency that runs before every request. The default, `amc_sdk.auth.no_auth`, authorizes everyone. It is a placeholder, not a security control. On the registration probe and on run calls, AMC sends the key in an `api-key` header when you register with **API key**, and `Authorization: Bearer <token>` with **Managed Identity**. Check whichever applies in your own `verify_caller` and raise `fastapi.HTTPException` to reject a caller.

## Checklist

- `app.py` on `amc_sdk.build_app`; the plain-HTTP server files skipped.
- `step_completed` carries the input and output split, `model`, and `cached_input_tokens`, plus `model_usage` when sub-agents ran on another model.
- On failure, `set_step_tokens` and `set_cost` run before `raise`.
- The `health=` probe checks credentials and the skill's `SKILL.md`.
- The container listens on port `8080`, and the AMC registration points at it.

## Related

- [The integration contract](https://www.agentmissioncontrol.ai/developers/integration-contract/)
- [Deploy and register in AMC](https://www.agentmissioncontrol.ai/developers/self-hosted-deploy-and-register/)
- [The agent core and prompt strategies](https://www.agentmissioncontrol.ai/developers/self-hosted-agent-core/)

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