diff --git a/.gitignore b/.gitignore
index 6a1667a1..fe2cbe00 100644
--- a/.gitignore
+++ b/.gitignore
@@ -156,6 +156,11 @@ uv.lock
 # Temporary schema processing directory
 .schema_temp/
 
+# Skill snapshot directories written by scripts/sync_schemas.py when a
+# tarball-managed skill is updated. These are local working copies for
+# diffing; they should not be committed.
+skills/*.previous
+
 # RegistrySync cursor file (written by FileCursorStore)
 .adcp-sync-cursor.json
 
diff --git a/Makefile b/Makefile
index 047a6165..09146874 100644
--- a/Makefile
+++ b/Makefile
@@ -1,4 +1,4 @@
-.PHONY: help format lint typecheck test regenerate-schemas pre-push ci-local clean install-dev
+.PHONY: help format lint typecheck test regenerate-schemas pre-push ci-local clean install-dev check-schema-drift
 
 # Detect Python and use venv if available
 PYTHON := $(shell if [ -f .venv/bin/python ]; then echo .venv/bin/python; else echo python3; fi)
@@ -46,8 +46,8 @@ regenerate-registry: ## Regenerate registry types from OpenAPI spec
 	$(PYTHON) scripts/generate_registry_types.py
 	@echo "✓ Registry types regenerated"
 
-regenerate-schemas: ## Download latest schemas and regenerate models
-	@echo "Downloading latest schemas..."
+regenerate-schemas: ## Download latest schemas and skills from bundle, then regenerate models
+	@echo "Downloading latest schemas and skills..."
 	$(PYTHON) scripts/sync_schemas.py
 	@echo "Fixing schema references..."
 	$(PYTHON) scripts/fix_schema_refs.py
@@ -109,7 +109,7 @@ full-check: pre-push ## Alias for pre-push (full check before committing)
 
 check-schema-drift: ## Check if schemas are out of sync with upstream
 	@echo "Checking for schema drift..."
-	@$(PYTHON) scripts/sync_schemas.py
+	@$(PYTHON) scripts/sync_schemas.py --no-skills
 	@$(PYTHON) scripts/fix_schema_refs.py
 	@$(PYTHON) scripts/generate_types.py
 	@if git diff --exit-code src/adcp/types/_generated.py schemas/cache/; then \
diff --git a/scripts/sync_schemas.py b/scripts/sync_schemas.py
index d5c9f350..5629ceaa 100755
--- a/scripts/sync_schemas.py
+++ b/scripts/sync_schemas.py
@@ -1,11 +1,11 @@
 #!/usr/bin/env python3
-"""
-Sync AdCP JSON schemas from the authoritative protocol bundle.
+"""Sync AdCP JSON schemas and agent skills from the authoritative protocol bundle.
 
 Downloads a single gzipped tarball at
 https://adcontextprotocol.org/protocol/{version}.tgz, verifies the bundle
 against its SHA-256 sidecar, and extracts the `schemas/` tree into
-`schemas/cache/`. Replaces the prior per-file sync.
+`schemas/cache/` and protocol-managed skills into `skills/`. Replaces the
+prior per-file sync.
 
 Pinned releases additionally ship Sigstore keyless sidecars (`.tgz.sig` +
 `.tgz.crt`). When present, this script shells out to `cosign verify-blob`
@@ -17,13 +17,16 @@
 bundle is not published, sync falls back to `latest.tgz` (the dev snapshot).
 
 Usage:
-    python scripts/sync_schemas.py
+    python scripts/sync_schemas.py              # sync schemas + skills
+    python scripts/sync_schemas.py --no-skills  # schemas only (e.g. drift checks)
 """
 
 from __future__ import annotations
 
+import argparse
 import hashlib
 import io
+import json
 import os
 import shutil
 import subprocess
@@ -36,6 +39,7 @@
 
 REPO_ROOT = Path(__file__).parent.parent
 CACHE_DIR = REPO_ROOT / "schemas" / "cache"
+SKILLS_DIR = REPO_ROOT / "skills"
 VERSION_FILE = REPO_ROOT / "src" / "adcp" / "ADCP_VERSION"
 BUNDLE_BASE_URL = "https://adcontextprotocol.org/protocol"
 USER_AGENT = "adcp-python-sdk/3.0"
@@ -163,35 +167,119 @@ def verify_cosign_signature(
         )
 
 
-def replace_cache_from_bundle(tgz_bytes: bytes, effective_version: str) -> int:
+def _extract_bundle(
+    tgz_bytes: bytes, effective_version: str
+) -> tuple[Path, tempfile.TemporaryDirectory[str]]:
+    """Extract the bundle to a temporary directory and return the bundle root.
+
+    The caller is responsible for closing the returned TemporaryDirectory.
+    Use as a context manager: ``with tmpdir: ...``
+    """
+    tmpdir: tempfile.TemporaryDirectory[str] = tempfile.TemporaryDirectory()
+    try:
+        with tarfile.open(fileobj=io.BytesIO(tgz_bytes), mode="r:gz") as tf:
+            tf.extractall(tmpdir.name, filter="data")
+    except Exception:
+        tmpdir.cleanup()
+        raise
+    bundle_root = Path(tmpdir.name) / f"adcp-{effective_version}"
+    return bundle_root, tmpdir
+
+
+def replace_cache_from_bundle(bundle_root: Path) -> int:
     """Extract the bundle's `schemas/` tree into CACHE_DIR, replacing its contents.
 
     Returns the number of files written.
     """
-    with tempfile.TemporaryDirectory() as tmpdir:
-        with tarfile.open(fileobj=io.BytesIO(tgz_bytes), mode="r:gz") as tf:
-            tf.extractall(tmpdir, filter="data")
+    schemas_src = bundle_root / "schemas"
+    if not schemas_src.is_dir():
+        raise RuntimeError(
+            f"Bundle missing expected directory: {bundle_root.name}/schemas/"
+        )
 
-        bundle_root = Path(tmpdir) / f"adcp-{effective_version}"
-        schemas_src = bundle_root / "schemas"
-        if not schemas_src.is_dir():
-            raise RuntimeError(
-                f"Bundle missing expected directory: adcp-{effective_version}/schemas/"
-            )
+    if CACHE_DIR.exists():
+        shutil.rmtree(CACHE_DIR)
+    CACHE_DIR.parent.mkdir(parents=True, exist_ok=True)
+    shutil.copytree(schemas_src, CACHE_DIR)
 
-        if CACHE_DIR.exists():
-            shutil.rmtree(CACHE_DIR)
-        CACHE_DIR.parent.mkdir(parents=True, exist_ok=True)
-        shutil.copytree(schemas_src, CACHE_DIR)
+    return sum(1 for _ in CACHE_DIR.rglob("*") if _.is_file())
 
-        return sum(1 for _ in CACHE_DIR.rglob("*") if _.is_file())
+
+def sync_skills_from_bundle(bundle_root: Path, skills_dir: Path) -> int:
+    """Sync protocol-managed skills from the bundle into skills_dir.
+
+    Reads manifest.json to enumerate canonical skills, then copies each
+    skill directory, excluding nested schemas/ subdirs (the SDK already has
+    those in schemas/cache/). SDK-local skills not in the manifest are left
+    untouched. Previous versions are snapshotted as <name>.previous siblings.
+
+    Returns the number of skill files written.
+    """
+    manifest_path = bundle_root / "manifest.json"
+    if not manifest_path.exists():
+        print("  ! No manifest.json in bundle — skipping skill sync")
+        return 0
+
+    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+    skill_names = manifest.get("contents", {}).get("skills", [])
+    if not isinstance(skill_names, list) or not skill_names:
+        print("  ! No skills listed in bundle manifest — skipping skill sync")
+        return 0
+
+    skills_dir.mkdir(parents=True, exist_ok=True)
+    count = 0
+
+    for name in skill_names:
+        if not isinstance(name, str):
+            print(f"  ! Skipping non-string skill entry: {name!r}")
+            continue
+
+        # Guard against path traversal: reject names containing "/" or that
+        # resolve to a different basename (e.g. "good/../evil" or "../evil").
+        if "/" in name or name != Path(name).name:
+            raise RuntimeError(f"Unsafe skill name rejected: {name!r}")
+
+        src = bundle_root / "skills" / name
+        if not src.is_dir():
+            print(f"  ! Skill directory missing in bundle: skills/{name}/ — skipping")
+            continue
+
+        dst = skills_dir / name
+        prev = skills_dir / f"{name}.previous"
+        if dst.exists():
+            if prev.is_dir():
+                shutil.rmtree(prev)
+            elif prev.exists():
+                prev.unlink()
+            shutil.copytree(dst, prev)
+            shutil.rmtree(dst)
+
+        # Copy the skill tree, excluding embedded schemas/ subdirs —
+        # those duplicate the canonical schemas/cache/ tree the SDK already has.
+        shutil.copytree(src, dst, ignore=shutil.ignore_patterns("schemas"))
+        count += sum(1 for _ in dst.rglob("*") if _.is_file())
+
+    return count
 
 
 def main() -> None:
+    parser = argparse.ArgumentParser(
+        description="Sync AdCP schemas and skills from the protocol bundle."
+    )
+    parser.add_argument(
+        "--no-skills",
+        action="store_true",
+        help="Skip skill sync (useful for schema-only drift checks).",
+    )
+    args = parser.parse_args()
+
     target_version = get_target_adcp_version()
-    print(f"Syncing AdCP schema bundle from {BUNDLE_BASE_URL}...")
+    print(f"Syncing AdCP protocol bundle from {BUNDLE_BASE_URL}...")
     print(f"Target version: {target_version}")
-    print(f"Cache directory: {CACHE_DIR}\n")
+    print(f"Schema cache: {CACHE_DIR}")
+    if not args.no_skills:
+        print(f"Skills dir:   {SKILLS_DIR}")
+    print()
 
     try:
         print(f"Fetching {target_version}.tgz + checksum...")
@@ -239,14 +327,33 @@ def main() -> None:
             )
 
     try:
-        file_count = replace_cache_from_bundle(tgz_bytes, effective_version)
+        bundle_root, tmpdir = _extract_bundle(tgz_bytes, effective_version)
     except (tarfile.TarError, RuntimeError) as exc:
         print(f"\n✗ Failed to extract bundle: {exc}", file=sys.stderr)
         sys.exit(1)
 
-    print(f"\n✓ Successfully synced {file_count} schema files")
+    with tmpdir:
+        try:
+            schema_count = replace_cache_from_bundle(bundle_root)
+        except (OSError, shutil.Error, RuntimeError) as exc:
+            print(f"\n✗ Failed to extract schemas: {exc}", file=sys.stderr)
+            sys.exit(1)
+
+        skill_count = 0
+        if not args.no_skills:
+            try:
+                skill_count = sync_skills_from_bundle(bundle_root, SKILLS_DIR)
+            except (OSError, shutil.Error, RuntimeError) as exc:
+                print(f"\n✗ Failed to sync skills: {exc}", file=sys.stderr)
+                sys.exit(1)
+
+    print(f"\n✓ Successfully synced {schema_count} schema files")
+    if not args.no_skills:
+        print(f"✓ Successfully synced {skill_count} skill files")
     print(f"  Effective version: adcp-{effective_version}")
-    print(f"  Location: {CACHE_DIR}")
+    print(f"  Schema location:  {CACHE_DIR}")
+    if not args.no_skills:
+        print(f"  Skills location:  {SKILLS_DIR}")
 
 
 if __name__ == "__main__":
diff --git a/tests/test_sync_schemas.py b/tests/test_sync_schemas.py
new file mode 100644
index 00000000..9a098b35
--- /dev/null
+++ b/tests/test_sync_schemas.py
@@ -0,0 +1,348 @@
+"""Tests for scripts/sync_schemas.py — skills sync functions."""
+
+from __future__ import annotations
+
+# Import the functions under test directly from the script module.
+# sync_schemas.py lives under scripts/, which is not on sys.path by default,
+# so we load it via importlib.
+import importlib.util
+import json
+import sys
+import tempfile
+from pathlib import Path
+
+import pytest
+
+_SCRIPT = Path(__file__).parent.parent / "scripts" / "sync_schemas.py"
+_spec = importlib.util.spec_from_file_location("sync_schemas", _SCRIPT)
+assert _spec is not None and _spec.loader is not None
+_mod = importlib.util.module_from_spec(_spec)
+sys.modules["sync_schemas"] = _mod
+_spec.loader.exec_module(_mod)  # type: ignore[union-attr]
+
+replace_cache_from_bundle = _mod.replace_cache_from_bundle
+sync_skills_from_bundle = _mod.sync_skills_from_bundle
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _make_bundle(
+    tmp: Path,
+    skills: dict[str, dict[str, str]] | None = None,
+    manifest_skills: list[object] | None = None,
+    include_manifest: bool = True,
+    schemas_in_bundle: bool = False,
+) -> Path:
+    """Build a fake bundle_root directory for tests.
+
+    Args:
+        tmp: Parent temp directory.
+        skills: Mapping of skill_name → {filename: content} for files to create.
+        manifest_skills: List value for manifest["contents"]["skills"]. Defaults
+            to the keys of *skills* when provided.
+        include_manifest: Whether to write manifest.json at all.
+        schemas_in_bundle: Whether to add a schemas/ subdir inside each skill.
+    """
+    bundle_root = tmp / "adcp-test"
+    bundle_root.mkdir(exist_ok=True)
+
+    if skills:
+        skills_root = bundle_root / "skills"
+        skills_root.mkdir(exist_ok=True)
+        for skill_name, files in skills.items():
+            skill_dir = skills_root / skill_name
+            skill_dir.mkdir()
+            for fname, content in files.items():
+                (skill_dir / fname).write_text(content)
+            if schemas_in_bundle:
+                (skill_dir / "schemas").mkdir()
+                (skill_dir / "schemas" / "schema.json").write_text("{}")
+
+    if include_manifest:
+        names: list[object]
+        if manifest_skills is not None:
+            names = manifest_skills
+        elif skills:
+            names = list(skills.keys())
+        else:
+            names = []
+        manifest = {"contents": {"skills": names}}
+        (bundle_root / "manifest.json").write_text(json.dumps(manifest))
+
+    return bundle_root
+
+
+# ---------------------------------------------------------------------------
+# replace_cache_from_bundle
+# ---------------------------------------------------------------------------
+
+
+class TestReplaceCacheFromBundle:
+    def test_copies_schemas_to_cache_dir(self, tmp_path: Path) -> None:
+        bundle_root = tmp_path / "adcp-test"
+        schemas_src = bundle_root / "schemas"
+        schemas_src.mkdir(parents=True)
+        (schemas_src / "request.json").write_text('{"type":"object"}')
+
+        cache_dir = tmp_path / "cache"
+        # Monkeypatch CACHE_DIR for this test
+        original = _mod.CACHE_DIR
+        _mod.CACHE_DIR = cache_dir
+        try:
+            count = replace_cache_from_bundle(bundle_root)
+        finally:
+            _mod.CACHE_DIR = original
+
+        assert count == 1
+        assert (cache_dir / "request.json").read_text() == '{"type":"object"}'
+
+    def test_raises_if_schemas_dir_missing(self, tmp_path: Path) -> None:
+        bundle_root = tmp_path / "adcp-test"
+        bundle_root.mkdir()
+
+        with pytest.raises(RuntimeError, match="Bundle missing expected directory"):
+            replace_cache_from_bundle(bundle_root)
+
+
+# ---------------------------------------------------------------------------
+# sync_skills_from_bundle
+# ---------------------------------------------------------------------------
+
+
+class TestSyncSkillsFromBundle:
+    def test_no_manifest_returns_zero(self, capsys: pytest.CaptureFixture[str]) -> None:
+        with tempfile.TemporaryDirectory() as tmp_str:
+            tmp = Path(tmp_str)
+            bundle_root = _make_bundle(tmp, include_manifest=False)
+            skills_dir = tmp / "skills"
+
+            result = sync_skills_from_bundle(bundle_root, skills_dir)
+
+            assert result == 0
+        assert "No manifest.json" in capsys.readouterr().out
+
+    def test_empty_skills_list_returns_zero(
+        self, capsys: pytest.CaptureFixture[str]
+    ) -> None:
+        with tempfile.TemporaryDirectory() as tmp_str:
+            tmp = Path(tmp_str)
+            bundle_root = _make_bundle(tmp, manifest_skills=[])
+            skills_dir = tmp / "skills"
+
+            result = sync_skills_from_bundle(bundle_root, skills_dir)
+
+            assert result == 0
+        assert "No skills listed" in capsys.readouterr().out
+
+    def test_skill_copied_to_skills_dir(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_str:
+            tmp = Path(tmp_str)
+            bundle_root = _make_bundle(
+                tmp, skills={"call-adcp-agent": {"SKILL.md": "# Call AdCP Agent"}}
+            )
+            skills_dir = tmp / "skills"
+
+            count = sync_skills_from_bundle(bundle_root, skills_dir)
+
+            assert count == 1
+            assert (skills_dir / "call-adcp-agent" / "SKILL.md").read_text() == (
+                "# Call AdCP Agent"
+            )
+
+    def test_schemas_subdir_excluded(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_str:
+            tmp = Path(tmp_str)
+            bundle_root = _make_bundle(
+                tmp,
+                skills={"adcp-brand": {"SKILL.md": "# Brand"}},
+                schemas_in_bundle=True,
+            )
+            skills_dir = tmp / "skills"
+
+            sync_skills_from_bundle(bundle_root, skills_dir)
+
+            # SKILL.md should be copied; schemas/ subdir must be excluded
+            assert (skills_dir / "adcp-brand" / "SKILL.md").exists()
+            assert not (skills_dir / "adcp-brand" / "schemas").exists()
+
+    def test_previous_snapshot_created_on_update(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_str:
+            tmp = Path(tmp_str)
+            skills_dir = tmp / "skills"
+            skills_dir.mkdir()
+
+            # Pre-populate an existing skill
+            existing = skills_dir / "adcp-brand"
+            existing.mkdir()
+            (existing / "SKILL.md").write_text("# Old Brand")
+
+            bundle_root = _make_bundle(
+                tmp, skills={"adcp-brand": {"SKILL.md": "# New Brand"}}
+            )
+            sync_skills_from_bundle(bundle_root, skills_dir)
+
+            assert (skills_dir / "adcp-brand" / "SKILL.md").read_text() == "# New Brand"
+            assert (skills_dir / "adcp-brand.previous" / "SKILL.md").read_text() == (
+                "# Old Brand"
+            )
+
+    def test_previous_snapshot_replaced_on_second_update(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_str:
+            tmp = Path(tmp_str)
+            skills_dir = tmp / "skills"
+            skills_dir.mkdir()
+
+            # Set up an existing .previous dir from a prior run
+            prev = skills_dir / "adcp-brand.previous"
+            prev.mkdir()
+            (prev / "SKILL.md").write_text("# Very Old Brand")
+
+            existing = skills_dir / "adcp-brand"
+            existing.mkdir()
+            (existing / "SKILL.md").write_text("# Old Brand")
+
+            bundle_root = _make_bundle(
+                tmp, skills={"adcp-brand": {"SKILL.md": "# New Brand"}}
+            )
+            sync_skills_from_bundle(bundle_root, skills_dir)
+
+            assert (skills_dir / "adcp-brand.previous" / "SKILL.md").read_text() == (
+                "# Old Brand"
+            )
+
+    def test_local_only_skill_untouched(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_str:
+            tmp = Path(tmp_str)
+            skills_dir = tmp / "skills"
+            skills_dir.mkdir()
+
+            # A locally-managed skill not in the manifest
+            local = skills_dir / "build-seller-agent"
+            local.mkdir()
+            (local / "SKILL.md").write_text("# Seller Agent (SDK-local)")
+
+            bundle_root = _make_bundle(
+                tmp,
+                skills={"call-adcp-agent": {"SKILL.md": "# Call"}},
+                manifest_skills=["call-adcp-agent"],
+            )
+            sync_skills_from_bundle(bundle_root, skills_dir)
+
+            assert (skills_dir / "build-seller-agent" / "SKILL.md").read_text() == (
+                "# Seller Agent (SDK-local)"
+            )
+
+    def test_path_traversal_dotdot_rejected(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_str:
+            tmp = Path(tmp_str)
+            bundle_root = _make_bundle(
+                tmp,
+                skills={},
+                manifest_skills=["../evil"],
+                include_manifest=True,
+            )
+            skills_dir = tmp / "skills"
+
+            with pytest.raises(RuntimeError, match="Unsafe skill name rejected"):
+                sync_skills_from_bundle(bundle_root, skills_dir)
+
+    def test_path_traversal_slash_in_name_rejected(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_str:
+            tmp = Path(tmp_str)
+            bundle_root = _make_bundle(
+                tmp,
+                skills={},
+                manifest_skills=["good/../evil"],
+                include_manifest=True,
+            )
+            skills_dir = tmp / "skills"
+
+            with pytest.raises(RuntimeError, match="Unsafe skill name rejected"):
+                sync_skills_from_bundle(bundle_root, skills_dir)
+
+    def test_non_string_name_skipped(
+        self, capsys: pytest.CaptureFixture[str]
+    ) -> None:
+        with tempfile.TemporaryDirectory() as tmp_str:
+            tmp = Path(tmp_str)
+            bundle_root = _make_bundle(
+                tmp,
+                skills={"call-adcp-agent": {"SKILL.md": "# Call"}},
+                manifest_skills=[42, "call-adcp-agent"],
+            )
+            skills_dir = tmp / "skills"
+
+            count = sync_skills_from_bundle(bundle_root, skills_dir)
+
+            assert count == 1  # only the valid string entry is synced
+        assert "Skipping non-string" in capsys.readouterr().out
+
+    def test_missing_bundle_skill_dir_skipped(
+        self, capsys: pytest.CaptureFixture[str]
+    ) -> None:
+        with tempfile.TemporaryDirectory() as tmp_str:
+            tmp = Path(tmp_str)
+            # Manifest lists a skill that has no corresponding directory in the bundle
+            bundle_root = _make_bundle(
+                tmp,
+                skills={},
+                manifest_skills=["adcp-brand"],
+                include_manifest=True,
+            )
+            skills_dir = tmp / "skills"
+
+            count = sync_skills_from_bundle(bundle_root, skills_dir)
+
+            assert count == 0
+        assert "missing in bundle" in capsys.readouterr().out
+
+    def test_missing_bundle_skill_preserves_existing_dst(
+        self, capsys: pytest.CaptureFixture[str]
+    ) -> None:
+        with tempfile.TemporaryDirectory() as tmp_str:
+            tmp = Path(tmp_str)
+            skills_dir = tmp / "skills"
+            skills_dir.mkdir()
+
+            # Pre-existing skill in dst
+            existing = skills_dir / "adcp-brand"
+            existing.mkdir()
+            (existing / "SKILL.md").write_text("# Existing Brand")
+
+            # Manifest lists the skill but bundle has no source dir for it
+            bundle_root = _make_bundle(
+                tmp,
+                skills={},
+                manifest_skills=["adcp-brand"],
+                include_manifest=True,
+            )
+            sync_skills_from_bundle(bundle_root, skills_dir)
+
+            # dst must not be touched when src is absent
+            assert (skills_dir / "adcp-brand" / "SKILL.md").read_text() == (
+                "# Existing Brand"
+            )
+        assert "missing in bundle" in capsys.readouterr().out
+
+    def test_multiple_skills_synced(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_str:
+            tmp = Path(tmp_str)
+            bundle_root = _make_bundle(
+                tmp,
+                skills={
+                    "adcp-brand": {"SKILL.md": "# Brand"},
+                    "adcp-creative": {"SKILL.md": "# Creative"},
+                    "call-adcp-agent": {"SKILL.md": "# Agent"},
+                },
+            )
+            skills_dir = tmp / "skills"
+
+            count = sync_skills_from_bundle(bundle_root, skills_dir)
+
+            assert count == 3
+            assert (skills_dir / "adcp-brand" / "SKILL.md").exists()
+            assert (skills_dir / "adcp-creative" / "SKILL.md").exists()
+            assert (skills_dir / "call-adcp-agent" / "SKILL.md").exists()