Module adcp.server.responses

Response builder helpers for ADCP servers.

These functions produce correctly-shaped AdCP response dicts that match the generated Pydantic response schemas. They reduce boilerplate and ensure schema compliance.

Every builder here matches the field names in the corresponding generated response type (e.g., SyncAccountsResponse uses "accounts", SyncCreativesResponse uses "creatives").

Usage

from adcp.server.responses import capabilities_response, products_response

@mcp.tool() async def get_adcp_capabilities(): return capabilities_response(["media_buy"])

@mcp.tool() async def get_products(): return products_response(MY_PRODUCTS)

Functions

def activate_signal_response(deployments: list[dict[str, Any]], *, sandbox: bool = True) ‑> dict[str, typing.Any]

Expand source code

def activate_signal_response(
    deployments: list[dict[str, Any]],
    *,
    sandbox: bool = True,
) -> dict[str, Any]:
    """Build an activate_signal success response.

    Each deployment should include: type, is_live, activation_key.
    For platform: platform, account.
    For agent: agent_url.
    Matches ActivateSignalResponse1 (success) schema.
    """
    return {
        "deployments": deployments,
        "sandbox": sandbox,
    }

Build an activate_signal success response.

Each deployment should include: type, is_live, activation_key. For platform: platform, account. For agent: agent_url. Matches ActivateSignalResponse1 (success) schema.

def build_creative_response(creative_manifest: dict[str, Any] | list[dict[str, Any]],
*,
sandbox: bool = True) ‑> dict[str, typing.Any]

Expand source code

def build_creative_response(
    creative_manifest: dict[str, Any] | list[dict[str, Any]],
    *,
    sandbox: bool = True,
) -> dict[str, Any]:
    """Build a build_creative success response.

    Accepts either a single manifest dict or a list of manifests.
    Each manifest should include: format_id, name, assets.

    Single manifest matches BuildCreativeResponse1.
    List matches BuildCreativeResponse2 (multi-format).
    """
    if isinstance(creative_manifest, list):
        return {
            "creative_manifests": creative_manifest,
            "sandbox": sandbox,
        }
    return {
        "creative_manifest": creative_manifest,
        "sandbox": sandbox,
    }

Build a build_creative success response.

Accepts either a single manifest dict or a list of manifests. Each manifest should include: format_id, name, assets.

Single manifest matches BuildCreativeResponse1. List matches BuildCreativeResponse2 (multi-format).

def capabilities_response(supported_protocols: list[str],
*,
major_versions: list[int] | None = None,
sandbox: bool = True,
features: dict[str, Any] | None = None,
idempotency: dict[str, Any] | None = None,
compliance_testing: dict[str, Any] | None = None) ‑> dict[str, typing.Any]

Expand source code

