Module adcp.validation.schema_validator

Schema-driven validation for AdCP tool requests and responses.

The client uses this pre-send and post-receive; the opt-in server middleware uses the same core to reject drift at the dispatcher.

Issues carry an RFC 6901 JSON Pointer to the offending field so callers can index every failure programmatically instead of parsing free text.

Issue messages are sanitized: jsonschema's built-in ValidationError.message embeds the offending value verbatim ("'Bearer sk-...' is not of type 'integer'"). That string flows to the wire envelope and to logs, so we never let raw payload values escape. The sanitized form keeps the structural facts (keyword, expected type, enum size, constraint bound) and drops the value.

Functions

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

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

Classes

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