Module adcp.validation

AdCP validation helpers.

Two independent pieces live here:

• Discriminator / mutual-exclusivity checks for adagents.json and product.json raw-dict payloads (legacy.py). These complement Pydantic models when parsing third-party JSON that hasn't yet been coerced.

• Schema-driven validation against the bundled AdCP JSON schemas (adcp.validation.schema_loader, adcp.validation.schema_validator, adcp.validation.schema_errors, adcp.validation.client_hooks). Used pre-send + post-receive on the client, and opt-in at the server dispatcher, to catch field-name drift before it reaches a storyboard run.

Sub-modules

adcp.validation.client_hooks

Client-side hooks that run the schema validator around every AdCP tool call. Pre-send validation blocks malformed requests; post-receive validation …

adcp.validation.envelope

Wire-version detection for inbound AdCP requests …

adcp.validation.legacy

Runtime validation for AdCP data structures …

adcp.validation.oneof_hints

Heuristic hint strings for oneOf near-miss validation failures …

adcp.validation.schema_errors

Convert schema validation failures into thrown errors and the AdCP VALIDATION_ERROR envelope used by server middleware.

adcp.validation.schema_loader

JSON Schema loader for AdCP tool request/response validation …

adcp.validation.schema_validator

Schema-driven validation for AdCP tool requests and responses …

adcp.validation.version

Bundle-key resolution for per-version schema validation …

Functions

def build_adcp_validation_error_payload(tool: str,
side: str,
issues: list[ValidationIssue]) ‑> dict[str, typing.Any]

Expand source code

def build_adcp_validation_error_payload(
    tool: str, side: str, issues: list[ValidationIssue]
) -> dict[str, Any]:
    """Serialize issues into the kwargs expected by the AdCP ``Error`` model.

    Returns a dict with ``code`` / ``message`` / optional ``field`` /
    ``details`` keys — ready to splat into
    ``Error(**build_adcp_validation_error_payload(...))`` or into the
    server's ``adcp_error`` response envelope.

    Messages on every ``ValidationIssue`` are already sanitized (see
    :func:`adcp.validation.schema_validator._safe_message`) — they do
    not echo user-supplied values, so the wire envelope cannot leak
    bearer tokens / PII / prompt-injection strings from the offending
    payload back to the peer.
    """
    first = issues[0] if issues else None
    if first is not None:
        message = f"{tool} {side} failed schema validation at {first.pointer}: {first.message}"
    else:
        message = f"{tool} {side} failed schema validation"

    payload: dict[str, Any] = {
        "code": "VALIDATION_ERROR",
        "message": message,
        "details": {
            "tool": tool,
            "side": side,
            "issues": [_issue_to_wire(i) for i in issues],
        },
    }
    if first is not None and first.pointer:
        payload["field"] = first.pointer
    return payload

Serialize issues into the kwargs expected by the AdCP Error model.

Returns a dict with code / message / optional field / details keys — ready to splat into Error(**build_adcp_validation_error_payload(...)) or into the server's adcp_error response envelope.

Messages on every ValidationIssue are already sanitized (see :func:adcp.validation.schema_validator._safe_message) — they do not echo user-supplied values, so the wire envelope cannot leak bearer tokens / PII / prompt-injection strings from the offending payload back to the peer.

def build_validation_error(tool: str,
side: str,
issues: list[ValidationIssue]) ‑> SchemaValidationError

Expand source code

def build_validation_error(
    tool: str, side: str, issues: list[ValidationIssue]
) -> SchemaValidationError:
    """Build a :class:`SchemaValidationError` carrying every failure.

    Strict-mode client hooks raise this so callers can inspect the full
    pointer list via ``.issues`` and the ``details`` dict.
    """
    return SchemaValidationError(tool, side, issues)

Build a :class:SchemaValidationError carrying every failure.

Strict-mode client hooks raise this so callers can inspect the full pointer list via .issues and the details dict.

def format_issues(issues: list[ValidationIssue],
limit: int = 3) ‑> str

Expand source code

def format_issues(issues: list[ValidationIssue], limit: int = 3) -> str:
    """Render a compact one-line summary of failures — useful for logs.

    Issues already carry sanitized messages (see :func:`_safe_message`),
    so this output is safe to emit to stdlib loggers and debug buffers.
    """
    head = "; ".join(f"{i.pointer} {i.message}" for i in issues[:limit])
    rest = len(issues) - limit
    return f"{head} (+{rest} more)" if rest > 0 else head

Render a compact one-line summary of failures — useful for logs.

Issues already carry sanitized messages (see :func:_safe_message), so this output is safe to emit to stdlib loggers and debug buffers.

def get_validator(tool_name: str, direction: Direction, *, version: str | None = None) ‑> typing.Any | None

Expand source code

