live_eval

Instructions

Implementation Reference

• src/server.py:260-268 (registration)

  The tool 'live_eval' is registered as an MCP tool with schema requiring 'expr' (string), optional 'ref', and response controls. It uses forward('eval') which delegates to the bridge.

  server.add_tool(Tool("live_eval", (
      "Evaluate a Python expression inside Live with song, app, obj, and Live bindings. "
      + ABLETON_AGENT_GUIDE
      + " Use live_exec for statements; prefer installed browser/library assets before generated assets unless asked."
  ), schema({
      "expr": {"type": "string"},
      "ref": ref,
      **response_controls,
  }, ["expr"]), forward("eval")))

• src/server.py:33-34 (handler)

  The handler for live_eval is a lambda created by forward('eval'), which calls bridge.request('eval', args). This sends a JSON-RPC request to the Ableton bridge.

  def forward(method: str):
      return lambda args: bridge.request(method, args)

• src/server.py:264-268 (schema)

  Input schema for live_eval: requires 'expr' (string), optional 'ref' (object with path/id), plus optional response controls (detail, max_items, max_depth, max_string_length, timeout).

  ), schema({
      "expr": {"type": "string"},
      "ref": ref,
      **response_controls,
  }, ["expr"]), forward("eval")))

• src/bridge.py:40-109 (helper)

  The bridge request method sends the 'eval' JSON-RPC method to the Ableton Live bridge over a socket connection and returns the result.

  def request(self, method: str, params: dict[str, Any] | None = None) -> Any:
      params = params or {}
      payload = {
          "jsonrpc": "2.0",
          "id": next(self._ids),
          "method": method,
          "params": params,
      }
      with self._lock:
          try:
              response = self._send(payload)
          except OSError:
              self.close()
              try:
                  response = self._send(payload)
              except OSError as exc:
                  self.close()
                  raise AbletonBridgeError(f"Could not connect to Ableton bridge at {self.config.host}:{self.config.port}: {exc}") from exc
      message = json.loads(response.decode("utf-8"))
      if "error" in message:
          err = message["error"]
          detail = err.get("data") if os.environ.get("ABLETON_MCP_TRACEBACK") else ""
          suffix = f": {detail}" if detail else ""
          raise AbletonBridgeError(f"{err.get('code', -32000)} {err.get('message', 'Bridge error')}{suffix}")
      return message.get("result")
  
  def close(self) -> None:
      if self._sock is not None:
          try:
              self._sock.close()
          except OSError:
              pass
          self._sock = None
  
  def _send(self, payload: dict[str, Any]) -> bytes:
      sock = self._socket()
      request_timeout = self.config.timeout
      params = payload.get("params") or {}
      if isinstance(params, dict) and params.get("timeout") is not None:
          request_timeout = max(request_timeout, float(params["timeout"]) + 1.0)
      sock.settimeout(request_timeout)
      line = (json.dumps(payload, separators=(",", ":")) + "\n").encode("utf-8")
      sock.sendall(line)
      return self._read_line(sock, self.config.max_response_bytes)
  
  def _socket(self) -> socket.socket:
      if self._sock is None:
          self._sock = socket.create_connection((self.config.host, self.config.port), self.config.timeout)
          self._sock.settimeout(self.config.timeout)
      return self._sock
  
  @staticmethod
  def _read_line(sock: socket.socket, max_bytes: int = 8 * 1024 * 1024) -> bytes:
      chunks: list[bytes] = []
      total = 0
      while True:
          chunk = sock.recv(4096)
          if not chunk:
              break
          total += len(chunk)
          if max_bytes >= 0 and total > max_bytes:
              raise OSError(f"Ableton bridge response exceeds {max_bytes} bytes")
          chunks.append(chunk)
          if b"\n" in chunk:
              break
      data = b"".join(chunks)
      if not data:
          raise OSError("No response from Ableton bridge")
      return data.split(b"\n", 1)[0]