def capabilities_response(
    supported_protocols: list[str],
    *,
    major_versions: list[int] | None = None,
    sandbox: bool = True,
    features: dict[str, Any] | None = None,
    idempotency: dict[str, Any] | None = None,
    compliance_testing: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Build a get_adcp_capabilities response.

    Args:
        supported_protocols: e.g. ["media_buy"], ["media_buy", "signals"].
            Valid values: media_buy, signals, governance, creative, brand,
            sponsored_intelligence. ``compliance_testing`` is NOT a protocol —
            pass it via the ``compliance_testing`` kwarg.
        major_versions: AdCP major versions. Defaults to [3].
        sandbox: Whether this is a sandbox agent. Defaults to True.
        features: Additional feature flags.
        idempotency: Optional idempotency declaration, nested under
            ``adcp.idempotency`` per AdCP #2315. Pass the output of
            :meth:`adcp.server.idempotency.IdempotencyStore.capability` here
            to declare the seller's ``replay_ttl_seconds``.
        compliance_testing: Optional top-level ``compliance_testing`` block
            to advertise compliance-testing capabilities. When provided,
            emitted as a sibling of ``adcp`` in the response.

    Example::

        from adcp.server.responses import capabilities_response
        from adcp.server.idempotency import IdempotencyStore, MemoryBackend

        store = IdempotencyStore(backend=MemoryBackend(), ttl_seconds=86400)
        return capabilities_response(
            ["media_buy"],
            idempotency=store.capability(),
        )
    """
    adcp_info: dict[str, Any] = {"major_versions": major_versions or [3]}
    if idempotency:
        adcp_info["idempotency"] = idempotency
    resp: dict[str, Any] = {
        "adcp": adcp_info,
        "supported_protocols": supported_protocols,
        "sandbox": sandbox,
    }
    if features:
        resp["features"] = features
    if compliance_testing is not None:
        resp["compliance_testing"] = compliance_testing
    return resp

Build a get_adcp_capabilities response.

Args

supported_protocols

e.g. ["media_buy"], ["media_buy", "signals"]. Valid values: media_buy, signals, governance, creative, brand, sponsored_intelligence. compliance_testing is NOT a protocol — pass it via the compliance_testing kwarg.

major_versions

AdCP major versions. Defaults to [3].

sandbox

Whether this is a sandbox agent. Defaults to True.

features

Additional feature flags.

idempotency

Optional idempotency declaration, nested under adcp.idempotency per AdCP #2315. Pass the output of :meth:IdempotencyStore.capability() here to declare the seller's replay_ttl_seconds.

compliance_testing

Optional top-level compliance_testing block to advertise compliance-testing capabilities. When provided, emitted as a sibling of adcp in the response.

Example::

from adcp.server.responses import capabilities_response
from adcp.server.idempotency import IdempotencyStore, MemoryBackend

store = IdempotencyStore(backend=MemoryBackend(), ttl_seconds=86400)
return capabilities_response(
    ["media_buy"],
    idempotency=store.capability(),
)

def creative_formats_response(formats: list[Any], *, sandbox: bool = True) ‑> dict[str, typing.Any]

Expand source code

def creative_formats_response(
    formats: list[Any],
    *,
    sandbox: bool = True,
) -> dict[str, Any]:
    """Build a list_creative_formats response.

    Each format should include: format_id ({agent_url, id}), name.
    Matches ListCreativeFormatsResponse schema.
    """
    return {
        "formats": _serialize(formats),
        "sandbox": sandbox,
    }

Build a list_creative_formats response.

Each format should include: format_id ({agent_url, id}), name. Matches ListCreativeFormatsResponse schema.

def delivery_response(media_buy_deliveries: list[dict[str, Any]],
*,
reporting_period: dict[str, str] | None = None,
currency: str = 'USD',
sandbox: bool = True) ‑> dict[str, typing.Any]

Expand source code

def delivery_response(
    media_buy_deliveries: list[dict[str, Any]],
    *,
    reporting_period: dict[str, str] | None = None,
    currency: str = "USD",
    sandbox: bool = True,
) -> dict[str, Any]:
    """Build a get_media_buy_delivery response.

    Each media_buy_delivery should include:
        media_buy_id, status, totals (impressions, spend, etc.), by_package.

    Matches GetMediaBuyDeliveryResponse schema.

    Args:
        media_buy_deliveries: Array of delivery data per media buy.
        reporting_period: {"start": ISO timestamp, "end": ISO timestamp}.
            Defaults to current timestamp for both.
        currency: ISO 4217 currency code.
        sandbox: Whether this is simulated data.
    """
    now = datetime.now(timezone.utc).isoformat()
    return {
        "reporting_period": reporting_period or {"start": now, "end": now},
        "media_buy_deliveries": media_buy_deliveries,
        "currency": currency,
        "sandbox": sandbox,
    }

Build a get_media_buy_delivery response.

Each media_buy_delivery should include: media_buy_id, status, totals (impressions, spend, etc.), by_package.

Matches GetMediaBuyDeliveryResponse schema.

Args

media_buy_deliveries

Array of delivery data per media buy.

reporting_period

{"start": ISO timestamp, "end": ISO timestamp}. Defaults to current timestamp for both.

currency

ISO 4217 currency code.

sandbox

Whether this is simulated data.

def error_response(code: str, message: str) ‑> dict[str, typing.Any]

Expand source code

def error_response(code: str, message: str) -> dict[str, Any]:
    """Build a single AdCP error object (not a full error response).

    .. deprecated::
        Use ``adcp_error()`` from ``adcp.server.helpers`` instead.
        It returns a properly wrapped ``{"errors": [...]}`` response with
        auto-recovery classification. This function returns an unwrapped
        single error dict ``{"code": ..., "message": ...}`` which is not
        a valid ADCP error response on its own.
    """
    return {"code": code, "message": message}

Build a single AdCP error object (not a full error response).

Deprecated

Use adcp_error() from adcp.server.helpers instead. It returns a properly wrapped {"errors": [...]} response with auto-recovery classification. This function returns an unwrapped single error dict {"code": ..., "message": ...} which is not a valid ADCP error response on its own.

def list_creatives_response(creatives: list[Any],
*,
pagination: dict[str, Any] | None = None,
sandbox: bool = True) ‑> dict[str, typing.Any]

Expand source code

def list_creatives_response(
    creatives: list[Any],
    *,
    pagination: dict[str, Any] | None = None,
    sandbox: bool = True,
) -> dict[str, Any]:
    """Build a list_creatives response.

    Each creative should include: creative_id, name, format_id, status.
    Matches ListCreativesResponse schema.

    Timestamp defaults: every Creative item in the spec requires
    ``created_date`` and ``updated_date`` (ISO 8601 UTC). For any dict
    item that omits either field, this helper fills it with the current
    UTC timestamp (``datetime.now(timezone.utc).isoformat()``). Both
    fields default to the same value when neither is provided, which
    matches the intuitive meaning for a freshly-listed item. Explicit
    caller-provided values are always preserved. Pydantic model items
    are passed through ``_serialize`` unchanged — callers using typed
    Creative models should set timestamps on the model.
    """
    now = datetime.now(timezone.utc).isoformat()
    filled: list[Any] = []
    for item in creatives:
        if isinstance(item, dict):
            has_created = "created_date" in item and item["created_date"] is not None
            has_updated = "updated_date" in item and item["updated_date"] is not None
            if has_created and has_updated:
                filled.append(item)
                continue
            patched = dict(item)
            if not has_created:
                patched["created_date"] = now
            if not has_updated:
                patched["updated_date"] = now
            filled.append(patched)
        else:
            filled.append(item)

    count = len(filled)
    return {
        "creatives": _serialize(filled),
        "pagination": pagination or {"total": count, "has_more": False},
        "query_summary": {"total_results": count, "total_matching": count, "returned": count},
        "sandbox": sandbox,
    }

Build a list_creatives response.

Each creative should include: creative_id, name, format_id, status. Matches ListCreativesResponse schema.

Timestamp defaults: every Creative item in the spec requires created_date and updated_date (ISO 8601 UTC). For any dict item that omits either field, this helper fills it with the current UTC timestamp (datetime.now(timezone.utc).isoformat()). Both fields default to the same value when neither is provided, which matches the intuitive meaning for a freshly-listed item. Explicit caller-provided values are always preserved. Pydantic model items are passed through _serialize unchanged — callers using typed Creative models should set timestamps on the model.

def log_event_response(events_received: int, events_processed: int, *, sandbox: bool = True) ‑> dict[str, typing.Any]

Expand source code

def log_event_response(
    events_received: int,
    events_processed: int,
    *,
    sandbox: bool = True,
) -> dict[str, Any]:
    """Build a log_event success response.

    Matches LogEventResponse1 (success) schema.
    """
    return {
        "events_received": events_received,
        "events_processed": events_processed,
        "sandbox": sandbox,
    }

Build a log_event success response.

Matches LogEventResponse1 (success) schema.

def media_buy_error_response(errors: list[dict[str, str]]) ‑> dict[str, typing.Any]

Expand source code

def media_buy_error_response(errors: list[dict[str, str]]) -> dict[str, Any]:
    """Build a create_media_buy error response.

    Each error dict: {"code": "...", "message": "..."}.
    Matches CreateMediaBuyResponse2 (error) schema.
    """
    return {"errors": errors}

Build a create_media_buy error response.

Each error dict: {"code": "…", "message": "…"}. Matches CreateMediaBuyResponse2 (error) schema.

def media_buy_response(media_buy_id: str,
packages: list[Any],
*,
buyer_ref: str | None = None,
status: str | None = None,
valid_actions: list[str] | None = None,
revision: int | None = None,
confirmed_at: str | None = None,
sandbox: bool = True) ‑> dict[str, typing.Any]

Expand source code

def media_buy_response(
    media_buy_id: str,
    packages: list[Any],
    *,
    buyer_ref: str | None = None,
    status: str | None = None,
    valid_actions: list[str] | None = None,
    revision: int | None = None,
    confirmed_at: str | None = None,
    sandbox: bool = True,
) -> dict[str, Any]:
    """Build a create_media_buy success response.

    Each package should include: package_id, product_id, pricing_option_id, budget.
    Matches CreateMediaBuyResponse1 (success) schema.

    Auto-populates valid_actions from status if not provided.
    Auto-sets revision to 1 and confirmed_at to now if not provided.
    """
    resp: dict[str, Any] = {
        "media_buy_id": media_buy_id,
        "packages": _serialize(packages),
        "revision": revision if revision is not None else 1,
        "confirmed_at": confirmed_at or datetime.now(timezone.utc).isoformat(),
        "sandbox": sandbox,
    }
    if buyer_ref is not None:
        resp["buyer_ref"] = buyer_ref
    if status is not None:
        resp["status"] = status
        if valid_actions is None:
            resp["valid_actions"] = valid_actions_for_status(status)
        else:
            resp["valid_actions"] = valid_actions
    elif valid_actions is not None:
        resp["valid_actions"] = valid_actions
    return resp

Build a create_media_buy success response.

Each package should include: package_id, product_id, pricing_option_id, budget. Matches CreateMediaBuyResponse1 (success) schema.

Auto-populates valid_actions from status if not provided. Auto-sets revision to 1 and confirmed_at to now if not provided.

def media_buys_response(media_buys: list[Any], *, sandbox: bool = True) ‑> dict[str, typing.Any]

Expand source code

def media_buys_response(
    media_buys: list[Any],
    *,
    sandbox: bool = True,
) -> dict[str, Any]:
    """Build a get_media_buys response.

    Each media buy should include: media_buy_id, status, currency, packages.
    Matches GetMediaBuysResponse schema.
    """
    return {
        "media_buys": _serialize(media_buys),
        "sandbox": sandbox,
    }

Build a get_media_buys response.

Each media buy should include: media_buy_id, status, currency, packages. Matches GetMediaBuysResponse schema.

def preview_creative_response(previews: list[dict[str, Any]],
*,
expires_at: str | None = None,
sandbox: bool = True) ‑> dict[str, typing.Any]

Expand source code

def preview_creative_response(
    previews: list[dict[str, Any]],
    *,
    expires_at: str | None = None,
    sandbox: bool = True,
) -> dict[str, Any]:
    """Build a preview_creative single response.

    Each preview should include:
        preview_id, input ({format_id, name, assets}),
        renders ([{render_id, output_format, preview_url, role, dimensions}]).

    Matches PreviewCreativeResponse1 (single) schema.
    """
    return {
        "response_type": "single",
        "previews": previews,
        "expires_at": expires_at or "2099-12-31T23:59:59Z",
        "sandbox": sandbox,
    }

Build a preview_creative single response.

Each preview should include: preview_id, input ({format_id, name, assets}), renders ([{render_id, output_format, preview_url, role, dimensions}]).

Matches PreviewCreativeResponse1 (single) schema.

def products_response(products: list[Any], *, item_count: int | None = None, sandbox: bool = True) ‑> dict[str, typing.Any]

Expand source code

def products_response(
    products: list[Any],
    *,
    item_count: int | None = None,
    sandbox: bool = True,
) -> dict[str, Any]:
    """Build a get_products response.

    Matches GetProductsResponse schema.
    """
    serialized = _serialize(products)
    resp: dict[str, Any] = {
        "products": serialized,
        "item_count": item_count if item_count is not None else len(serialized),
        "sandbox": sandbox,
    }
    return resp

Build a get_products response.

Matches GetProductsResponse schema.

def signals_response(signals: list[Any], *, sandbox: bool = True) ‑> dict[str, typing.Any]

Expand source code

def signals_response(
    signals: list[Any],
    *,
    sandbox: bool = True,
) -> dict[str, Any]:
    """Build a get_signals response.

    Each signal should include: signal_agent_segment_id, name, description,
    signal_type, data_provider, coverage_percentage, deployments, pricing_options, signal_id.
    Matches GetSignalsResponse schema.
    """
    return {
        "signals": _serialize(signals),
        "sandbox": sandbox,
    }

Build a get_signals response.

Each signal should include: signal_agent_segment_id, name, description, signal_type, data_provider, coverage_percentage, deployments, pricing_options, signal_id. Matches GetSignalsResponse schema.

def sync_accounts_response(accounts: list[dict[str, Any]], *, sandbox: bool = True) ‑> dict[str, typing.Any]

Expand source code

def sync_accounts_response(
    accounts: list[dict[str, Any]],
    *,
    sandbox: bool = True,
) -> dict[str, Any]:
    """Build a sync_accounts success response.

    Each account dict should include: account_id, brand, operator,
    action ("created"|"updated"), status ("active"|"pending_approval").

    Matches SyncAccountsResponse1 schema (field: "accounts").
    """
    return {"accounts": accounts, "sandbox": sandbox}

Build a sync_accounts success response.

Each account dict should include: account_id, brand, operator, action ("created"|"updated"), status ("active"|"pending_approval").

Matches SyncAccountsResponse1 schema (field: "accounts").

def sync_catalogs_response(catalogs: list[dict[str, Any]], *, sandbox: bool = True) ‑> dict[str, typing.Any]

Expand source code

def sync_catalogs_response(
    catalogs: list[dict[str, Any]],
    *,
    sandbox: bool = True,
) -> dict[str, Any]:
    """Build a sync_catalogs success response.

    Each catalog should include: catalog_id, action, item_count, items_approved.
    Matches SyncCatalogsResponse1 (success) schema.
    """
    return {
        "catalogs": catalogs,
        "sandbox": sandbox,
    }

Build a sync_catalogs success response.

Each catalog should include: catalog_id, action, item_count, items_approved. Matches SyncCatalogsResponse1 (success) schema.

def sync_creatives_response(creatives: list[dict[str, Any]], *, sandbox: bool = True) ‑> dict[str, typing.Any]

Expand source code

def sync_creatives_response(
    creatives: list[dict[str, Any]],
    *,
    sandbox: bool = True,
) -> dict[str, Any]:
    """Build a sync_creatives success response.

    Each creative dict should include: creative_id, action ("created"|"updated").
    Optionally: status ("processing"|"pending_review"|"approved"|"rejected"|"archived").
    Matches SyncCreativesResponse1 schema (field: "creatives").
    """
    return {"creatives": creatives, "sandbox": sandbox}

Build a sync_creatives success response.

Each creative dict should include: creative_id, action ("created"|"updated"). Optionally: status ("processing"|"pending_review"|"approved"|"rejected"|"archived"). Matches SyncCreativesResponse1 schema (field: "creatives").

def sync_governance_response(accounts: list[dict[str, Any]], *, sandbox: bool = True) ‑> dict[str, typing.Any]

Expand source code

def sync_governance_response(
    accounts: list[dict[str, Any]],
    *,
    sandbox: bool = True,
) -> dict[str, Any]:
    """Build a sync_governance response.

    Each account dict should include: account, status ("synced"),
    governance_agents ([{url, categories}]).
    """
    return {"accounts": accounts, "sandbox": sandbox}

Build a sync_governance response.

Each account dict should include: account, status ("synced"), governance_agents ([{url, categories}]).

def update_media_buy_response(media_buy_id: str,
*,
affected_packages: list[Any] | None = None,
status: str | None = None,
valid_actions: list[str] | None = None,
revision: int | None = None,
sandbox: bool = True) ‑> dict[str, typing.Any]

Expand source code

def update_media_buy_response(
    media_buy_id: str,
    *,
    affected_packages: list[Any] | None = None,
    status: str | None = None,
    valid_actions: list[str] | None = None,
    revision: int | None = None,
    sandbox: bool = True,
) -> dict[str, Any]:
    """Build an update_media_buy success response.

    Matches UpdateMediaBuyResponse1 (success) schema.
    Auto-populates valid_actions from status if not provided.
    """
    resp: dict[str, Any] = {
        "media_buy_id": media_buy_id,
        "sandbox": sandbox,
    }
    if affected_packages is not None:
        resp["affected_packages"] = _serialize(affected_packages)
    if status is not None:
        resp["status"] = status
        if valid_actions is None:
            resp["valid_actions"] = valid_actions_for_status(status)
        else:
            resp["valid_actions"] = valid_actions
    elif valid_actions is not None:
        resp["valid_actions"] = valid_actions
    if revision is not None:
        resp["revision"] = revision
    return resp

Build an update_media_buy success response.

Matches UpdateMediaBuyResponse1 (success) schema. Auto-populates valid_actions from status if not provided.