Module adcp.server.translate

Error translation and request normalization for proxy and custom-transport servers.

Standard servers using serve() or ADCPAgentExecutor do not need these helpers — the framework handles error translation and request normalization internally.

These are for proxy servers that catch ADCPError from a downstream agent call and need to format it for their own transport, or custom multi-transport servers that bypass the standard framework.

Not exported from adcp.server — import directly::

from adcp.server.translate import translate_error, normalize_request

# In a proxy catching errors from a downstream agent:
try:
    result = await downstream_client.create_media_buy(params)
except ADCPError as e:
    raise translate_error(e, protocol="a2a")
    # Raises: InternalError(message="...", data={...})

# Normalize deprecated field names from older callers:
params = normalize_request(params, task_name="create_media_buy")

Functions

def build_mcp_error_result(exc: ADCPError | Error | Any,
*,
params: dict[str, Any] | None = None,
method_name: str = '',
response_enhancer: ResponseEnhancer | None = None,
context: ToolContext | None = None) ‑> CallToolResult
Expand source code 
def build_mcp_error_result(
    exc: ADCPError | Error | Any,
    *,
    params: dict[str, Any] | None = None,
    method_name: str = "",
    response_enhancer: ResponseEnhancer | None = None,
    context: ToolContext | None = None,
) -> CallToolResult:
    """Build an MCP ``CallToolResult`` carrying the structured ``adcp_error`` envelope.

    The framework dispatcher returns this when a platform method raises a
    structured AdCP error. The result has ``isError=True`` AND
    ``structuredContent={"adcp_error": {...}}`` on the same envelope —
    matching the spec's transport-errors.mdx §MCP Binding shape that the
    storyboard runner's ``/adcp_error/code`` JSON-pointer assertion
    expects.

    The text fallback in ``content[]`` preserves human-readable display
    for clients that do not consume ``structuredContent`` (LLM tool-use
    surfaces, log viewers).

    Buyer agents read the structured envelope first; the text fallback
    is only consulted when ``structuredContent`` is absent, per the
    spec's structured-error precedence rules.

    When ``params`` is supplied and carries a ``context`` field, that
    field is echoed onto the structuredContent envelope alongside
    ``adcp_error`` — symmetric with the success path's
    :func:`adcp.server.helpers.inject_context` call. Without this echo,
    error responses violate the AdCP context-passthrough contract and
    buyers lose correlation IDs and idempotency hints across the
    raise-AdcpError boundary.

    When ``response_enhancer`` is supplied it runs against the structured
    envelope after the context echo — the same
    :data:`~adcp.server.ResponseEnhancer` the success path uses, so a
    seller stamping cross-cutting fields covers error responses (including
    credential-policy errors) too. ``method_name`` and ``context`` are
    forwarded to the context-aware enhancer arity.
    """
    from adcp.server.helpers import inject_context

    code, message, recovery, field, suggestion, details, _errors = _extract_structured_fields(exc)

    adcp_error: dict[str, Any] = {
        "code": code,
        "message": message,
        "recovery": recovery,
    }
    if field is not None:
        adcp_error["field"] = field
    if suggestion is not None:
        adcp_error["suggestion"] = suggestion
    # ``retry_after`` lives on decisioning AdcpError; project it when present.
    retry_after = getattr(exc, "retry_after", None)
    if retry_after is not None:
        adcp_error["retry_after"] = retry_after
    if details:
        adcp_error["details"] = dict(details)

    # Text fallback for clients that don't read structuredContent.
    if field:
        text = f"{code}[{field}]: {message}"
    else:
        text = f"{code}: {message}"
    if suggestion:
        text += f"\nSuggestion: {suggestion}"

    structured: dict[str, Any] = {"adcp_error": adcp_error}
    if params is not None:
        inject_context(params, structured)

    # Run the seller's response enhancer on the error envelope AFTER the
    # context echo (so a stripped credential can't be re-introduced) —
    # symmetric with the success path in ``create_tool_caller``. Error
    # responses are not schema-validated, so the enhancer's output ships
    # as-is; a buggy enhancer is caught and logged inside the helper.
    _apply_response_enhancer(response_enhancer, method_name, structured, context)

    return CallToolResult(
        content=[TextContent(type="text", text=text)],
        structuredContent=structured,
        isError=True,
    )

