sync_a2a_schema.py
1 """ 2 This script synchronizes the local a2a.json schema with the version corresponding 3 to the installed a2a-sdk package. It fetches the schema from the official A2A 4 GitHub repository using a version-specific tag. 5 """ 6 7 import importlib.metadata 8 import importlib.util 9 import re 10 import sys 11 from pathlib import Path 12 from typing import Optional 13 14 import httpx 15 16 17 # Assuming this script is run from the project root 18 PROJECT_ROOT = Path(__file__).parent.parent 19 SCHEMA_DIR = PROJECT_ROOT / "src" / "solace_agent_mesh" / "common" / "a2a_spec" 20 SCHEMA_PATH = SCHEMA_DIR / "a2a.json" 21 22 23 def get_sdk_version() -> str: 24 """Gets the installed version of a2a-sdk.""" 25 try: 26 version = importlib.metadata.version("a2a-sdk") 27 print(f"Found a2a-sdk version: {version}") 28 return version 29 except importlib.metadata.PackageNotFoundError: 30 print("Error: 'a2a-sdk' package not found.", file=sys.stderr) 31 print("Please ensure the project dependencies are installed.", file=sys.stderr) 32 sys.exit(1) 33 34 35 def construct_git_tag(version: str) -> str: 36 """Constructs a Git tag from a version string (e.g., '0.5.1' -> 'v0.5.1').""" 37 return f"v{version}" 38 39 40 def find_sdk_types_file() -> Path: 41 """Finds the path to the installed a2a/types.py file.""" 42 try: 43 spec = importlib.util.find_spec("a2a.types") 44 if spec and spec.origin: 45 print(f"Found a2a.types at: {spec.origin}") 46 return Path(spec.origin) 47 except Exception as e: 48 print(f"Error finding 'a2a.types' module: {e}", file=sys.stderr) 49 sys.exit(1) 50 51 print("Error: Could not find the installed 'a2a.types' module.", file=sys.stderr) 52 sys.exit(1) 53 54 55 def parse_url_from_header(types_file_path: Path) -> str: 56 """Parses the source URL from the header of the types.py file.""" 57 try: 58 with open(types_file_path, "r", encoding="utf-8") as f: 59 # Read the first few lines to find the filename URL 60 for _ in range(5): 61 line = f.readline() 62 match = re.search(r"#\s*filename:\s*(https?://\S+)", line) 63 if match: 64 url = match.group(1) 65 print(f"Found source URL in header: {url}") 66 return url 67 except Exception as e: 68 print(f"Error reading or parsing {types_file_path}: {e}", file=sys.stderr) 69 sys.exit(1) 70 71 print( 72 f"Error: Could not find the source URL in the header of {types_file_path}.", 73 file=sys.stderr, 74 ) 75 sys.exit(1) 76 77 78 def modify_url_with_tag(url: str, tag: str) -> str: 79 """Replaces the branch/commit part of the URL with a specific Git tag.""" 80 # This regex is designed to find a commit hash or a branch ref like 'refs/heads/main' 81 modified_url, count = re.subn(r"/(?:[a-f0-9]{40}|refs/heads/\w+)/", f"/{tag}/", url) 82 if count == 0: 83 print( 84 f"Warning: Could not substitute tag '{tag}' into URL '{url}'. The URL format may have changed.", 85 file=sys.stderr, 86 ) 87 # Fallback for a simpler structure if the main regex fails 88 modified_url, count = re.subn(r"/main/", f"/{tag}/", url) 89 if count == 0: 90 print( 91 "Error: Fallback URL modification also failed. Cannot proceed.", 92 file=sys.stderr, 93 ) 94 sys.exit(1) 95 96 print(f"Modified URL for version tag '{tag}': {modified_url}") 97 return modified_url 98 99 100 def parse_protocol_version_from_file(types_file_path: Path) -> Optional[str]: 101 """ 102 Parses the default protocol_version from the AgentCard definition in the types.py file. 103 """ 104 try: 105 with open(types_file_path, "r", encoding="utf-8") as f: 106 content = f.read() 107 # Use a regex that is less sensitive to whitespace and type hint changes 108 match = re.search( 109 r"class AgentCard\(.*?protocol_version:.*?=\s*'([^']+)'", 110 content, 111 re.DOTALL, # Allow . to match newlines 112 ) 113 if match: 114 version = match.group(1) 115 print(f"Found protocol_version in AgentCard: {version}") 116 return version 117 print( 118 "Warning: Could not find 'protocol_version' default in AgentCard definition.", 119 file=sys.stderr, 120 ) 121 return None 122 except Exception as e: 123 print( 124 f"Error parsing protocol_version from {types_file_path}: {e}", 125 file=sys.stderr, 126 ) 127 return None 128 129 130 def download_schema_with_fallback(base_url: str, version: str, save_path: Path): 131 """ 132 Attempts to download the schema for the given version, falling back to 133 earlier patch versions if a tag is not found. 134 """ 135 version_parts = version.split(".") 136 if len(version_parts) < 3: 137 print( 138 f"Error: Could not parse version string '{version}'. Expected at least 'X.Y.Z'.", 139 file=sys.stderr, 140 ) 141 sys.exit(1) 142 143 try: 144 major, minor, patch = map(int, version_parts[:3]) 145 except ValueError: 146 print( 147 f"Error: Could not parse major.minor.patch from version string '{version}'.", 148 file=sys.stderr, 149 ) 150 sys.exit(1) 151 152 for p in range(patch, -1, -1): 153 current_version = f"{major}.{minor}.{p}" 154 current_tag = construct_git_tag(current_version) 155 print(f"Attempting to find schema for tag: {current_tag}") 156 157 versioned_url = modify_url_with_tag(base_url, current_tag) 158 159 try: 160 with httpx.Client() as client: 161 response = client.get(versioned_url, follow_redirects=True) 162 if response.status_code == 200: 163 print(f"Success: Found schema at {versioned_url}") 164 save_path.parent.mkdir(parents=True, exist_ok=True) 165 with open(save_path, "w", encoding="utf-8") as f: 166 f.write(response.text) 167 print(f"Successfully saved schema to: {save_path}") 168 return # Success, exit the function 169 elif response.status_code == 404: 170 if p > 0: 171 print( 172 f"Info: Schema not found for tag {current_tag} (HTTP 404). Trying next patch version..." 173 ) 174 else: 175 print( 176 f"Info: Schema not found for tag {current_tag} (HTTP 404). This was the last attempt." 177 ) 178 continue # Try next patch version 179 else: 180 # For other errors (500, 403, etc.), fail fast. 181 print( 182 f"Error: Received unexpected status code {response.status_code} from {versioned_url}", 183 file=sys.stderr, 184 ) 185 response.raise_for_status() 186 except httpx.RequestError as e: 187 print(f"Error downloading schema: {e}", file=sys.stderr) 188 sys.exit(1) 189 except IOError as e: 190 print(f"Error saving schema file to {save_path}: {e}", file=sys.stderr) 191 sys.exit(1) 192 193 # If loop finishes without returning, no version was found 194 print( 195 f"Error: Could not find a valid schema for version {major}.{minor}.x (tried patches from {patch} down to 0).", 196 file=sys.stderr, 197 ) 198 sys.exit(1) 199 200 201 def main(): 202 """Main script execution.""" 203 print("--- Starting A2A Schema Synchronization ---") 204 types_py_path = find_sdk_types_file() 205 206 # Try to get the precise protocol version from the AgentCard model 207 schema_version = parse_protocol_version_from_file(types_py_path) 208 209 # Fallback to SDK package version if parsing fails 210 if not schema_version: 211 print("Warning: Falling back to a2a-sdk package version.", file=sys.stderr) 212 schema_version = get_sdk_version() 213 214 base_url = parse_url_from_header(types_py_path) 215 download_schema_with_fallback(base_url, schema_version, SCHEMA_PATH) 216 print("--- A2A Schema Synchronization Complete ---") 217 218 219 if __name__ == "__main__": 220 main()