Module adcp.validation.schema_loader

JSON Schema loader for AdCP tool request/response validation.

Loads the bundled per-tool schemas shipped with the SDK plus the core/ schemas that async response variants $ref, then prepares validator specs lazily by (tool_name, direction, bundle_key).

Schemas live under a per-version bundle key (see :func:resolve_bundle_key()) so multiple AdCP spec versions can coexist. Callers pass an optional version to :func:get_validator(); None defaults to the SDK's compile-time pin (ADCP_VERSION). Each bundle key gets its own _LoaderState — file index, validator specs, core registry — so cross-version traffic doesn't share compilation state.

Discovery paths (first hit wins, per bundle key):

• Installed package — importlib.resources.files("adcp") / "_schemas" / {bundle_key}<code> populated by </code>scripts/bundle_schemas.py before wheel build.
• Dev checkout — <repo>/schemas/cache/{bundle_key}/ (where scripts/sync_schemas.py writes the canonical bundle). Tried when the packaged copy is absent, so editable installs against a fresh clone validate against the repo's schemas.

Functions

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_named_schema_document(relative_path: str, *, version: str | None = None) ‑> dict[str, typing.Any] | None

Expand source code

def get_named_schema_document(
    relative_path: str,
    *,
    version: str | None = None,
) -> dict[str, Any] | None:
    """Return a bundled non-task schema document, uncompiled.

    The document counterpart to :func:`get_named_validator`. Some SDK helpers
    need a schema's *data* rather than its validation behavior -- notably the
    normative ``enumMetadata`` blocks the spec requires SDKs to dispatch on,
    which carry no validation semantics at all.

    ``relative_path`` is relative to the versioned schema root, e.g.
    ``"enums/media-buy-valid-action.json"``. Path traversal is refused and a
    missing document returns ``None``, so an older pin degrades rather than
    raising.
    """
    path = Path(relative_path)
    if path.is_absolute() or not path.parts or ".." in path.parts:
        return None
    state = _ensure_state(version)
    if state is None:
        return None
    file = state.root.root.joinpath(*path.parts)
    try:
        if not file.is_file() or not file.resolve().is_relative_to(state.root.root.resolve()):
            return None
        document = json.loads(file.read_text())
    except (OSError, json.JSONDecodeError):
        return None
    return deepcopy(document) if isinstance(document, dict) else None

Return a bundled non-task schema document, uncompiled.

The document counterpart to :func:get_named_validator(). Some SDK helpers need a schema's data rather than its validation behavior – notably the normative enumMetadata blocks the spec requires SDKs to dispatch on, which carry no validation semantics at all.

relative_path is relative to the versioned schema root, e.g. "enums/media-buy-valid-action.json". Path traversal is refused and a missing document returns None, so an older pin degrades rather than raising.

def get_named_validator(relative_path: str, *, version: str | None = None) ‑> typing.Any | None

Expand source code

def get_named_validator(
    relative_path: str,
    *,
    version: str | None = None,
) -> Any | None:
    """Return a fresh offline validator for a non-task schema in the bundle.

    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.

    Task validation normally goes through :func:`get_validator`.  Some SDK
    helpers also consume standalone protocol documents (for example a
    seller-hosted acceptance-policy catalog) and must validate them against
    the exact schema version shipped by the SDK.  This loader uses the same
    canonical-ID registry and format checker as task validation, so external
    ``$ref`` values are resolved only from the signed local bundle.

    ``relative_path`` is relative to the versioned schema root, such as
    ``"media-buy/acceptance-policy-catalog.json"``.  Invalid or missing paths
    return ``None`` rather than escaping the bundle root.
    """
    path = Path(relative_path)
    if path.is_absolute() or not path.parts or ".." in path.parts:
        return None
    state = _ensure_state(version)
    if state is None:
        return None
    cache_key = path.as_posix()
    cached = state.named_compiled.get(cache_key)
    if cached is not None:
        return cached.make()
    file = state.root.root.joinpath(*path.parts)
    try:
        if not file.is_file() or not file.resolve().is_relative_to(state.root.root.resolve()):
            return None
        schema = json.loads(file.read_text())
    except (OSError, json.JSONDecodeError):
        return None
    if not isinstance(schema, dict):
        return None

    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:
        cached = state.named_compiled.get(cache_key)
        if cached is not None:
            return cached.make()
        try:
            spec = _ValidatorSpec(
                schema=schema,
                base_uri=file.resolve().as_uri(),
                store=_reachable_schema_store(state, file, schema),
                format_checker=_build_format_checker(),
            )
            validator = spec.make()
        except (OSError, json.JSONDecodeError, SchemaError, ValueError):
            return None
        state.named_compiled[cache_key] = spec
        return validator

Return a fresh offline validator for a non-task schema in the bundle.

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.

Task validation normally goes through :func:get_validator(). Some SDK helpers also consume standalone protocol documents (for example a seller-hosted acceptance-policy catalog) and must validate them against the exact schema version shipped by the SDK. This loader uses the same canonical-ID registry and format checker as task validation, so external $ref values are resolved only from the signed local bundle.

relative_path is relative to the versioned schema root, such as "media-buy/acceptance-policy-catalog.json". Invalid or missing paths return None rather than escaping the bundle root.

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