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 canonical 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. Public server factories enforce it strictly by default; validation=None is the explicit compatibility escape hatch.

Generated Pydantic models are structural application types. They preserve common field constraints but do not promise complete parity with JSON Schema conditionals or protocol-specific x-adcp-validation rules. Canonical schema validation is therefore authoritative at wire boundaries.

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 any(issue.keyword == "schema_unavailable" for issue in issues):
        from adcp.validation.envelope import SUPPORTED_WIRE_VERSIONS

        payload["code"] = "VERSION_UNSUPPORTED"
        payload["field"] = "adcp_version"
        payload["recovery"] = "correctable"
        payload["details"]["supported_versions"] = list(SUPPORTED_WIRE_VERSIONS)
        return payload
    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_bundle_adcp_version(*, version: str | None = None) ‑> str | None

Expand source code

def get_bundle_adcp_version(*, version: str | None = None) -> str | None:
    """Return the exact AdCP release declared by a resolved schema bundle.

    Stable validation normally collapses patch releases to a shared
    ``MAJOR.MINOR`` cache key. Exact-source compatibility coordinators compare
    this value with the negotiated patch and fail closed on a mismatch.
    """

    state = _ensure_state(version)
    if state is None:
        return None
    index_file = state.root.root / "index.json"
    try:
        index = json.loads(index_file.read_text())
    except (OSError, json.JSONDecodeError) as exc:
        logger.warning("Failed to load schema bundle index %s: %s", index_file, exc)
        return None
    declared = index.get("adcp_version")
    return declared if isinstance(declared, str) else None

Return the exact AdCP release declared by a resolved schema bundle.

Stable validation normally collapses patch releases to a shared MAJOR.MINOR cache key. Exact-source compatibility coordinators compare this value with the negotiated patch and fail closed on a mismatch.

def get_mcp_schema(tool_name: str,
direction: "Literal['request', 'sync']",
*,
version: str | None = None) ‑> dict[str, typing.Any] | None

Expand source code

def get_mcp_schema(
    tool_name: str,
    direction: Literal["request", "sync"],
    *,
    version: str | None = None,
) -> dict[str, Any] | None:
    """Return the compact transport schema used for MCP ``tools/list``.

    Newer bundles provide self-contained production-profile schemas that
    remove duplicated descriptions and definitions. Releases without those
    artifacts fall back to their canonical versioned schema.

    Successful materializations belong to the immutable versioned loader
    state. Each caller receives an independent, alias-free JSON tree; missing
    or invalid schemas are not added to the materialization cache.
    """
    state = _ensure_state(version)
    if state is None:
        return None
    key = (tool_name, direction)
    cached = state.mcp_schemas.get(key)
    if cached is None:
        with state.mcp_schema_lock:
            # Only one concurrent first caller traverses the reference graph.
            cached = state.mcp_schemas.get(key)
            if cached is None:
                file = (
                    state.mcp_index.get(key)
                    or state.source_index.get(key)
                    or 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 MCP schema %s for %s::%s: %s",
                        file,
                        tool_name,
                        direction,
                        exc,
                    )
                    return None
                if not isinstance(schema, dict):
                    logger.warning("MCP schema %s is not a JSON object", file)
                    return None
                try:
                    portable = _self_contained_schema(
                        state,
                        file,
                        _effective_task_schema(
                            schema, tool_name, direction, bundle_key=state.bundle_key
                        ),
                    )
                except (OSError, json.JSONDecodeError, KeyError, ValueError) as exc:
                    logger.warning("Failed to make MCP schema %s portable: %s", file, exc)
                    return None
                compact = _strip_schema_annotations(portable)
                if not isinstance(compact, dict):
                    return None
                cached = json.dumps(compact, separators=(",", ":"))
                state.mcp_schemas[key] = cached
    return cast(dict[str, Any], json.loads(cached))

Return the compact transport schema used for MCP tools/list.

Newer bundles provide self-contained production-profile schemas that remove duplicated descriptions and definitions. Releases without those artifacts fall back to their canonical versioned schema.

Successful materializations belong to the immutable versioned loader state. Each caller receives an independent, alias-free JSON tree; missing or invalid schemas are not added to the materialization cache.

def get_portable_schema(tool_name: str, direction: Direction, *, version: str | None = None) ‑> dict[str, typing.Any] | None

Expand source code

def get_portable_schema(
    tool_name: str,
    direction: Direction,
    *,
    version: str | None = None,
) -> dict[str, Any] | None:
    """Return a self-contained schema safe outside its source directory."""
    state = _ensure_state(version)
    if state is None:
        return None
    key = (tool_name, direction)
    cached = state.portable.get(key)
    if cached is not None:
        return deepcopy(cached)
    file = state.source_index.get(key) or state.file_index.get(key)
    if file is None:
        return None
    try:
        schema = json.loads(file.read_text())
        if not isinstance(schema, dict):
            raise ValueError("schema root is not an object")
        portable = _self_contained_schema(
            state,
            file,
            _effective_task_schema(schema, tool_name, direction, bundle_key=state.bundle_key),
        )
    except (OSError, json.JSONDecodeError, KeyError, ValueError) as exc:
        logger.warning("Failed to make schema %s portable for %s: %s", file, key, exc)
        return None
    state.portable[key] = portable
    return deepcopy(portable)

