feat(validation): schema-driven validation with client hooks and opt-in server middleware

.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

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

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

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"
 

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"

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

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

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

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)

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

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,