execute

Instructions

Implementation Reference

• src/build123d_mcp/server.py:15-18 (handler)

  MCP tool registration and handler for 'execute'. Decorated with @mcp.tool() by FastMCP, calls WorkerSession.execute(code) which proxies to the worker subprocess.

  @mcp.tool()
  def execute(code: str) -> str:
      """Execute build123d Python code in the persistent session. Use show(shape, name) to register named objects (name defaults to 'shape'); show() immediately prints volume and face count confirming the shape is non-empty. After any boolean operation (-, +, &) call measure(topology) or measure(volume) to confirm it succeeded before calling render_view."""
      return _session.execute(code)

• src/build123d_mcp/tools/execute.py:1-2 (helper)

  Thin helper that delegates to Session.execute(code). Used by test code for direct session execution without going through MCP.

  def execute_code(session, code: str) -> str:
      return session.execute(code)

• src/build123d_mcp/session.py:61-148 (handler)

  Core execution logic: AST security check, compile, exec with SIGALRM timeout, rollback on errors, automatic current_shape detection, and quick diagnostics output.

  def execute(self, code: str) -> str:
      # Layer 1: AST check before anything runs
      try:
          check_ast(code)
      except ValueError as e:
          self.last_error_detail = {"type": "SecurityError", "message": str(e), "line": None, "excerpt": None}
          return f"Error: SecurityError: {e}"
  
      try:
          compiled = compile(code, "<mcp>", "exec")
      except SyntaxError as e:
          excerpt = self._syntax_excerpt(code, e.lineno)
          self.last_error_detail = {"type": "SyntaxError", "message": str(e), "line": e.lineno, "excerpt": excerpt}
          return f"Error: SyntaxError: {e}"
  
      values_before = {
          k: v.copy() if isinstance(v, (list, dict, set)) else v
          for k, v in self.namespace.items()
          if k not in ("__builtins__", "show")
      }
      shape_before = self.current_shape
      objects_before = dict(self.objects)
  
      buf = io.StringIO()
      exc: Exception | None = None
  
      # Layer 3: SIGALRM timeout (Unix main-thread only; silently skipped otherwise).
      # Fire 2s before the parent's conn.poll() deadline so the worker always
      # returns a response before the parent kills it.
      _alarm_set = False
      _old_handler: Any = None
      try:
          def _timeout_handler(signum: int, frame: Any) -> None:
              raise ExecutionTimeout(
                  f"Code exceeded the {self.exec_timeout}s execution time limit."
              )
          _old_handler = signal.signal(signal.SIGALRM, _timeout_handler)
          signal.alarm(max(1, self.exec_timeout - 2))
          _alarm_set = True
      except (OSError, ValueError, AttributeError):
          pass  # Windows (no SIGALRM) or non-main thread; no timeout protection
  
      try:
          with redirect_stdout(buf), redirect_stderr(buf):
              exec(compiled, self.namespace)  # noqa: S102
          # Cancel alarm immediately so it cannot fire during post-exec shape detection.
          if _alarm_set:
              signal.alarm(0)
      except ExecutionTimeout as e:
          self._rollback_namespace(values_before)
          self.current_shape = shape_before
          self.objects.clear()
          self.objects.update(objects_before)
          self.last_error_detail = {"type": "ExecutionTimeout", "message": str(e), "line": None, "excerpt": None}
          return f"Error: ExecutionTimeout: {e}"
      except AssertionError as e:
          self._rollback_namespace(values_before)
          self.current_shape = shape_before
          self.objects.clear()
          self.objects.update(objects_before)
          msg = str(e) or "Constraint failed"
          self.last_error_detail = {"type": "AssertionError", "message": msg, "line": None, "excerpt": None}
          return f"Constraint failed: {e}" if str(e) else "Constraint failed"
      except Exception as e:
          exc = e
      finally:
          if _alarm_set:
              signal.alarm(0)
              signal.signal(signal.SIGALRM, _old_handler)
  
      if exc is not None:
          self._rollback_namespace(values_before)
          self.current_shape = shape_before
          self.objects.clear()
          self.objects.update(objects_before)
          self.last_error_detail = self._make_error_detail(exc, code)
          return f"Error: {type(exc).__name__}: {exc}"
  
      self.last_error_detail = None
      new_keys = {k for k in self.namespace if k not in ("__builtins__", "show")} - values_before.keys()
      self._update_current_shape(new_keys)
  
      output = buf.getvalue() or "OK"
      if self.current_shape is not None and self.current_shape is not shape_before:
          diag = self._quick_diagnostics(self.current_shape)
          if diag:
              output = output.rstrip("\n") + "\n" + diag
      return output

• src/build123d_mcp/server.py:15-18 (registration)

  Tool registration via FastMCP @mcp.tool() decorator exposing the 'execute' tool to MCP clients.

  @mcp.tool()
  def execute(code: str) -> str:
      """Execute build123d Python code in the persistent session. Use show(shape, name) to register named objects (name defaults to 'shape'); show() immediately prints volume and face count confirming the shape is non-empty. After any boolean operation (-, +, &) call measure(topology) or measure(volume) to confirm it succeeded before calling render_view."""
      return _session.execute(code)

