# Harden an agent or workflow for AMC

Agent Mission Control developer docs > Microsoft Agent Framework (Python)

> Copy the marker helpers that carry steps, tokens, model and failures to AMC, wrap a single agent for both host paths, and turn on process-loss recovery for workflows.

**Before you start:** An agent or workflow that runs locally; see Build a single agent or Build a workflow.

Foundry hosting takes away what AMC needs to show a run properly. These helpers put it back. They are not in Microsoft's docs or samples; they come from agents and workflows running against AMC. Copy them into every new hosted agent or workflow.

## What hosting takes away

| You lose | Because | AMC needs it for |
| --- | --- | --- |
| Per-executor `author_name` | Consecutive text merges into a single message item. | Labelling each step. |
| Token counts | Run-level `usage` is always empty for a hosted agent or workflow, streaming or not, and the host drops the SDK's mid-stream `usage` content. | Cost, ROI and budgets. |
| Which model ran | It is not on the wire. | Pricing each step by its own model. |
| Why a step failed | An exception aborts the stream with no structured reason. | Showing the failure reason. |

The one channel that survives is an informational `function_call` output item. Each executor, or the single agent, emits one with everything AMC needs in its `arguments`. `informational_only=True` marks it transcript-only, so Foundry never tries to run it as a tool.

> **Note:** AMC tells a marker from a real tool call by `call_id == name`. A genuine tool call has a generated `call_id` that differs from its name and a paired `function_call_output`.

## The marker arguments

