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 compiles validators 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, compiled validators, 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_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, 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.