• src/build123d_mcp/worker.py:60-213 (helper)

  WorkerSession.execute() proxies the execute call to the worker subprocess via Pipe, sending op='execute' with the code argument. This is the bridge between server.py and the worker process.

          return session.execute(args["code"])
  
      if op == "render_view":
          from build123d_mcp.tools.render import render_view
          return render_view(session, **args)
  
      if op == "export_file":
          from build123d_mcp.tools.export import export_file
          return export_file(session, **args)
  
      if op == "interference":
          from build123d_mcp.tools.interference import interference
          return interference(session, **args)
  
      if op == "measure":
          from build123d_mcp.tools.measure import measure
          return measure(session, **args)
  
      if op == "list_objects":
          from build123d_mcp.tools.list_objects import list_objects
          return list_objects(session)
  
      if op == "save_snapshot":
          name = args["name"]
          session.save_snapshot(name)
          saved = (["current_shape"] if session.current_shape is not None else []) + list(session.snapshots[name]["objects"].keys())
          return f"Snapshot '{name}' saved. Geometry captured: {', '.join(saved) if saved else 'none'}."
  
      if op == "restore_snapshot":
          name = args["name"]
          try:
              session.restore_snapshot(name)
          except KeyError as e:
              return f"Error: {e}"
          restored = (["current_shape"] if session.current_shape is not None else []) + list(session.objects.keys())
          return f"Snapshot '{name}' restored. Active geometry: {', '.join(restored) if restored else 'none'}."
  
      if op == "reset":
          session.reset()
          return "Session reset."
  
      if op == "search_library":
          if library_index is None:
              return "No part library configured."
          from build123d_mcp.tools.library import search_library
          return search_library(library_index, args.get("query", ""))
  
      if op == "load_part":
          if library_index is None:
              return "No part library configured."
          from build123d_mcp.tools.library import load_part
          return load_part(session, library_index, args["name"], args.get("params", ""))
  
      if op == "diff_snapshot":
          from build123d_mcp.tools.diff import diff_snapshot
          return diff_snapshot(session, args["snapshot_a"], args.get("snapshot_b", ""), args.get("format", "text"))
  
      if op == "session_state":
          from build123d_mcp.tools.session_state import session_state
          return session_state(session)
  
      if op == "health_check":
          from build123d_mcp.tools.health_check import health_check
          return health_check(session)
  
      if op == "version":
          from importlib.metadata import version
          return version("build123d-mcp")
  
      if op == "last_error":
          from build123d_mcp.tools.last_error import last_error
          return last_error(session)
  
      if op == "shape_compare":
          from build123d_mcp.tools.shape_compare import shape_compare
          return shape_compare(session, args["object_a"], args["object_b"])
  
      raise ValueError(f"Unknown operation: '{op}'")
  
  
  class WorkerSession:
      """Parent-side proxy to the persistent worker subprocess.
  
      Exposes the same interface as Session so server.py can use either.
      """
  
      _RENDER_TIMEOUT = 120
      _EXPORT_TIMEOUT = 60
      _INTERFERENCE_TIMEOUT = 30
      _SHORT_TIMEOUT = 10
  
      def __init__(self, exec_timeout: int = 60, library_path: str = "", allow_all_imports: bool = False) -> None:
          self._exec_timeout = exec_timeout
          self._library_path = library_path
          self._allow_all_imports = allow_all_imports
          self._conn: Any = None
          self._proc: Any = None
          self._start_worker()
  
      def _start_worker(self) -> None:
          ctx = multiprocessing.get_context("spawn")
          parent_conn, child_conn = ctx.Pipe()
          self._proc = ctx.Process(
              target=worker_main,
              args=(child_conn, self._library_path, self._exec_timeout, self._allow_all_imports),
              daemon=True,
          )
          self._proc.start()
          child_conn.close()
          self._conn = parent_conn
  
          if not self._conn.poll(_WORKER_READY_TIMEOUT):
              self._proc.kill()
              self._proc.join(5)
              raise RuntimeError("Worker process failed to start within timeout.")
          self._conn.recv()  # consume the ready signal
  
      def _kill_worker(self) -> None:
          try:
              self._proc.kill()
              self._proc.join(5)
          except Exception:
              pass
  
      def _call(self, op: str, args: dict, timeout: int) -> Any:
          if not self._proc.is_alive():
              self._start_worker()
              raise RuntimeError("Worker crashed; session restarted. Re-run your setup code.")
  
          self._conn.send({"op": op, "args": args})
  
          if not self._conn.poll(timeout):
              self._kill_worker()
              self._start_worker()
              from build123d_mcp.security import ExecutionTimeout
              if op == "execute":
                  raise ExecutionTimeout(
                      f"Code exceeded the {timeout}s execution time limit."
                  )
              raise RuntimeError(f"Operation '{op}' timed out after {timeout}s.")
  
          try:
              response = self._conn.recv()
          except EOFError:
              self._start_worker()
              raise RuntimeError("Worker process crashed unexpectedly; session restarted.")
  
          if response["ok"]:
              return response["result"]
          raise RuntimeError(response["error"])
  
      # --- Session-compatible interface ---
  
      def execute(self, code: str) -> str: