diff --git a/src/adcp/types/versioned.py b/src/adcp/types/versioned.py
index a6578a785..50bc4a57d 100644
--- a/src/adcp/types/versioned.py
+++ b/src/adcp/types/versioned.py
@@ -6,19 +6,29 @@
 an older peer in the same SDK process. Generated ``.pyi`` files expose exact
 field, nested-value, and direct constructor requiredness information to type
 checkers; conditional JSON Schema constraints remain runtime-only. At runtime
-these remain exact-schema ``RootModel[dict[str, Any]]`` boundary validators, so
-subclassing them to add Pydantic fields is not supported; compose adopter-only
-state beside the versioned model instead.
+these remain exact-schema ``RootModel[dict[str, Any]]`` boundary validators.
+Use :func:`make_versioned_base` when one normal Pydantic model must combine a
+pinned protocol shape with adopter-defined, excluded internal fields.
 """
 
 from __future__ import annotations
 
 import copy
+import importlib
 import re
 from functools import cache
-from typing import Any, ClassVar, Literal
-
-from pydantic import GetJsonSchemaHandler, RootModel, model_validator
+from types import GenericAlias
+from typing import Any, ClassVar, Literal, Union
+
+from pydantic import (
+    BaseModel,
+    ConfigDict,
+    Field,
+    GetJsonSchemaHandler,
+    RootModel,
+    create_model,
+    model_validator,
+)
 from pydantic.json_schema import JsonSchemaValue
 from pydantic_core import CoreSchema
 
@@ -122,6 +132,86 @@ def expand(value: Any, stack: frozenset[str]) -> Any:
     return expanded
 
 
+def _resolve_local_ref(reference: str, document: dict[str, Any]) -> dict[str, Any] | None:
+    if not reference.startswith("#/"):
+        return None
+    target: Any = document
+    for raw_part in reference.removeprefix("#/").split("/"):
+        part = raw_part.replace("~1", "/").replace("~0", "~")
+        if not isinstance(target, dict) or part not in target:
+            return None
+        target = target[part]
+    return target if isinstance(target, dict) else None
+
+
+def _fallback_annotation(
+    schema: Any,
+    document: dict[str, Any],
+    seen: frozenset[str] = frozenset(),
+) -> Any:
+    """Return a conservative annotation when the current model has no field."""
+    if not isinstance(schema, dict):
+        return Any
+    reference = schema.get("$ref")
+    if isinstance(reference, str) and reference not in seen:
+        target = _resolve_local_ref(reference, document)
+        if target is not None:
+            return _fallback_annotation(target, document, seen | {reference})
+    values = schema.get("enum")
+    if isinstance(values, list) and values:
+        return Literal.__getitem__(tuple(values))
+    if "const" in schema:
+        return Literal.__getitem__((schema["const"],))
+    alternatives = schema.get("oneOf") or schema.get("anyOf")
+    if isinstance(alternatives, list) and alternatives:
+        annotations = list(
+            dict.fromkeys(
+                _fallback_annotation(part, document, seen)
+                for part in alternatives
+                if isinstance(part, dict)
+            )
+        )
+        if len(annotations) == 1:
+            return annotations[0]
+        if annotations:
+            return Union.__getitem__(tuple(annotations))
+    all_of = schema.get("allOf")
+    if isinstance(all_of, list):
+        annotations = [
+            annotation
+            for part in all_of
+            if isinstance(part, dict)
+            and (annotation := _fallback_annotation(part, document, seen)) is not Any
+        ]
+        if annotations:
+            return annotations[0]
+    schema_type = schema.get("type")
+    if isinstance(schema_type, list):
+        annotations = [
+            _fallback_annotation({**schema, "type": item}, document, seen) for item in schema_type
+        ]
+        return Union.__getitem__(tuple(dict.fromkeys(annotations)))
+    if schema_type == "array":
+        item_type = _fallback_annotation(schema.get("items", {}), document, seen)
+        return GenericAlias(list, item_type)
+    if schema_type == "object" or "properties" in schema:
+        additional = schema.get("additionalProperties")
+        value_type = (
+            _fallback_annotation(additional, document, seen)
+            if isinstance(additional, dict)
+            else Any
+        )
+        return GenericAlias(dict, (str, value_type))
+    primitive_types = {
+        "string": str,
+        "integer": int,
+        "number": float,
+        "boolean": bool,
+        "null": type(None),
+    }
+    return primitive_types.get(schema_type, Any) if isinstance(schema_type, str) else Any
+
+
 class VersionedSchemaModel(RootModel[dict[str, Any]]):
     """Dict-shaped Pydantic model that enforces one bundled schema version.
 
