live_get

Instructions

Implementation Reference

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

  The 'forward' function creates a closure (lambda) that calls bridge.request(method, args). For live_get, the method is 'get', so the handler is: lambda args: bridge.request('get', args). This is the actual execution logic — it sends a 'get' request to the Ableton bridge with the tool's arguments.

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

• src/server.py:59-68 (registration)

  The 'live_get' tool is registered with the MCP server at line 59. It has name 'live_get', description 'Resolve object; read selected properties/children.', accepts parameters: ref (required object with path/id), properties (array of strings), children (array or dict), child_limit (integer), plus optional response controls (detail, max_items, max_depth, max_string_length). The handler is set to forward('get').

  server.add_tool(Tool("live_get", "Resolve object; read selected properties/children.", schema({
      "ref": ref,
      "properties": {"type": "array", "items": {"type": "string"}},
      "children": {"oneOf": [
          {"type": "array", "items": {"type": "string"}},
          {"type": "object", "additionalProperties": {"type": "integer"}},
      ]},
      "child_limit": {"type": "integer", "minimum": 0},
      **response_controls,
  }, ["ref"]), forward("get")))

• src/server.py:36-43 (schema)

  The 'ref' schema used by live_get defines the reference to a Live object, requiring either a 'path' (string) or an 'id' (integer).

  ref = {
      "type": "object",
      "properties": {
          "path": {"type": "string"},
          "id": {"type": "integer"},
      },
      "additionalProperties": False,
  }

• src/server.py:46-53 (schema)

  The 'response_controls' dict defines shared optional parameters (detail, max_items, max_depth, max_string_length) that are spread into live_get's schema via **response_controls.

  response_controls = {
      "detail": {"type": "boolean"},
      "max_items": {"type": "integer"},
      "max_depth": {"type": "integer"},
      "max_string_length": {"type": "integer"},
      "timeout": {"type": "number", "description": "Seconds to wait for Live's main thread."},
  }
  mutation_controls = {

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

  The 'request' method on AbletonBridgeClient is the underlying helper that the live_get handler calls. It sends a JSON-RPC request with the method name (e.g., 'get') and params to the Ableton Live bridge over a TCP socket, 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: