Update to AdCP v2.4 schemas with automated compliance validation

.github/workflows/ci.yml

@@ -47,6 +47,9 @@ jobs:
       - name: Run smoke tests
         run: uv run pytest tests/smoke/ -v --no-cov -m smoke
 
+      - name: Run schema compliance tests
+        run: uv run pytest tests/schema_compliance/ -v --no-cov
+
       - name: Run all tests with coverage
         run: uv run pytest --cov=src/creative_agent --cov-fail-under=10 --cov-report=term-missing
 

.pre-commit-config.yaml

@@ -17,6 +17,9 @@ repos:
         pass_filenames: false
         stages: [manual]
 
+      # Schema compliance runs in CI only (requires uv + full test environment)
+      # See .github/workflows/ci.yml for schema compliance validation
+
   - repo: https://github.com/pre-commit/pre-commit-hooks
     rev: v6.0.0
     hooks:

scripts/generate_schemas.py

@@ -160,8 +160,8 @@ def generate_schemas_from_json(schema_dir: Path, output_file: Path):
     temp_dir.mkdir(exist_ok=True)
 
     try:
-        # Process each JSON schema file
-        schema_files = list(schema_dir.glob("*.json"))
+        # Process each JSON schema file in sorted order for deterministic output
+        schema_files = sorted(schema_dir.glob("*.json"))
         print(f"📝 Found {len(schema_files)} schema files")
 
         # Skip these non-schema files
@@ -208,6 +208,7 @@ def generate_schemas_from_json(schema_dir: Path, output_file: Path):
             "--target-python-version",
             "3.12",
             "--disable-timestamp",
+            "--reuse-model",  # Reuse models with same content for deterministic class names
         ]
 
         result = subprocess.run(cmd, capture_output=True, text=True, check=False)

scripts/update_schemas.py