@@ -206,6 +296,84 @@ def __get_pydantic_json_schema__(
         return _inline_local_refs(cls.schema_document)
 
 
+class _VersionedExtensionModel(BaseModel):
+    """Normal Pydantic base carrying one pinned protocol boundary shape."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    schema_version: ClassVar[str]
+    schema_tool_name: ClassVar[str]
+    schema_direction: ClassVar[VersionedDirection]
+    schema_document: ClassVar[dict[str, Any]]
+
+    @model_validator(mode="before")
+    @classmethod
+    def _apply_schema_defaults(cls, value: Any) -> Any:
+        if not isinstance(value, dict):
+            return value
+        result = dict(value)
+        raw_properties = cls.schema_document.get("properties", {})
+        properties = raw_properties if isinstance(raw_properties, dict) else {}
+        for name, field_schema in properties.items():
+            if name not in result and isinstance(field_schema, dict) and "default" in field_schema:
+                result[name] = copy.deepcopy(field_schema["default"])
+        return result
+
+    def _protocol_payload(self) -> dict[str, Any]:
+        return BaseModel.model_dump(
+            self,
+            mode="json",
+            by_alias=True,
+            exclude_unset=True,
+        )
+
+    @model_validator(mode="after")
+    def _validate_schema_document(self) -> _VersionedExtensionModel:
+        validator = get_validator(
+            self.schema_tool_name,
+            self.schema_direction,
+            version=self.schema_version,
+        )
+        if validator is None:
+            raise ValueError(
+                f"no {self.schema_version} schema for "
+                f"{self.schema_tool_name}::{self.schema_direction}"
+            )
+        issues = sorted(
+            validator.iter_errors(self._protocol_payload()),
+            key=lambda error: list(error.path),
+        )
+        if issues:
+            issue = issues[0]
+            path = ".".join(str(part) for part in issue.absolute_path) or "<root>"
+            raise ValueError(f"{path}: {issue.message}")
+        return self
+
+    def model_dump(self, *args: Any, **kwargs: Any) -> dict[str, Any]:
+        """Keep absent optional fields out of the protocol wire payload."""
+        kwargs.setdefault("exclude_unset", True)
+        return super().model_dump(*args, **kwargs)
+
+    def model_dump_json(self, *args: Any, **kwargs: Any) -> str:
+        """JSON form of :meth:`model_dump` with the same boundary semantics."""
+        kwargs.setdefault("exclude_unset", True)
+        return super().model_dump_json(*args, **kwargs)
+
+    @classmethod
+    def model_json_schema(cls, *args: Any, **kwargs: Any) -> dict[str, Any]:
+        del args, kwargs
+        return copy.deepcopy(cls.schema_document)
+
+    @classmethod
+    def __get_pydantic_json_schema__(
+        cls,
+        core_schema: CoreSchema,
+        handler: GetJsonSchemaHandler,
+    ) -> JsonSchemaValue:
+        del core_schema, handler
+        return _inline_local_refs(cls.schema_document)
+
+
 def _pascal_case(tool_name: str) -> str:
     return "".join(part.capitalize() for part in tool_name.split("_"))
 
@@ -215,6 +383,106 @@ def _snake_case(model_stem: str) -> str:
     return re.sub(r"([a-z0-9])([A-Z])", r"\1_\2", step1).lower()
 
 
+def _schema_key_for_model_name(model_name: str) -> tuple[str, VersionedDirection]:
+    direction: VersionedDirection
+    if model_name.endswith("SubmittedResponse"):
+        direction = "submitted"
+        stem = model_name[: -len("SubmittedResponse")]
+    elif model_name.endswith("WorkingResponse"):
+        direction = "working"
+        stem = model_name[: -len("WorkingResponse")]
+    elif model_name.endswith("InputRequiredResponse"):
+        direction = "input-required"
+        stem = model_name[: -len("InputRequiredResponse")]
+    elif model_name.endswith("Request"):
+        direction = "request"
+        stem = model_name[: -len("Request")]
+    elif model_name.endswith("Response"):
+        direction = "sync"
+        stem = model_name[: -len("Response")]
+    else:
+        raise AttributeError(
+            f"version-scoped model names must end in Request or Response: {model_name}"
+        )
+    return _snake_case(stem), direction
+
+
+def _current_model_annotations(model_name: str) -> dict[str, Any]:
+    current_types = importlib.import_module("adcp.types")
+    current_model = getattr(current_types, model_name, None)
+    if not isinstance(current_model, type) or not issubclass(current_model, BaseModel):
+        return {}
+    return {
+        name: field.annotation if field.annotation is not None else Any
+        for name, field in current_model.model_fields.items()
+    }
+
+
+@cache
+def make_versioned_base(version: str, model_name: str) -> type[BaseModel]:
+    """Create a subclassable Pydantic base for one bundled protocol model.
+
+    Example::
+
+        ListCreatives31 = make_versioned_base("3.1", "ListCreativesRequest")
+
+        class SellerListCreativesRequest(ListCreatives31):
+            internal_tenant_id: str = Field(exclude=True)
+
+    The returned class has real top-level Pydantic fields, reuses nested
+    runtime annotations from the SDK's current public model where available,
+    and validates its serialized protocol payload against the requested
+    bundled schema. Adopter subclasses may add fields declared with
+    ``Field(exclude=True)``; those fields never enter schema validation or the
+    wire payload. Unknown undeclared fields are rejected even when a protocol
+    schema permits extension keys, keeping version-only fields explicit.
+    """
+    tool_name, direction = _schema_key_for_model_name(model_name)
+    schema = get_portable_schema(tool_name, direction, version=version)
+    if schema is None:
+        raise LookupError(f"no {version} schema for {tool_name}::{direction}")
+
+    shapes = _object_shapes(schema, schema)
+    properties = {
+        name: field_schema
+        for shape_properties, _required in shapes
+        for name, field_schema in shape_properties.items()
+    }
+    guaranteed_fields = set.intersection(*(required for _properties, required in shapes))
+    current_annotations = _current_model_annotations(model_name)
+    fields: dict[str, Any] = {}
+    for name, field_schema in properties.items():
+        annotation = current_annotations.get(
+            name,
+            _fallback_annotation(field_schema, schema),
+        )
+        description = field_schema.get("description") if isinstance(field_schema, dict) else None
+        if isinstance(field_schema, dict) and "default" in field_schema:
+            default = Field(
+                default=copy.deepcopy(field_schema["default"]),
+                description=description,
+            )
+        elif name in guaranteed_fields:
+            default = Field(description=description)
+        else:
+            default = Field(default=None, description=description)
+        fields[name] = (annotation, default)
+
+    version_token = re.sub(r"[^A-Za-z0-9]+", "_", version).strip("_")
+    generated_name = f"{model_name}V{version_token}Base"
+    model: type[_VersionedExtensionModel] = create_model(
+        generated_name,
+        __base__=_VersionedExtensionModel,
+        __module__=__name__,
+        **fields,
+    )
+    model.schema_version = version
+    model.schema_tool_name = tool_name
+    model.schema_direction = direction
+    model.schema_document = schema
+    return model
+
+
 @cache
 def schema_model_for_version(
     version: str,
@@ -261,29 +529,10 @@ def schema_model_for_version(
 
 def model_for_version(version: str, model_name: str) -> type[VersionedSchemaModel]:
     """Resolve ``ListCreativesRequest``-style names for a protocol release."""
-    direction: VersionedDirection
-    if model_name.endswith("SubmittedResponse"):
-        direction = "submitted"
-        stem = model_name[: -len("SubmittedResponse")]
-    elif model_name.endswith("WorkingResponse"):
-        direction = "working"
-        stem = model_name[: -len("WorkingResponse")]
-    elif model_name.endswith("InputRequiredResponse"):
-        direction = "input-required"
-        stem = model_name[: -len("InputRequiredResponse")]
-    elif model_name.endswith("Request"):
-        direction = "request"
-        stem = model_name[: -len("Request")]
-    elif model_name.endswith("Response"):
-        direction = "sync"
-        stem = model_name[: -len("Response")]
-    else:
-        raise AttributeError(
-            f"version-scoped model names must end in Request or Response: {model_name}"
-        )
+    tool_name, direction = _schema_key_for_model_name(model_name)
     return schema_model_for_version(
         version,
-        _snake_case(stem),
+        tool_name,
         direction,
         model_name=model_name,
     )
@@ -325,6 +574,7 @@ def directory() -> list[str]:
 __all__ = [
     "VersionedDirection",
     "VersionedSchemaModel",
+    "make_versioned_base",
     "model_for_version",
     "schema_model_for_version",
     "versioned_surface",
diff --git a/tests/test_version_scoped_models.py b/tests/test_version_scoped_models.py
index 29e02ec24..49c54dea7 100644
--- a/tests/test_version_scoped_models.py
+++ b/tests/test_version_scoped_models.py
@@ -2,11 +2,12 @@
 
 from __future__ import annotations
 
+import json
 from pathlib import Path
 
 import pytest
 from jsonschema.validators import validator_for
-from pydantic import TypeAdapter, ValidationError
+from pydantic import Field, TypeAdapter, ValidationError
 
 from adcp.server import (
     ADCPHandler,
@@ -23,7 +24,8 @@
 from adcp.types.v31 import PackageRequest as PackageRequest31
 from adcp.types.v32 import ListCreativesRequest as ListCreativesRequest32
 from adcp.types.v32 import PackageRequest as PackageRequest32
-from adcp.validation import get_mcp_schema, get_validator
+from adcp.types.versioned import make_versioned_base
+from adcp.validation import get_mcp_schema, get_portable_schema, get_validator
 
 ROOT = Path(__file__).parents[1]
 
@@ -93,6 +95,91 @@ def test_generated_stubs_preserve_nested_all_of_requiredness() -> None:
     assert "coverage_rate: Required[" in metrics
 
 
+def test_versioned_base_supports_excluded_adopter_fields() -> None:
+    base = make_versioned_base("3.1", "ListCreativesRequest")
+
+    class SellerListCreativesRequest(base):
+        internal_tenant_id: str = Field(exclude=True)
+
+    request = SellerListCreativesRequest(
+        internal_tenant_id="tenant-1",
+        include_assignments=True,
+    )
+    payload = request.model_dump(mode="json")
+
+    assert "include_assignments" in SellerListCreativesRequest.model_fields
+    assert "internal_tenant_id" in SellerListCreativesRequest.model_fields
+    assert payload["include_assignments"] is True
+    assert "internal_tenant_id" not in payload
+    assert "internal_tenant_id" not in json.loads(request.model_dump_json())
+
+
+def test_versioned_base_uses_current_nested_runtime_models() -> None:
+    from adcp.types import CreativeFilters
+
+    base = make_versioned_base("3.1", "ListCreativesRequest")
+    request = base(filters={"statuses": ["approved"]})
+
+    assert isinstance(request.filters, CreativeFilters)
+    assert request.model_dump(mode="json")["filters"] == {"statuses": ["approved"]}
+
+
+def test_versioned_base_emits_only_the_canonical_pinned_schema() -> None:
+    base = make_versioned_base("3.1", "ListCreativesRequest")
+
+    class SellerListCreativesRequest(base):
+        internal_tenant_id: str = Field(exclude=True)
+
+    canonical = get_portable_schema("list_creatives", "request", version="3.1")
+    assert canonical is not None
+    assert SellerListCreativesRequest.model_json_schema() == canonical
+
+    adapter_schema = TypeAdapter(SellerListCreativesRequest).json_schema()
+    encoded = json.dumps(adapter_schema)
+    assert "internal_tenant_id" not in encoded
+    assert "assignment_projection" not in encoded
+
+
+def test_versioned_base_enforces_version_delta_fields() -> None:
+    request31 = make_versioned_base("3.1", "ListCreativesRequest")
+    request32 = make_versioned_base("3.2-beta.0", "ListCreativesRequest")
+
+    assert "assignment_projection" not in request31.model_fields
+    assert "assignment_limit" not in request31.model_fields
+    assert "assignment_projection" in request32.model_fields
+    assert "assignment_limit" in request32.model_fields
+
+    with pytest.raises(ValidationError, match="assignment_projection"):
+        request31(assignment_projection="all")
+    valid32 = request32(
+        assignment_projection="matching",
+        filters={"indicator_types": ["creative_fatigue"]},
+    )
+    assert valid32.assignment_projection == "matching"
+
+
+def test_versioned_base_keeps_31_package_budget_required() -> None:
+    package = make_versioned_base("3.1", "PackageRequest")
+
+    assert package.model_fields["budget"].is_required()
+    assert "format_ids" in package.model_fields
+    with pytest.raises(ValidationError, match="budget"):
+        package(product_id="product-1", pricing_option_id="fixed")
+    with pytest.raises(ValidationError, match="budget"):
+        package(product_id="product-1", pricing_option_id="fixed", budget=None)
+
+    valid = package(product_id="product-1", pricing_option_id="fixed", budget=100.0)
+    assert valid.budget == 100.0
+
+
+def test_versioned_base_is_cached_and_rejects_unknown_models() -> None:
+    assert make_versioned_base("3.1", "ListCreativesRequest") is make_versioned_base(
+        "3.1", "ListCreativesRequest"
+    )
+    with pytest.raises(LookupError, match="no 3.1 schema"):
+        make_versioned_base("3.1", "NotAProtocolRequest")
+
+
 def test_versioned_models_keep_generated_model_ergonomics() -> None:
     request = ListCreativesRequest31(include_assignments=True)
     assert request.include_assignments is True