Serve the agent to AMC with amc_sdk

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.

What build_app serves

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

ToolResultBlocks 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.

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)

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 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.

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.

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.