def get_validator(
    tool_name: str,
    direction: Direction,
    *,
    version: str | None = None,
) -> Any | None:
    """Return a compiled validator for ``(tool_name, direction, version)``.

    Returns ``None`` when no schema ships for this pair — callers should
    skip validation (e.g., custom tools outside the AdCP catalog, or
    sync-only tools asked for an async variant that doesn't exist, or a
    version whose bundle isn't on disk).

    ``version=None`` resolves to the SDK's compile-time pin
    (``ADCP_VERSION``). Pass a wire-version string (e.g. ``"3.0.7"``,
    ``"2.5"``, ``"3.1.0-beta.1"``) to validate against a non-current
    schema — :func:`adcp.validation.version.resolve_bundle_key` collapses
    it to the cache key.
    """
    state = _ensure_state(version)
    if state is None:
        return None
    key = (tool_name, direction)
    cached = state.compiled.get(key)
    if cached is not None:
        return cached
    file = state.file_index.get(key)
    if file is None:
        return None
    try:
        schema = json.loads(file.read_text())
    except (OSError, json.JSONDecodeError) as exc:
        logger.warning("Failed to load schema %s for %s: %s", file, key, exc)
        return None

    try:
        from jsonschema import Draft7Validator, FormatChecker
        from jsonschema.exceptions import SchemaError
    except ImportError as exc:  # pragma: no cover
        raise RuntimeError(
            "jsonschema is required for AdCP schema validation. "
            "Install with: pip install 'jsonschema>=4.0.0'"
        ) from exc

    with _compile_lock:
        # Re-check: another thread may have compiled the validator for
        # this key while we were loading the schema off disk.
        cached = state.compiled.get(key)
        if cached is not None:
            return cached
        try:
            resolver = _make_ref_resolver(state, file, schema)
            format_checker = FormatChecker()
            format_checker.checks("date-time")(_is_rfc3339_date_time)
            validator = Draft7Validator(
                schema,
                resolver=resolver,
                format_checker=format_checker,
            )
        except SchemaError as exc:
            logger.warning("Invalid schema %s for %s: %s", file, key, exc)
            return None
        state.compiled[key] = validator
        return validator

Return a compiled validator for (tool_name, direction, adcp.validation.version).

Returns None when no schema ships for this pair — callers should skip validation (e.g., custom tools outside the AdCP catalog, or sync-only tools asked for an async variant that doesn't exist, or a version whose bundle isn't on disk).

version=None resolves to the SDK's compile-time pin (ADCP_VERSION). Pass a wire-version string (e.g. "3.0.7", "2.5", "3.1.0-beta.1") to validate against a non-current schema — :func:resolve_bundle_key() collapses it to the cache key.

def list_validator_keys(*, version: str | None = None) ‑> list[str]

Expand source code

def list_validator_keys(*, version: str | None = None) -> list[str]:
    """Every ``tool::direction`` pair with a shipped schema. Used by tests."""
    state = _ensure_state(version)
    if state is None:
        return []
    return sorted(f"{tool}::{direction}" for (tool, direction) in state.file_index)

Every tool::direction pair with a shipped schema. Used by tests.

def resolve_validation_modes(config: ValidationHookConfig | None = None) ‑> tuple[typing.Literal['strict', 'warn', 'off'], typing.Literal['strict', 'warn', 'off']]

Expand source code

def resolve_validation_modes(
    config: ValidationHookConfig | None = None,
) -> tuple[ValidationMode, ValidationMode]:
    """Return the effective ``(requests, responses)`` modes.

    Resolution order (per side):

    1. Explicit ``config.requests`` / ``config.responses`` (when set).
    2. ``ADCP_VALIDATION_MODE`` env var — applies to both sides.
    3. ``ADCP_ENV=prod|production`` flips the response default to
       ``warn``; requests fall back to ``warn`` (the type default).
    4. Hard defaults: ``requests="warn"``, ``responses="strict"``.

    Read at call time (not import time) so tests that mutate env vars
    via ``patch.dict`` work without a module-level reset hook.
    """
    explicit_req = config.requests if config is not None else None
    explicit_resp = config.responses if config is not None else None
    env_mode = _env_validation_mode()

    req: ValidationMode = explicit_req or env_mode or "warn"
    resp: ValidationMode = explicit_resp or env_mode or _default_response_mode()
    return req, resp

Return the effective (requests, responses) modes.

Resolution order (per side):

1. Explicit config.requests / config.responses (when set).
2. ADCP_VALIDATION_MODE env var — applies to both sides.
3. ADCP_ENV=prod|production flips the response default to warn; requests fall back to warn (the type default).
4. Hard defaults: requests="warn", responses="strict".

Read at call time (not import time) so tests that mutate env vars via patch.dict work without a module-level reset hook.

def validate_adagents(adagents: dict[str, Any]) ‑> None

Expand source code