@@ -0,0 +1,246 @@
+#!/usr/bin/env python3
+"""
+Update local schema cache from AdCP website.
+
+This script downloads all AdCP JSON schemas from adcontextprotocol.org
+and updates the local cache in tests/schemas/v1/.
+
+Usage:
+    python scripts/update_schemas.py [--dry-run]
+"""
+
+import argparse
+import json
+import sys
+from pathlib import Path
+
+import httpx
+
+
+def filename_to_ref(filename: str) -> str:
+    """Convert our flattened filename format to a $ref path."""
+    # _schemas_v1_core_format_json.json -> /schemas/v1/core/format.json
+    name = filename.replace(".json", "").replace("_json", ".json").replace("_", "/", 1)
+    return name
+
+
+def ref_to_filename(ref: str) -> str:
+    """Convert $ref path to our flattened filename format."""
+    # /schemas/v1/core/format.json -> _schemas_v1_core_format_json.json
+    return ref.replace("/", "_").replace(".", "_") + ".json"
+
+
+def download_schema(ref: str, base_url: str = "https://adcontextprotocol.org") -> dict | None:
+    """
+    Download a schema from AdCP website.
+
+    Returns schema dict if successful, None if not found or error.
+    """
+    schema_url = f"{base_url}{ref}"
+
+    try:
+        print(f"  Fetching: {ref}")
+        response = httpx.get(schema_url, timeout=10.0, follow_redirects=True)
+        response.raise_for_status()
+
+        # Check if we got JSON (not HTML)
+        content_type = response.headers.get("content-type", "")
+        if "json" not in content_type.lower():
+            print(f"  ⚠️  Skipping {ref}: Got {content_type} instead of JSON")
+            return None
+
+        schema = response.json()
+        return schema
+
+    except httpx.HTTPStatusError as e:
+        if e.response.status_code == 404:
+            print(f"  ⚠️  Not found: {ref}")
+        else:
+            print(f"  ❌ HTTP {e.response.status_code}: {ref}")
+        return None
+    except Exception as e:
+        print(f"  ❌ Error downloading {ref}: {e}")
+        return None
+
+
+def is_creative_agent_schema(ref: str) -> bool:
+    """
+    Check if a schema is relevant for a Creative Agent.
+
+    Creative agents only need schemas related to creative formats, assets,
+    and creative agent tools - not media buy, signals, or other protocol areas.
+    """
+    creative_patterns = [
+        "/schemas/v1/core/assets/",  # All asset types
+        "/schemas/v1/core/creative-",  # Creative-specific schemas
+        "/schemas/v1/core/format",  # Format and format-id
+        "/schemas/v1/core/brand-manifest",  # Brand manifest schemas
+        "/schemas/v1/creative/",  # Creative agent tool schemas
+        "/schemas/v1/enums/",  # Shared enums (needed by assets and formats)
+        "/schemas/v1/standard-formats/",  # Standard format definitions
+        "/schemas/v1/adagents.json",  # Agent capabilities
+        "/schemas/v1/core/response.json",  # Protocol response wrapper
+        "/schemas/v1/core/error.json",  # Error schema
+        "/schemas/v1/core/sub-asset.json",  # Sub-asset for carousels
+    ]
+
+    return any(pattern in ref for pattern in creative_patterns)
+
+
+def discover_schemas(schema_dir: Path, creative_only: bool = True) -> list[str]:
+    """
+    Discover all schema $refs from existing cache.
+
+    Args:
+        schema_dir: Directory containing cached schemas
+        creative_only: If True, only return creative-agent-relevant schemas
+
+    Returns list of unique $ref paths found in existing schemas.
+    """
+    refs = set()
+
+    for schema_file in schema_dir.glob("*.json"):
+        try:
+            with open(schema_file) as f:
+                schema = json.load(f)
+
+            # Extract $ref from this schema
+            if "$id" in schema:
+                schema_ref = schema["$id"]
+                if not creative_only or is_creative_agent_schema(schema_ref):
+                    refs.add(schema_ref)
+
+            # Recursively find all $refs in the schema
+            all_refs = find_refs_in_schema(schema)
+            if creative_only:
+                all_refs = {r for r in all_refs if is_creative_agent_schema(r)}
+            refs.update(all_refs)
+
+        except Exception as e:
+            print(f"  ⚠️  Error reading {schema_file.name}: {e}")
+
+    return sorted(refs)
+
+
+def find_refs_in_schema(obj: dict | list) -> set[str]:
+    """Recursively find all $ref values in a schema."""
+    refs = set()
+
+    if isinstance(obj, dict):
+        if "$ref" in obj:
+            refs.add(obj["$ref"])
+        for value in obj.values():
+            refs.update(find_refs_in_schema(value))
+    elif isinstance(obj, list):
+        for item in obj:
+            refs.update(find_refs_in_schema(item))
+
+    return refs
+
+
+def update_schemas(schema_dir: Path, dry_run: bool = False, creative_only: bool = True):
+    """
+    Update schemas from AdCP website.
+
+    Discovers schema refs from existing cache, downloads latest versions,
+    and updates local files.
+
+    Args:
+        schema_dir: Directory containing cached schemas
+        dry_run: If True, show what would change without modifying files
+        creative_only: If True, only update creative-agent-relevant schemas
+    """
+    print(f"📂 Schema directory: {schema_dir}")
+    if creative_only:
+        print("🎨 Filtering to creative-agent-relevant schemas only")
+
+    if not schema_dir.exists():
+        print(f"❌ Directory not found: {schema_dir}")
+        sys.exit(1)
+
+    # Discover all schema refs
+    print("\n🔍 Discovering schemas from existing cache...")
+    refs = discover_schemas(schema_dir, creative_only=creative_only)
+    print(f"   Found {len(refs)} unique schema refs")
+
+    # Download and update each schema
+    print("\n📥 Downloading latest schemas...")
+    updated = 0
+    unchanged = 0
+    failed = 0
+
+    for ref in refs:
+        # Validate ref
+        if not ref.startswith("/schemas/v1/"):
+            print(f"  ⚠️  Skipping invalid ref: {ref}")
+            continue
+
+        # Download latest version
+        latest_schema = download_schema(ref)
+        if latest_schema is None:
+            failed += 1
+            continue
+
+        # Compare with local version
+        filename = ref_to_filename(ref)
+        local_path = schema_dir / filename
+
+        if local_path.exists():
+            with open(local_path) as f:
+                local_schema = json.load(f)
+
+            if local_schema == latest_schema:
+                print(f"  ✓ No changes: {filename}")
+                unchanged += 1
+                continue
+
+        # Update local file
+        if dry_run:
+            print(f"  🔄 Would update: {filename}")
+            updated += 1
+        else:
+            with open(local_path, "w") as f:
+                json.dump(latest_schema, f, indent=2)
+                f.write("\n")  # Add trailing newline
+            print(f"  ✅ Updated: {filename}")
+            updated += 1
+
+    # Summary
+    print(f"\n📊 Summary:")
+    print(f"   Updated: {updated}")
+    print(f"   Unchanged: {unchanged}")
+    print(f"   Failed: {failed}")
+
+    if dry_run:
+        print("\n   (Dry run - no files were modified)")
+
+    if updated > 0 and not dry_run:
+        print("\n💡 Next steps:")
+        print("   1. Review changes: git diff tests/schemas/v1/")
+        print("   2. Regenerate Python models: python scripts/generate_schemas.py")
+        print("   3. Run tests: pytest")
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="Update AdCP schemas from website (creative-agent-relevant schemas only by default)"
+    )
+    parser.add_argument("--dry-run", action="store_true", help="Show what would be updated without making changes")
+    parser.add_argument(
+        "--schema-dir",
+        type=Path,
+        default=Path("tests/schemas/v1"),
+        help="Directory containing JSON schemas (default: tests/schemas/v1)",
+    )
+    parser.add_argument(
+        "--all-schemas",
+        action="store_true",
+        help="Include all AdCP schemas (media buy, signals, etc.), not just creative-agent schemas",
+    )
+    args = parser.parse_args()
+
+    update_schemas(args.schema_dir, dry_run=args.dry_run, creative_only=not args.all_schemas)
+
+
+if __name__ == "__main__":
+    main()

