Skip to content

feat(adcp): support ADCP_BASE_URL env override in sync_schemas.py #285

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
bokelley wants to merge 3 commits into
base: main
Choose a base branch
from
+90 −38
Open

feat(adcp): support ADCP_BASE_URL env override in sync_schemas.py #285

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
4d8d92a
feat(adcp): support ADCP_BASE_URL env override in sync_schemas.py
claude Apr 26, 2026
4495f30
fix(test): guard test_default_value against ADCP_BASE_URL in env; use…
claude Apr 26, 2026
7d3fc9e
feat(adcp): reject ADCP_BASE_URL values ending in /protocol
bokelley Apr 29, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 23 additions & 14 deletions scripts/sync_schemas.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,12 @@
The target version comes from `src/adcp/ADCP_VERSION`. If that version's
bundle is not published, sync falls back to `latest.tgz` (the dev snapshot).

Environment variables:
ADCP_BASE_URL Override the protocol host (default: https://adcontextprotocol.org).
Set to point at a fixture CDN for cross-SDK CI or pre-release testing.
Trailing slashes are stripped automatically. Do NOT include "/protocol".
ADCP_SKIP_SIGNATURE Set to "1" to skip Sigstore verification and trust the SHA-256 only.

Usage:
python scripts/sync_schemas.py # sync schemas + skills
python scripts/sync_schemas.py --no-skills # schemas only (e.g. drift checks)
Expand All @@ -41,7 +47,17 @@
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"
_ADCP_BASE = os.environ.get("ADCP_BASE_URL", "https://adcontextprotocol.org").rstrip("/")
# Reject overrides ending in /protocol — appending our own /protocol below
# would silently produce //protocol and 404 against any sensible CDN. Fail
# loud at module import so the typo surfaces immediately.
if _ADCP_BASE.endswith("/protocol"):
raise ValueError(
f"ADCP_BASE_URL={_ADCP_BASE!r} ends with '/protocol'. The script "
"appends '/protocol' itself — pass only the protocol host "
"(e.g. https://adcontextprotocol.org)."
)
BUNDLE_BASE_URL = _ADCP_BASE + "/protocol"
USER_AGENT = "adcp-python-sdk/3.0"

# Sigstore keyless verification identity. Must match the upstream release
Expand Down Expand Up @@ -118,9 +134,7 @@ def fetch_signature_sidecars(version: str) -> tuple[bytes | None, bytes | None]:
return sig, crt


def verify_cosign_signature(
tgz_bytes: bytes, sig_bytes: bytes, crt_bytes: bytes
) -> None:
def verify_cosign_signature(tgz_bytes: bytes, sig_bytes: bytes, crt_bytes: bytes) -> None:
"""Verify the bundle with `cosign verify-blob`.

Raises RuntimeError if cosign is not installed or verification fails.
Expand Down Expand Up @@ -193,9 +207,7 @@ def replace_cache_from_bundle(bundle_root: Path) -> int:
"""
schemas_src = bundle_root / "schemas"
if not schemas_src.is_dir():
raise RuntimeError(
f"Bundle missing expected directory: {bundle_root.name}/schemas/"
)
raise RuntimeError(f"Bundle missing expected directory: {bundle_root.name}/schemas/")

if CACHE_DIR.exists():
shutil.rmtree(CACHE_DIR)
Expand Down Expand Up @@ -274,6 +286,8 @@ def main() -> None:
args = parser.parse_args()

target_version = get_target_adcp_version()
if "ADCP_BASE_URL" in os.environ:
print(f" ! ADCP_BASE_URL override active: {_ADCP_BASE}")
print(f"Syncing AdCP protocol bundle from {BUNDLE_BASE_URL}...")
print(f"Target version: {target_version}")
print(f"Schema cache: {CACHE_DIR}")
Expand All @@ -283,9 +297,7 @@ def main() -> None:

try:
print(f"Fetching {target_version}.tgz + checksum...")
tgz_bytes, expected_sha, effective_version = fetch_bundle_with_fallback(
target_version
)
tgz_bytes, expected_sha, effective_version = fetch_bundle_with_fallback(target_version)
except (HTTPError, URLError) as exc:
print(f"\n✗ Failed to download bundle: {exc}", file=sys.stderr)
sys.exit(1)
Expand All @@ -311,10 +323,7 @@ def main() -> None:
sys.exit(1)

if sig_bytes is None or crt_bytes is None:
print(
f" ! No Sigstore sidecars for adcp-{effective_version} "
"(checksum-only trust)"
)
print(f" ! No Sigstore sidecars for adcp-{effective_version} " "(checksum-only trust)")
else:
try:
verify_cosign_signature(tgz_bytes, sig_bytes, crt_bytes)
Expand Down
91 changes: 67 additions & 24 deletions tests/test_sync_schemas.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -124,9 +124,7 @@ def test_no_manifest_returns_zero(self, capsys: pytest.CaptureFixture[str]) -> N
assert result == 0
assert "No manifest.json" in capsys.readouterr().out

def test_empty_skills_list_returns_zero(
self, capsys: pytest.CaptureFixture[str]
) -> None:
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=[])
Expand Down Expand Up @@ -179,15 +177,11 @@ def test_previous_snapshot_created_on_update(self) -> None:
existing.mkdir()
(existing / "SKILL.md").write_text("# Old Brand")