Build an MCP CallToolResult carrying the structured adcp_error envelope.

The framework dispatcher returns this when a platform method raises a structured AdCP error. The result has isError=True AND structuredContent={"adcp_error": {...}} on the same envelope — matching the spec's transport-errors.mdx §MCP Binding shape that the storyboard runner's /adcp_error/code JSON-pointer assertion expects.

The text fallback in content[] preserves human-readable display for clients that do not consume structuredContent (LLM tool-use surfaces, log viewers).

Buyer agents read the structured envelope first; the text fallback is only consulted when structuredContent is absent, per the spec's structured-error precedence rules.

When params is supplied and carries a context field, that field is echoed onto the structuredContent envelope alongside adcp_error — symmetric with the success path's :func:inject_context() call. Without this echo, error responses violate the AdCP context-passthrough contract and buyers lose correlation IDs and idempotency hints across the raise-AdcpError boundary.

When response_enhancer is supplied it runs against the structured envelope after the context echo — the same :data:~adcp.server.ResponseEnhancer the success path uses, so a seller stamping cross-cutting fields covers error responses (including credential-policy errors) too. method_name and context are forwarded to the context-aware enhancer arity.

def normalize_request(params: dict[str, Any], task_name: str | None = None) ‑> dict[str, typing.Any]
Expand source code 
def normalize_request(
    params: dict[str, Any],
    task_name: str | None = None,
) -> dict[str, Any]:
    """Normalize deprecated field names and structures in request params.

    Applies known transforms so servers can accept both old and new field
    formats without duplicating normalization logic in every handler.

    Transforms applied:

    - ``account_id: "123"`` → ``account: {account_id: "123"}`` (structural)
    - ``brand_manifest: "https://..."`` → ``brand: {domain: "..."}`` (URL parse)
    - ``promoted_offerings`` → ``catalogs`` (rename)
    - ``campaign_ref`` → ``buyer_campaign_ref`` (create_media_buy only)
    - Package-level ``optimization_goal`` → ``optimization_goals`` (scalar→array)
    - Package-level ``catalog`` → ``catalogs`` (scalar→array)

    If both the deprecated and current field name are present, the current
    name takes precedence and the deprecated name is removed.

    Args:
        params: Request parameters dict.
        task_name: ADCP task/tool name (e.g. ``"create_media_buy"``).
            Enables tool-scoped renames when provided.

    Returns:
        New dict with deprecated field names replaced by current names.
        Original dict is not mutated (top-level copy; packages list is
        copied if package-level transforms apply).
    """
    result = dict(params)

    # Structural transforms
    _normalize_account(result)
    _normalize_brand_manifest(result)

    # Package-level transforms (deep copy the packages list)
    if "packages" in result and isinstance(result["packages"], list):
        result["packages"] = [
            dict(pkg) if isinstance(pkg, dict) else pkg for pkg in result["packages"]
        ]
        _normalize_packages(result)

    # Global renames
    for old_name, new_name in _GLOBAL_RENAMES.items():
        if old_name in result:
            if new_name not in result:
                result[new_name] = result.pop(old_name)
            else:
                del result[old_name]

    # Tool-scoped renames
    if task_name:
        tool_renames = _TOOL_RENAMES.get(task_name, {})
        for old_name, new_name in tool_renames.items():
            if old_name in result:
                if new_name not in result:
                    result[new_name] = result.pop(old_name)
                else:
                    del result[old_name]

    return result

Normalize deprecated field names and structures in request params.

Applies known transforms so servers can accept both old and new field formats without duplicating normalization logic in every handler.