def validate_adagents(adagents: dict[str, Any]) -> None:
    """Validate an adagents.json structure.

    Args:
        adagents: The adagents.json dict

    Raises:
        ValidationError: If validation fails
    """
    authorized_agents = adagents.get("authorized_agents")
    if isinstance(authorized_agents, list):
        for agent in authorized_agents:
            if isinstance(agent, dict):
                validate_agent_authorization(agent)

    revoked = adagents.get("revoked_publisher_domains")
    if revoked is not None:
        if not isinstance(revoked, list):
            raise ValidationError("'revoked_publisher_domains' must be an array")
        for entry in revoked:
            if not isinstance(entry, dict):
                raise ValidationError("revoked_publisher_domains entry must be an object")
            validate_revoked_publisher_domain_entry(entry)

Validate an adagents.json structure.

Args

adagents

The adagents.json dict

Raises

ValidationError

If validation fails

def validate_agent_authorization(agent: dict[str, Any]) ‑> None

Expand source code

def validate_agent_authorization(agent: dict[str, Any]) -> None:
    """Validate agent authorization discriminated union.

    AdCP v2.4.0+ uses discriminated unions with authorization_type discriminator:
    - authorization_type: "property_ids" requires property_ids
    - authorization_type: "property_tags" requires property_tags
    - authorization_type: "inline_properties" requires properties
    - authorization_type: "publisher_properties" requires publisher_properties

    For backward compatibility, also validates the old mutual exclusivity constraint.

    Args:
        agent: An agent dict from adagents.json

    Raises:
        ValidationError: If discriminator or field constraints are violated
    """
    authorization_type = agent.get("authorization_type")
    auth_fields = ["properties", "property_ids", "property_tags", "publisher_properties"]
    present_fields = [field for field in auth_fields if field in agent and agent[field] is not None]

    # If authorization_type discriminator is present, validate discriminated union
    if authorization_type:
        if authorization_type == "property_ids" and "property_ids" not in present_fields:
            raise ValidationError(
                "Agent with authorization_type='property_ids' must have property_ids"
            )
        elif authorization_type == "property_tags" and "property_tags" not in present_fields:
            raise ValidationError(
                "Agent with authorization_type='property_tags' must have property_tags"
            )
        elif authorization_type == "inline_properties" and "properties" not in present_fields:
            raise ValidationError(
                "Agent with authorization_type='inline_properties' must have properties"
            )
        elif (
            authorization_type == "publisher_properties"
            and "publisher_properties" not in present_fields
        ):
            raise ValidationError(
                "Agent with authorization_type='publisher_properties' "
                "must have publisher_properties"
            )
        elif authorization_type not in (
            "property_ids",
            "property_tags",
            "inline_properties",
            "publisher_properties",
        ):
            raise ValidationError(f"Agent has invalid authorization_type: {authorization_type}")

    # Validate mutual exclusivity (for both old and new formats)
    if len(present_fields) > 1:
        raise ValidationError(
            f"Agent authorization cannot have multiple fields: {', '.join(present_fields)}. "
            f"Only one of {', '.join(auth_fields)} is allowed."
        )

    if len(present_fields) == 0:
        raise ValidationError(
            f"Agent authorization must have exactly one of: {', '.join(auth_fields)}."
        )

    # If using publisher_properties, validate each item
    if "publisher_properties" in present_fields:
        for pub_prop in agent["publisher_properties"]:
            validate_publisher_properties_item(pub_prop)

Validate agent authorization discriminated union.

AdCP v2.4.0+ uses discriminated unions with authorization_type discriminator: - authorization_type: "property_ids" requires property_ids - authorization_type: "property_tags" requires property_tags - authorization_type: "inline_properties" requires properties - authorization_type: "publisher_properties" requires publisher_properties

For backward compatibility, also validates the old mutual exclusivity constraint.

Args

agent

An agent dict from adagents.json

Raises

ValidationError

If discriminator or field constraints are violated

def validate_incoming_response(tool_name: str,
data: Any,
mode: ValidationMode,
debug_logs: list[DebugLogEntry] | None = None) ‑> ValidationOutcome

Expand source code

def validate_incoming_response(
    tool_name: str,
    data: Any,
    mode: ValidationMode,
    debug_logs: list[DebugLogEntry] | None = None,
) -> ValidationOutcome:
    """Run response validation per the configured mode.

    * ``off`` — no-op (returns a valid skipped outcome).
    * ``warn`` — log + return the invalid outcome so the caller can
      surface details without failing the task.
    * ``strict`` — return the invalid outcome so the caller fails the task.

    Never raises — matches the existing Python response contract where a
    validation failure turns a task into ``status=FAILED`` rather than
    raising out of the adapter.
    """
    if mode == "off":
        return ValidationOutcome(valid=True, issues=[], variant="skipped")
    outcome = validate_response(tool_name, data)
    if not outcome.valid and mode == "warn":
        _log_warning(debug_logs, tool_name, "response", outcome)
    return outcome

