adcontextprotocol
diff --git a/‎CLAUDE.md‎
Lines changed: 44 additions & 0 deletions b/‎CLAUDE.md‎
Lines changed: 44 additions & 0 deletions
diff --git a/‎Makefile‎
Lines changed: 2 additions & 2 deletions b/‎Makefile‎
Lines changed: 2 additions & 2 deletions
diff --git a/‎pyproject.toml‎
Lines changed: 1 addition & 0 deletions b/‎pyproject.toml‎
Lines changed: 1 addition & 0 deletions
diff --git a/‎schemas/cache/1.0.0/format-id.json‎
Lines changed: 24 additions & 0 deletions b/‎schemas/cache/1.0.0/format-id.json‎
Lines changed: 24 additions & 0 deletions
diff --git a/‎scripts/generate_models.py‎
Lines changed: 12 additions & 6 deletions b/‎scripts/generate_models.py‎
Lines changed: 12 additions & 6 deletions
diff --git a/‎scripts/generate_models_simple.py‎
Lines changed: 71 additions & 3 deletions b/‎scripts/generate_models_simple.py‎
Lines changed: 71 additions & 3 deletions
diff --git a/‎src/adcp/__main__.py‎
Lines changed: 107 additions & 8 deletions b/‎src/adcp/__main__.py‎
Lines changed: 107 additions & 8 deletions
@@ -95,6 +95,50 @@ async def _get_client(self) -> httpx.AsyncClient:
     return self._client
 ```
 
+**Format Definition Caching (Production Consideration)**
+
+Creative formats rarely change, so production implementations should cache format definitions to avoid redundant `list_creative_formats` calls:
+
+```python
+from functools import lru_cache
+
+# Option 1: Simple LRU cache for format lookups
+@lru_cache(maxsize=100)
+async def get_format_definition(agent_url: str, format_id: str) -> Format:
+    """Cached format definition lookup."""
+    agent = get_agent_for_url(agent_url)
+    result = await agent.list_creative_formats()
+    for fmt in result.data.formats:
+        if fmt.format_id.id == format_id:
+            return fmt
+    raise ValueError(f"Format not found: {format_id}")
+
+# Option 2: TTL-based cache with periodic refresh
+class FormatCache:
+    def __init__(self, ttl_seconds: int = 3600):
+        self._cache: dict[tuple[str, str], tuple[Format, float]] = {}
+        self._ttl = ttl_seconds
+
+    async def get_format(self, agent_url: str, format_id: str) -> Format:
+        """Get format with TTL-based caching."""
+        key = (agent_url, format_id)
+        if key in self._cache:
+            fmt, timestamp = self._cache[key]
+            if time.time() - timestamp < self._ttl:
+                return fmt
+
+        # Cache miss or expired - fetch fresh
+        fmt = await self._fetch_format(agent_url, format_id)
+        self._cache[key] = (fmt, time.time())
+        return fmt
+```
+
+**Cache Invalidation Strategies:**
+1. **TTL-based** (recommended): Formats rarely change, 1-hour TTL is reasonable
+2. **Version-based**: Use format version in ID (e.g., `banner_300x250_v2`)
+3. **ETags**: Check `Last-Modified` headers if agents support it
+4. **Explicit invalidation**: Clear cache when format changes detected
+
 ## Common Pitfalls to Avoid
 
 **String Escaping in Code Generation**
 
@@ -17,9 +17,9 @@ help: ## Show this help message
 install-dev: ## Install package in development mode with dev dependencies
 	$(PIP) install -e ".[dev]"
 
-format: ## Format code with black
+format: ## Format code with black (excludes generated files)
 	$(BLACK) src/ tests/ scripts/
-	@echo "✓ Code formatted successfully"
+	@echo "✓ Code formatted successfully (generated.py excluded via pyproject.toml)"
 
 lint: ## Run linter (ruff) on source code
 	$(RUFF) check src/ tests/
 
@@ -58,6 +58,7 @@ where = ["src"]
 [tool.black]
 line-length = 100
 target-version = ["py310", "py311", "py312"]
+extend-exclude = "/(generated|tasks)\\.py$"
 
 [tool.ruff]
 line-length = 100
 
@@ -0,0 +1,24 @@
+{
+  "$schema": "http://json-schema.org/draft-07/schema#",
+  "$id": "/schemas/v1/core/format-id.json",
+  "title": "Format ID",
+  "description": "Structured format identifier with agent URL and format name",
+  "type": "object",
+  "properties": {
+    "agent_url": {
+      "type": "string",
+      "format": "uri",
+      "description": "URL of the agent that defines this format (e.g., 'https://creatives.adcontextprotocol.org' for standard formats, or 'https://publisher.com/.well-known/adcp/sales' for custom formats)"
+    },
+    "id": {
+      "type": "string",
+      "pattern": "^[a-zA-Z0-9_-]+$",
+      "description": "Format identifier within the agent's namespace (e.g., 'display_300x250', 'video_standard_30s')"
+    }
+  },
+  "required": [
+    "agent_url",
+    "id"
+  ],
+  "additionalProperties": false
+}
@@ -30,17 +30,23 @@ def main():
     # Generate models using datamodel-code-generator
     cmd = [
         "datamodel-codegen",
-        "--input", str(SCHEMAS_DIR),
-        "--input-file-type", "jsonschema",
-        "--output", str(OUTPUT_FILE),
-        "--output-model-type", "pydantic_v2.BaseModel",
+        "--input",
+        str(SCHEMAS_DIR),
+        "--input-file-type",
+        "jsonschema",
+        "--output",
+        str(OUTPUT_FILE),
+        "--output-model-type",
+        "pydantic_v2.BaseModel",
         "--use-standard-collections",
         "--use-schema-description",
         "--use-field-description",
         "--field-constraints",
         "--use-default",
-        "--enum-field-as-literal", "all",
-        "--target-python-version", "3.10",
+        "--enum-field-as-literal",
+        "all",
+        "--target-python-version",
+        "3.10",
         "--collapse-root-models",
         "--allow-extra-fields",
         "--enable-version-header",
 
@@ -81,7 +81,7 @@ def generate_model_for_schema(schema_file: Path) -> str:
         lines = [f"# Type alias for {schema.get('title', model_name)}"]
         if "description" in schema:
             desc = escape_string_for_python(schema["description"])
-            lines.append(f'# {desc}')
+            lines.append(f"# {desc}")
         lines.append(f"{model_name} = {python_type}")
         return "\n".join(lines)
 
@@ -223,6 +223,70 @@ def validate_imports(output_file: Path) -> tuple[bool, str]:
         return False, f"Import validation error: {e}"
 
 
+def add_format_id_validation(code: str) -> str:
+    """
+    Add field validator to FormatId class for pattern enforcement.
+
+    The format-id.json schema specifies a pattern, but Pydantic v2 requires
+    explicit field_validator to enforce it.
+    """
+    # Find the FormatId class - match the class and find where to insert
+    # Look for the pattern: class FormatId(...): ... id: str = Field(...)
+    # Then add the validator after the last field
+
+    lines = code.split("\n")
+    result_lines = []
+    in_format_id = False
+    found_id_field = False
+    indent = ""
+
+    for i, line in enumerate(lines):
+        result_lines.append(line)
+
+        # Detect start of FormatId class
+        if "class FormatId(BaseModel):" in line:
+            in_format_id = True
+            # Detect indent level (usually 4 spaces)
+            if i + 1 < len(lines):
+                next_line = lines[i + 1]
+                indent = next_line[: len(next_line) - len(next_line.lstrip())]
+
+        # Detect the id field in FormatId class
+        if in_format_id and line.strip().startswith("id: str"):
+            found_id_field = True
+
+        # After the id field, add the validator
+        if (
+            in_format_id
+            and found_id_field
+            and (
+                line.strip() == ""
+                or (
+                    i + 1 < len(lines)
+                    and not lines[i + 1].strip().startswith(("agent_url:", "id:"))
+                )
+            )
+        ):
+            # Add validator here
+            validator_lines = [
+                "",
+                f'{indent}@field_validator("id")',
+                f"{indent}@classmethod",
+                f"{indent}def validate_id_pattern(cls, v: str) -> str:",
+                f'{indent}    """Validate format ID contains only alphanumeric characters, hyphens, and underscores."""',
+                f'{indent}    if not re.match(r"^[a-zA-Z0-9_-]+$", v):',
+                f"{indent}        raise ValueError(",
+                f'{indent}            f"Invalid format ID: {{v!r}}. Must contain only alphanumeric characters, hyphens, and underscores"',
+                f"{indent}        )",
+                f"{indent}    return v",
+            ]
+            result_lines.extend(validator_lines)
+            in_format_id = False
+            found_id_field = False
+
+    return "\n".join(result_lines)
+
+
 def main():
     """Generate models for core types and task request/response schemas."""
     if not SCHEMAS_DIR.exists():
@@ -233,6 +297,7 @@ def main():
 
     # Core domain types that are referenced by task schemas
     core_types = [
+        "format-id.json",  # Must come before format.json (which references it)
         "product.json",
         "media-buy.json",
         "package.json",
@@ -294,9 +359,10 @@ def main():
         "",
         "from __future__ import annotations",
         "",
+        "import re",
         "from typing import Any, Literal",
         "",
-        "from pydantic import BaseModel, Field",
+        "from pydantic import BaseModel, Field, field_validator",
         "",
         "",
         "# ============================================================================",
@@ -305,7 +371,6 @@ def main():
         "",
         "# These types are referenced in schemas but don't have schema files",
         "# Defining them as type aliases to maintain type safety",
-        "FormatId = str",
         "PackageRequest = dict[str, Any]",
         "PushNotificationConfig = dict[str, Any]",
         "ReportingCapabilities = dict[str, Any]",
@@ -353,6 +418,9 @@ def main():
     # Join all lines into final code
     generated_code = "\n".join(output_lines)
 
+    # Add custom validation for FormatId
+    generated_code = add_format_id_validation(generated_code)
+
     # Validate syntax before writing
     print("\nValidating generated code...")
     is_valid, error_msg = validate_python_syntax(generated_code, "generated.py")
 
@@ -36,13 +36,15 @@ def print_result(result: Any, json_output: bool = False) -> None:
                 "data": result.data,
                 "error": result.error,
                 "metadata": result.metadata,
-                "debug_info": {
-                    "request": result.debug_info.request,
-                    "response": result.debug_info.response,
-                    "duration_ms": result.debug_info.duration_ms,
-                }
-                if result.debug_info
-                else None,
+                "debug_info": (
+                    {
+                        "request": result.debug_info.request,
+                        "response": result.debug_info.response,
+                        "duration_ms": result.debug_info.duration_ms,
+                    }
+                    if result.debug_info
+                    else None
+                ),
             }
         )
     else:
@@ -73,10 +75,107 @@ async def execute_tool(
     config = AgentConfig(**agent_config)
 
     async with ADCPClient(config) as client:
-        result = await client.call_tool(tool_name, payload)
+        # Dispatch to specific method based on tool name
+        result = await _dispatch_tool(client, tool_name, payload)
         print_result(result, json_output)
 
 
+# Tool dispatch mapping - single source of truth for ADCP methods
+# Types are filled at runtime to avoid circular imports
+TOOL_DISPATCH: dict[str, tuple[str, type | None]] = {
+    "get_products": ("get_products", None),
+    "list_creative_formats": ("list_creative_formats", None),
+    "sync_creatives": ("sync_creatives", None),
+    "list_creatives": ("list_creatives", None),
+    "get_media_buy_delivery": ("get_media_buy_delivery", None),
+    "list_authorized_properties": ("list_authorized_properties", None),
+    "get_signals": ("get_signals", None),
+    "activate_signal": ("activate_signal", None),
+    "provide_performance_feedback": ("provide_performance_feedback", None),
+}
+
+
+async def _dispatch_tool(client: ADCPClient, tool_name: str, payload: dict[str, Any]) -> Any:
+    """Dispatch tool call to appropriate client method.
+
+    Args:
+        client: ADCP client instance
+        tool_name: Name of the tool to invoke
+        payload: Request payload as dict
+
+    Returns:
+        TaskResult with typed response or error
+
+    Raises:
+        ValidationError: If payload doesn't match request schema (caught and returned as TaskResult)
+    """
+    from pydantic import ValidationError
+
+    from adcp.types import generated as gen
+    from adcp.types.core import TaskResult, TaskStatus
+
+    # Lazy initialization of request types (avoid circular imports)
+    if TOOL_DISPATCH["get_products"][1] is None:
+        TOOL_DISPATCH["get_products"] = ("get_products", gen.GetProductsRequest)
+        TOOL_DISPATCH["list_creative_formats"] = (
+            "list_creative_formats",
+            gen.ListCreativeFormatsRequest,
+        )
+        TOOL_DISPATCH["sync_creatives"] = ("sync_creatives", gen.SyncCreativesRequest)
+        TOOL_DISPATCH["list_creatives"] = ("list_creatives", gen.ListCreativesRequest)
+        TOOL_DISPATCH["get_media_buy_delivery"] = (
+            "get_media_buy_delivery",
+            gen.GetMediaBuyDeliveryRequest,
+        )
+        TOOL_DISPATCH["list_authorized_properties"] = (
+            "list_authorized_properties",
+            gen.ListAuthorizedPropertiesRequest,
+        )
+        TOOL_DISPATCH["get_signals"] = ("get_signals", gen.GetSignalsRequest)
+        TOOL_DISPATCH["activate_signal"] = ("activate_signal", gen.ActivateSignalRequest)
+        TOOL_DISPATCH["provide_performance_feedback"] = (
+            "provide_performance_feedback",
+            gen.ProvidePerformanceFeedbackRequest,
+        )
+
+    # Check if tool exists
+    if tool_name not in TOOL_DISPATCH:
+        available = ", ".join(sorted(TOOL_DISPATCH.keys()))
+        return TaskResult(
+            status=TaskStatus.FAILED,
+            error=f"Unknown tool: {tool_name}. Available tools: {available}",
+        )
+
+    # Get method and request type
+    method_name, request_type = TOOL_DISPATCH[tool_name]
+
+    # Type guard - request_type should be initialized by this point
+    if request_type is None:
+        return TaskResult(
+            status=TaskStatus.FAILED,
+            error=f"Internal error: {tool_name} request type not initialized",
+        )
+
+    method = getattr(client, method_name)
+
+    # Validate and invoke
+    try:
+        request = request_type(**payload)
+        return await method(request)
+    except ValidationError as e:
+        # User-friendly error for invalid payloads
+        error_details = []
+        for error in e.errors():
+            field = ".".join(str(loc) for loc in error["loc"])
+            msg = error["msg"]
+            error_details.append(f"  - {field}: {msg}")
+
+        return TaskResult(
+            status=TaskStatus.FAILED,
+            error=f"Invalid request payload for {tool_name}:\n" + "\n".join(error_details),
+        )
+
+
 def load_payload(payload_arg: str | None) -> dict[str, Any]:
     """Load payload from argument (JSON, @file, or stdin)."""
     if not payload_arg: