#!/usr/bin/env python3 """Zero-dependency stdio MCP server bridging to the Orca/Orca-CAD control socket. Speaks MCP (JSON-RPC 2.0 over newline-delimited stdio) to an MCP client (Claude Code), and forwards each tool call to the app's Unix-domain control socket (opened by the GUI when launched with ORCA_CAD_MCP set). The tool list is built *live* from the app's own `describe_tools` reply — introspection drives the schema, so new kernel methods surface without touching this file. Usage: orca_cad_mcp_bridge.py [SOCKET_PATH] (default /tmp/orca-cad-mcp.sock) The app must be running with ORCA_CAD_MCP set; if the socket is down, tools/list falls back to the slice-1 set and tool calls report the connection error (never crash). """ import sys, os, json, socket, itertools SOCK_PATH = sys.argv[1] if len(sys.argv) > 1 else "/tmp/orca-cad-mcp.sock" SERVER_INFO = {"name": "orca-cad", "version": "0.1"} _app_id = itertools.count(1) # --- app control-socket round-trip -------------------------------------------- def app_call(method, params=None): """One request to the app over the unix socket. Raises on transport failure.""" req = {"jsonrpc": "2.0", "id": next(_app_id), "method": method} if params is not None: req["params"] = params s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) s.settimeout(20) try: s.connect(SOCK_PATH) s.sendall((json.dumps(req) + "\n").encode()) buf = b"" while not buf.endswith(b"\n"): chunk = s.recv(65536) if not chunk: break buf += chunk finally: s.close() return json.loads(buf.decode()) # --- describe_tools -> MCP tool schemas --------------------------------------- _TYPE_MAP = {"number": "number", "integer": "integer", "string": "string", "boolean": "boolean", "array": "array", "object": "object"} def _param_schema(p): sch = {"type": _TYPE_MAP.get(p.get("type", "string"), "string")} desc = p.get("description") if "unit" in p: desc = (desc + " " if desc else "") + f"in {p['unit']}" if desc: sch["description"] = desc if "enum" in p: sch["enum"] = p["enum"] if "min" in p: sch["minimum"] = p["min"] if "max" in p: sch["maximum"] = p["max"] if "default" in p: sch["default"] = p["default"] return sch # Fallback when the app socket is unreachable at list time (e.g. the GUI isn't up yet when the # MCP client lists tools at session start). The LIVE describe_tools reply is authoritative — this # mirrors its 16-method surface so the agent still sees the full toolset; params without a # `default` are surfaced as required. Keep in sync with McpControl.cpp describe_tools(). def _p(name, typ="number", **kw): return dict(name=name, type=typ, **kw) _PLANE = _p("plane", "string", enum=["XY", "XZ", "YZ"], default="XY") _FALLBACK_TOOLS = {"tools": [ {"name": "describe_tools", "summary": "List callable tools and their parameters.", "params": []}, {"name": "describe_scene", "summary": "Feature tree + per-body bounding boxes.", "params": []}, {"name": "extrude", "summary": "Extrude a profile (or width x height rectangle) to a depth; Onshape end conditions.", "params": [_p("width", default=20), _p("height", default=20), _p("distance", default=10), _PLANE, _p("profile", "array", default=[], description="closed [[x,y],...] overrides width/height"), _p("boolean", "string", enum=["new", "union", "subtract", "intersect"], default="new"), _p("end", "string", enum=["blind", "symmetric", "two_sided", "through_all", "up_to_face"], default="blind"), _p("distance2", default=0, description="second side when end=two_sided"), _p("up_to_face", "integer", default=-1, description="target face id when end=up_to_face"), _p("taper", default=0), _p("flip", "boolean", default=False)]}, {"name": "revolve", "summary": "Revolve a profile about a plane axis.", "params": [_p("width", default=20), _p("height", default=10), _p("angle", default=360), _p("axis", "integer", enum=[0, 1], default=0), _p("flip", "boolean", default=False), _PLANE, _p("profile", "array", default=[], description="closed [[x,y],...] overrides width/height"), _p("boolean", "string", enum=["new", "union", "subtract", "intersect"], default="new")]}, {"name": "fillet", "summary": "Round a measured edge of a body.", "params": [_p("edge", "integer"), _p("radius", default=1), _p("body", "integer", default=-1, description="target body; omit for last")]}, {"name": "chamfer", "summary": "Chamfer a measured edge of a body.", "params": [_p("edge", "integer"), _p("distance", default=1), _p("body", "integer", default=-1, description="target body; omit for last")]}, {"name": "hole", "summary": "Drill a circular hole at (x,y) on a plane.", "params": [_p("diameter", default=5), _p("depth", default=10), _p("through", "boolean", default=False), _p("x", default=0), _p("y", default=0), _PLANE]}, {"name": "boolean", "summary": "Combine two bodies: union | subtract | intersect.", "params": [_p("op", "string", enum=["union", "subtract", "intersect"], default="subtract"), _p("target", "integer", default=0), _p("tool", "integer", default=1), _p("keep_tool", "boolean", default=False), _p("tolerance", default=0)]}, {"name": "pattern", "summary": "Replicate a body: linear or circular.", "params": [_p("circular", "boolean", default=False), _p("count", "integer", default=3), _p("spacing", default=10), _p("dir", "integer", enum=[0, 1], default=0), _p("angle", default=360), _PLANE, _p("body", "integer", default=-1, description="target body; omit for last")]}, {"name": "shell", "summary": "Hollow a body to a wall thickness; optionally open one face.", "params": [_p("thickness", default=1), _p("face", "integer", default=-1, description="face id to leave open; omit for closed"), _p("body", "integer", default=-1, description="target body; omit for last")]}, {"name": "draft", "summary": "Taper a body face by an angle (pull +Z).", "params": [_p("face", "integer"), _p("angle", default=5), _p("body", "integer", default=-1, description="target body; omit for last")]}, {"name": "query_topology", "summary": "Measured faces and edges of a body.", "params": [_p("body", "integer", default=0)]}, {"name": "measure", "summary": "Distance/angle between two refs {face|edge|point} on a body.", "params": [_p("body", "integer", default=0), _p("a", "object"), _p("b", "object")]}, {"name": "slice_body", "summary": "Cross-section of a body; ordered closed/open contours.", "params": [_p("body", "integer", default=0), _PLANE, _p("offset", default=0)]}, {"name": "import_step", "summary": "Import a STEP file as native B-rep bodies.", "params": [_p("path", "string")]}, {"name": "validate_against", "summary": "Volume + bbox + surface deviation of a body vs a reference {step|body}.", "params": [_p("body", "integer", default=0), _p("reference", "object")]}, ]} def list_tools(): try: desc = app_call("describe_tools").get("result", {}) if "tools" not in desc: desc = _FALLBACK_TOOLS except Exception: desc = _FALLBACK_TOOLS out = [] for t in desc["tools"]: params = t.get("params", []) props = {p["name"]: _param_schema(p) for p in params} required = [p["name"] for p in params if "default" not in p] out.append({ "name": t["name"], "description": t.get("summary", ""), "inputSchema": {"type": "object", "properties": props, "required": required}, }) return out # --- MCP method handlers ------------------------------------------------------ def handle(req): m = req.get("method") rid = req.get("id") if m == "initialize": ver = (req.get("params") or {}).get("protocolVersion", "2024-11-05") return {"jsonrpc": "2.0", "id": rid, "result": { "protocolVersion": ver, "capabilities": {"tools": {"listChanged": False}}, "serverInfo": SERVER_INFO}} if m == "ping": return {"jsonrpc": "2.0", "id": rid, "result": {}} if m == "tools/list": return {"jsonrpc": "2.0", "id": rid, "result": {"tools": list_tools()}} if m == "tools/call": p = req.get("params") or {} name = p.get("name") args = p.get("arguments") or {} try: reply = app_call(name, args) except Exception as e: return {"jsonrpc": "2.0", "id": rid, "result": { "content": [{"type": "text", "text": f"control socket unreachable ({SOCK_PATH}): {e}"}], "isError": True}} if "error" in reply: return {"jsonrpc": "2.0", "id": rid, "result": { "content": [{"type": "text", "text": json.dumps(reply["error"])}], "isError": True}} return {"jsonrpc": "2.0", "id": rid, "result": { "content": [{"type": "text", "text": json.dumps(reply.get("result"), indent=2)}]}} if rid is not None: # unknown *request* return {"jsonrpc": "2.0", "id": rid, "error": {"code": -32601, "message": f"method not found: {m}"}} return None # notification (e.g. notifications/initialized) -> no reply def main(): for line in sys.stdin: line = line.strip() if not line: continue try: req = json.loads(line) except Exception: continue resp = handle(req) if resp is not None: sys.stdout.write(json.dumps(resp) + "\n") sys.stdout.flush() if __name__ == "__main__": main()