Run response validation per the configured mode.

• off — no-op (returns a valid skipped outcome).
• warn — log + return the invalid outcome so the caller can surface details without failing the task.
• strict — return the invalid outcome so the caller fails the task.

Never raises — matches the existing Python response contract where a validation failure turns a task into status=FAILED rather than raising out of the adapter.

def validate_outgoing_request(tool_name: str,
params: Any,
mode: ValidationMode,
debug_logs: list[DebugLogEntry] | None = None) ‑> ValidationOutcome | None

Expand source code

def validate_outgoing_request(
    tool_name: str,
    params: Any,
    mode: ValidationMode,
    debug_logs: list[DebugLogEntry] | None = None,
) -> ValidationOutcome | None:
    """Run request validation per the configured mode.

    * ``off`` — no-op (returns ``None``; validator is not consulted).
    * ``warn`` — log + continue; returns the outcome.
    * ``strict`` — raise :class:`SchemaValidationError` on failure.
    """
    if mode == "off":
        return None
    outcome = validate_request(tool_name, params)
    if outcome.valid:
        return outcome
    if mode == "warn":
        _log_warning(debug_logs, tool_name, "request", outcome)
        return outcome
    raise build_validation_error(tool_name, "request", outcome.issues)

Run request validation per the configured mode.

• off — no-op (returns None; validator is not consulted).
• warn — log + continue; returns the outcome.
• strict — raise :class:SchemaValidationError on failure.

def validate_product(product: dict[str, Any]) ‑> None

Expand source code

def validate_product(product: dict[str, Any]) -> None:
    """Validate a Product object.

    Args:
        product: Product dict

    Raises:
        ValidationError: If validation fails
    """
    if "publisher_properties" in product and product["publisher_properties"]:
        for item in product["publisher_properties"]:
            validate_publisher_properties_item(item)

Validate a Product object.

Args

product

Product dict

Raises

ValidationError

If validation fails

def validate_publisher_properties_item(item: Any) ‑> None

Expand source code

def validate_publisher_properties_item(item: Any) -> None:
    """Validate a single ``publisher_properties[]`` entry.

    Accepts either a raw ``dict`` (the wire form) or a parsed Pydantic
    model instance (``PublisherPropertySelector1`` / ``…2`` / ``…3``).
    For Pydantic instances the model is coerced via
    ``.model_dump(exclude_none=False)`` and the same checks apply.

    Two XORs are enforced per the publisher-property-selector JSON Schema
    (adcp#4504):

    * Selector XOR: exactly one of ``property_ids`` / ``property_tags``
      is present for ``by_id`` / ``by_tag`` (``all`` requires neither).
    * Publisher XOR: exactly one of ``publisher_domain`` (singular) or
      ``publisher_domains`` (compact array) is present — both or neither
      both fail. ``publisher_domains`` is NOT allowed on
      ``selection_type='by_id'`` since property IDs are publisher-scoped;
      callers wanting per-publisher ID sets must use one entry per
      publisher.

    Why the Pydantic input form matters: ``datamodel-code-generator``
    cannot translate the JSON Schema's
    ``allOf[not[required[both]]] + anyOf[required[either]]`` construct
    into Pydantic field constraints, so the typed surface (selector 1/3
    direct instantiation) is laxer than the schema. Consumers parsing
    via Pydantic should call this helper post-construction to close the
    gap.

    Args:
        item: A single item from publisher_properties array — either a
            ``dict`` or a Pydantic ``BaseModel`` instance.

    Raises:
        ValidationError: If discriminator or field constraints are violated
    """
    if hasattr(item, "model_dump"):
        item = item.model_dump(exclude_none=False)
    if not isinstance(item, dict):
        raise ValidationError(
            "publisher_properties item must be a dict or a Pydantic model "
            f"instance, got {type(item).__name__}"
        )
    selection_type = item.get("selection_type")
    has_property_ids = "property_ids" in item and item["property_ids"] is not None
    has_property_tags = "property_tags" in item and item["property_tags"] is not None
    has_publisher_domain = "publisher_domain" in item and item["publisher_domain"] is not None
    publisher_domains = item.get("publisher_domains")
    has_publisher_domains = publisher_domains is not None

    if selection_type:
        if selection_type == "by_id" and not has_property_ids:
            raise ValidationError(
                "publisher_properties item with selection_type='by_id' must have property_ids"
            )
        elif selection_type == "by_tag" and not has_property_tags:
            raise ValidationError(
                "publisher_properties item with selection_type='by_tag' must have property_tags"
            )
        elif selection_type not in ("all", "by_id", "by_tag"):
            raise ValidationError(
                f"publisher_properties item has invalid selection_type: {selection_type}"
            )

    if has_property_ids and has_property_tags:
        raise ValidationError(
            "publisher_properties item cannot have both property_ids and property_tags. "
            "These fields are mutually exclusive."
        )

    # selection_type='all' carries neither selector array; older callers
    # without the discriminator must still provide one of the two.
    if selection_type not in ("all",) and not has_property_ids and not has_property_tags:
        raise ValidationError(
            "publisher_properties item must have either property_ids or property_tags. "
            "At least one is required."
        )

    if has_publisher_domain and has_publisher_domains:
        raise ValidationError(
            "publisher_properties item cannot have both publisher_domain and "
            "publisher_domains. These fields are mutually exclusive (XOR)."
        )

    if not has_publisher_domain and not has_publisher_domains:
        raise ValidationError(
            "publisher_properties item must have exactly one of publisher_domain "
            "or publisher_domains."
        )

    if has_publisher_domains and selection_type == "by_id":
        # by_id is single-publisher only — property IDs are publisher-scoped,
        # so fanning the same ID set across multiple publishers is meaningless.
        raise ValidationError(
            "publisher_properties item with selection_type='by_id' cannot use "
            "publisher_domains[]; property IDs are publisher-scoped. Use one "
            "entry per publisher with publisher_domain."
        )

    if has_publisher_domains:
        if not isinstance(publisher_domains, list) or len(publisher_domains) == 0:
            raise ValidationError(
                "publisher_properties item publisher_domains must be a non-empty array"
            )
        if any(not isinstance(d, str) or not d for d in publisher_domains):
            raise ValidationError(
                "publisher_properties item publisher_domains entries must be non-empty strings"
            )
        if len(set(publisher_domains)) != len(publisher_domains):
            raise ValidationError(
                "publisher_properties item publisher_domains entries must be unique"
            )