Return a self-contained schema safe outside its source directory.

def get_schema(tool_name: str, direction: Direction, *, version: str | None = None) ‑> dict[str, typing.Any] | None

Expand source code

def get_schema(
    tool_name: str,
    direction: Direction,
    *,
    version: str | None = None,
) -> dict[str, Any] | None:
    """Return a defensive copy of a bundled version-specific JSON Schema.

    This is the non-compiled counterpart to :func:`get_validator`. It powers
    version-scoped public models and MCP ``tools/list`` advertisement, both of
    which need the schema document itself rather than only a validator.
    """
    state = _ensure_state(version)
    if state is None:
        return None
    file = state.file_index.get((tool_name, direction))
    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: %s",
            file,
            tool_name,
            direction,
            exc,
        )
        return None
    if not isinstance(schema, dict):
        logger.warning("Schema %s is not a JSON object", file)
        return None
    return deepcopy(
        _effective_task_schema(schema, tool_name, direction, bundle_key=state.bundle_key)
    )

Return a defensive copy of a bundled version-specific JSON Schema.

This is the non-compiled counterpart to :func:get_validator(). It powers version-scoped public models and MCP tools/list advertisement, both of which need the schema document itself rather than only a validator.

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 fresh validator for ``(tool_name, direction, version)``.

    Schema loading and reference discovery are cached. Each lookup creates
    an independent resolver so callers can validate concurrently. Obtain a
    validator per caller rather than sharing one instance between threads.

    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.make()
    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
    if file.is_relative_to(state.root.bundled):
        schema = _normalize_bundled_schema_for_validation(schema)
    schema = _effective_task_schema(schema, tool_name, direction, bundle_key=state.bundle_key)

    try:
        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.20.0'"
        ) from exc

    with _compile_lock:
        # Re-check: another thread may have prepared the validator spec for
        # this key while we were loading the schema off disk.
        cached = state.compiled.get(key)
        if cached is not None:
            return cached.make()
        try:
            _load_schema_registry(state)
            base_uri = file.resolve().as_uri()
            schema_id = schema.get("$id") if isinstance(schema, dict) else None
            # Flattened bundles have no nested $id scopes after normalization.
            # Their fragment refs only need the root document. Avoid copying
            # the entire modular registry into every fresh resolver.
            if (
                file.is_relative_to(state.root.bundled)
                and isinstance(schema, dict)
                and not _has_external_refs(schema)
                and not (isinstance(schema_id, str) and urlsplit(schema_id).fragment)
            ):
                store = {base_uri: schema}
                if isinstance(schema_id, str):
                    store[schema_id] = schema
                    store[urljoin(base_uri, schema_id)] = schema
            else:
                store = _reachable_registry_store(state.registry, base_uri, schema)
            spec = _ValidatorSpec(
                schema=schema,
                base_uri=base_uri,
                store=store,
                format_checker=_build_format_checker(),
                bundle_key=state.bundle_key,
            )
            validator = spec.make()
        except SchemaError as exc:
            logger.warning("Invalid schema %s for %s: %s", file, key, exc)
            return None
        state.compiled[key] = spec
        return validator

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

Schema loading and reference discovery are cached. Each lookup creates an independent resolver so callers can validate concurrently. Obtain a validator per caller rather than sharing one instance between threads.

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,
*,
version: str | None = None) ‑> ValidationOutcome

Expand source code

def validate_incoming_response(
    tool_name: str,
    data: Any,
    mode: ValidationMode,
    debug_logs: list[DebugLogEntry] | None = None,
    *,
    version: str | 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, version=version)
    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,
*,
version: str | None = None) ‑> ValidationOutcome | None

Expand source code

def validate_outgoing_request(
    tool_name: str,
    params: Any,
    mode: ValidationMode,
    debug_logs: list[DebugLogEntry] | None = None,
    *,
    version: str | 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, version=version)
    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.
    """
    payload = _as_wire_payload(payload)
    validator = get_validator(tool_name, "request", version=version)
    if validator is None:
        missing = _missing_explicit_schema_outcome(tool_name, "request", version)
        if missing is not None:
            return missing
        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, _as_wire_payload(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:
        missing = _missing_explicit_schema_outcome(tool_name, used_variant, version)
        if missing is not None:
            return missing
        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: ``"VERSION_UNSUPPORTED"`` when no explicit-version schema
            exists; otherwise ``"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 any(issue.keyword == "schema_unavailable" for issue in issues):
            from adcp.validation.envelope import SUPPORTED_WIRE_VERSIONS

            self.code = "VERSION_UNSUPPORTED"
            self.details["supported_versions"] = list(SUPPORTED_WIRE_VERSIONS)
        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

"VERSION_UNSUPPORTED" when no explicit-version schema exists; otherwise "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