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: ServerError(InternalError(message="...", data={...}))

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

Functions

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

Expand source code

def translate_error(
    exc: ADCPError | Error,
    protocol: Literal["mcp", "a2a"] | Protocol,
) -> ToolError | ServerError:
    """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 ``ServerError`` wrapping ``InvalidParamsError``
    (for correctable errors) or ``InternalError`` (for transient/terminal).

    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, ``ServerError`` 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}")

    # Extract structured fields from the input
    if isinstance(exc, Error):
        code = exc.code
        message = exc.message
        suggestion = exc.suggestion
        details = exc.details
        recovery = _recovery_for_code(code)
        errors = None
    elif isinstance(exc, ADCPError):
        code = _error_code_for_exception(exc)
        message = exc.message
        suggestion = exc.suggestion
        recovery = _recovery_for_code(code)
        details = None
        errors = getattr(exc, "errors", None)
    else:
        raise TypeError(f"Expected ADCPError or Error, got {type(exc).__name__}")

    if proto == "mcp":
        return _to_mcp(code, message, suggestion=suggestion)
    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 ServerError wrapping InvalidParamsError (for correctable errors) or InternalError (for transient/terminal).

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