Validate a single publisher_properties[] entry.

Accepts either a raw dict (the wire form) or a parsed Pydantic model instance (PublisherPropertySelector1 / …2 / …3). For Pydantic instances the model is coerced via .model_dump(exclude_none=False) and the same checks apply.

Two XORs are enforced per the publisher-property-selector JSON Schema (adcp#4504):

• Selector XOR: exactly one of property_ids / property_tags is present for by_id / by_tag (all requires neither).
• Publisher XOR: exactly one of publisher_domain (singular) or publisher_domains (compact array) is present — both or neither both fail. publisher_domains is NOT allowed on selection_type='by_id' since property IDs are publisher-scoped; callers wanting per-publisher ID sets must use one entry per publisher.

Why the Pydantic input form matters: datamodel-code-generator cannot translate the JSON Schema's allOf[not[required[both]]] + anyOf[required[either]] construct into Pydantic field constraints, so the typed surface (selector 1/3 direct instantiation) is laxer than the schema. Consumers parsing via Pydantic should call this helper post-construction to close the gap.

Args

item

A single item from publisher_properties array — either a dict or a Pydantic BaseModel instance.

Raises

ValidationError

If discriminator or field constraints are violated

def validate_request(tool_name: str, payload: Any, *, version: str | None = None) ‑> ValidationOutcome

Expand source code

def validate_request(
    tool_name: str,
    payload: Any,
    *,
    version: str | None = None,
) -> ValidationOutcome:
    """Validate an outgoing request against ``{tool}-request.json``.

    ``version=None`` validates against the SDK's compile-time-pinned
    schema. Pass an explicit wire version (e.g. ``"2.5"``, ``"3.0.7"``)
    to validate against a non-current bundle — the dispatcher uses this
    when the buyer claims a legacy ``adcp_major_version`` so the request
    is checked against the schema the buyer actually targets.
    """
    validator = get_validator(tool_name, "request", version=version)
    if validator is None:
        return _OK_SKIPPED
    if _count_nodes(payload, _MAX_PAYLOAD_NODES) >= _MAX_PAYLOAD_NODES:
        return ValidationOutcome(
            valid=False,
            issues=[
                ValidationIssue(
                    pointer="/",
                    message=f"payload exceeds validator size limit ({_MAX_PAYLOAD_NODES} nodes)",
                    keyword="payload_size",
                    schema_path="",
                )
            ],
            variant="request",
        )
    errors = _iter_errors_bounded(validator, payload)
    if not errors:
        return ValidationOutcome(valid=True, issues=[], variant="request")
    root_schema = getattr(validator, "schema", None)
    return ValidationOutcome(
        valid=False,
        issues=[_format_error(e, root_schema, payload) for e in errors],
        variant="request",
    )

Validate an outgoing request against {tool}-request.json.

version=None validates against the SDK's compile-time-pinned schema. Pass an explicit wire version (e.g. "2.5", "3.0.7") to validate against a non-current bundle — the dispatcher uses this when the buyer claims a legacy adcp_major_version so the request is checked against the schema the buyer actually targets.

def validate_response(tool_name: str, payload: Any, *, version: str | None = None) ‑> ValidationOutcome

Expand source code

def validate_response(
    tool_name: str,
    payload: Any,
    *,
    version: str | None = None,
) -> ValidationOutcome:
    """Validate an incoming response, selecting the variant by payload shape.

    ``version`` semantics match :func:`validate_request` — defaults to the
    SDK pin; pass a wire version to validate against a legacy schema.
    """
    payload = _normalize_response_for_validation(tool_name, payload)
    variant: ResponseVariant = _select_response_variant(payload)
    validator = get_validator(tool_name, variant, version=version)
    used_variant: Direction = variant
    if validator is None and variant != "sync":
        validator = get_validator(tool_name, "sync", version=version)
        used_variant = "sync"
    if validator is None:
        return _OK_SKIPPED
    if _count_nodes(payload, _MAX_PAYLOAD_NODES) >= _MAX_PAYLOAD_NODES:
        return ValidationOutcome(
            valid=False,
            issues=[
                ValidationIssue(
                    pointer="/",
                    message=f"payload exceeds validator size limit ({_MAX_PAYLOAD_NODES} nodes)",
                    keyword="payload_size",
                    schema_path="",
                )
            ],
            variant=used_variant,
        )
    errors = _iter_errors_bounded(validator, payload)
    if not errors:
        return ValidationOutcome(valid=True, issues=[], variant=used_variant)
    root_schema = getattr(validator, "schema", None)
    return ValidationOutcome(
        valid=False,
        issues=[_format_error(e, root_schema, payload) for e in errors],
        variant=used_variant,
    )

Validate an incoming response, selecting the variant by payload shape.

adcp.validation.version semantics match :func:validate_request() — defaults to the SDK pin; pass a wire version to validate against a legacy schema.

def validate_revoked_publisher_domain_entry(entry: dict[str, Any]) ‑> None

Expand source code

def validate_revoked_publisher_domain_entry(entry: dict[str, Any]) -> None:
    """Validate a single ``revoked_publisher_domains[]`` entry.

    Required fields: ``publisher_domain`` (non-empty string) and
    ``revoked_at`` (RFC 3339 / ISO 8601 string). Optional ``reason``
    must be one of the four enum values when present. Wall-clock parsing
    of ``revoked_at`` is left to Pydantic (``AwareDatetime`` in the
    generated model) — this helper only enforces shape on raw dicts.
    """
    publisher_domain = entry.get("publisher_domain")
    if not isinstance(publisher_domain, str) or not publisher_domain:
        raise ValidationError(
            "revoked_publisher_domains entry must have a non-empty " "'publisher_domain' string"
        )

    revoked_at = entry.get("revoked_at")
    if not isinstance(revoked_at, str) or not revoked_at:
        raise ValidationError(
            "revoked_publisher_domains entry must have a non-empty "
            "'revoked_at' ISO 8601 timestamp string"
        )

    reason = entry.get("reason")
    if reason is not None and reason not in _REVOCATION_REASONS:
        raise ValidationError(
            f"revoked_publisher_domains entry has invalid reason={reason!r} "
            f"(expected one of: {', '.join(sorted(_REVOCATION_REASONS))})"
        )

Validate a single revoked_publisher_domains[] entry.

Required fields: publisher_domain (non-empty string) and revoked_at (RFC 3339 / ISO 8601 string). Optional reason must be one of the four enum values when present. Wall-clock parsing of revoked_at is left to Pydantic (AwareDatetime in the generated model) — this helper only enforces shape on raw dicts.

Classes

class AdcpValidationErrorDetails (tool: str,
side: str,
issues: list[ValidationIssue])

Expand source code

@dataclass(frozen=True)
class AdcpValidationErrorDetails:
    """Shape of ``adcp_error.details`` inside a server-side
    ``VALIDATION_ERROR`` envelope. Shipped so buyers can index every
    pointer programmatically instead of parsing the free-text message."""

    tool: str
    side: str
    issues: list[ValidationIssue]

Shape of adcp_error.details inside a server-side VALIDATION_ERROR envelope. Shipped so buyers can index every pointer programmatically instead of parsing the free-text message.

Instance variables

var issues : list[ValidationIssue]

var side : str

var tool : str

class DebugLogEntry (*args, **kwargs)

Expand source code

class DebugLogEntry(TypedDict, total=False):
    """Append-only entry shape for the ``debug_logs`` list threaded by
    the client and server call paths. ``total=False`` so callers can
    still construct partial entries."""

    type: str
    message: str
    timestamp: str
    schema_variant: str
    issues: list[dict[str, Any]]

Append-only entry shape for the debug_logs list threaded by the client and server call paths. total=False so callers can still construct partial entries.

Ancestors

• builtins.dict

Class variables

var issues : list[dict[str, typing.Any]]

var message : str

var schema_variant : str

var timestamp : str

var type : str

class SchemaValidationError (tool: str,
side: str,
issues: list[ValidationIssue],
message: str | None = None)

Expand source code

class SchemaValidationError(Exception):
    """Raised by strict-mode client hooks when a payload fails schema.

    Carries the full issue list via :attr:`issues` so callers can inspect
    every JSON Pointer, not just the first. Mirrors the shape of the AdCP
    L3 ``VALIDATION_ERROR`` error envelope.

    Attributes:
        tool: AdCP tool name that was being validated.
        side: ``"request"`` or ``"response"``.
        issues: Every failure, each with a sanitized message.
        code: Always ``"VALIDATION_ERROR"``.
        details: Structured payload mirroring the wire error envelope's
            ``details`` shape — tool/side/issues, ready for programmatic
            inspection by callers that don't want to parse the exception
            message.
    """

    tool: str
    side: str
    issues: list[ValidationIssue]
    code: str
    details: dict[str, Any]

    def __init__(
        self,
        tool: str,
        side: str,
        issues: list[ValidationIssue],
        message: str | None = None,
    ) -> None:
        self.tool = tool
        self.side = side
        self.issues = issues
        self.code = "VALIDATION_ERROR"
        self.details = {
            "tool": tool,
            "side": side,
            "issues": [_issue_to_wire(i) for i in issues],
        }
        if message is None:
            first = issues[0] if issues else None
            if first is not None:
                message = (
                    f"{tool} {side} failed schema validation at "
                    f"{first.pointer}: {first.message}"
                )
            else:
                message = f"{tool} {side} failed schema validation"
        super().__init__(message)

Raised by strict-mode client hooks when a payload fails schema.

Carries the full issue list via :attr:issues so callers can inspect every JSON Pointer, not just the first. Mirrors the shape of the AdCP L3 VALIDATION_ERROR error envelope.

Attributes

tool

AdCP tool name that was being validated.

side

"request" or "response".

issues

Every failure, each with a sanitized message.

code

Always "VALIDATION_ERROR".

details

Structured payload mirroring the wire error envelope's details shape — tool/side/issues, ready for programmatic inspection by callers that don't want to parse the exception message.

Ancestors

• builtins.Exception
• builtins.BaseException

Class variables

var code : str

var details : dict[str, typing.Any]

var issues : list[ValidationIssue]

var side : str

var tool : str

class UnknownFieldPolicy (*args, **kwds)

Expand source code

class UnknownFieldPolicy(str, Enum):
    """Server-side policy for unknown top-level tool arguments.

    Runs at the transport boundary before Pydantic request-model coercion
    can silently accept or drop extra fields.
    """

    REJECT = "reject"
    STRIP = "strip"
    IGNORE = "ignore"

Server-side policy for unknown top-level tool arguments.

Runs at the transport boundary before Pydantic request-model coercion can silently accept or drop extra fields.

Ancestors

• builtins.str
• enum.Enum

Class variables

var IGNORE

var REJECT

var STRIP

class ValidationError (*args, **kwargs)

Expand source code

class ValidationError(ValueError):
    """Raised when runtime validation fails."""

    pass

Raised when runtime validation fails.

Ancestors

• builtins.ValueError
• builtins.Exception
• builtins.BaseException

class ValidationErrorDetails (tool: str,
side: str,
issues: list[ValidationIssue])

Expand source code

@dataclass(frozen=True)
class ValidationErrorDetails:
    """Mirror of :attr:`SchemaValidationError.details` as a typed dataclass.

    Retained as a public type so callers can annotate their own
    intermediate structures without depending on the exception class.
    """

    tool: str
    side: str
    issues: list[ValidationIssue]

Mirror of :attr:SchemaValidationError.details as a typed dataclass.

Retained as a public type so callers can annotate their own intermediate structures without depending on the exception class.

Instance variables

var issues : list[ValidationIssue]

var side : str

var tool : str

class ValidationHookConfig (requests: ValidationMode | None = None,
responses: ValidationMode | None = None,
unknown_fields: "UnknownFieldPolicy | Literal['reject', 'strip', 'ignore'] | None" = None)

Expand source code

@dataclass(frozen=True)
class ValidationHookConfig:
    """Per-side client validation modes.

    Defaults match the TS port (adcontextprotocol/adcp-client#694):

    * ``requests``: ``"warn"`` — strict would break callers that
      intentionally send partial payloads (error-path tests, exploratory
      probes). Storyboards and compliance runners that want hard-stop
      enforcement pass ``requests="strict"`` explicitly.
    * ``responses``: ``"strict"`` in dev/test, ``"warn"`` when
      ``ADCP_ENV`` is set to ``production`` / ``prod``. Strict-by-default
      makes the SDK a compliance harness: drift from an agent fails the
      task on the first call, not the Nth storyboard run.

    Resolution order for both sides at call time:

    1. Explicit value on this config (``requests=`` / ``responses=``).
    2. ``ADCP_VALIDATION_MODE`` env var (``strict`` / ``warn`` / ``off``)
       — applies to both sides unless overridden by an explicit value.
       Matches the TS port (adcontextprotocol/adcp-client).
    3. ``ADCP_ENV=prod|production`` flips the response default to
       ``warn``; requests fall back to the type default.
    4. Defaults: ``requests="warn"``, ``responses="strict"``.

    Only ``ADCP_ENV`` and ``ADCP_VALIDATION_MODE`` are consulted —
    generic ``ENV`` / ``ENVIRONMENT`` would collide with unrelated
    tooling (rails, postgres, 12-factor) and silently flip the SDK's
    default.
    """

    requests: ValidationMode | None = None
    responses: ValidationMode | None = None
    #: Server-side policy for unsupported top-level tool arguments.
    #: ``None`` preserves existing permissive behavior.
    unknown_fields: UnknownFieldPolicy | Literal["reject", "strip", "ignore"] | None = None

Per-side client validation modes.

Defaults match the TS port (adcontextprotocol/adcp-client#694):

• requests: "warn" — strict would break callers that intentionally send partial payloads (error-path tests, exploratory probes). Storyboards and compliance runners that want hard-stop enforcement pass requests="strict" explicitly.
• responses: "strict" in dev/test, "warn" when ADCP_ENV is set to production / prod. Strict-by-default makes the SDK a compliance harness: drift from an agent fails the task on the first call, not the Nth storyboard run.

Resolution order for both sides at call time:

1. Explicit value on this config (requests= / responses=).
2. ADCP_VALIDATION_MODE env var (strict / warn / off) — applies to both sides unless overridden by an explicit value. Matches the TS port (adcontextprotocol/adcp-client).
3. ADCP_ENV=prod|production flips the response default to warn; requests fall back to the type default.
4. Defaults: requests="warn", responses="strict".

Only ADCP_ENV and ADCP_VALIDATION_MODE are consulted — generic ENV / ENVIRONMENT would collide with unrelated tooling (rails, postgres, 12-factor) and silently flip the SDK's default.

Instance variables

var requests : Literal['strict', 'warn', 'off'] | None

var responses : Literal['strict', 'warn', 'off'] | None

var unknown_fields : UnknownFieldPolicy | Literal['reject', 'strip', 'ignore'] | None

Server-side policy for unsupported top-level tool arguments. None preserves existing permissive behavior.

class ValidationIssue (pointer: str, message: str, keyword: str, schema_path: str, hint: str | None = None)

Expand source code

@dataclass(frozen=True)
class ValidationIssue:
    """A single validation failure.

    Attributes:
        pointer: RFC 6901 JSON Pointer to the offending field.
        message: Sanitized, value-free description of the failure.
            Safe to return over the wire; does not echo input data.
        keyword: jsonschema keyword that rejected the payload
            (``required``, ``type``, ``enum``, etc.).
        schema_path: Path inside the schema that rejected the payload.
        hint: Optional near-miss diagnostic naming the closest matching
            ``oneOf`` variant and the wrong discriminator key. Only
            populated when the heuristic in :mod:`adcp.validation.oneof_hints`
            picks a clear winner; ``None`` otherwise. Additive — clients
            that ignore the field behave as before.
    """

    pointer: str
    message: str
    keyword: str
    schema_path: str
    hint: str | None = None

A single validation failure.

Attributes

pointer

RFC 6901 JSON Pointer to the offending field.

message

Sanitized, value-free description of the failure. Safe to return over the wire; does not echo input data.

keyword

jsonschema keyword that rejected the payload (required, type, enum, etc.).

schema_path

Path inside the schema that rejected the payload.

hint

Optional near-miss diagnostic naming the closest matching oneOf variant and the wrong discriminator key. Only populated when the heuristic in :mod:adcp.validation.oneof_hints picks a clear winner; None otherwise. Additive — clients that ignore the field behave as before.

Instance variables

var hint : str | None

var keyword : str

var message : str

var pointer : str

var schema_path : str

class ValidationOutcome (valid: bool,
issues: list[ValidationIssue] = <factory>,
variant: str = 'skipped')

Expand source code

@dataclass(frozen=True)
class ValidationOutcome:
    valid: bool
    issues: list[ValidationIssue] = field(default_factory=list)
    variant: str = "skipped"

ValidationOutcome(valid: 'bool', issues: 'list[ValidationIssue]' = , variant: 'str' = 'skipped')

Instance variables

var issues : list[ValidationIssue]

var valid : bool

var variant : str