From e54721cd7c1ffd24c50d009c61496782b5f152af Mon Sep 17 00:00:00 2001 From: Samuel Price Date: Sun, 19 Apr 2026 11:20:13 -0400 Subject: [PATCH] feat: add generate_tool_annotations.py script for Claude-powered proto annotation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reads KiCad's IPC API proto definitions and uses the Claude API to generate rich, user-facing descriptions for MCP tool metadata. The output JSON file can be loaded by an MCP server at startup to annotate auto-generated tools. Features: - Fetches proto files from KiCad GitLab or loads from a local checkout - Pure-Python proto parser (no protoc dependency) - Prompt caching on the static proto context block — re-runs are cheap - --resume to skip already-annotated messages after an interrupted run - --dry-run to preview what would be annotated without calling the API - Reports cache hit/write/read token counts after each API call Usage: python scripts/generate_tool_annotations.py --fetch-from-gitlab python scripts/generate_tool_annotations.py --proto-dir /path/to/kicad/api/proto Co-Authored-By: Claude Sonnet 4.6 --- scripts/generate_tool_annotations.py | 666 +++++++++++++++++++++++++++ 1 file changed, 666 insertions(+) create mode 100644 scripts/generate_tool_annotations.py diff --git a/scripts/generate_tool_annotations.py b/scripts/generate_tool_annotations.py new file mode 100644 index 0000000..cf584fd --- /dev/null +++ b/scripts/generate_tool_annotations.py @@ -0,0 +1,666 @@ +#!/usr/bin/env python3 +""" +generate_tool_annotations.py — Annotate KiCad IPC API proto messages with Claude + +Reads KiCad's protobuf API definitions and uses the Claude API to generate rich, +user-facing descriptions suitable for MCP tool metadata. The output JSON file can +be loaded by an MCP server to annotate auto-generated tools with descriptions that +go beyond what's in the proto files (e.g., unit conventions, commit ownership +semantics, blocking/interactive behavior). + +Because the proto content is large and static, it is sent once as a cached prompt +block; only the lightweight annotation-request portion is billed at full rate. +Re-running the script against the same proto revision is therefore very cheap. + +Usage +----- +Annotate from a local KiCad source checkout:: + + python scripts/generate_tool_annotations.py \\ + --proto-dir /path/to/kicad/api/proto \\ + --output data/tool_annotations.json + +Fetch proto files directly from GitLab (no checkout needed):: + + python scripts/generate_tool_annotations.py \\ + --fetch-from-gitlab \\ + --kicad-ref master \\ + --output data/tool_annotations.json + +Resume an interrupted run (skips messages already in the output file):: + + python scripts/generate_tool_annotations.py \\ + --proto-dir ./api/proto \\ + --output data/tool_annotations.json \\ + --resume + +Preview what would be annotated without calling the API:: + + python scripts/generate_tool_annotations.py \\ + --proto-dir ./api/proto \\ + --dry-run + +Environment variables +--------------------- +ANTHROPIC_API_KEY + Required. Your Anthropic API key. + +Dependencies +------------ + anthropic>=0.40.0 + requests>=2.28.0 (only needed with --fetch-from-gitlab) +""" + +from __future__ import annotations + +import argparse +import json +import os +import re +import sys +import textwrap +from dataclasses import dataclass, field +from pathlib import Path +from typing import Optional + +# --------------------------------------------------------------------------- +# Constants +# --------------------------------------------------------------------------- + +GITLAB_RAW_BASE = "https://gitlab.com/kicad/code/kicad/-/raw" + +# Relative paths inside the KiCad repo that contain API proto definitions. +# Extend this list when KiCad adds new proto files. +PROTO_RELATIVE_PATHS: list[str] = [ + "api/proto/board/board_commands.proto", + "api/proto/board/board.proto", + "api/proto/board/board_types.proto", + "api/proto/schematic/schematic_commands.proto", + "api/proto/schematic/schematic_types.proto", + "api/proto/common/commands/base_commands.proto", + "api/proto/common/commands/editor_commands.proto", + "api/proto/common/commands/project_commands.proto", + "api/proto/common/types/base_types.proto", + "api/proto/common/types/enums.proto", +] + +DEFAULT_MODEL = "claude-opus-4-7" +DEFAULT_OUTPUT = "tool_annotations.json" + +# --------------------------------------------------------------------------- +# Data model +# --------------------------------------------------------------------------- + + +@dataclass +class ProtoField: + name: str + type_name: str + number: int + comment: str = "" + repeated: bool = False + optional: bool = False + + def summary(self) -> str: + qualifier = "repeated " if self.repeated else ("optional " if self.optional else "") + parts = [f"{qualifier}{self.type_name} {self.name}"] + if self.comment: + parts.append(f" // {self.comment}") + return "".join(parts) + + +@dataclass +class ProtoMessage: + name: str + comment: str + fields: list[ProtoField] = field(default_factory=list) + source_file: str = "" + is_response: bool = False + + def as_text(self) -> str: + lines = [] + if self.comment: + for line in textwrap.wrap(self.comment, width=80): + lines.append(f"// {line}") + lines.append(f"message {self.name} {{") + for f in self.fields: + lines.append(f" {f.summary()}") + lines.append("}") + return "\n".join(lines) + + +# --------------------------------------------------------------------------- +# Proto file fetching +# --------------------------------------------------------------------------- + + +def fetch_proto_from_gitlab(ref: str) -> dict[str, str]: + """Fetch proto files from KiCad's GitLab. Returns {relative_path: content}.""" + try: + import requests + except ImportError: + sys.exit("requests is required for --fetch-from-gitlab. Install with: pip install requests") + + files: dict[str, str] = {} + session = requests.Session() + + for rel_path in PROTO_RELATIVE_PATHS: + url = f"{GITLAB_RAW_BASE}/{ref}/{rel_path}" + print(f" Fetching {rel_path} ...", flush=True) + resp = session.get(url, timeout=30) + if resp.status_code == 200: + files[rel_path] = resp.text + elif resp.status_code == 404: + print(f" WARNING: {rel_path} not found at ref '{ref}' — skipping", file=sys.stderr) + else: + sys.exit(f"HTTP {resp.status_code} fetching {url}") + + return files + + +def load_proto_from_dir(proto_dir: Path) -> dict[str, str]: + """Load proto files from a local directory. Returns {relative_path: content}.""" + files: dict[str, str] = {} + for proto_file in sorted(proto_dir.rglob("*.proto")): + rel = str(proto_file.relative_to(proto_dir)) + files[rel] = proto_file.read_text(encoding="utf-8") + if not files: + sys.exit(f"No .proto files found under {proto_dir}") + return files + + +# --------------------------------------------------------------------------- +# Proto parser +# --------------------------------------------------------------------------- + +# Matches license/copyright headers so we can suppress them from comment text. +_LICENSE_KEYWORDS = frozenset(["copyright", "gnu general public", "program source code", "free software"]) + +# Matches proto field declarations (handles repeated/optional qualifiers). +_FIELD_RE = re.compile( + r"^(repeated\s+|optional\s+)?([\w.]+)\s+(\w+)\s*=\s*(\d+)\s*;?\s*(?://(.*))?$" +) + + +def _is_license_comment(text: str) -> bool: + lower = text.lower() + return any(kw in lower for kw in _LICENSE_KEYWORDS) + + +def parse_proto_text(text: str, source_name: str = "") -> list[ProtoMessage]: + """ + Extract top-level message definitions from proto3 source text. + + Returns a list of ProtoMessage objects in declaration order. + Comments immediately preceding a message declaration are captured + as its docstring. Field-level comments (both inline and preceding) + are attached to each field. + """ + lines = text.splitlines() + messages: list[ProtoMessage] = [] + + i = 0 + pending_comments: list[str] = [] + in_block = False + block_buf: list[str] = [] + + while i < len(lines): + raw = lines[i] + stripped = raw.strip() + + # ── block comment handling ────────────────────────────────────────── + if "/*" in stripped and "*/" not in stripped: + in_block = True + block_buf = [re.sub(r"^\s*/\*+", "", stripped).strip()] + i += 1 + continue + + if in_block: + if "*/" in stripped: + in_block = False + tail = re.sub(r"\*+/.*$", "", stripped).strip().lstrip("* ").strip() + if tail: + block_buf.append(tail) + comment_text = " ".join(l for l in block_buf if l) + if not _is_license_comment(comment_text): + pending_comments = block_buf[:] + else: + pending_comments = [] + block_buf = [] + else: + block_buf.append(stripped.lstrip("* ").strip()) + i += 1 + continue + + # Inline block comment on one line: /* ... */ + if "/*" in stripped and "*/" in stripped: + m = re.search(r"/\*(.*?)\*/", stripped) + if m: + comment_text = m.group(1).strip() + if not _is_license_comment(comment_text): + pending_comments = [comment_text] + else: + pending_comments = [] + i += 1 + continue + + # ── line comment ──────────────────────────────────────────────────── + if stripped.startswith("//"): + comment_text = stripped.lstrip("/").strip() + pending_comments.append(comment_text) + i += 1 + continue + + # ── message declaration ───────────────────────────────────────────── + msg_match = re.match(r"^message\s+(\w+)\s*\{?\s*$", stripped) + if msg_match: + msg_name = msg_match.group(0).split()[1] + doc = " ".join(l for l in pending_comments if l).strip() + if _is_license_comment(doc): + doc = "" + pending_comments = [] + + # Collect fields inside the message body + proto_fields: list[ProtoField] = [] + brace_depth = 1 if "{" in stripped else 0 + field_comments: list[str] = [] + j = i + 1 + + # If the opening brace is on the next line + if brace_depth == 0 and j < len(lines) and "{" in lines[j]: + brace_depth = 1 + j += 1 + + while j < len(lines) and brace_depth > 0: + fraw = lines[j] + fstripped = fraw.strip() + + if fstripped.startswith("//"): + field_comments.append(fstripped.lstrip("/").strip()) + j += 1 + continue + + opens = fstripped.count("{") + closes = fstripped.count("}") + brace_depth += opens - closes + + if brace_depth <= 0: + j += 1 + break + + if brace_depth == 1: + fm = _FIELD_RE.match(fstripped) + if fm: + qualifier = fm.group(1) or "" + type_name = (fm.group(2) or "").split(".")[-1] + field_name = fm.group(3) or "" + field_num = int(fm.group(4) or 0) + inline = (fm.group(5) or "").strip() + + combined = " ".join(field_comments).strip() + if inline: + combined = (combined + " " + inline).strip() + + proto_fields.append( + ProtoField( + name=field_name, + type_name=type_name, + number=field_num, + comment=combined, + repeated="repeated" in qualifier, + optional="optional" in qualifier, + ) + ) + field_comments = [] + elif fstripped and not fstripped.startswith(("/*", "*", "enum", "oneof", "map")): + field_comments = [] + + j += 1 + + messages.append( + ProtoMessage( + name=msg_name, + comment=doc, + fields=proto_fields, + source_file=source_name, + is_response=msg_name.endswith(("Response", "Result")), + ) + ) + i = j + continue + + # ── anything else resets pending comments ─────────────────────────── + if stripped and not stripped.startswith(("syntax", "package", "import", "option")): + pending_comments = [] + + i += 1 + + return messages + + +def parse_all_protos(files: dict[str, str]) -> dict[str, ProtoMessage]: + """Parse all proto file contents and return a flat dict of message_name -> ProtoMessage.""" + all_messages: dict[str, ProtoMessage] = {} + for source_name, content in files.items(): + for msg in parse_proto_text(content, source_name): + all_messages[msg.name] = msg + return all_messages + + +# --------------------------------------------------------------------------- +# Annotation generation via Claude +# --------------------------------------------------------------------------- + +_SYSTEM_PROMPT = """\ +You are a technical writer generating MCP (Model Context Protocol) tool annotations +for the KiCad IPC API. The KiCad IPC API is a protobuf-based API for scripting and +automating the KiCad EDA suite. + +Your task: given a set of protobuf message definitions, produce a JSON object mapping +each REQUEST message name to a structured annotation. Skip pure response messages +(those whose names end in Response or Result). + +Important KiCad API conventions to include when relevant: +- All coordinates and distances are in **nanometers** (nm). Multiply mm values by 1e6. +- `DocumentSpecifier` identifies which open KiCad document to target (PCB, schematic, etc.). +- `ItemHeader` wraps a DocumentSpecifier plus an optional container KIID and field mask. +- `KIID` is a UUID string identifying a specific design object. +- `BeginCommit`/`EndCommit` must bracket any write operations that should be undoable. +- Messages marked "blocking" cause KiCad to return AS_BUSY until the operation completes. +- Messages marked "interactive" transfer control to the user; KiCad becomes unresponsive + to further API calls until the user confirms or cancels. +- `WARNING:` comments in the proto indicate destructive or irreversible operations. + +Output format — a single JSON object, no markdown fences, no explanation:: + +{ + "MessageName": { + "description": "One or two sentences. What does this command do? Who calls it and why?", + "parameters": { + "field_name": "What this field controls. Include units, allowed values, or defaults." + }, + "returns": "What the paired response message contains. Omit if obvious.", + "warnings": ["Any WARNING or irreversibility notes from the proto, verbatim or paraphrased."], + "blocking": true, + "interactive": false + } +} + +Rules: +- Omit `warnings` if the array would be empty. +- Set `blocking` true only for operations explicitly documented as blocking. +- Set `interactive` true only for operations that hand control to the user. +- Keep `description` under 120 characters when possible. +- Field descriptions should mention units (nanometers for coordinates/distances) where applicable. +- If a field has an obvious name and no comment, a one-word description is fine. +""" + + +def _build_proto_context(messages: dict[str, ProtoMessage]) -> str: + """Render all parsed messages as structured text for the prompt.""" + sections: list[str] = [] + by_file: dict[str, list[ProtoMessage]] = {} + for msg in messages.values(): + by_file.setdefault(msg.source_file, []).append(msg) + + for source in sorted(by_file): + sections.append(f"# {source}") + for msg in by_file[source]: + sections.append(msg.as_text()) + sections.append("") + + return "\n".join(sections) + + +def call_claude( + messages: dict[str, ProtoMessage], + model: str, + existing: dict, + resume: bool, +) -> dict: + """ + Call the Claude API to generate annotations for all request messages. + + Uses prompt caching on the proto context block so repeated runs against + the same proto definitions only bill the small annotation-request portion. + """ + try: + import anthropic + except ImportError: + sys.exit("anthropic SDK is required. Install with: pip install anthropic") + + api_key = os.environ.get("ANTHROPIC_API_KEY") + if not api_key: + sys.exit("ANTHROPIC_API_KEY environment variable is not set") + + client = anthropic.Anthropic(api_key=api_key) + + # Decide which messages need annotation + # Command messages live in *_commands.proto files; type/settings messages in *_types.proto + # and base_types.proto are data structures, not callable commands. + request_messages = { + name: msg + for name, msg in messages.items() + if "_commands" in msg.source_file and not msg.is_response + } + if resume: + already_done = set(existing.get("annotations", {}).keys()) + todo = {n: m for n, m in request_messages.items() if n not in already_done} + print(f" Resuming: {len(already_done)} already annotated, {len(todo)} remaining") + else: + todo = request_messages + + if not todo: + print(" Nothing to annotate.") + return existing + + proto_context = _build_proto_context(messages) + target_names = sorted(todo.keys()) + + print(f" Sending {len(target_names)} messages to {model} for annotation ...") + + response = client.messages.create( + model=model, + max_tokens=8192, + system=_SYSTEM_PROMPT, + messages=[ + { + "role": "user", + "content": [ + # Cache the large, static proto context + { + "type": "text", + "text": ( + "## KiCad IPC API — proto definitions\n\n" + + proto_context + ), + "cache_control": {"type": "ephemeral"}, + }, + # Small, non-cached request specifying what to annotate + { + "type": "text", + "text": ( + "## Annotation request\n\n" + "Generate MCP annotations for the following request messages:\n" + + "\n".join(f"- {n}" for n in target_names) + + "\n\nReturn only the JSON object described in your instructions." + ), + }, + ], + } + ], + ) + + # Report cache stats when available + usage = response.usage + if hasattr(usage, "cache_creation_input_tokens"): + print( + f" Tokens — input: {usage.input_tokens}, " + f"cache_write: {usage.cache_creation_input_tokens}, " + f"cache_read: {usage.cache_read_input_tokens}, " + f"output: {usage.output_tokens}" + ) + + raw = response.content[0].text.strip() + + # Strip markdown fences if the model wrapped the output anyway + if raw.startswith("```"): + raw = re.sub(r"^```[a-z]*\n?", "", raw).rstrip("`").strip() + + try: + new_annotations: dict = json.loads(raw) + except json.JSONDecodeError as exc: + print(f"ERROR: Claude returned invalid JSON: {exc}", file=sys.stderr) + print("--- raw response ---", file=sys.stderr) + print(raw[:2000], file=sys.stderr) + sys.exit(1) + + # Merge into existing output + result = dict(existing) + result.setdefault("annotations", {}).update(new_annotations) + return result + + +# --------------------------------------------------------------------------- +# Output helpers +# --------------------------------------------------------------------------- + + +def load_existing(output_path: Path) -> dict: + """Load an existing annotations file, returning an empty structure if absent.""" + if output_path.exists(): + try: + return json.loads(output_path.read_text(encoding="utf-8")) + except (json.JSONDecodeError, OSError): + pass + return {"annotations": {}} + + +def write_output(data: dict, output_path: Path, kicad_ref: str) -> None: + data["_meta"] = { + "kicad_ref": kicad_ref, + "generator": "generate_tool_annotations.py", + } + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8") + print(f" Written: {output_path} ({len(data.get('annotations', {}))} annotations)") + + +def dry_run(messages: dict[str, ProtoMessage]) -> None: + print(f"\nDry run — {len(messages)} command messages found:\n") + for name in sorted(messages): + msg = messages[name] + comment = (msg.comment[:72] + "…") if len(msg.comment) > 75 else msg.comment + source = f" [{msg.source_file}]" + print(f" {name:<40} {comment or '(no comment)'}") + print(f" {'':40} {source}") + if msg.fields: + for f in msg.fields[:3]: + print(f" {f.name}: {f.type_name}") + if len(msg.fields) > 3: + print(f" … and {len(msg.fields) - 3} more field(s)") + print() + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + + +def build_parser() -> argparse.ArgumentParser: + p = argparse.ArgumentParser( + prog="generate_tool_annotations", + description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + + source = p.add_mutually_exclusive_group(required=True) + source.add_argument( + "--proto-dir", + metavar="PATH", + type=Path, + help="Local directory containing KiCad proto files (e.g. /path/to/kicad/api/proto).", + ) + source.add_argument( + "--fetch-from-gitlab", + action="store_true", + help="Download proto files directly from KiCad's GitLab repository.", + ) + + p.add_argument( + "--kicad-ref", + metavar="REF", + default="master", + help="Git ref (branch, tag, or commit) to fetch from GitLab. Default: master.", + ) + p.add_argument( + "--output", + metavar="FILE", + type=Path, + default=Path(DEFAULT_OUTPUT), + help=f"Output JSON file. Default: {DEFAULT_OUTPUT}.", + ) + p.add_argument( + "--model", + metavar="MODEL", + default=DEFAULT_MODEL, + help=f"Claude model to use for annotation. Default: {DEFAULT_MODEL}.", + ) + p.add_argument( + "--resume", + action="store_true", + help="Skip messages that already have annotations in the output file.", + ) + p.add_argument( + "--dry-run", + action="store_true", + help="Parse proto files and list what would be annotated; do not call the API.", + ) + + return p + + +def main(argv: Optional[list[str]] = None) -> int: + args = build_parser().parse_args(argv) + + # ── load proto files ───────────────────────────────────────────────────── + if args.fetch_from_gitlab: + print(f"Fetching proto files from GitLab (ref={args.kicad_ref}) ...") + proto_files = fetch_proto_from_gitlab(args.kicad_ref) + kicad_ref = args.kicad_ref + else: + proto_dir = args.proto_dir.expanduser().resolve() + if not proto_dir.is_dir(): + sys.exit(f"--proto-dir does not exist: {proto_dir}") + print(f"Loading proto files from {proto_dir} ...") + proto_files = load_proto_from_dir(proto_dir) + kicad_ref = "local" + + print(f" Loaded {len(proto_files)} proto file(s)") + + # ── parse ──────────────────────────────────────────────────────────────── + messages = parse_all_protos(proto_files) + request_count = sum(1 for m in messages.values() if not m.is_response) + print(f" Parsed {len(messages)} messages ({request_count} request, " + f"{len(messages) - request_count} response/type)") + + if args.dry_run: + dry_run_cmd = { + name: msg + for name, msg in messages.items() + if "_commands" in msg.source_file and not msg.is_response + } + dry_run(dry_run_cmd) + return 0 + + # ── annotate ───────────────────────────────────────────────────────────── + existing = load_existing(args.output) if args.resume else {"annotations": {}} + result = call_claude(messages, args.model, existing, resume=args.resume) + + # ── write ──────────────────────────────────────────────────────────────── + write_output(result, args.output, kicad_ref) + return 0 + + +if __name__ == "__main__": + sys.exit(main())