diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e25e194e5..130b92267 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -274,6 +274,10 @@ jobs: if: steps.version-check.outputs.is_prerelease != 'true' run: python scripts/fix_schema_refs.py + - name: Bundle schemas into package + if: steps.version-check.outputs.is_prerelease != 'true' + run: python scripts/bundle_schemas.py + - name: Generate models if: steps.version-check.outputs.is_prerelease != 'true' run: python scripts/generate_types.py diff --git a/.gitignore b/.gitignore index 59c8b6bff..bee6d64b9 100644 --- a/.gitignore +++ b/.gitignore @@ -11,6 +11,10 @@ __pycache__/ build/ develop-eggs/ dist/ +# Bundled schema copy — populated by scripts/bundle_schemas.py before +# wheel build, mirrors schemas/cache/ so setuptools package-data ships +# them. Committed copies would duplicate ~13MB across both trees. +src/adcp/_schemas/ downloads/ eggs/ .eggs/ diff --git a/MANIFEST.in b/MANIFEST.in index 640a95357..3f91b3776 100644 --- a/MANIFEST.in +++ b/MANIFEST.in @@ -2,3 +2,7 @@ include ADCP_VERSION include README.md include LICENSE recursive-include src/adcp py.typed +# Bundled AdCP JSON schemas. ``scripts/bundle_schemas.py`` mirrors +# ``schemas/cache/`` into ``src/adcp/_schemas/`` before ``python -m +# build`` so the validator ships with the wheel. +recursive-include src/adcp/_schemas *.json diff --git a/Makefile b/Makefile index 1334f2849..047a61656 100644 --- a/Makefile +++ b/Makefile @@ -51,6 +51,8 @@ regenerate-schemas: ## Download latest schemas and regenerate models $(PYTHON) scripts/sync_schemas.py @echo "Fixing schema references..." $(PYTHON) scripts/fix_schema_refs.py + @echo "Bundling schemas into package..." + $(PYTHON) scripts/bundle_schemas.py @echo "Generating Pydantic models..." $(PYTHON) scripts/generate_types.py @echo "Consolidating exports..." @@ -92,6 +94,7 @@ clean: ## Clean generated files and caches @echo "✓ Cleaned all generated files and caches" build: ## Build distribution packages + $(PYTHON) scripts/bundle_schemas.py python -m build @echo "✓ Distribution packages built" diff --git a/pyproject.toml b/pyproject.toml index f3e6e2cb4..e4ccfe15e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -52,6 +52,11 @@ dependencies = [ # idempotency middleware to compute the spec-mandated payload hash for # replay detection. Tiny pure-Python dep, no transitive weight. "rfc8785>=0.1.4", + # JSON Schema validation — backs adcp.validation.schema_validator for + # pre-send request / post-receive response drift detection against the + # bundled AdCP schemas. Pin to the Draft 7 generation the schemas + # declare via ``$schema``. + "jsonschema>=4.0.0", ] [project.scripts] @@ -97,7 +102,15 @@ Issues = "https://github.com/adcontextprotocol/adcp-client-python/issues" where = ["src"] [tool.setuptools.package-data] -adcp = ["py.typed", "ADCP_VERSION", "signing/pg/*.sql"] +adcp = [ + "py.typed", + "ADCP_VERSION", + "signing/pg/*.sql", + # AdCP JSON schemas, mirrored from ``schemas/cache/`` by + # ``scripts/bundle_schemas.py`` so the wheel ships them for + # ``adcp.validation.schema_loader``. + "_schemas/**/*.json", +] [tool.black] line-length = 100 @@ -153,6 +166,13 @@ ignore_errors = true module = ["psycopg", "psycopg.*", "psycopg_pool", "psycopg_pool.*"] ignore_missing_imports = true +# jsonschema doesn't ship PEP 561 stubs in the default package; the +# ``types-jsonschema`` stub package is optional and we use +# :class:`Any`-typed returns across the validator boundary anyway. +[[tool.mypy.overrides]] +module = ["jsonschema", "jsonschema.*"] +ignore_missing_imports = true + [tool.pytest.ini_options] testpaths = ["tests"] asyncio_mode = "auto" diff --git a/scripts/bundle_schemas.py b/scripts/bundle_schemas.py new file mode 100644 index 000000000..ecc26d06e --- /dev/null +++ b/scripts/bundle_schemas.py @@ -0,0 +1,46 @@ +#!/usr/bin/env python3 +"""Mirror ``schemas/cache/`` into ``src/adcp/_schemas/`` so the packaged +wheel ships the JSON schemas that ``adcp.validation.schema_loader`` needs. + +The canonical copy lives in ``schemas/cache/`` (populated by +``scripts/sync_schemas.py``). That tree is outside the package, so +setuptools can't include it via ``package-data``. This script copies it +into the package right before build / regenerate, keeping both trees in +lockstep so editable installs and wheel builds both resolve schemas. + +Runs as part of ``make regenerate-schemas`` after ``sync_schemas.py`` and +before ``generate_types.py``. +""" + +from __future__ import annotations + +import shutil +import sys +from pathlib import Path + +REPO_ROOT = Path(__file__).parent.parent +SRC_CACHE = REPO_ROOT / "schemas" / "cache" +DEST = REPO_ROOT / "src" / "adcp" / "_schemas" + + +def main() -> int: + if not SRC_CACHE.is_dir(): + print(f"error: {SRC_CACHE} does not exist — run sync_schemas.py first", file=sys.stderr) + return 1 + + if DEST.exists(): + shutil.rmtree(DEST) + + shutil.copytree( + SRC_CACHE, + DEST, + ignore=shutil.ignore_patterns("*.md", ".hashes.json"), + ) + + count = sum(1 for _ in DEST.rglob("*.json")) + print(f"Bundled {count} schemas into {DEST.relative_to(REPO_ROOT)}") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/adcp/__init__.py b/src/adcp/__init__.py index ea914dce8..39b93fb94 100644 --- a/src/adcp/__init__.py +++ b/src/adcp/__init__.py @@ -415,7 +415,12 @@ uses_deprecated_assets_field, ) from adcp.validation import ( + SchemaValidationError, ValidationError, + ValidationHookConfig, + ValidationIssue, + ValidationMode, + ValidationOutcome, validate_adagents, validate_agent_authorization, validate_product, @@ -832,7 +837,12 @@ def get_adcp_version() -> str: "AdagentsTimeoutError", "RegistryError", # Validation utilities + "SchemaValidationError", "ValidationError", + "ValidationHookConfig", + "ValidationIssue", + "ValidationMode", + "ValidationOutcome", "validate_adagents", "validate_agent_authorization", "validate_product", diff --git a/src/adcp/client.py b/src/adcp/client.py index 84546b599..2dea9f789 100644 --- a/src/adcp/client.py +++ b/src/adcp/client.py @@ -291,6 +291,7 @@ from adcp.types.generated_poc.tmp.identity_match_request import IdentityMatchRequest from adcp.types.generated_poc.tmp.identity_match_response import IdentityMatchResponse from adcp.utils.operation_id import create_operation_id +from adcp.validation.client_hooks import ValidationHookConfig logger = logging.getLogger(__name__) @@ -310,6 +311,7 @@ def __init__( strict_idempotency: bool = False, signing: SigningConfig | None = None, context_id: str | None = None, + validation: ValidationHookConfig | None = None, ): """ Initialize ADCP client for a single agent. @@ -344,6 +346,20 @@ def __init__( ``jwks_uri``. Supported on both A2A and MCP (``mcp_transport="streamable_http"``); SSE-transport MCP logs a warning and falls through unsigned. + validation: Schema-driven validation modes for outgoing + requests and incoming responses against the bundled AdCP + JSON schemas. Defaults (matching the TS port): requests + in ``warn`` mode (drift logged but not blocked — partial + payloads in error-path tests still work) and responses + in ``strict`` mode (agent drift fails the task). The + response mode flips to ``warn`` when any of ``ADCP_ENV`` + / ``PYTHON_ENV`` / ``ENV`` / ``ENVIRONMENT`` is set to + ``production`` / ``prod``. Storyboards and compliance + runners that want hard-stop enforcement everywhere pass + ``validation=ValidationHookConfig(requests="strict", + responses="strict")``; high-throughput callers can set + either side to ``"off"`` to skip the validator entirely + with zero overhead. context_id: A2A-only. Seed the A2A conversation context. Pass a previously-returned ``context_id`` to resume a session across process restarts, or a self-assigned UUID to name @@ -394,6 +410,9 @@ def __init__( self.adapter.idempotency_capability_check = self._ensure_idempotency_capability if signing is not None: self.adapter.signing_request_hook = self._sign_outgoing_request + # Apply schema validation modes (default: requests=warn, responses=strict + # in dev/test, warn in production — see ``ValidationHookConfig`` docs). + self.adapter.configure_validation(validation) if context_id is not None: if not isinstance(self.adapter, A2AAdapter): diff --git a/src/adcp/protocols/a2a.py b/src/adcp/protocols/a2a.py index ec312b2c1..72aea1c33 100644 --- a/src/adcp/protocols/a2a.py +++ b/src/adcp/protocols/a2a.py @@ -31,6 +31,11 @@ from adcp.protocols.base import ProtocolAdapter from adcp.signing.autosign import current_operation as _signing_operation from adcp.types.core import AgentConfig, DebugInfo, TaskResult, TaskStatus +from adcp.validation.client_hooks import ( + validate_incoming_response, + validate_outgoing_request, +) +from adcp.validation.schema_validator import SchemaValidationError, format_issues logger = logging.getLogger(__name__) @@ -241,6 +246,20 @@ async def _call_a2a_tool( params, idempotency_key = _idempotency.inject_key( tool_name, params, client_token=self.idempotency_client_token ) + + # Pre-send schema validation. Matches the MCP adapter: strict mode + # surfaces as TaskStatus.FAILED so the SDK's unified failure model + # is preserved; warn mode logs and continues; off short-circuits. + try: + validate_outgoing_request(tool_name, params, self.request_validation_mode) + except SchemaValidationError as exc: + return TaskResult[Any]( + status=TaskStatus.FAILED, + error=str(exc), + success=False, + idempotency_key=idempotency_key, + ) + a2a_client = await self._get_a2a_client() # Build A2A message @@ -355,6 +374,26 @@ async def _call_a2a_tool( _idempotency.raise_for_idempotency_error( tool_name, task_result.data, self.agent_config.id ) + # Post-receive schema validation. Only runs when the task + # carries data (terminal completion); async interim states + # with ``data=None`` skip naturally. Strict mode flips the + # TaskResult to FAILED; warn mode logs and passes through. + if task_result.success and task_result.data is not None: + response_outcome = validate_incoming_response( + tool_name, task_result.data, self.response_validation_mode + ) + if not response_outcome.valid and self.response_validation_mode == "strict": + task_result = TaskResult[Any]( + status=TaskStatus.FAILED, + error=( + f"Schema validation failed for {tool_name}: " + f"{format_issues(response_outcome.issues)}" + ), + message=task_result.message, + success=False, + debug_info=task_result.debug_info, + idempotency_key=task_result.idempotency_key, + ) return _idempotency.annotate_result(task_result, idempotency_key) else: # Message response (shouldn't happen for send_message, but handle it) diff --git a/src/adcp/protocols/base.py b/src/adcp/protocols/base.py index 819da9df9..a624b233e 100644 --- a/src/adcp/protocols/base.py +++ b/src/adcp/protocols/base.py @@ -10,6 +10,7 @@ from adcp.types.core import AgentConfig, TaskResult, TaskStatus from adcp.utils.response_parser import parse_json_or_text, parse_mcp_content +from adcp.validation.client_hooks import ValidationHookConfig, ValidationMode if TYPE_CHECKING: import httpx @@ -44,6 +45,23 @@ def __init__(self, agent_config: AgentConfig): # via its httpx client's event_hooks; MCP consumes it via a custom # httpx_client_factory passed to streamablehttp_client. self.signing_request_hook: Callable[[httpx.Request], Awaitable[None]] | None = None + # Schema validation modes — resolved by the owning ADCPClient via + # ``configure_validation``. Class defaults match the TS port: warn + # on requests (don't block partial payloads in error-path tests), + # strict on responses (agent drift fails the task on first call). + # Adapters instantiated directly without an ADCPClient inherit + # these defaults; production callers flip responses to warn via + # ``ADCPClient(validation=...)`` or an env override. + self.request_validation_mode: ValidationMode = "warn" + self.response_validation_mode: ValidationMode = "strict" + + def configure_validation(self, config: ValidationHookConfig | None) -> None: + """Apply a client's :class:`ValidationHookConfig` to this adapter.""" + from adcp.validation.client_hooks import resolve_validation_modes + + req, resp = resolve_validation_modes(config) + self.request_validation_mode = req + self.response_validation_mode = resp # ======================================================================== # Helper methods for response parsing diff --git a/src/adcp/protocols/mcp.py b/src/adcp/protocols/mcp.py index e6ed53667..2c6bae12b 100644 --- a/src/adcp/protocols/mcp.py +++ b/src/adcp/protocols/mcp.py @@ -57,6 +57,11 @@ from adcp.protocols.base import ProtocolAdapter from adcp.signing.autosign import current_operation as _signing_operation from adcp.types.core import DebugInfo, TaskResult, TaskStatus +from adcp.validation.client_hooks import ( + validate_incoming_response, + validate_outgoing_request, +) +from adcp.validation.schema_validator import SchemaValidationError, format_issues # Spec-defined limits from docs/building/implementation/mcp-response-extraction.mdx # and docs/building/implementation/transport-errors.mdx. @@ -145,10 +150,7 @@ def extract_adcp_success(result: Any) -> dict[str, Any] | None: parsed = json.loads(text) except (json.JSONDecodeError, ValueError): continue - if ( - isinstance(parsed, dict) - and not (len(parsed) == 1 and "adcp_error" in parsed) - ): + if isinstance(parsed, dict) and not (len(parsed) == 1 and "adcp_error" in parsed): return parsed return None @@ -351,8 +353,8 @@ async def _get_session(self) -> ClientSession: streamable_http_extra: dict[str, Any] = {} if self.signing_request_hook is not None: if self.agent_config.mcp_transport == "streamable_http": - streamable_http_extra["httpx_client_factory"] = ( - _make_signing_http_factory(self.signing_request_hook) + streamable_http_extra["httpx_client_factory"] = _make_signing_http_factory( + self.signing_request_hook ) else: logger.warning( @@ -496,6 +498,19 @@ async def _call_mcp_tool(self, tool_name: str, params: dict[str, Any]) -> TaskRe ) try: + # Pre-send schema validation — throws in strict, logs in warn, + # skips in off. Runs before session setup so a drifted payload + # doesn't even open a connection. + try: + validate_outgoing_request(tool_name, params, self.request_validation_mode) + except SchemaValidationError as exc: + return TaskResult[Any]( + status=TaskStatus.FAILED, + error=str(exc), + success=False, + idempotency_key=idempotency_key, + ) + session = await self._get_session() if self.agent_config.debug: @@ -609,6 +624,27 @@ async def _call_mcp_tool(self, tool_name: str, params: dict[str, Any]) -> TaskRe tool_name, data_to_return, self.agent_config.id ) + # Post-receive schema validation — catches field-name drift from + # agents. Strict mode fails the task; warn mode logs and returns + # the data unchanged; off short-circuits without invoking the + # validator. Never raises — mirrors the existing contract where + # response-side failures surface as TaskStatus.FAILED. + response_outcome = validate_incoming_response( + tool_name, data_to_return, self.response_validation_mode + ) + if not response_outcome.valid and self.response_validation_mode == "strict": + return TaskResult[Any]( + status=TaskStatus.FAILED, + error=( + f"Schema validation failed for {tool_name}: " + f"{format_issues(response_outcome.issues)}" + ), + message=message_text, + success=False, + debug_info=debug_info, + idempotency_key=idempotency_key, + ) + # Return both the structured data and the human-readable message task_result = TaskResult[Any]( status=TaskStatus.COMPLETED, diff --git a/src/adcp/server/mcp_tools.py b/src/adcp/server/mcp_tools.py index 5c8fbbadb..e5a306352 100644 --- a/src/adcp/server/mcp_tools.py +++ b/src/adcp/server/mcp_tools.py @@ -24,6 +24,7 @@ from typing import Any from adcp.server.base import ADCPHandler, ToolContext +from adcp.validation.client_hooks import ValidationHookConfig logger = logging.getLogger(__name__) @@ -1530,6 +1531,8 @@ class when the annotation is: def create_tool_caller( handler: ADCPHandler[Any], method_name: str, + *, + validation: ValidationHookConfig | None = None, ) -> Callable[..., Any]: """Create a tool caller function for an ADCP handler method. @@ -1548,9 +1551,23 @@ def create_tool_caller( ``INVALID_REQUEST`` AdCP error so callers see a spec-typed recovery classification rather than a raw stack trace. + **Schema-driven validation (issue #249).** When ``validation`` is + supplied, the dispatcher validates incoming requests and outgoing + responses against the bundled AdCP JSON schemas. Request failures + raise ``ADCPTaskError(VALIDATION_ERROR)`` before the handler runs, + so malformed payloads never hit business logic. Response failures + either raise ``VALIDATION_ERROR`` (strict) or log a warning + (warn). Defaults to off on the server side — the client-side + hooks already catch drift for SDK-built clients, and enabling + server validation is a deliberate opt-in for authors who want + dispatcher-level enforcement. + Args: handler: The ADCP handler instance method_name: Name of the method to call + validation: Optional :class:`ValidationHookConfig` with + per-side modes (``strict`` / ``warn`` / ``off``). Omitting + it disables server-side schema validation entirely. Returns: Async callable ``call_tool(params, context=None)``. The ``context`` @@ -1566,13 +1583,44 @@ def create_tool_caller( from adcp.exceptions import ADCPTaskError from adcp.server.helpers import inject_context from adcp.types import Error + from adcp.validation.schema_errors import build_adcp_validation_error_payload + from adcp.validation.schema_validator import ( + format_issues, + validate_request, + validate_response, + ) method = getattr(handler, method_name) params_model = _resolve_params_pydantic_model(method) + # Opt-in server-side schema modes. ``None`` keeps validation off + # entirely (zero overhead on the hot path) — the TS-port default for + # ``createAdcpServer`` is the same: validation is an explicit opt-in. + request_mode = validation.requests if validation is not None else None + response_mode = validation.responses if validation is not None else None + async def call_tool(params: dict[str, Any], context: ToolContext | None = None) -> Any: ctx = context if context is not None else ToolContext() raw_params = params # Preserve the original dict for context echo. + + if request_mode is not None and request_mode != "off": + outcome = validate_request(method_name, params) + if not outcome.valid: + summary = format_issues(outcome.issues) + if request_mode == "strict": + payload = build_adcp_validation_error_payload( + method_name, "request", outcome.issues + ) + raise ADCPTaskError( + operation=method_name, + errors=[Error(**payload)], + ) + logger.warning( + "Schema validation warning (request) for %s: %s", + method_name, + summary, + ) + call_params: Any = params if params_model is not None and isinstance(params, dict): try: @@ -1623,6 +1671,30 @@ async def call_tool(params: dict[str, Any], context: ToolContext | None = None) # model (which won't carry the wire ``context`` field). if isinstance(result, dict): inject_context(raw_params, result) + + if response_mode is not None and response_mode != "off" and isinstance(result, dict): + # Skip validation when the handler returned the AdCP L3 + # error envelope (``{"adcp_error": {...}}``). That envelope + # has its own shape enforced by the ``Error`` builder; the + # per-tool response schema would false-positive on it and + # convert a real protocol error into a fake VALIDATION_ERROR. + if "adcp_error" not in result: + outcome = validate_response(method_name, result) + if not outcome.valid: + summary = format_issues(outcome.issues) + logger.warning( + "Schema validation warning (response) for %s: %s", + method_name, + summary, + ) + if response_mode == "strict": + payload = build_adcp_validation_error_payload( + method_name, "response", outcome.issues + ) + raise ADCPTaskError( + operation=method_name, + errors=[Error(**payload)], + ) return result return call_tool @@ -1634,7 +1706,13 @@ class MCPToolSet: Provides tool definitions and handlers for registering with an MCP server. """ - def __init__(self, handler: ADCPHandler[Any], *, advertise_all: bool = False): + def __init__( + self, + handler: ADCPHandler[Any], + *, + advertise_all: bool = False, + validation: ValidationHookConfig | None = None, + ): """Create tool set from handler. Args: @@ -1644,6 +1722,8 @@ def __init__(self, handler: ADCPHandler[Any], *, advertise_all: bool = False): SDK's ``not_supported`` default. See :func:`get_tools_for_handler` for the default behavior (override-filtered advertisement). + validation: Opt-in schema validation config applied to every + tool caller. See :func:`create_tool_caller`. """ self.handler = handler self._filtered_definitions = get_tools_for_handler(handler, advertise_all=advertise_all) @@ -1652,7 +1732,7 @@ def __init__(self, handler: ADCPHandler[Any], *, advertise_all: bool = False): # Create tool callers only for filtered tools for tool_def in self._filtered_definitions: name = tool_def["name"] - self._tools[name] = create_tool_caller(handler, name) + self._tools[name] = create_tool_caller(handler, name, validation=validation) @property def tool_definitions(self) -> list[dict[str, Any]]: @@ -1681,7 +1761,12 @@ def get_tool_names(self) -> list[str]: return list(self._tools.keys()) -def create_mcp_tools(handler: ADCPHandler[Any], *, advertise_all: bool = False) -> MCPToolSet: +def create_mcp_tools( + handler: ADCPHandler[Any], + *, + advertise_all: bool = False, + validation: ValidationHookConfig | None = None, +) -> MCPToolSet: """Create MCP tools from an ADCP handler. This is the main entry point for MCP server integration. @@ -1711,8 +1796,12 @@ async def call_tool(name: str, arguments: dict): advertise_all: When True, advertise every tool the handler type supports — even those whose method is still the SDK's ``not_supported`` default. See :func:`get_tools_for_handler`. + validation: Opt-in schema validation config. When supplied, + every tool caller validates requests and responses against + the bundled AdCP JSON schemas. See + :func:`create_tool_caller` for mode semantics. Returns: MCPToolSet with tool definitions and handlers. """ - return MCPToolSet(handler, advertise_all=advertise_all) + return MCPToolSet(handler, advertise_all=advertise_all, validation=validation) diff --git a/src/adcp/validation/__init__.py b/src/adcp/validation/__init__.py new file mode 100644 index 000000000..d1ec913b3 --- /dev/null +++ b/src/adcp/validation/__init__.py @@ -0,0 +1,84 @@ +"""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 + (``schema_loader``, ``schema_validator``, ``schema_errors``, + ``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. +""" + +from __future__ import annotations + +from adcp.validation.client_hooks import ( + DebugLogEntry, + ValidationHookConfig, + ValidationMode, + resolve_validation_modes, + validate_incoming_response, + validate_outgoing_request, +) +from adcp.validation.legacy import ( + ValidationError, + validate_adagents, + validate_agent_authorization, + validate_product, + validate_publisher_properties_item, +) +from adcp.validation.schema_errors import ( + AdcpValidationErrorDetails, + ValidationErrorDetails, + build_adcp_validation_error_payload, + build_validation_error, +) +from adcp.validation.schema_loader import ( + Direction, + ResponseVariant, + get_validator, + list_validator_keys, +) +from adcp.validation.schema_validator import ( + SchemaValidationError, + ValidationIssue, + ValidationOutcome, + format_issues, + validate_request, + validate_response, +) + +__all__ = [ + # Legacy (adagents / product) + "ValidationError", + "validate_adagents", + "validate_agent_authorization", + "validate_product", + "validate_publisher_properties_item", + # Schema core + "Direction", + "ResponseVariant", + "SchemaValidationError", + "ValidationIssue", + "ValidationOutcome", + "format_issues", + "get_validator", + "list_validator_keys", + "validate_request", + "validate_response", + # Errors + "AdcpValidationErrorDetails", + "ValidationErrorDetails", + "build_adcp_validation_error_payload", + "build_validation_error", + # Client hooks + "DebugLogEntry", + "ValidationHookConfig", + "ValidationMode", + "resolve_validation_modes", + "validate_incoming_response", + "validate_outgoing_request", +] diff --git a/src/adcp/validation/client_hooks.py b/src/adcp/validation/client_hooks.py new file mode 100644 index 000000000..24ca64e37 --- /dev/null +++ b/src/adcp/validation/client_hooks.py @@ -0,0 +1,161 @@ +"""Client-side hooks that run the schema validator around every AdCP tool +call. Pre-send validation blocks malformed requests; post-receive +validation catches field-name drift from agents (issue #249).""" + +from __future__ import annotations + +import logging +import os +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Any, Literal, TypedDict + +from adcp.validation.schema_errors import build_validation_error +from adcp.validation.schema_validator import ( + ValidationOutcome, + format_issues, + validate_request, + validate_response, +) + +logger = logging.getLogger(__name__) + +ValidationMode = Literal["strict", "warn", "off"] + + +@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. + + Only ``ADCP_ENV`` is 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 + + +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]] + + +def _default_response_mode() -> ValidationMode: + """Response default: ``strict`` unless ``ADCP_ENV`` declares production. + + Read at call time (not import time) so tests that ``patch.dict`` the + environment work without a module-level reset hook. + """ + val = os.environ.get("ADCP_ENV") + if val and val.lower() in {"prod", "production"}: + return "warn" + return "strict" + + +def resolve_validation_modes( + config: ValidationHookConfig | None = None, +) -> tuple[ValidationMode, ValidationMode]: + """Return the effective ``(requests, responses)`` modes.""" + req: ValidationMode = (config.requests if config is not None else None) or "warn" + resp: ValidationMode = ( + config.responses if config is not None else None + ) or _default_response_mode() + return req, resp + + +def _log_warning( + debug_logs: list[DebugLogEntry] | None, + tool_name: str, + side: str, + outcome: ValidationOutcome, +) -> None: + # Issue messages are sanitized (see schema_validator._safe_message) so + # this summary is safe to emit to logs without leaking user values. + summary = format_issues(outcome.issues) + logger.warning("Schema validation warning (%s) for %s: %s", side, tool_name, summary) + if debug_logs is None: + return + debug_logs.append( + DebugLogEntry( + type="warning", + message=f"Schema validation warning for {tool_name}: {summary}", + timestamp=datetime.now(timezone.utc).isoformat(), + schema_variant=outcome.variant, + issues=[ + { + "pointer": i.pointer, + "message": i.message, + "keyword": i.keyword, + "schema_path": i.schema_path, + } + for i in outcome.issues + ], + ) + ) + + +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) + + +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 diff --git a/src/adcp/validation.py b/src/adcp/validation/legacy.py similarity index 100% rename from src/adcp/validation.py rename to src/adcp/validation/legacy.py diff --git a/src/adcp/validation/schema_errors.py b/src/adcp/validation/schema_errors.py new file mode 100644 index 000000000..2ad200b8c --- /dev/null +++ b/src/adcp/validation/schema_errors.py @@ -0,0 +1,88 @@ +"""Convert schema validation failures into thrown errors and the AdCP +``VALIDATION_ERROR`` envelope used by server middleware.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +from adcp.validation.schema_validator import SchemaValidationError, ValidationIssue + + +@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] + + +@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] + + +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) + + +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": [ + { + "pointer": i.pointer, + "message": i.message, + "keyword": i.keyword, + "schema_path": i.schema_path, + } + for i in issues + ], + }, + } + if first is not None and first.pointer: + payload["field"] = first.pointer + return payload diff --git a/src/adcp/validation/schema_loader.py b/src/adcp/validation/schema_loader.py new file mode 100644 index 000000000..dd62cdc18 --- /dev/null +++ b/src/adcp/validation/schema_loader.py @@ -0,0 +1,246 @@ +"""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)``. + +Schemas are discovered via two paths: + +* **Installed package** — ``importlib.resources.files("adcp") / "_schemas"`` + populated by ``scripts/bundle_schemas.py`` before wheel build. +* **Dev checkout** — ``/schemas/cache/`` (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. +""" + +from __future__ import annotations + +import json +import logging +import threading +import warnings +from importlib.resources import as_file, files +from pathlib import Path +from typing import Any, Literal + +logger = logging.getLogger(__name__) + +# Serialize first-time init and validator compilation. Concurrent callers +# on a fresh process can otherwise both walk the schema tree or compile +# the same validator twice. Result is the same either way, but the lock +# keeps behaviour deterministic and avoids redundant filesystem walks. +_init_lock = threading.Lock() +_compile_lock = threading.Lock() + +ResponseVariant = Literal["sync", "submitted", "working", "input-required"] +Direction = Literal["request", "sync", "submitted", "working", "input-required"] + + +class _SchemaRoot: + """Filesystem view of the schema tree, regardless of packaged vs dev.""" + + def __init__(self, root: Path) -> None: + self.root = root + self.bundled = root / "bundled" + self.core = root / "core" + + def exists(self) -> bool: + return self.bundled.is_dir() + + +def _resolve_schema_root() -> _SchemaRoot | None: + """Locate the schema tree. Packaged copy wins; fall back to repo layout. + + Returns ``None`` when neither location is populated — the validator + degrades to ``skipped`` for every tool, matching the TS behavior for + tools outside the AdCP catalog. + """ + try: + packaged = files("adcp") / "_schemas" + with as_file(packaged) as p: + packaged_path = Path(p) + if (packaged_path / "bundled").is_dir(): + return _SchemaRoot(packaged_path) + except (ModuleNotFoundError, FileNotFoundError, OSError): + pass + + here = Path(__file__).resolve() + for ancestor in here.parents: + candidate = ancestor / "schemas" / "cache" + if (candidate / "bundled").is_dir(): + return _SchemaRoot(candidate) + if ancestor.parent == ancestor: + break + return None + + +class _LoaderState: + def __init__(self, root: _SchemaRoot) -> None: + self.root = root + self.file_index: dict[tuple[str, Direction], Path] = {} + self.compiled: dict[tuple[str, Direction], Any] = {} + self.registry: dict[str, dict[str, Any]] = {} + self._core_loaded = False + + +_state: _LoaderState | None = None + + +def _walk_json(dir_: Path) -> list[Path]: + if not dir_.is_dir(): + return [] + return sorted(p for p in dir_.rglob("*.json") if p.is_file()) + + +def _build_index(root: _SchemaRoot) -> dict[tuple[str, Direction], Path]: + index: dict[tuple[str, Direction], Path] = {} + + for file in _walk_json(root.bundled): + base = file.stem + if base.endswith("-request"): + tool = base[: -len("-request")].replace("-", "_") + index[(tool, "request")] = file + elif base.endswith("-response"): + tool = base[: -len("-response")].replace("-", "_") + index[(tool, "sync")] = file + + for entry in sorted(root.root.iterdir()): + if not entry.is_dir() or entry.name in ("bundled", "core"): + continue + for file in _walk_json(entry): + base = file.stem + if base.endswith("-async-response-submitted"): + tool = base[: -len("-async-response-submitted")].replace("-", "_") + index[(tool, "submitted")] = file + elif base.endswith("-async-response-working"): + tool = base[: -len("-async-response-working")].replace("-", "_") + index[(tool, "working")] = file + elif base.endswith("-async-response-input-required"): + tool = base[: -len("-async-response-input-required")].replace("-", "_") + index[(tool, "input-required")] = file + + return index + + +def _ensure_state() -> _LoaderState | None: + global _state + if _state is not None: + return _state + with _init_lock: + # Double-checked pattern: re-read inside the lock in case another + # thread initialized while we were waiting. + if _state is not None: + return _state + root = _resolve_schema_root() + if root is None: + logger.debug("AdCP schemas not found; schema validation will skip all tools") + return None + new_state = _LoaderState(root) + new_state.file_index = _build_index(root) + _state = new_state + return _state + + +def _load_core_registry(state: _LoaderState) -> None: + if state._core_loaded: + return + for file in _walk_json(state.root.core): + try: + schema = json.loads(file.read_text()) + except (OSError, json.JSONDecodeError) as exc: + logger.warning("Failed to load core schema %s: %s", file, exc) + continue + schema_id = schema.get("$id") + if isinstance(schema_id, str): + state.registry[schema_id] = schema + state._core_loaded = True + + +def _make_ref_resolver(state: _LoaderState, base_file: Path) -> Any: + """Build a jsonschema ``RefResolver`` rooted at the file's directory. + + Async variant schemas use relative refs like ``../core/context.json``; + giving the resolver a ``file://`` base URI lets those resolve against + disk. Also seeds the core ``$id``-keyed registry so bundled schemas + that reference a core type by canonical id still resolve. + + ``RefResolver`` is deprecated in jsonschema 4.18+ (to be replaced by + the ``referencing`` library). Suppress the warning locally so + downstream projects running ``-W error::DeprecationWarning`` don't + crash on import; migration tracked as a follow-up. + """ + with warnings.catch_warnings(): + warnings.simplefilter("ignore", DeprecationWarning) + try: + from jsonschema import RefResolver + except ImportError as exc: # pragma: no cover - guarded by dep install + raise RuntimeError( + "jsonschema is required for AdCP schema validation. " + "Install with: pip install 'jsonschema>=4.0.0'" + ) from exc + + _load_core_registry(state) + base_uri = base_file.resolve().parent.as_uri() + "/" + return RefResolver(base_uri=base_uri, referrer={}, store=dict(state.registry)) + + +def get_validator(tool_name: str, direction: Direction) -> Any | None: + """Return a compiled validator for ``(tool_name, direction)`` or ``None``. + + ``None`` means 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). + """ + state = _ensure_state() + 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 + 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) + validator = Draft7Validator(schema, resolver=resolver) + except SchemaError as exc: + logger.warning("Invalid schema %s for %s: %s", file, key, exc) + return None + state.compiled[key] = validator + return validator + + +def list_validator_keys() -> list[str]: + """Every ``tool::direction`` pair with a shipped schema. Used by tests.""" + state = _ensure_state() + if state is None: + return [] + return sorted(f"{tool}::{direction}" for (tool, direction) in state.file_index) + + +def _reset_for_tests() -> None: + """Clear cached state so a fresh resolve runs. Test-only.""" + global _state + _state = None diff --git a/src/adcp/validation/schema_validator.py b/src/adcp/validation/schema_validator.py new file mode 100644 index 000000000..5bec81970 --- /dev/null +++ b/src/adcp/validation/schema_validator.py @@ -0,0 +1,350 @@ +"""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. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass, field +from itertools import islice +from typing import Any + +from adcp.validation.schema_loader import Direction, ResponseVariant, get_validator + +# Cap the number of issues returned. A hostile peer sending a deeply- +# nested payload can otherwise force jsonschema to walk and sort the +# entire error tree before we format. The cap is well above what a +# human caller would ever debug against. +_MAX_ISSUES = 50 + +# Cap input size before validation. Protects against a peer posting a +# multi-megabyte object designed to chew CPU in ``iter_errors``. Transport +# layers typically enforce their own limits; this is defense-in-depth. +_MAX_PAYLOAD_NODES = 10_000 + +_REQUIRED_MSG = re.compile(r"^'(?P.+)' is a required property$") + + +@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. + """ + + pointer: str + message: str + keyword: str + schema_path: str + + +@dataclass(frozen=True) +class ValidationOutcome: + valid: bool + issues: list[ValidationIssue] = field(default_factory=list) + variant: str = "skipped" + + +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": [ + { + "pointer": i.pointer, + "message": i.message, + "keyword": i.keyword, + "schema_path": i.schema_path, + } + 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) + + +_OK_SKIPPED = ValidationOutcome(valid=True, issues=[], variant="skipped") + + +def _path_to_pointer(path: Any) -> str: + """Convert a jsonschema ``deque(['packages', 0, 'targeting'])`` to + ``/packages/0/targeting``. Empty path maps to ``/`` per RFC 6901 + convention used in the TS SDK (AJV's ``instancePath='' -> '/'``).""" + if not path: + return "/" + + def escape(seg: Any) -> str: + s = str(seg) + return s.replace("~", "~0").replace("/", "~1") + + return "/" + "/".join(escape(seg) for seg in path) + + +def _safe_message(err: Any, keyword: str) -> str: + """Value-free description of the failure. + + Never echoes the offending payload value — that would leak bearer + tokens, PII, or prompt-injection strings through the error envelope. + Keeps only the structural facts the caller needs to locate the + offending field and understand why the schema rejected it. + """ + if keyword == "required": + # ``err.message`` contains a schema-declared key name, not user + # data. Pattern: "'X' is a required property". Safe to pass through. + return str(err.message) + if keyword == "type": + expected = err.validator_value + return f"expected type {expected!r}" + if keyword == "enum": + try: + size = len(err.validator_value) + except TypeError: + size = 0 + return f"value not in allowed enum ({size} option{'s' if size != 1 else ''})" + if keyword == "const": + return "value does not match the required constant" + if keyword in { + "minLength", + "maxLength", + "minimum", + "maximum", + "exclusiveMinimum", + "exclusiveMaximum", + "minItems", + "maxItems", + "minProperties", + "maxProperties", + "multipleOf", + }: + return f"{keyword} constraint failed (bound={err.validator_value!r})" + if keyword == "pattern": + return "string does not match required pattern" + if keyword == "format": + return f"value does not match required format ({err.validator_value!r})" + if keyword == "additionalProperties": + return "unexpected property" + if keyword in {"oneOf", "anyOf", "allOf", "not"}: + return f"{keyword} composition failed" + return f"{keyword} constraint failed" + + +def _missing_required_key(err: Any) -> str | None: + """Extract the missing property name from a ``required`` failure. + + Prefer the regex-parsed name; fall back to diffing the schema's + required list against the payload's keys when parsing fails. + Returns ``None`` if both paths fail — the caller keeps the pointer + at the parent object. + """ + match = _REQUIRED_MSG.match(err.message or "") + if match: + return match.group("name") + required = err.validator_value or [] + instance = err.instance if isinstance(err.instance, dict) else None + if instance is not None: + for key in required: + if key not in instance: + return str(key) + return None + + +def _format_error(err: Any) -> ValidationIssue: + """Turn a ``jsonschema.exceptions.ValidationError`` into a ``ValidationIssue``.""" + pointer = _path_to_pointer(list(err.absolute_path)) + keyword = str(err.validator or "validation") + + if keyword == "required": + name = _missing_required_key(err) + if name is not None: + pointer = pointer.rstrip("/") + "/" + name if pointer != "/" else "/" + name + + schema_path = "#/" + "/".join(str(seg) for seg in err.absolute_schema_path) + + return ValidationIssue( + pointer=pointer, + message=_safe_message(err, keyword), + keyword=keyword, + schema_path=schema_path, + ) + + +def _count_nodes(payload: Any, limit: int) -> int: + """Count payload nodes up to ``limit``; stops early once the cap is hit. + + Works as a cheap proxy for "how much work will jsonschema do on this + tree" — good enough to reject obviously-hostile payloads before + ``iter_errors`` walks them. + """ + stack: list[Any] = [payload] + n = 0 + while stack: + node = stack.pop() + n += 1 + if n >= limit: + return n + if isinstance(node, dict): + stack.extend(node.values()) + elif isinstance(node, list): + stack.extend(node) + return n + + +def _iter_errors_bounded(validator: Any, payload: Any) -> list[Any]: + """Collect up to ``_MAX_ISSUES`` errors, sorted by path depth first, + then lexically — callers see the shallowest failure first.""" + errors = list(islice(validator.iter_errors(payload), _MAX_ISSUES)) + errors.sort(key=lambda e: (len(e.absolute_path), list(e.absolute_path))) + return errors + + +def validate_request(tool_name: str, payload: Any) -> ValidationOutcome: + """Validate an outgoing request against ``{tool}-request.json``.""" + validator = get_validator(tool_name, "request") + 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") + return ValidationOutcome( + valid=False, + issues=[_format_error(e) for e in errors], + variant="request", + ) + + +def _select_response_variant(payload: Any) -> ResponseVariant: + """Pick the response variant by payload shape per AdCP 3.0 async contract. + + Per issue #688: choose by ``status`` field, not just tool name. + ``submitted`` / ``working`` / ``input-required`` are the three + async variants; everything else (``completed``, no status, terminal + errors) routes to the sync schema. A2A adapters pre-extract the + artifact data before validation, which normally doesn't carry a + top-level ``status`` — sync-fallback is load-bearing in that path. + """ + if isinstance(payload, dict): + status = payload.get("status") + if status == "submitted": + return "submitted" + if status == "working": + return "working" + if status == "input-required": + return "input-required" + return "sync" + + +def validate_response(tool_name: str, payload: Any) -> ValidationOutcome: + """Validate an incoming response, selecting the variant by payload shape.""" + variant: ResponseVariant = _select_response_variant(payload) + validator = get_validator(tool_name, variant) + used_variant: Direction = variant + if validator is None and variant != "sync": + validator = get_validator(tool_name, "sync") + 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) + return ValidationOutcome( + valid=False, + issues=[_format_error(e) for e in errors], + variant=used_variant, + ) + + +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 diff --git a/tests/fixtures/public_api_snapshot.json b/tests/fixtures/public_api_snapshot.json index ed77c7e25..a5bcf2ccb 100644 --- a/tests/fixtures/public_api_snapshot.json +++ b/tests/fixtures/public_api_snapshot.json @@ -254,6 +254,7 @@ "ReportUsageResponse", "ResolvedBrand", "ResolvedProperty", + "SchemaValidationError", "SegmentIdActivationKey", "SiSendActionResponseRequest", "SiSendTextMessageRequest", @@ -316,6 +317,10 @@ "ValidateContentDeliveryErrorResponse", "ValidateContentDeliverySuccessResponse", "ValidationError", + "ValidationHookConfig", + "ValidationIssue", + "ValidationMode", + "ValidationOutcome", "ValidationResult", "VcpmAuctionPricingOption", "VcpmFixedRatePricingOption", diff --git a/tests/integration/test_a2a_context_id.py b/tests/integration/test_a2a_context_id.py index dd1ef46b0..957bafbd8 100644 --- a/tests/integration/test_a2a_context_id.py +++ b/tests/integration/test_a2a_context_id.py @@ -69,16 +69,17 @@ class _EchoHandler(ADCPHandler): """Minimal handler for the happy-path tests — the assertions are at - the protocol layer, not the handler. Returns empty payloads.""" + the protocol layer, not the handler. Returns spec-compliant empty + payloads so the client's strict response validator passes.""" async def get_adcp_capabilities(self, params: Any, context: Any = None) -> dict[str, Any]: return {"adcp": {"major_versions": [3]}, "supported_protocols": ["media_buy"]} async def get_products(self, params: Any, context: Any = None) -> dict[str, Any]: - return {"products": [{"id": "p1", "name": "Display"}]} + return {"products": []} async def create_media_buy(self, params: Any, context: Any = None) -> dict[str, Any]: - return {"media_buy_id": "mb-1"} + return {"media_buy_id": "mb-1", "packages": []} class _Observer: @@ -256,6 +257,21 @@ async def execute(self, context: RequestContext, event_queue: EventQueue) -> Non state = TaskState.completed text = "approved" + # The completion turn must carry a spec-compliant ``create_media_buy`` + # response so the client's strict schema validator passes — this + # test is exercising task_id echo semantics, not the response + # payload shape. Turn 1's ``input-required`` state has no data + # requirement (it's an interim state). The ``approved`` flag is + # a test-only marker and rides as an additional property, which + # the schema permits (``additionalProperties: true``). + if state == TaskState.completed: + data: dict[str, Any] = { + "media_buy_id": "mb-1", + "packages": [], + "approved": True, + } + else: + data = {"approved": False} task = Task( id=context.task_id or str(uuid4()), context_id=context.context_id or str(uuid4()), @@ -265,7 +281,7 @@ async def execute(self, context: RequestContext, event_queue: EventQueue) -> Non artifact_id=str(uuid4()), parts=[ Part(root=TextPart(text=text)), - Part(root=DataPart(data={"approved": state == TaskState.completed})), + Part(root=DataPart(data=data)), ], ) ], diff --git a/tests/test_idempotency.py b/tests/test_idempotency.py index 8764f3cf7..fad73a06f 100644 --- a/tests/test_idempotency.py +++ b/tests/test_idempotency.py @@ -518,7 +518,17 @@ async def test_replayed_surfaces_on_result(self) -> None: artifacts=[ Artifact( artifact_id="a1", - parts=[Part(root=DataPart(data={"replayed": True, "media_buy_id": "mb_1"}))], + parts=[ + Part( + root=DataPart( + data={ + "replayed": True, + "media_buy_id": "mb_1", + "packages": [], + } + ) + ) + ], ) ], ) @@ -634,7 +644,11 @@ async def test_replayed_surfaces_via_mcp(self) -> None: mock_result = MagicMock() mock_result.isError = False mock_result.content = [] - mock_result.structuredContent = {"replayed": True, "media_buy_id": "mb_1"} + mock_result.structuredContent = { + "replayed": True, + "media_buy_id": "mb_1", + "packages": [], + } session.call_tool = AsyncMock(return_value=mock_result) with patch.object(adapter, "_get_session", AsyncMock(return_value=session)): result = await adapter._call_mcp_tool("create_media_buy", {"brand": "acme"}) diff --git a/tests/test_idempotency_storyboard.py b/tests/test_idempotency_storyboard.py index 5f95f7615..69379c40a 100644 --- a/tests/test_idempotency_storyboard.py +++ b/tests/test_idempotency_storyboard.py @@ -134,6 +134,7 @@ async def mock_send(request: Any) -> SendMessageSuccessResponse: fresh = { "media_buy_id": f"mb_{uuid.uuid4().hex[:8]}", "idempotency_key": key, + "packages": [], } seller_cache[key] = fresh return SendMessageSuccessResponse(result=_task_with_data(fresh)) @@ -207,6 +208,7 @@ async def mock_send(request: Any) -> SendMessageSuccessResponse: fresh = { "media_buy_id": f"mb_{uuid.uuid4().hex[:8]}", "idempotency_key": key, + "packages": [], } seller_cache[key] = fresh return SendMessageSuccessResponse(result=_task_with_data(fresh)) @@ -242,7 +244,7 @@ async def mock_send(request: Any) -> SendMessageSuccessResponse: return SendMessageSuccessResponse(result=_task_with_data(data)) mb_id = f"mb_{uuid.uuid4().hex[:8]}" created_ids.add(mb_id) - fresh = {"media_buy_id": mb_id, "idempotency_key": key} + fresh = {"media_buy_id": mb_id, "idempotency_key": key, "packages": []} seller_cache[key] = fresh return SendMessageSuccessResponse(result=_task_with_data(fresh)) diff --git a/tests/test_protocols.py b/tests/test_protocols.py index a88c8dd96..4e5d7a4a8 100644 --- a/tests/test_protocols.py +++ b/tests/test_protocols.py @@ -173,11 +173,14 @@ async def test_call_tool_multiple_data_parts(self, a2a_config): """Test that last DataPart is authoritative when multiple exist.""" adapter = A2AAdapter(a2a_config) - # Simulates streaming scenario with intermediate + final DataParts + # Simulates streaming scenario with intermediate + final DataParts. + # The final DataPart must be a spec-compliant get_products response + # so the adapter's strict post-receive validation passes — this + # test is exercising DataPart-merge ordering, not tolerant parsing. mock_task = create_mock_a2a_task( parts=[ DataPart(data={"status": "processing", "progress": 50}), - DataPart(data={"status": "completed", "result": "final"}), + DataPart(data={"products": []}), ] ) mock_response = SendMessageSuccessResponse(result=mock_task) @@ -190,15 +193,19 @@ async def test_call_tool_multiple_data_parts(self, a2a_config): assert result.success is True # Should use last DataPart, not first - assert result.data == {"status": "completed", "result": "final"} + assert result.data == {"products": []} @pytest.mark.asyncio async def test_call_tool_multiple_artifacts_uses_last(self, a2a_config): """Test that last artifact is used when multiple artifacts exist (streaming scenario).""" adapter = A2AAdapter(a2a_config) - # Simulates streaming with multiple artifacts - # A2A spec doesn't define artifact.status, so we use the last (most recent) one + # Simulates streaming with multiple artifacts. A2A spec doesn't + # define artifact.status, so the adapter picks the last (most + # recent) one. The last artifact's DataPart must be a spec- + # compliant get_products response so strict post-receive + # validation passes — empty products[] keeps the test focused + # on artifact-ordering semantics, not schema drift. mock_task = Task( id="task_123", context_id="ctx_456", @@ -215,14 +222,14 @@ async def test_call_tool_multiple_artifacts_uses_last(self, a2a_config): artifact_id="artifact_2", parts=[ TextPart(text="Processing complete"), - DataPart(data={"status": "completed", "products": ["prod1"]}), + DataPart(data={"products": []}), ], ), Artifact( artifact_id="artifact_3", parts=[ TextPart(text="Final result"), - DataPart(data={"status": "completed", "products": ["prod2"]}), + DataPart(data={"products": []}), ], ), ], @@ -237,7 +244,7 @@ async def test_call_tool_multiple_artifacts_uses_last(self, a2a_config): assert result.success is True # Should use last artifact (most recent) - assert result.data == {"status": "completed", "products": ["prod2"]} + assert result.data == {"products": []} assert result.message == "Final result" @pytest.mark.asyncio @@ -245,11 +252,12 @@ async def test_call_tool_with_response_wrapper(self, a2a_config): """Test handling ADK-style response wrapper {"response": {...}}.""" adapter = A2AAdapter(a2a_config) - # ADK wraps the actual response in {"response": {...}} + # ADK wraps the actual response in {"response": {...}}. Empty + # products[] keeps the unwrapped payload spec-compliant. mock_task = create_mock_a2a_task( parts=[ TextPart(text="Products retrieved"), - DataPart(data={"response": {"products": [{"id": "prod1", "name": "TV Ad"}]}}), + DataPart(data={"response": {"products": []}}), ] ) mock_response = SendMessageSuccessResponse(result=mock_task) @@ -262,7 +270,7 @@ async def test_call_tool_with_response_wrapper(self, a2a_config): assert result.success is True # Should unwrap the "response" wrapper - assert result.data == {"products": [{"id": "prod1", "name": "TV Ad"}]} + assert result.data == {"products": []} assert result.message == "Products retrieved" @pytest.mark.asyncio @@ -270,13 +278,15 @@ async def test_call_tool_with_response_wrapper_and_metadata(self, a2a_config): """Test handling response wrapper with additional metadata keys.""" adapter = A2AAdapter(a2a_config) - # Some ADK responses have both "response" and other metadata + # Some ADK responses have both "response" and other metadata. Keep + # the wrapped payload spec-compliant (empty products[]) so the + # unwrap path is the only thing under test. mock_task = create_mock_a2a_task( parts=[ TextPart(text="Products retrieved"), DataPart( data={ - "response": {"products": [{"id": "prod1"}]}, + "response": {"products": []}, "metadata": {"cache_hit": True}, } ), @@ -292,7 +302,7 @@ async def test_call_tool_with_response_wrapper_and_metadata(self, a2a_config): assert result.success is True # Should still unwrap and return the "response" content - assert result.data == {"products": [{"id": "prod1"}]} + assert result.data == {"products": []} @pytest.mark.asyncio async def test_interim_response_working(self, a2a_config): @@ -770,9 +780,11 @@ async def test_call_tool_success(self, mcp_config): # Mock MCP session mock_session = AsyncMock() mock_result = MagicMock() - # Mock MCP result with structuredContent (required for AdCP) + # Mock MCP result with structuredContent (required for AdCP). Empty + # products[] keeps the payload spec-compliant without having to + # enumerate every required product field. mock_result.content = [{"type": "text", "text": "Success"}] - mock_result.structuredContent = {"products": [{"id": "prod1"}]} + mock_result.structuredContent = {"products": []} mock_result.isError = False mock_session.call_tool.return_value = mock_result @@ -790,7 +802,7 @@ async def test_call_tool_success(self, mcp_config): # Verify result uses structuredContent assert result.success is True assert result.status == TaskStatus.COMPLETED - assert result.data == {"products": [{"id": "prod1"}]} + assert result.data == {"products": []} @pytest.mark.asyncio async def test_call_tool_with_structured_content(self, mcp_config): @@ -800,9 +812,11 @@ async def test_call_tool_with_structured_content(self, mcp_config): # Mock MCP session mock_session = AsyncMock() mock_result = MagicMock() - # Mock MCP result with structuredContent (preferred over content) - mock_result.content = [{"type": "text", "text": "Found 42 creative formats"}] - mock_result.structuredContent = {"formats": [{"id": "format1"}, {"id": "format2"}]} + # Mock MCP result with structuredContent (preferred over content). + # Empty formats[] is spec-compliant without enumerating every + # required Format field. + mock_result.content = [{"type": "text", "text": "Found 0 creative formats"}] + mock_result.structuredContent = {"formats": []} mock_result.isError = False mock_session.call_tool.return_value = mock_result @@ -812,9 +826,9 @@ async def test_call_tool_with_structured_content(self, mcp_config): # Verify result uses structuredContent, not content array assert result.success is True assert result.status == TaskStatus.COMPLETED - assert result.data == {"formats": [{"id": "format1"}, {"id": "format2"}]} + assert result.data == {"formats": []} # Verify message extraction from content array - assert result.message == "Found 42 creative formats" + assert result.message == "Found 0 creative formats" @pytest.mark.asyncio async def test_call_tool_no_structured_adcp_data(self, mcp_config): diff --git a/tests/test_schema_validation.py b/tests/test_schema_validation.py new file mode 100644 index 000000000..43e9b31a4 --- /dev/null +++ b/tests/test_schema_validation.py @@ -0,0 +1,243 @@ +"""Unit tests for the schema-driven validator (issue #249). + +Exercises the real bundled schemas shipped with the SDK — no mocking, +so a schema regeneration that breaks the validator's assumptions +surfaces here rather than at storyboard time. +""" + +from __future__ import annotations + +import os +from unittest.mock import patch + +import pytest + +from adcp.validation import ( + SchemaValidationError, + ValidationHookConfig, + build_adcp_validation_error_payload, + build_validation_error, + format_issues, + list_validator_keys, + resolve_validation_modes, + validate_incoming_response, + validate_outgoing_request, + validate_request, + validate_response, +) +from adcp.validation.schema_validator import ValidationIssue + + +class TestValidateRequest: + def test_flags_missing_required_fields_with_json_pointer(self) -> None: + outcome = validate_request("get_products", {}) + assert outcome.valid is False + pointers = [i.pointer for i in outcome.issues] + assert "/buying_mode" in pointers, f"expected /buying_mode in {pointers}" + + def test_returns_skipped_for_tools_outside_adcp_catalog(self) -> None: + outcome = validate_request("custom_seller_extension", {"anything": True}) + assert outcome.valid is True + assert outcome.variant == "skipped" + + def test_accepts_extension_fields_without_error(self) -> None: + outcome = validate_request( + "get_products", + { + "brief": "campaign brief", + "promoted_offering": "product", + "buying_mode": "brief", + # ext + unknown vendor field — additionalProperties must tolerate both + "ext": {"gam": {"custom_field": 1}}, + "unknown_vendor_field": {"ok": True}, + }, + ) + # Schema may require other fields; only assert unknown extensions + # themselves don't appear as failures. + for issue in outcome.issues: + assert issue.pointer != "/unknown_vendor_field" + assert not issue.pointer.startswith("/ext") + + +class TestValidateResponse: + def test_selects_submitted_variant_on_status(self) -> None: + outcome = validate_response("create_media_buy", {"status": "submitted", "task_id": "t_1"}) + assert outcome.valid is True + assert outcome.variant == "submitted" + + def test_selects_working_variant(self) -> None: + outcome = validate_response("create_media_buy", {"status": "working", "task_id": "t_2"}) + assert outcome.valid is True + assert outcome.variant == "working" + + def test_selects_input_required_variant(self) -> None: + outcome = validate_response( + "create_media_buy", {"status": "input-required", "task_id": "t_3"} + ) + assert outcome.valid is True + assert outcome.variant == "input-required" + + def test_falls_back_to_sync_without_status(self) -> None: + outcome = validate_response("create_media_buy", {"media_buy_id": "mb_1"}) + assert outcome.variant == "sync" + + def test_surfaces_errors_with_pointer_keyword_schema_path(self) -> None: + outcome = validate_response("get_products", {"products": "not-an-array"}) + assert outcome.valid is False + products_issue = next((i for i in outcome.issues if i.pointer == "/products"), None) + assert products_issue is not None, "expected an issue at /products" + assert products_issue.keyword == "type" + assert products_issue.schema_path + # Sanitized message MUST NOT echo the offending value — the + # whole point of the sanitizer is keeping tokens/PII out of the + # wire envelope and logs. + assert "not-an-array" not in products_issue.message + assert "expected type" in products_issue.message + + def test_sanitizes_offending_value_out_of_message(self) -> None: + """Hostile or buggy payloads can carry secrets in the wrong slot. + The error message the caller sees must not echo them back.""" + secret = "Bearer sk-should-never-appear-in-any-error" + outcome = validate_response("get_products", {"products": secret}) + assert outcome.valid is False + for issue in outcome.issues: + assert secret not in issue.message + assert secret not in issue.schema_path + + +class TestFormatIssues: + def test_caps_verbose_failures_and_notes_overflow(self) -> None: + issues = [ + ValidationIssue( + pointer=f"/{c}", + message="oops", + keyword="required", + schema_path="#/required", + ) + for c in "abcd" + ] + summary = format_issues(issues, limit=2) + assert "/a" in summary + assert "/b" in summary + assert "(+2 more)" in summary + + +class TestErrorBuilders: + def test_build_validation_error_carries_details(self) -> None: + issues = [ + ValidationIssue( + pointer="/foo/bar", + message="bad", + keyword="type", + schema_path="#/properties/foo/bar", + ) + ] + err = build_validation_error("get_products", "request", issues) + assert isinstance(err, SchemaValidationError) + assert err.code == "VALIDATION_ERROR" + assert err.tool == "get_products" + assert err.side == "request" + assert err.issues == issues + # ``details`` is a declared attribute on the class — not a bolt-on. + assert err.details["tool"] == "get_products" + assert err.details["side"] == "request" + assert err.details["issues"][0]["pointer"] == "/foo/bar" + + def test_build_adcp_validation_error_payload(self) -> None: + issues = [ + ValidationIssue( + pointer="/media_buy_id", + message="is required", + keyword="required", + schema_path="#/required", + ) + ] + payload = build_adcp_validation_error_payload("create_media_buy", "response", issues) + assert payload["code"] == "VALIDATION_ERROR" + assert "/media_buy_id" in payload["message"] + assert payload["field"] == "/media_buy_id" + assert payload["details"]["tool"] == "create_media_buy" + assert payload["details"]["side"] == "response" + assert payload["details"]["issues"][0]["pointer"] == "/media_buy_id" + + +class TestListValidatorKeys: + def test_exposes_every_shipped_pair(self) -> None: + keys = list_validator_keys() + assert len(keys) > 0 + for expected in ( + "get_products::request", + "get_products::sync", + "create_media_buy::submitted", + "create_media_buy::working", + "create_media_buy::input-required", + ): + assert expected in keys, f"missing {expected}" + + +class TestClientHooks: + def test_outgoing_strict_raises(self) -> None: + with pytest.raises(SchemaValidationError) as info: + validate_outgoing_request("create_media_buy", {}, "strict") + assert info.value.code == "VALIDATION_ERROR" + + def test_outgoing_warn_logs_and_returns_invalid(self) -> None: + logs: list = [] + outcome = validate_outgoing_request("create_media_buy", {}, "warn", logs) + assert outcome is not None + assert outcome.valid is False + assert len(logs) == 1 + assert logs[0]["type"] == "warning" + + def test_outgoing_off_short_circuits(self) -> None: + outcome = validate_outgoing_request("create_media_buy", {}, "off") + assert outcome is None + + def test_incoming_off_is_noop_valid(self) -> None: + outcome = validate_incoming_response("get_products", {"products": "not-array"}, "off") + assert outcome.valid is True + + def test_incoming_warn_logs_and_returns_invalid(self) -> None: + logs: list = [] + outcome = validate_incoming_response( + "get_products", {"products": "not-array"}, "warn", logs + ) + assert outcome.valid is False + assert len(logs) == 1 + + +class TestResolveValidationModes: + def test_requests_default_to_warn(self) -> None: + with patch.dict(os.environ, {}, clear=True): + req, _ = resolve_validation_modes() + assert req == "warn" + + def test_responses_default_to_strict(self) -> None: + with patch.dict(os.environ, {}, clear=True): + _, resp = resolve_validation_modes() + assert resp == "strict" + + def test_responses_flip_warn_when_adcp_env_is_production(self) -> None: + with patch.dict(os.environ, {"ADCP_ENV": "production"}, clear=True): + _, resp = resolve_validation_modes() + assert resp == "warn" + + def test_responses_flip_warn_when_adcp_env_is_prod_shorthand(self) -> None: + with patch.dict(os.environ, {"ADCP_ENV": "prod"}, clear=True): + _, resp = resolve_validation_modes() + assert resp == "warn" + + def test_generic_env_vars_do_not_flip_default(self) -> None: + """Only ``ADCP_ENV`` is consulted. Generic ``ENV`` / ``ENVIRONMENT`` + are set by unrelated tooling (rails, postgres, 12-factor) and + must not silently flip the SDK's default — that's a footgun.""" + for name in ("ENV", "ENVIRONMENT", "PYTHON_ENV"): + with patch.dict(os.environ, {name: "production"}, clear=True): + _, resp = resolve_validation_modes() + assert resp == "strict", f"{name} should not flip the default" + + def test_explicit_config_overrides_defaults(self) -> None: + req, resp = resolve_validation_modes( + ValidationHookConfig(requests="strict", responses="off") + ) + assert (req, resp) == ("strict", "off") diff --git a/tests/test_schema_validation_client.py b/tests/test_schema_validation_client.py new file mode 100644 index 000000000..0d1f030f3 --- /dev/null +++ b/tests/test_schema_validation_client.py @@ -0,0 +1,118 @@ +"""Client-side integration tests for schema-driven validation (issue #249). + +Exercises the wiring between :class:`adcp.ADCPClient`, the protocol +adapter, and the validation hook. The adapter's external transport is +mocked so these stay fast and deterministic; the validator itself runs +unmocked against the real bundled schemas. +""" + +from __future__ import annotations + +import os +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from adcp import ADCPClient, ValidationHookConfig +from adcp.types import AgentConfig, Protocol + + +def _agent_config() -> AgentConfig: + return AgentConfig( + id="test_agent", + agent_uri="https://test.example.com", + protocol=Protocol.MCP, + ) + + +class TestConfigPropagation: + def test_defaults_apply_when_omitted(self) -> None: + with patch.dict(os.environ, {}, clear=True): + client = ADCPClient(_agent_config()) + assert client.adapter.request_validation_mode == "warn" + assert client.adapter.response_validation_mode == "strict" + + def test_production_env_flips_responses_to_warn(self) -> None: + with patch.dict(os.environ, {"ADCP_ENV": "production"}, clear=True): + client = ADCPClient(_agent_config()) + assert client.adapter.response_validation_mode == "warn" + + def test_explicit_config_wins(self) -> None: + with patch.dict(os.environ, {}, clear=True): + client = ADCPClient( + _agent_config(), + validation=ValidationHookConfig(requests="strict", responses="off"), + ) + assert client.adapter.request_validation_mode == "strict" + assert client.adapter.response_validation_mode == "off" + + +class TestMCPAdapterHooks: + @pytest.mark.asyncio + async def test_strict_request_blocks_call_before_transport(self) -> None: + client = ADCPClient( + _agent_config(), + validation=ValidationHookConfig(requests="strict"), + ) + session = MagicMock() + session.call_tool = AsyncMock() + + with patch.object(client.adapter, "_get_session", AsyncMock(return_value=session)): + result = await client.adapter._call_mcp_tool("get_products", {}) + + assert result.success is False + assert "VALIDATION_ERROR" in (result.error or "") or "failed schema validation" in ( + result.error or "" + ) + session.call_tool.assert_not_called() + + @pytest.mark.asyncio + async def test_strict_response_fails_task_on_drift(self) -> None: + client = ADCPClient( + _agent_config(), + validation=ValidationHookConfig(requests="off", responses="strict"), + ) + + # Fake MCP tool result: drift — products should be an array, not a string. + fake_result = MagicMock() + fake_result.isError = False + fake_result.structuredContent = {"products": "oops"} + fake_result.content = [] + + session = MagicMock() + session.call_tool = AsyncMock(return_value=fake_result) + + with patch.object(client.adapter, "_get_session", AsyncMock(return_value=session)): + result = await client.adapter._call_mcp_tool( + "get_products", + { + "brief": "b", + "promoted_offering": "shoes", + "buying_mode": "brief", + }, + ) + + assert result.success is False + assert "Schema validation failed" in (result.error or "") + + @pytest.mark.asyncio + async def test_off_mode_skips_validator(self) -> None: + client = ADCPClient( + _agent_config(), + validation=ValidationHookConfig(requests="off", responses="off"), + ) + + # Drift response — in off mode it should pass through unchanged. + fake_result = MagicMock() + fake_result.isError = False + fake_result.structuredContent = {"products": "oops"} + fake_result.content = [] + + session = MagicMock() + session.call_tool = AsyncMock(return_value=fake_result) + + with patch.object(client.adapter, "_get_session", AsyncMock(return_value=session)): + result = await client.adapter._call_mcp_tool("get_products", {}) + + assert result.success is True + assert result.data == {"products": "oops"} diff --git a/tests/test_schema_validation_server.py b/tests/test_schema_validation_server.py new file mode 100644 index 000000000..c357f2629 --- /dev/null +++ b/tests/test_schema_validation_server.py @@ -0,0 +1,163 @@ +"""Server-middleware integration tests for schema-driven validation (issue #249). + +Exercises the opt-in request/response validator on ``create_tool_caller`` +against real handlers, using the MCP-facing dispatcher shape. +""" + +from __future__ import annotations + +from typing import Any + +import pytest + +from adcp.exceptions import ADCPTaskError +from adcp.server.base import ADCPHandler, ToolContext +from adcp.server.mcp_tools import create_tool_caller +from adcp.validation import ValidationHookConfig + + +class _StubHandler(ADCPHandler[Any]): + """Minimal handler that records whether it was invoked and returns the + payload supplied in the constructor.""" + + def __init__(self, response: dict[str, Any]) -> None: + self._response = response + self.called = False + + async def get_products(self, params: dict[str, Any], ctx: ToolContext) -> dict[str, Any]: + self.called = True + return dict(self._response) + + +VALID_GET_PRODUCTS = { + "brief": "test campaign", + "promoted_offering": "shoes", + "buying_mode": "brief", +} + + +class TestRequestsStrict: + @pytest.mark.asyncio + async def test_rejects_malformed_with_validation_error_before_dispatch( + self, + ) -> None: + handler = _StubHandler({"products": []}) + caller = create_tool_caller( + handler, + "get_products", + validation=ValidationHookConfig(requests="strict"), + ) + with pytest.raises(ADCPTaskError) as info: + await caller({}) + assert handler.called is False + first = info.value.errors[0] + assert first.code == "VALIDATION_ERROR" + assert first.field, "expected field pointer on error" + assert first.details + assert first.details["side"] == "request" + + @pytest.mark.asyncio + async def test_accepts_valid_requests(self) -> None: + handler = _StubHandler({"products": []}) + caller = create_tool_caller( + handler, + "get_products", + validation=ValidationHookConfig(requests="strict"), + ) + result = await caller(dict(VALID_GET_PRODUCTS)) + assert isinstance(result["products"], list) + + +class TestRequestsWarn: + @pytest.mark.asyncio + async def test_logs_warning_but_dispatches(self, caplog: pytest.LogCaptureFixture) -> None: + handler = _StubHandler({"products": []}) + caller = create_tool_caller( + handler, + "get_products", + validation=ValidationHookConfig(requests="warn"), + ) + with caplog.at_level("WARNING", logger="adcp.server.mcp_tools"): + await caller({}) + assert handler.called is True + messages = [r.message for r in caplog.records] + assert any( + "Schema validation warning (request) for get_products" in m for m in messages + ), f"no warning log: {messages}" + + +class TestNoValidationConfig: + @pytest.mark.asyncio + async def test_unconfigured_does_not_validate(self) -> None: + handler = _StubHandler({"products": []}) + caller = create_tool_caller(handler, "get_products") + await caller({}) + assert handler.called is True + + +class TestResponses: + @pytest.mark.asyncio + async def test_strict_drift_surfaces_validation_error(self) -> None: + handler = _StubHandler({"products": "oops"}) + caller = create_tool_caller( + handler, + "get_products", + validation=ValidationHookConfig(responses="strict"), + ) + with pytest.raises(ADCPTaskError) as info: + await caller(dict(VALID_GET_PRODUCTS)) + first = info.value.errors[0] + assert first.code == "VALIDATION_ERROR" + assert first.details["side"] == "response" + + @pytest.mark.asyncio + async def test_warn_logs_but_returns_response_unchanged( + self, caplog: pytest.LogCaptureFixture + ) -> None: + handler = _StubHandler({"products": "oops"}) + caller = create_tool_caller( + handler, + "get_products", + validation=ValidationHookConfig(responses="warn"), + ) + with caplog.at_level("WARNING", logger="adcp.server.mcp_tools"): + result = await caller(dict(VALID_GET_PRODUCTS)) + assert result["products"] == "oops" + messages = [r.message for r in caplog.records] + assert any( + "Schema validation warning (response) for get_products" in m for m in messages + ), f"no warning log: {messages}" + + @pytest.mark.asyncio + async def test_valid_response_passes_strict(self) -> None: + handler = _StubHandler({"products": []}) + caller = create_tool_caller( + handler, + "get_products", + validation=ValidationHookConfig(responses="strict"), + ) + result = await caller(dict(VALID_GET_PRODUCTS)) + assert result["products"] == [] + + @pytest.mark.asyncio + async def test_adcp_error_envelope_skips_response_validation(self) -> None: + """Handler-returned ``adcp_error`` envelopes have their own shape + enforced by the ``Error`` builder; validating them against the + per-tool success schema would convert a real protocol error + (e.g. ``NOT_FOUND``) into a fake ``VALIDATION_ERROR``.""" + handler = _StubHandler( + { + "adcp_error": { + "code": "NOT_FOUND", + "message": "no products match the brief", + } + } + ) + caller = create_tool_caller( + handler, + "get_products", + validation=ValidationHookConfig(responses="strict"), + ) + result = await caller(dict(VALID_GET_PRODUCTS)) + # Envelope passes through unchanged — no VALIDATION_ERROR raised. + assert result["adcp_error"]["code"] == "NOT_FOUND"