src/creative_agent/schemas/__init__.py

@@ -27,8 +27,8 @@
 
 # Format schemas
 from ..schemas_generated._schemas_v1_core_format_json import Format as CreativeFormat
-from ..schemas_generated._schemas_v1_media_buy_list_creative_formats_response_json import (
-    ListCreativeFormatsResponse,
+from ..schemas_generated._schemas_v1_creative_list_creative_formats_response_json import (
+    ListCreativeFormatsResponseCreativeAgent as ListCreativeFormatsResponse,
 )
 
 # Build schemas (agent-specific, not part of AdCP)

src/creative_agent/schemas_generated/_schemas_v1_core_brand_manifest_json.py

@@ -251,68 +251,10 @@ class BrandManifest1(BaseModel):
     ] = None
 
 
-class Asset3(BaseModel):
-    model_config = ConfigDict(
-        extra="forbid",
-    )
-    asset_id: Annotated[str, Field(description="Unique identifier for this asset")]
-    asset_type: Annotated[AssetType, Field(description="Type of asset")]
-    url: Annotated[AnyUrl, Field(description="URL to CDN-hosted asset file")]
-    tags: Annotated[
-        Optional[list[str]],
-        Field(
-            description="Tags for asset discovery (e.g., 'holiday', 'lifestyle', 'product_shot')"
-        ),
-    ] = None
-    name: Annotated[Optional[str], Field(description="Human-readable asset name")] = (
-        None
-    )
-    description: Annotated[
-        Optional[str], Field(description="Asset description or usage notes")
-    ] = None
-    width: Annotated[
-        Optional[int], Field(description="Image/video width in pixels")
-    ] = None
-    height: Annotated[
-        Optional[int], Field(description="Image/video height in pixels")
-    ] = None
-    duration_seconds: Annotated[
-        Optional[float], Field(description="Video/audio duration in seconds")
-    ] = None
-    file_size_bytes: Annotated[
-        Optional[int], Field(description="File size in bytes")
-    ] = None
-    format: Annotated[
-        Optional[str], Field(description="File format (e.g., 'jpg', 'mp4', 'mp3')")
-    ] = None
-    metadata: Annotated[
-        Optional[dict[str, Any]],
-        Field(description="Additional asset-specific metadata"),
-    ] = None
+Asset3 = Asset
 
 
-class ProductCatalog3(BaseModel):
-    model_config = ConfigDict(
-        extra="forbid",
-    )
-    feed_url: Annotated[AnyUrl, Field(description="URL to product catalog feed")]
-    feed_format: Annotated[
-        Optional[FeedFormat], Field(description="Format of the product feed")
-    ] = "google_merchant_center"
-    categories: Annotated[
-        Optional[list[str]],
-        Field(
-            description="Product categories available in the catalog (for filtering)"
-        ),
-    ] = None
-    last_updated: Annotated[
-        Optional[AwareDatetime],
-        Field(description="When the product catalog was last updated"),
-    ] = None
-    update_frequency: Annotated[
-        Optional[UpdateFrequency],
-        Field(description="How frequently the product catalog is updated"),
-    ] = None
+ProductCatalog3 = ProductCatalog
 
 
 class BrandManifest2(BaseModel):