Transforms applied:

  • account_id: "123" → account: {account_id: "123"} (structural)
  • brand_manifest: "https://..." → brand: {domain: "..."} (URL parse)
  • promoted_offerings → catalogs (rename)
  • campaign_ref → buyer_campaign_ref (create_media_buy only)
  • Package-level optimization_goal → optimization_goals (scalar→array)
  • Package-level catalog → catalogs (scalar→array)

If both the deprecated and current field name are present, the current name takes precedence and the deprecated name is removed.

Args

params
Request parameters dict.
task_name
ADCP task/tool name (e.g. "create_media_buy"). Enables tool-scoped renames when provided.

Returns

New dict with deprecated field names replaced by current names. Original dict is not mutated (top-level copy; packages list is copied if package-level transforms apply).

def translate_error(exc: ADCPError | Error, protocol: "Literal['mcp', 'a2a'] | Protocol") ‑> mcp.server.fastmcp.exceptions.ToolError | a2a.utils.errors.A2AError
Expand source code 
def translate_error(
    exc: ADCPError | Error,
    protocol: Literal["mcp", "a2a"] | Protocol,
) -> ToolError | A2AError:
    """Translate an AdCP error to a protocol SDK error type.

    Returns an error that can be directly raised in a protocol handler::

        try:
            result = await handler.create_media_buy(params)
        except ADCPError as e:
            raise translate_error(e, protocol="mcp")

    For MCP, returns ``ToolError`` (from ``mcp.server.fastmcp``).
    For A2A, returns an :class:`~a2a.utils.errors.A2AError` subclass:
    :class:`~a2a.utils.errors.InvalidParamsError` for correctable errors
    (client can fix) or :class:`~a2a.utils.errors.InternalError` for
    transient/terminal (server-side or unfixable).

    The ``data`` field on A2A errors preserves recovery classification,
    error_code, suggestion, and details so buyer agents can make
    retry/fix/abandon decisions.

    Args:
        exc: An ADCPError exception or an Error Pydantic model.
        protocol: Target protocol - ``"mcp"`` or ``"a2a"``.

    Returns:
        ``ToolError`` for MCP, :class:`~a2a.utils.errors.A2AError`
        subclass for A2A. Raise the result.

    Raises:
        ValueError: If protocol is not ``"mcp"`` or ``"a2a"``.

    Warning:
        Error details are passed through to the caller. Do not include
        internal state (stack traces, SQL queries, internal URLs) in
        Error objects passed to this function.
    """
    proto = protocol.value if isinstance(protocol, Protocol) else str(protocol)
    proto = proto.lower()
    if proto not in ("mcp", "a2a"):
        raise ValueError(f"protocol must be 'mcp' or 'a2a', got {protocol!r}")

    code, message, recovery, field, suggestion, details, errors = _extract_structured_fields(exc)

    if proto == "mcp":
        return _to_mcp(code, message, suggestion=suggestion, field=field, details=details)
    return _to_a2a(
        code,
        message,
        recovery=recovery,
        suggestion=suggestion,
        details=details,
        errors=errors,
    )

Translate an AdCP error to a protocol SDK error type.

Returns an error that can be directly raised in a protocol handler::

try:
    result = await handler.create_media_buy(params)
except ADCPError as e:
    raise translate_error(e, protocol="mcp")

For MCP, returns ToolError (from mcp.server.fastmcp). For A2A, returns an :class:~a2a.utils.errors.A2AError subclass: :class:~a2a.utils.errors.InvalidParamsError for correctable errors (client can fix) or :class:~a2a.utils.errors.InternalError for transient/terminal (server-side or unfixable).

The data field on A2A errors preserves recovery classification, error_code, suggestion, and details so buyer agents can make retry/fix/abandon decisions.

Args

exc
An ADCPError exception or an Error Pydantic model.
protocol
Target protocol - "mcp" or "a2a".

Returns

ToolError for MCP, :class:~a2a.utils.errors.A2AError subclass for A2A. Raise the result.

Raises

ValueError
If protocol is not "mcp" or "a2a".

Warning

Error details are passed through to the caller. Do not include internal state (stack traces, SQL queries, internal URLs) in Error objects passed to this function.