bundle_root = _make_bundle(
tmp, skills={"adcp-brand": {"SKILL.md": "# New 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"
)
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:
Expand All @@ -204,14 +198,10 @@ def test_previous_snapshot_replaced_on_second_update(self) -> None:
existing.mkdir()
(existing / "SKILL.md").write_text("# Old Brand")

bundle_root = _make_bundle(
tmp, skills={"adcp-brand": {"SKILL.md": "# New 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"
)
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:
Expand Down Expand Up @@ -263,9 +253,7 @@ def test_path_traversal_slash_in_name_rejected(self) -> None:
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:
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(
Expand All @@ -280,9 +268,7 @@ def test_non_string_name_skipped(
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:
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
Expand Down Expand Up @@ -322,9 +308,7 @@ def test_missing_bundle_skill_preserves_existing_dst(
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 (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:
Expand All @@ -346,3 +330,62 @@ def test_multiple_skills_synced(self) -> None:
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()


# ---------------------------------------------------------------------------
# BUNDLE_BASE_URL env override (ADCP_BASE_URL)
# ---------------------------------------------------------------------------


class TestBundleBaseUrl:
def test_default_value(self, monkeypatch: pytest.MonkeyPatch) -> None:
# Fresh load with env var absent — guards against shell having ADCP_BASE_URL set.
monkeypatch.delenv("ADCP_BASE_URL", raising=False)
fresh_spec = importlib.util.spec_from_file_location("sync_schemas_default", _SCRIPT)
assert fresh_spec is not None and fresh_spec.loader is not None
fresh_mod = importlib.util.module_from_spec(fresh_spec)
fresh_spec.loader.exec_module(fresh_mod) # type: ignore[union-attr]
assert fresh_mod.BUNDLE_BASE_URL == "https://adcontextprotocol.org/protocol"

def test_env_override(self, monkeypatch: pytest.MonkeyPatch) -> None:
# Fresh module load with ADCP_BASE_URL set to verify override is applied.
monkeypatch.setenv("ADCP_BASE_URL", "https://fixture.example.com")
fresh_spec = importlib.util.spec_from_file_location("sync_schemas_fresh", _SCRIPT)
assert fresh_spec is not None and fresh_spec.loader is not None
fresh_mod = importlib.util.module_from_spec(fresh_spec)
fresh_spec.loader.exec_module(fresh_mod) # type: ignore[union-attr]
assert fresh_mod.BUNDLE_BASE_URL == "https://fixture.example.com/protocol"

def test_env_override_strips_trailing_slash(self, monkeypatch: pytest.MonkeyPatch) -> None:
# Trailing slash on ADCP_BASE_URL must not produce "//protocol".
monkeypatch.setenv("ADCP_BASE_URL", "https://fixture.example.com/")
fresh_spec = importlib.util.spec_from_file_location("sync_schemas_fresh2", _SCRIPT)
assert fresh_spec is not None and fresh_spec.loader is not None
fresh_mod = importlib.util.module_from_spec(fresh_spec)
fresh_spec.loader.exec_module(fresh_mod) # type: ignore[union-attr]
assert fresh_mod.BUNDLE_BASE_URL == "https://fixture.example.com/protocol"
assert "//protocol" not in fresh_mod.BUNDLE_BASE_URL

def test_env_override_rejects_protocol_suffix(self, monkeypatch: pytest.MonkeyPatch) -> None:
# Override ending in /protocol would double-append. Fail loud at
# import rather than silently 404-ing later.
monkeypatch.setenv("ADCP_BASE_URL", "https://fixture.example.com/protocol")
fresh_spec = importlib.util.spec_from_file_location("sync_schemas_protocol_suffix", _SCRIPT)
assert fresh_spec is not None and fresh_spec.loader is not None
fresh_mod = importlib.util.module_from_spec(fresh_spec)
with pytest.raises(ValueError, match="ends with '/protocol'"):
fresh_spec.loader.exec_module(fresh_mod) # type: ignore[union-attr]

def test_env_override_rejects_protocol_suffix_with_trailing_slash(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
# Same guard, but with a trailing slash on the override — rstrip
# runs first, so the /protocol still trips the check.
monkeypatch.setenv("ADCP_BASE_URL", "https://fixture.example.com/protocol/")
fresh_spec = importlib.util.spec_from_file_location(
"sync_schemas_protocol_trailing", _SCRIPT
)
assert fresh_spec is not None and fresh_spec.loader is not None
fresh_mod = importlib.util.module_from_spec(fresh_spec)
with pytest.raises(ValueError, match="ends with '/protocol'"):
fresh_spec.loader.exec_module(fresh_mod) # type: ignore[union-attr]
Loading