| Key | Type | Meaning |
| --- | --- | --- |
| `delivered` | string | The text this step produced. It is the step's headline in AMC. |
| `input_tokens`, `output_tokens` | integer | Counted by you from `usage_details`. `input_token_count` already includes cached tokens, and `output_token_count` includes reasoning tokens. |
| `total_tokens` | integer | `input + output`. |
| `model` | string | The model deployment id. AMC prices the step by this model. |
| `cached_input_tokens` | integer | The cached subset of input, billed at the cheaper cached rate. `0` when absent. |
| `model_usage` | object | Per-model breakdown: `{<model>: {"input": i, "output": o, "cached": c}}`. AMC prices each row at that model's rate and sums them. |
| `failure_reason` | string | `"<ExceptionType>: <message>"`. Its presence marks the step **failed**. |
| `activity_log` | array | Optional detail rows under the step; see [Report steps, tokens and human input](https://www.agentmissioncontrol.ai/developers/af-report-to-amc/). |
| `run_name` | string | Optional. Renames the run while completing the step. |

Azure OpenAI returns token counts, not cost, so don't send `cost_usd`. AMC's price is tokens times the model's configured rate, which means the counts must be right.

## Shared helpers

`workflow.py`
```python
import os

from agent_framework import Content, Message, WorkflowContext


def _model() -> str:
    # Local dev sets FOUNDRY_MODEL; Foundry hosting injects the other two.
    for var in ("FOUNDRY_MODEL", "FOUNDRY_MODEL_NAME", "AZURE_AI_MODEL_DEPLOYMENT_NAME"):
        value = os.getenv(var)
        if value:
            return value
    raise KeyError("Set FOUNDRY_MODEL, FOUNDRY_MODEL_NAME or AZURE_AI_MODEL_DEPLOYMENT_NAME.")


def _token_count(response) -> tuple[int, int, int]:
    """(input, output, cached_input) from an AgentResponse, or zeros."""
    usage = getattr(response, "usage_details", None)   # attribute on the response...
    if not usage:
        return 0, 0, 0
    return (                                            # ...but dict access on usage_details
        int(usage.get("input_token_count") or 0),
        int(usage.get("output_token_count") or 0),
        int(usage.get("cached_input_token_count")
            or usage.get("cached_token_count")
            or usage.get("cached_tokens") or 0),
    )


async def _emit_step_marker(ctx: WorkflowContext, step_id: str, delivered: str | None = None,
                            input_tokens: int | None = None, output_tokens: int | None = None,
                            model: str | None = None, cached_input_tokens: int | None = None,
                            model_usage: dict | None = None, failure_reason: str | None = None,
                            activity_log: list[dict] | None = None) -> None:
    arguments: dict = {}
    if delivered is not None:
        arguments["delivered"] = delivered
    if input_tokens is not None:
        arguments["input_tokens"] = input_tokens
    if output_tokens is not None:
        arguments["output_tokens"] = output_tokens
    if input_tokens is not None or output_tokens is not None:
        arguments["total_tokens"] = (input_tokens or 0) + (output_tokens or 0)
    if model:
        arguments["model"] = model
    if cached_input_tokens is not None:
        arguments["cached_input_tokens"] = cached_input_tokens
    if model_usage is None and model and (input_tokens is not None or output_tokens is not None):
        model_usage = {model: {"input": input_tokens or 0, "output": output_tokens or 0,
                               "cached": cached_input_tokens or 0}}
    if model_usage:
        arguments["model_usage"] = model_usage
    if failure_reason:
        arguments["failure_reason"] = failure_reason
    if activity_log:
        arguments["activity_log"] = activity_log
    await ctx.yield_output(Message(
        role="assistant",
        contents=[Content.from_function_call(
            call_id=step_id, name=step_id,          # call_id == name: the marker discriminator
            arguments=arguments, informational_only=True,
        )],
        author_name=step_id,
    ))
```

## Workflow executors

Emit the marker after the agent runs. On failure, emit a `failure_reason` marker and then re-raise, so the run still fails but AMC records why.

`workflow.py`
```python
class WriterExecutor(Executor):
    def __init__(self, agent, id: str = "writer") -> None:
        super().__init__(id=id)
        self._agent = agent

    async def _draft(self, prompt: str, ctx: WorkflowContext[Draft, Message]) -> None:
        try:
            response = await self._agent.run(prompt)
        except Exception as exc:
            await _emit_step_marker(ctx, "writer",
                                    failure_reason=f"{type(exc).__name__}: {exc}", model=_model())
            raise
        in_tok, out_tok, cached_tok = _token_count(response)
        await _emit_step_marker(ctx, "writer", delivered=response.text,
                                input_tokens=in_tok, output_tokens=out_tok,
                                model=_model(), cached_input_tokens=cached_tok)
        await ctx.send_message(Draft(prompt=prompt, draft=response.text))
```

- A step with no model call, such as a pure Python check or the human-input step, still emits a marker with `input_tokens=0, output_tokens=0` so every step is reported.
- Designate the executors in the builder: `WorkflowBuilder(start_executor=..., output_from=[terminal], intermediate_output_from=[every other executor])`. Undesignated executors' markers are not forwarded.
- If you also run the same graph in a local harness that reports to AMC some other way, pass that reporter into the executor and emit markers only when it is absent. Markers are the Foundry path.

### One executor, several models

Normally one executor wraps one agent on one deployment, and the helper builds the single-row `model_usage` for you. Only when one executor calls agents on different deployments do you accumulate it yourself:

```python
def _merge_usage(acc: dict, model: str, response) -> None:
    in_tok, out_tok, cached = _token_count(response)
    row = acc.setdefault(model, {"input": 0, "output": 0, "cached": 0})
    row["input"] += in_tok
    row["output"] += out_tok
    row["cached"] += cached

usage: dict = {}
r1 = await self._planner.run(prompt);  _merge_usage(usage, PLANNER_MODEL, r1)
r2 = await self._writer.run(r1.text);  _merge_usage(usage, WRITER_MODEL, r2)
await _emit_step_marker(ctx, "compose", delivered=r2.text,
                        input_tokens=sum(u["input"] for u in usage.values()),
                        output_tokens=sum(u["output"] for u in usage.values()),
                        model=WRITER_MODEL,      # a display label; the cost comes from model_usage
                        model_usage=usage)
```

- Key each row by the **pricing** model id, the id AMC has a rate for. Otherwise that slice is priced at AMC's default rate.
- Add per-turn deltas. Don't also add a cumulative total on top.
- The simpler alternative is one executor per model, so each step is single-model.

## A single agent

A hosted single agent reports no tokens either. Subclass `Agent` and append one trailing marker whose `call_id` and `name` are both the agent's name. `ResponsesHostServer` may drive the agent streaming or not, so wrap both paths. The subclass calls `super().run()`, so middleware, tools and `response_format` still apply.

`agent.py`
```python
from agent_framework import (
    Agent, AgentResponseUpdate, Content, Message, UsageDetails, add_usage_details,
)

AGENT_NAME = "contoso-orders-agent"
MARKER_STEP = AGENT_NAME                     # call_id == name == AGENT_NAME


def _split_usage(usage: UsageDetails | None) -> tuple[int, int, int]:
    if not usage:
        return 0, 0, 0
    return (int(usage.get("input_token_count") or 0),
            int(usage.get("output_token_count") or 0),
            int(usage.get("cached_input_token_count") or usage.get("cached_token_count")
                or usage.get("cached_tokens") or 0))


def _marker_content(delivered: str, input_tokens: int, output_tokens: int,
                    model: str | None = None, cached_input_tokens: int = 0,
                    activity_log: list[dict] | None = None) -> Content:
    arguments: dict[str, object] = {
        "delivered": delivered,
        "input_tokens": input_tokens,
        "output_tokens": output_tokens,
        "total_tokens": input_tokens + output_tokens,
        "cached_input_tokens": cached_input_tokens,
    }
    if model:
        arguments["model"] = model
        arguments["model_usage"] = {model: {"input": input_tokens, "output": output_tokens,
                                            "cached": cached_input_tokens}}
    if activity_log:
        arguments["activity_log"] = activity_log
    return Content.from_function_call(call_id=MARKER_STEP, name=MARKER_STEP,
                                      arguments=arguments, informational_only=True)


class _TokenMarkerAgent(Agent):
    def run(self, messages=None, *, stream: bool = False, **kwargs):  # type: ignore[override]
        if stream:
            return self._run_streaming_with_marker(messages, **kwargs)
        return self._run_nonstreaming_with_marker(messages, **kwargs)

    async def _run_nonstreaming_with_marker(self, messages, **kwargs):
        response = await super().run(messages, stream=False, **kwargs)
        in_tok, out_tok, cached_tok = _split_usage(getattr(response, "usage_details", None))
        response.messages.append(Message(
            role="assistant",
            contents=[_marker_content(response.text or "", in_tok, out_tok, _model(), cached_tok)],
            author_name=MARKER_STEP,
        ))
        return response

    async def _run_streaming_with_marker(self, messages, **kwargs):
        usage: UsageDetails | None = None
        text_parts: list[str] = []
        async for update in super().run(messages, stream=True, **kwargs):
            for content in update.contents or []:
                content_usage = getattr(content, "usage_details", None)
                if content_usage:
                    usage = add_usage_details(usage, content_usage)
                if getattr(content, "type", None) == "text" and content.text:
                    text_parts.append(content.text)
            yield update                             # re-emit every real update first
        in_tok, out_tok, cached_tok = _split_usage(usage)
        yield AgentResponseUpdate(
            contents=[_marker_content("".join(text_parts), in_tok, out_tok, _model(), cached_tok)],
            role="assistant", author_name=MARKER_STEP,
        )
```

Build the agent from `_TokenMarkerAgent(client=..., name=AGENT_NAME, ...)` instead of `Agent(...)`. A single agent has no per-step failure branch: a hard failure raises and the marker never emits.

## Process-loss recovery for workflows

By default a hosted workflow run dies with the process serving its stream. A deploy, a replica recycle, a scale-in or a dropped client connection leaves the run stuck at `in_progress`, and a reconnect replays an empty stream. `resilient_background=True` fixes this: the runtime persists the input, leases the work, re-invokes the handler in a later process, and switches replay to a file-backed store so a reconnect replays the real events.

- **It is workflow-only.** `ResponsesHostServer` raises `RuntimeError` if you pass it for a single agent.
- **It is a runtime upgrade.** The flag does not exist before `azure-ai-agentserver-responses` 2.x and raises `TypeError`. Bump `agent-framework-core==1.19.0`, `agent-framework-foundry==1.13.1`, `agent-framework-foundry-hosting==1.0.0b260918` and `azure-ai-agentserver-responses>=2.2.0b1,<3` together.
- **It needs protocol 2.0.0.** Responses 2.x requires hosted-agent protocol `2.0.0`; the agent fails fast at startup if the environment doesn't support it. Deploy one workflow first to confirm before rolling out the rest.
- **Don't hand-roll replay.** Never call `streams.use_file_backed_replay()` yourself. The flag configures it, and a manual call overwrites that process-wide setting and breaks replay.

### Allow-list every checkpoint type

Recovery restores the workflow's checkpoint, and core 1.19 blocks deserializing any custom type that isn't named. List the `request_data` payload and **every** dataclass on an edge as `"<module>:<qualname>"`. A workflow with no human-input pause still needs its edge types listed. Framework types such as `Message` are already allowed.

`main.py`
```python
checkpoint_provider = CheckpointStoreProvider(
    allowed_checkpoint_types=[
        "workflow:TopicRequest",   # ctx.request_info(request_data=...) payload
        "workflow:Topic",          # each edge message type
        "workflow:Written",
    ]
)
```

The module is the one that holds the type at runtime. When `main.py` does `from workflow import ...`, it is `workflow`, exactly as the error message prints it. A missing type fails with `Checkpoint deserialization blocked for type 'workflow:TopicRequest'.`

## Checklist

- `_model()`, `_token_count()` and `_emit_step_marker()` copied in (or `_split_usage()`, `_marker_content()` and `_TokenMarkerAgent` for a single agent).
- Every executor catches exceptions, emits `failure_reason`, and re-raises.
- Every success path emits `delivered`, `input_tokens`, `output_tokens`, `model` and `cached_input_tokens`.
- Only an executor that drives more than one model builds `model_usage` itself.
- `output_from=[terminal]` and `intermediate_output_from=[every other executor]`.
- Markers use `call_id == name` and `informational_only=True`.
- Every model the agent can run has a token rate configured in AMC, or it is priced at AMC's default rate.
- For recovery: `ResponsesServerOptions(resilient_background=True)`, the responses 2.x pins, a `CheckpointStoreProvider` listing every custom type, and one workflow deployed first.

## Related

- [Report steps, tokens and human input](https://www.agentmissioncontrol.ai/developers/af-report-to-amc/)
- [Host and deploy on Foundry](https://www.agentmissioncontrol.ai/developers/af-host-and-deploy/)
- [Troubleshoot Agent Framework agents](https://www.agentmissioncontrol.ai/developers/af-troubleshooting/)

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