#!/usr/bin/env python3 # SPDX-License-Identifier: MIT # Copyright (c) 2026 Mutual Solutions, Inc. # # Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated # documentation files (the "Software"), to deal in the Software without restriction, including without limitation the # rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit # persons to whom the Software is furnished to do so, subject to the following conditions: # # The above copyright notice and this permission notice shall be included in all copies or substantial portions of the # Software. # # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE # WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR # COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR # OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. # (Full text: https://zap.mutual.solutions/downloads/LICENSE.txt. Using the Zap! service itself is governed by # https://zap.mutual.solutions/terms.) """zap_mcp.py: MCP server (stdio) for Zap!, the hosted KiCad auto-router at https://zap.mutual.solutions. Single file, Python 3.9 or newer, standard library only. Tools: check_board, route_board (check, submit, wait, download, report), get_report, get_job, list_jobs, delete_job, send_feedback (the user's opinion to the Zap! team, only after the user approved the exact text), board_snapshot (a picture of one small area of a routed board, to show the user and, with their yes, attach to the feedback). API key: environment variable ZAP_API_KEY, or the file named by ZAP_API_KEY_FILE, or ~/.config/zap/key. It is never taken from tool arguments and never written to output or logs. Claude Code: claude mcp add zap -e ZAP_API_KEY_FILE=$HOME/zap-key.txt -- python3 /path/to/zap_mcp.py Claude Desktop: {"mcpServers": {"zap": {"command": "python3", "args": ["/path/to/zap_mcp.py"], "env": {"ZAP_API_KEY_FILE": "/path/to/zap-key.txt"}}}} Guide for agents: https://zap.mutual.solutions/agents.md ZAP_URL (default https://zap.mutual.solutions) must be https; redirects are never followed, so the key only goes there. route_board saves the routed board, the full report (BOARD.routed.zap-report.json) and copies of the rules, then deletes the server copy (delete_after_download=false keeps it for 7 days), like zap.py run.""" import base64, gzip, json, os, stat, sys, time, urllib.error, urllib.request, uuid VERSION = "1.10" URL = os.environ.get("ZAP_URL", "https://zap.mutual.solutions").rstrip("/") class _NoRedirect(urllib.request.HTTPRedirectHandler): """A redirect becomes an error: the key is never sent on to another address.""" def redirect_request(self, req, fp, code, msg, headers, newurl): return None OPENER = urllib.request.build_opener(_NoRedirect) KEY_WARNING = [] # set once when the key file is readable by other users; shown with every tool result PROTOCOLS = ("2025-06-18", "2025-03-26", "2024-11-05") REVIEW = "Review the routed board and run KiCad's DRC (Inspect > Design Rules Checker) before ordering boards." FEEDBACK_RULE = ("Whenever the user gives an opinion about Zap! or a routing result (something they liked, a problem, a " "wish), ask once whether they would like to send it to the Zap! team. When you ask, say in one short " "sentence what happens, without overstating or understating it, for example: \"Send this to the Zap! " "team? Only this text (and the picture, if any) is used to improve Zap!, never your board file, and you " "can ask them to delete it.\" Show the exact text you would " "send (their words, or a summary they approve) and the job id. Call send_feedback only if they say " "yes; if they say no, do not ask again in this session. Never include board files or private details " "they did not approve." + " When the feedback is about a spot on the board, offer to attach a picture of just that area (board_snapshot), aimed with the report's issue positions; show the picture to the user before sending and attach it only if they say yes; if they do not want it, delete it (delete_snapshot). Never the whole board.") class ZapError(Exception): def __init__(self, status, body): self.status, self.body = status, body if isinstance(body, dict) else {} where = "; ".join("%s at x %s, y %s mm" % (f.get("ref"), (f.get("at_mm") or [None, None])[0], (f.get("at_mm") or [None, None])[1]) for f in (self.body.get("footprints") or [])[:50]) super().__init__("%s (%s)%s" % (self.body.get("message") or "HTTP %s" % status, self.body.get("error", status), ("\nFootprints: " + where) if where else "")) def load_key(): k = os.environ.get("ZAP_API_KEY", "").strip() if k: return k path = os.environ.get("ZAP_API_KEY_FILE") or os.path.expanduser("~/.config/zap/key") try: st = os.stat(path) if os.name == "posix" and st.st_mode & (stat.S_IRWXG | stat.S_IRWXO) and not KEY_WARNING: KEY_WARNING.append("Warning: the Zap! key file %s is readable by other users; run: chmod 600 %s" % (path, path)) print("zap_mcp: " + KEY_WARNING[0], file=sys.stderr, flush=True) with open(path) as f: return f.read().strip().split()[0] except (OSError, IndexError): raise ZapError(0, {"error": "no_key", "message": "No API key: set ZAP_API_KEY_FILE to the key file in the MCP " "server's environment (or ZAP_API_KEY, or save it in ~/.config/zap/key)"}) def call(method, path, body=None, ctype=None, raw=False, auth=True): """One API request with the documented retry rules: 429 -> wait Retry-After; edge 502/503/504 or a network error -> retry GETs with backoff (a POST is never repeated automatically: it could create a second job).""" if not URL.lower().startswith("https://"): raise ZapError(0, {"error": "bad_url", "message": "ZAP_URL must be an https:// address (the key is never sent over plain http)"}) headers = {"User-Agent": "zap-mcp/" + VERSION} if auth: headers["Authorization"] = "Bearer " + load_key() if ctype: headers["Content-Type"] = ctype for attempt in range(6): req = urllib.request.Request(URL + path, data=body, method=method, headers=headers) try: with OPENER.open(req, timeout=400) as r: # a pre-flight check may wait 60 s for a slot and run 300 s data = r.read() return data if raw else json.loads(data or b"{}") except urllib.error.HTTPError as e: try: b = json.loads(e.read()) except Exception: b = {} if 300 <= e.code < 400: raise ZapError(e.code, {"error": "redirect", "message": "%s answered with a redirect (HTTP %d), which is not " "followed so the key stays with Zap!: check ZAP_URL" % (URL, e.code)}) if e.code == 429 and b.get("error") in ("rate_limited",) and attempt < 5: time.sleep(min(60, int(e.headers.get("Retry-After", "5")))) continue if e.code == 503 and b.get("error") == "check_busy" and attempt < 5: time.sleep(min(60, int(e.headers.get("Retry-After", "30")))) continue if e.code in (502, 503, 504) and not b and method == "GET" and attempt < 5: time.sleep(10 + 10 * attempt) continue raise ZapError(e.code, b) except (urllib.error.URLError, OSError) as e: if method == "GET" and attempt < 5: time.sleep(3 + 3 * attempt) continue raise ZapError(0, {"error": "network", "message": "Could not reach %s: %s" % (URL, getattr(e, "reason", e))}) def read_file(path, what): path = os.path.expanduser(path) try: with open(path, "rb") as f: return os.path.basename(path), f.read() except OSError as e: raise ZapError(0, {"error": "local_file", "message": "Cannot read the %s %s: %s" % (what, path, e.strerror)}) def upload(a): """Multipart body from tool arguments; .kicad_pro / .kicad_dru next to the board are used when not given.""" board = a.get("board_path") or "" if not board.endswith(".kicad_pcb"): raise ZapError(0, {"error": "local_file", "message": "board_path must be a .kicad_pcb file"}) base = os.path.expanduser(board)[: -len(".kicad_pcb")] name, data = read_file(board, "board") # the board goes gzip-compressed (the service unpacks it; large boards then fit the 20 MiB upload limit) fields = {"board": (name + ".gz", gzip.compress(data, compresslevel=6, mtime=0))} pro = a.get("project_path") or (base + ".kicad_pro" if os.path.exists(base + ".kicad_pro") else None) dru = a.get("rules_path") or (base + ".kicad_dru" if os.path.exists(base + ".kicad_dru") else None) if pro: fields["project"] = read_file(pro, "project") if dru: fields["rules"] = read_file(dru, "rules") if a.get("currents_path"): fields["currents"] = read_file(a["currents_path"], "currents file") elif a.get("currents"): fields["currents"] = (None, json.dumps(a["currents"]).encode()) if a.get("router"): fields["router"] = (None, str(a["router"]).encode()) if a.get("consent_training") is True: fields["consent_training"] = (None, b"true") if a.get("keep_existing") is True: fields["keep_existing"] = (None, b"true") if a.get("strip_existing") is True: fields["strip_existing"] = (None, b"true") for k in OPTION_FIELDS: # routing options: only the ones the agent set if a.get(k) is not None: fields[k] = (None, str(a[k]).encode()) if a.get("fab_floor"): fields["fab_floor"] = (None, str(a["fab_floor"]).encode()) b = uuid.uuid4().hex parts = [] for name, (fname, data) in fields.items(): head = '--%s\r\nContent-Disposition: form-data; name="%s"' % (b, name) if fname is not None: head += '; filename="%s"\r\nContent-Type: application/octet-stream' % fname.replace('"', "") parts.append(head.encode() + b"\r\n\r\n" + data + b"\r\n") sent = [k for k in fields if k in ("board", "project", "rules", "currents")] return b"".join(parts) + ("--%s--\r\n" % b).encode(), "multipart/form-data; boundary=" + b, sent def summarize(st): """The numbers to show a human, all 'lower is better'.""" if not isinstance(st, dict): return {} m = st.get("metrics") or {} f = m.get("fab") or {} th = f.get("thermal") or {} out = {"job_id": st.get("id"), "status": st.get("status"), "router": (st.get("version") or {}).get("router"), "report_url": st.get("report_url"), "fab_floor": st.get("fab_floor"), "unconnected": m.get("unconnected"), "critical_unconnected": m.get("critical_unconnected"), "drc_errors": m.get("drc_errors"), "server_time_s": m.get("wall_time_s")} if f.get("computed"): out.update(shorts=f.get("shorts"), shorts_already_in_board=f.get("shorts_already_in_board"), jlc_violations=f.get("jlc_violations"), jlc_violations_by_type=f.get("jlc_violations_by_type"), jlc_items_first_10=(f.get("jlc_items") or [])[:10], jlc_items_truncated=f.get("jlc_items_truncated")) if th.get("computed"): out.update(hottest_rise_c="off the chart" if th.get("off_chart") else th.get("worst_rise_c"), rise_limit_c=th.get("limit_c"), sections_over_limit=th.get("sections_over_limit"), nets_over_limit=th.get("nets_over_limit"), currents=th.get("currents")) if m.get("existing_copper"): # keep_existing jobs: counts, not findings; *_removed must be 0 out["existing_copper"] = m["existing_copper"] if m.get("time_budget"): # minutes: how many attempts ran, which one was kept, the time used out["time_budget"] = m["time_budget"].get("summary") or m["time_budget"] if st.get("status") == "failed": out["error"] = st.get("error") return out def tool_check_board(a): body, ctype, sent = upload(a) r = call("POST", "/v1/check", body, ctype) text = r.get("summary", "") for kind in ("blockers", "warnings"): for f in r.get(kind) or []: text += "\n%s %s: %s" % ("PROBLEM" if kind == "blockers" else "warning", f.get("code"), f.get("message")) return r, text + "\n(files sent: %s; the check used no job quota)" % ", ".join(sent) def tool_route_board(a): board = a.get("board_path") or "" if a.get("check", True): r, text = tool_check_board(a) if not r.get("ok"): return {"submitted": False, "check": r}, "Not submitted: the pre-flight check found problems.\n" + text body, ctype, sent = upload(a) job = call("POST", "/v1/jobs", body, ctype) jid = job["id"] if not a.get("wait", True): return {"job_id": jid, "status": "queued"}, "Submitted job %s; poll it with get_job." % jid t0, deadline = time.time(), time.time() + float(a.get("timeout_s", 3900)) while True: st = call("GET", "/v1/jobs/" + jid) if st.get("status") in ("succeeded", "failed", "cancelled"): break if time.time() > deadline: return {"job_id": jid, "status": st.get("status")}, "Job %s is still %s; poll it with get_job.%s" % ( jid, st.get("status"), (" " + st["queue_note"]) if st.get("queue_note") else "") time.sleep(5 if time.time() - t0 < 60 else 10) res = summarize(st) if st["status"] == "succeeded": out = os.path.expanduser(a.get("out_path") or (os.path.expanduser(board)[: -len(".kicad_pcb")] + ".routed.kicad_pcb")) if os.path.abspath(out) == os.path.abspath(os.path.expanduser(board)): raise ZapError(0, {"error": "local_file", "message": "out_path would overwrite the input board"}) data = call("GET", "/v1/jobs/%s/result" % jid, raw=True) with open(out, "wb") as f: f.write(data) res["routed_file"] = out # KiCad opens a board with the project of the same name: give the routed board the input's rules too ibase, obase = os.path.expanduser(board)[: -len(".kicad_pcb")], out[: -len(".kicad_pcb")] if out.endswith(".kicad_pcb") else out for ext in (".kicad_pro", ".kicad_dru"): if os.path.exists(ibase + ext) and not os.path.exists(obase + ext): with open(ibase + ext, "rb") as f, open(obase + ext, "wb") as g: g.write(f.read()) res.setdefault("copied_rules", []).append(obase + ext) # the full report (every finding with its position and what to do) and the copper picture, before the server copy goes rpath = obase + ".zap-report.json" try: rep = call("GET", "/v1/jobs/%s/report.json" % jid) except ZapError: # an older server without reports: keep the job status rep = st with open(rpath, "w") as f: json.dump(rep, f, indent=1) res["report_file"] = rpath for lay in (rep.get("image") or {}).get("layers") or []: # one picture per copper layer p = "%s.%s.svg" % (obase, lay["name"].replace(".", "_")) with open(p, "wb") as f: f.write(call("GET", lay["url"], raw=True)) res.setdefault("copper_pictures", []).append(p) # problems to fix per kind (lower is better); the heat list's length is not a problem count names = {"hot_spots": "copper_sections_over_10c"} res["findings"] = {names.get(s.get("kind"), s.get("kind")): s.get("problems", s.get("total")) for s in rep.get("sections") or []} if (rep.get("summary") or {}).get("heat"): res["heat"] = rep["summary"]["heat"] if pours_line(rep.get("pours")): # the board's pours that Zap! replaced, and the ones it made: show the human res["pours"] = pours_line(rep.get("pours")) if a.get("delete_after_download", True): call("DELETE", "/v1/jobs/" + jid) res["server_files_deleted"] = True text = json.dumps(res, indent=1) + ("\n" + REVIEW if st["status"] == "succeeded" else "") return res, text def tool_get_job(a): st = call("GET", "/v1/jobs/" + a["job_id"]) res = dict(summarize(st), raw=st) if a.get("download_to") and st.get("status") == "succeeded": data = call("GET", "/v1/jobs/%s/result" % a["job_id"], raw=True) with open(os.path.expanduser(a["download_to"]), "wb") as f: f.write(data) res["routed_file"] = os.path.expanduser(a["download_to"]) return res, json.dumps(summarize(st), indent=1) def tool_service_info(a): """No key needed: service and router version, routing options on or off, queue and typical wait.""" v = call("GET", "/v1/version", auth=False) st = call("GET", "/v1/status", auth=False) return {"version": v, "status": st}, "Zap! %s, router %s; %s jobs waiting, typical routing time %s s." % ( v.get("service"), v.get("router"), st.get("jobs_waiting"), st.get("typical_run_s")) def tool_error_catalog(a): r = call("GET", "/v1/errors", auth=False) return r, "%d error codes, each with its HTTP status, whether retrying helps and what to change." % len(r.get("errors") or []) def tool_list_jobs(a): r = call("GET", "/v1/jobs") return r, "\n".join("%s %-9s %s" % (j["id"], j["status"], j["created_at"]) for j in r.get("jobs", [])) or "No jobs yet." def pours_line(p): """'Copper pours: yours replaced: +12V on F.Cu; Zap! made: GND on B.Cu' from the report's pours, or None.""" if not isinstance(p, dict) or not (p.get("replaced") or p.get("made")): return None t = lambda xs: "; ".join("%s on %s" % (x.get("net") or "no net", ", ".join(x.get("layers") or []) or "?") for x in xs) or "none" return "Copper pours: yours replaced: %s. Zap! made: %s." % (t(p.get("replaced") or []), t(p.get("made") or [])) def tool_get_report(a): rep = call("GET", "/v1/jobs/%s/report.json" % a["job_id"]) if rep.get("error"): return rep, "Job %s failed: %s What to do: %s" % (rep["id"], rep["error"].get("message"), rep["error"].get("what_to_do")) lines = ["Report of job %s (positions in mm as KiCad shows them; fewer findings is better):" % rep.get("id")] heat = (rep.get("summary") or {}).get("heat") if heat: # sections over the temperature-rise limit, and the hottest rise (the list below is the hottest sections) lines.append("Heat: " + heat) po = pours_line(rep.get("pours")) if po: lines.append(po) for s in rep.get("sections") or []: if s.get("kind") == "hot_spots": lines.append("Hottest copper sections (most within the limit): %s listed" % s.get("total")) else: lines.append("%s: %s" % (s.get("title"), s.get("total"))) seen = set() for x in (s.get("items") or [])[:5]: p = x.get("at_mm") or [None, None] lines.append(" at x %s, y %s%s%s" % (p[0], p[1], (" on " + x["layer"]) if x.get("layer") else "", (", net " + x["net"]) if x.get("net") else "")) if x.get("what_to_do") not in seen: seen.add(x.get("what_to_do")) lines.append(" what to do: " + str(x.get("what_to_do"))) return rep, "\n".join(lines) + "\nWeb page with the copper picture (needs the key): %s" % rep.get("report_url") FEEDBACK_KINDS = ("praise", "problem", "idea", "other") def tool_send_feedback(a): """POST /v1/feedback. Refused here, before anything is sent, unless the agent states the user approved the text.""" if a.get("user_approved") is not True: raise ZapError(0, {"error": "feedback_needs_user_approval", "message": "Not sent: show your user the exact text, ask whether to send it to the Zap! " "team, and call send_feedback with user_approved true only if they say yes."}) body = {"message": a.get("message"), "user_approved": True} for k in ("job_id", "kind", "contact_ok", "snapshot_ids"): if a.get(k) is not None: body[k] = a[k] r = call("POST", "/v1/feedback", json.dumps(body).encode(), "application/json") return r, "Sent to the Zap! team as feedback %s at %s. Tell the user it was sent." % (r.get("id"), r.get("received_at")) def tool_board_snapshot(a): """POST /v1/jobs/{id}/snapshot, then the PNG: saved locally and returned as an image to show the user.""" body = {k: a[k] for k in ("x_mm", "y_mm", "size_mm", "layers", "show_issues") if a.get(k) is not None} r = call("POST", "/v1/jobs/%s/snapshot" % a["job_id"], json.dumps(body).encode(), "application/json") png = call("GET", r["png_url"], raw=True) path = os.path.abspath(os.path.expanduser(a.get("save_to") or "zap-snapshot-%s.png" % r["snapshot_id"])) with open(path, "wb") as f: f.write(png) r["saved_to"] = path reg = r.get("region") or {} text = ("Picture %s saved to %s: x %s, y %s mm, %s mm square, layers %s; %d nets in the area; marked %s missing " "connections and %s design-rule errors. Show it to the user. Attach it (snapshot_ids) to send_feedback only " "if they say yes; otherwise it is deleted at %s." % ( r["snapshot_id"], path, reg.get("x_mm"), reg.get("y_mm"), reg.get("size_mm"), ", ".join(reg.get("layers") or []), len(r.get("nets_in_region") or []), (r.get("issues_marked") or {}).get("unconnected"), (r.get("issues_marked") or {}).get("drc"), r.get("expires_at"))) return r, text, png def tool_delete_snapshot(a): r = call("DELETE", "/v1/snapshots/%s.png" % a["snapshot_id"]) return r, "Deleted picture %s on the Zap! server." % a["snapshot_id"] def tool_delete_job(a): r = call("DELETE", "/v1/jobs/" + a["job_id"]) return r, ("Cancelled job %s." if r.get("status") == "cancelled" else "Deleted the files of job %s.") % a["job_id"] OPTION_FIELDS = { "signal_width_mm": {"type": "number", "description": "Signal track width in mm (default: the board's net class, else 0.2)."}, "via_diameter_mm": {"type": "number", "description": "Via diameter in mm; send with via_drill_mm (default: net class, else 0.6)."}, "via_drill_mm": {"type": "number", "description": "Via drill in mm; send with via_diameter_mm (default: net class, else 0.3)."}, "vias_in_pads": {"type": "string", "enum": ["forbid", "allow"], "description": "Vias inside SMD pads (default forbid; allow needs the board made with filled and capped vias, an extra fab option)."}, "ground_pour": {"type": "string", "enum": ["on", "off"], "description": "Ground copper pours (default on)."}, "size_rails_from_current": {"type": "string", "enum": ["on", "off"], "description": "Supply track width from current where room allows (default on; off: the net-class width, which may overheat)."}, "effort": {"type": "string", "enum": ["quick", "normal", "thorough"], "description": "Routing effort (default normal). thorough: about twice the routing work and keeps the best intermediate result; slower on large boards, usually fewer unrouted connections. quick: about one eighth of the work, faster, more unrouted connections."}, "minutes": {"type": "integer", "minimum": 1, "maximum": 60, "description": "Let Zap! keep trying for up to this many minutes (1 to 60) and return the best result; default is one round of a few minutes. Only when the human wants the best result and can wait; a job over 10 minutes may queue behind another one."}} FILES = {"board_path": {"type": "string", "description": "Path to the placed .kicad_pcb (outline on Edge.Cuts; no tracks or vias unless keep_existing or strip_existing is true)."}, "project_path": {"type": "string", "description": "Optional .kicad_pro with the design rules; default: the .kicad_pro next to the board if present."}, "rules_path": {"type": "string", "description": "Optional .kicad_dru; default: the .kicad_dru next to the board if present."}, "currents_path": {"type": "string", "description": "Optional JSON file of known currents: {\"nets\": {\"\": {\"amps\": A}}, \"ambient_c\": C}. State supply nets, not ground."}, "router": {"type": "string", "enum": ["stable", "next"], "description": "Router to use; default stable."}, "currents": {"type": "object", "description": "Known currents inline instead of currents_path: {\"nets\": {\"\": {\"amps\": A}}, \"ambient_c\": C}."}, "consent_training": {"type": "boolean", "description": "Records that the human is open to this board being used to improve Zap! (default false; only a preference: nothing is used without a separate written agreement)."}, "keep_existing": {"type": "boolean", "description": "true for a partly routed board: keep its tracks, vias and zones exactly and route only the missing connections (default false)."}, "strip_existing": {"type": "boolean", "description": "true to discard the board's existing tracks and vias and route everything (default false: a board with tracks or vias is refused with input_has_copper). The fix for that answer when the human has no KiCad at hand; not with keep_existing."}, **OPTION_FIELDS, "fab_floor": {"type": "string", "enum": ["jlc", "board"], "description": "jlc (default): never go below JLC's capability; nets that need finer copper are left unrouted and listed with the reason. board: follow the board's own minimums, for fabs finer than JLC; JLC findings are still reported."}} TOOLS = [ {"name": "check_board", "fn": tool_check_board, "description": "Pre-flight check of a KiCad board with Zap! (seconds, no routing, uses no job quota): lists every problem that would stop the job (blockers) and what would weaken it (warnings), with references, positions and the KiCad fix. Run this first.", "inputSchema": {"type": "object", "properties": FILES, "required": ["board_path"]}}, {"name": "route_board", "fn": tool_route_board, "description": "Route a placed KiCad board with Zap!: pre-flight check, submit (uses one of the key's daily jobs), wait (seconds for small boards, up to 60 minutes), download BOARD.routed.kicad_pcb next to the input (with copies of BOARD.kicad_pro/.kicad_dru so KiCad uses the same rules), delete the server copy. Returns the numbers to show the human; all are lower-is-better. The result needs a human review and KiCad DRC before fabrication.", "inputSchema": {"type": "object", "properties": dict(FILES, **{ "out_path": {"type": "string", "description": "Where to save the routed board; default BOARD.routed.kicad_pcb next to the input (never overwrites the input)."}, "check": {"type": "boolean", "description": "Run the pre-flight check first and do not submit on blockers (default true)."}, "wait": {"type": "boolean", "description": "Wait for the result (default true); false returns the job id at once."}, "timeout_s": {"type": "number", "description": "Stop waiting after this many seconds (default 3900); the job keeps running."}, "delete_after_download": {"type": "boolean", "description": "Delete the server copy after saving it (default true)."}}), "required": ["board_path"]}}, {"name": "get_report", "fn": tool_get_report, "description": "The result report of a finished Zap! job: every missing connection, design-rule error, JLC violation, hot spot and leftover copper, each with its position (mm, as KiCad shows them), layer, net and what to do in KiCad. Use it to tell the human exactly what to fix.", "inputSchema": {"type": "object", "properties": {"job_id": {"type": "string"}}, "required": ["job_id"]}}, {"name": "get_job", "fn": tool_get_job, "description": "Status and result numbers of a Zap! job; with download_to, also saves the routed board when it has succeeded.", "inputSchema": {"type": "object", "properties": {"job_id": {"type": "string"}, "download_to": {"type": "string"}}, "required": ["job_id"]}}, {"name": "service_info", "fn": tool_service_info, "description": "Zap! service status without a key: version, router build, whether routing options are on, jobs waiting and the typical wait and routing time.", "inputSchema": {"type": "object", "properties": {}}}, {"name": "error_catalog", "fn": tool_error_catalog, "description": "Every Zap! error code with its HTTP status, whether retrying helps, and the change that fixes it.", "inputSchema": {"type": "object", "properties": {}}}, {"name": "list_jobs", "fn": tool_list_jobs, "description": "This key's last 50 Zap! jobs (id, status, created time).", "inputSchema": {"type": "object", "properties": {}}}, {"name": "send_feedback", "fn": tool_send_feedback, "description": "Send the user's opinion about Zap! or a routing result to the Zap! team. Only call this after the " "user saw the exact text (and the job id) and said yes; if they said no, do not ask again in this " "session. The text is the user's words, or a summary they approved. Never send board files, file " "contents or anything the user did not approve. " + FEEDBACK_RULE, "inputSchema": {"type": "object", "additionalProperties": False, "required": ["message", "user_approved"], "properties": { "message": {"type": "string", "minLength": 1, "maxLength": 4000, "description": "Exactly the text the user approved: their words, or a summary they agreed to (1 to 4,000 characters)."}, "user_approved": {"type": "boolean", "enum": [True], "description": "true only after the user saw this exact text and said yes to sending it. Without it nothing is sent."}, "job_id": {"type": "string", "description": "Optional: the job the opinion is about (one of this key's jobs), shown to the user with the text."}, "kind": {"type": "string", "enum": list(FEEDBACK_KINDS), "description": "Optional: praise, problem, idea or other."}, "contact_ok": {"type": "boolean", "description": "Optional: true if the user agreed that the Zap! team may reply to the account's email address (default false)."}, "snapshot_ids": {"type": "array", "maxItems": 3, "items": {"type": "string"}, "description": "Optional: pictures from board_snapshot (same job) that the user saw and agreed to attach."}}}}, {"name": "board_snapshot", "fn": tool_board_snapshot, "description": "A picture (PNG) of one small square area, at most 40 mm, of a finished job's routed board, with the " "report's missing connections and design-rule errors marked: saved next to you and returned as an " "image. Use it when the user's feedback is about a spot on the board: aim it with the report's at_mm " "positions (get_report). Show the picture to the user before sending; attach it to send_feedback " "(snapshot_ids) only if they say yes. Never the whole board. A picture not attached is deleted after " "24 hours.", "inputSchema": {"type": "object", "additionalProperties": False, "required": ["job_id", "x_mm", "y_mm"], "properties": { "job_id": {"type": "string"}, "x_mm": {"type": "number", "description": "Centre of the area, x in mm as KiCad shows positions (for example a report item's at_mm)."}, "y_mm": {"type": "number", "description": "Centre of the area, y in mm."}, "size_mm": {"type": "number", "minimum": 1, "maximum": 40, "description": "Side of the square, mm (default 20)."}, "layers": {"type": "array", "items": {"type": "string"}, "description": "Copper layers to draw, e.g. [\"F.Cu\"] (default all)."}, "show_issues": {"type": "boolean", "description": "Mark the report's issues in the area (default true)."}, "save_to": {"type": "string", "description": "Where to save the PNG (default zap-snapshot-.png in the current folder)."}}}}, {"name": "delete_snapshot", "fn": tool_delete_snapshot, "description": "Delete a board picture the user does not want to attach, on the Zap! server, at once (otherwise it " "is deleted 24 hours after it was made). Pictures already attached to feedback are not deleted here.", "inputSchema": {"type": "object", "properties": {"snapshot_id": {"type": "string"}}, "required": ["snapshot_id"]}}, {"name": "delete_job", "fn": tool_delete_job, "description": "Cancel a queued or running Zap! job (it stops at once, nothing is kept), or delete a finished job's files on the server (otherwise they are deleted after 7 days).", "inputSchema": {"type": "object", "properties": {"job_id": {"type": "string"}}, "required": ["job_id"]}}, ] def respond(mid, result=None, error=None): msg = {"jsonrpc": "2.0", "id": mid} if error is not None: msg["error"] = error else: msg["result"] = result sys.stdout.write(json.dumps(msg) + "\n") sys.stdout.flush() def handle(m): method, mid, p = m.get("method"), m.get("id"), m.get("params") or {} if method == "initialize": want = p.get("protocolVersion") return respond(mid, {"protocolVersion": want if want in PROTOCOLS else PROTOCOLS[0], "capabilities": {"tools": {}}, "serverInfo": {"name": "zap", "version": VERSION}, "instructions": "Zap! routes placed KiCad boards. Use check_board first, then route_board. " "Report the numbers it returns to the human (lower is better) and end with: " + REVIEW + " " + FEEDBACK_RULE + " " + URL + "/agents.md"}) if method == "ping": return respond(mid, {}) if method == "tools/list": return respond(mid, {"tools": [{k: v for k, v in t.items() if k != "fn"} for t in TOOLS]}) if method == "tools/call": t = next((t for t in TOOLS if t["name"] == p.get("name")), None) if t is None: return respond(mid, error={"code": -32602, "message": "unknown tool %s" % p.get("name")}) try: out = t["fn"](p.get("arguments") or {}) data, text = out[0], out[1] if KEY_WARNING: text = KEY_WARNING[0] + "\n" + text content = [{"type": "text", "text": text}] if len(out) > 2 and out[2]: # a picture: shown to the agent (and by many clients to the user) as an image content.append({"type": "image", "data": base64.b64encode(out[2]).decode(), "mimeType": "image/png"}) return respond(mid, {"content": content, "structuredContent": data, "isError": False}) except ZapError as e: text = (KEY_WARNING[0] + "\n" if KEY_WARNING else "") + str(e) return respond(mid, {"content": [{"type": "text", "text": text}], "structuredContent": {"error": e.body}, "isError": True}) except Exception as e: # report, never crash the server return respond(mid, {"content": [{"type": "text", "text": "zap_mcp error: %s" % e}], "isError": True}) if mid is not None: # notifications (no id) get no answer respond(mid, error={"code": -32601, "message": "method not found: %s" % method}) def main(): for line in sys.stdin: line = line.strip() if not line: continue try: m = json.loads(line) except ValueError: respond(None, error={"code": -32700, "message": "parse error"}) continue for one in (m if isinstance(m, list) else [m]): if isinstance(one, dict): handle(one) if __name__ == "__main__": main()