FilesExpand file tree

 main

/

gen_all.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

142 lines (115 loc) · 4.56 KB

 main

/

gen_all.py

Copy path

Top

File metadata and controls

• Code
• Blame

142 lines (115 loc) · 4.56 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

70

71

72

73

74

75

76

77

78

79

80

81

82

83

84

85

86

87

88

89

90

91

92

93

94

95

96

97

98

99

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

#!/usr/bin/env python3

from __future__ import annotations

import argparse

import json

import os

import re

import sys

import urllib.error

import urllib.request

from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]

if str(ROOT) not in sys.path:

sys.path.append(str(ROOT))

from scripts import gen_meta, gen_schema, gen_signature # noqa: E402 pylint: disable=wrong-import-position

SCHEMA_DIR = ROOT / "schema"

SCHEMA_JSON = SCHEMA_DIR / "schema.json"

META_JSON = SCHEMA_DIR / "meta.json"

VERSION_FILE = SCHEMA_DIR / "VERSION"

DEFAULT_REPO = "agentclientprotocol/agent-client-protocol"

def parse_args() -> argparse.Namespace:

parser = argparse.ArgumentParser(description="Regenerate schema.py and meta.py from the ACP schema.")

parser.add_argument(

"--version",

"-v",

help=(

"Git ref (tag/branch) of agentclientprotocol/agent-client-protocol to fetch the schema from. "

"If omitted, uses the cached schema files or falls back to 'main' when missing."

),

)

parser.add_argument(

"--repo",

default=os.environ.get("ACP_SCHEMA_REPO", DEFAULT_REPO),

help="Source repository providing schema.json/meta.json (default: %(default)s)",

)

parser.add_argument(

"--no-download",

action="store_true",

help="Skip downloading schema files even when a version is provided.",

)

parser.set_defaults(format_output=True)

parser.add_argument(

"--force",

action="store_true",

help="Force schema download even if the requested ref is already cached locally.",

)

return parser.parse_args()

def main() -> None:

args = parse_args()

version = args.version or os.environ.get("ACP_SCHEMA_VERSION")

repo = args.repo

should_download = _should_download(args, version)

if should_download:

ref = resolve_ref(version)

download_schema(repo, ref)

else:

ref = resolve_ref(version) if version else _cached_ref()

if not (SCHEMA_JSON.exists() and META_JSON.exists()):

print("schema/schema.json or schema/meta.json missing; run with --version to fetch them.", file=sys.stderr)

sys.exit(1)

gen_schema.generate_schema()

gen_meta.generate_meta()

gen_signature.gen_signature(ROOT / "src" / "acp")

if ref:

print(f"Generated schema using ref: {ref}")

else:

print("Generated schema using local schema files")

def _should_download(args: argparse.Namespace, version: str | None) -> bool:

env_override = os.environ.get("ACP_SCHEMA_DOWNLOAD")

if env_override is not None:

return env_override.lower() in {"1", "true", "yes"}

if args.no_download:

return False

if version:

if not SCHEMA_JSON.exists() or not META_JSON.exists():

return True

cached = _cached_ref()

if args.force:

return True

return cached != resolve_ref(version)

return not (SCHEMA_JSON.exists() and META_JSON.exists())

def resolve_ref(version: str | None) -> str:

if not version:

return "refs/heads/main"

if version.startswith("refs/"):

return version

if re.fullmatch(r"v?\d+\.\d+\.\d+", version):

value = version if version.startswith("v") else f"v{version}"

return f"refs/tags/{value}"

return f"refs/heads/{version}"

def download_schema(repo: str, ref: str) -> None:

SCHEMA_DIR.mkdir(parents=True, exist_ok=True)

schema_url = f"https://raw.githubusercontent.com/{repo}/{ref}/schema/schema.unstable.json"

meta_url = f"https://raw.githubusercontent.com/{repo}/{ref}/schema/meta.unstable.json"

try:

schema_data = fetch_json(schema_url)

meta_data = fetch_json(meta_url)

except RuntimeError as exc: # pragma: no cover - network error path

print(exc, file=sys.stderr)

sys.exit(1)

SCHEMA_JSON.write_text(json.dumps(schema_data, indent=2), encoding="utf-8")

META_JSON.write_text(json.dumps(meta_data, indent=2), encoding="utf-8")

VERSION_FILE.write_text(ref + "\n", encoding="utf-8")

print(f"Fetched schema and meta from {repo}@{ref}")

def fetch_json(url: str) -> dict:

try:

with urllib.request.urlopen(url) as response: # noqa: S310 - trusted source configured by repo

return json.loads(response.read().decode("utf-8"))

except urllib.error.URLError as exc:

raise RuntimeError(f"Failed to fetch {url}: {exc}") from exc

def _cached_ref() -> str | None:

if VERSION_FILE.exists():

return VERSION_FILE.read_text(encoding="utf-8").strip() or None

return None

if __name__ == "__main__":

main()