Harden an agent or workflow for AMC

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 startAn agent or workflow that runs locally; see Build a single agent or Build a workflow.View as Markdown

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

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

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

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

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.

Last updated 2026-10-21