list_events

Instructions

Implementation Reference

• fmod_mcp/tools/discovery.py:30-40 (handler)

  Core implementation of list_events: builds JavaScript that finds all Event instances, filters by prefix, and returns path/guid/name.

  async def list_events(client: StudioClient, prefix: str | None = None) -> list[dict[str, str]]:
      """List events, optionally filtered by path prefix (e.g. ``event:/bumpers/``)."""
      prefix_json = json.dumps(prefix) if prefix else "null"
      js = f"""
          var prefix = {prefix_json};
          return studio.project.model.Event.findInstances()
              .filter(function(e) {{ return e.isValid; }})
              .map(function(e) {{ return {{ path: e.getPath(), guid: e.id, name: e.name }}; }})
              .filter(function(e) {{ return prefix == null || e.path.indexOf(prefix) === 0; }});
      """
      return await client.eval(js)

• fmod_mcp/server.py:47-50 (registration)

  MCP tool registration of 'list_events' via @mcp.tool() decorator, delegating to discovery.list_events.

  @mcp.tool()
  async def list_events(prefix: str | None = None) -> list[dict[str, str]]:
      """List events, optionally filtered by path prefix (e.g. 'event:/bumpers/')."""
      return await discovery.list_events(_studio(), prefix)

• fmod_mcp/server.py:48-49 (schema)

  Input schema: optional 'prefix' parameter (str | None) for filtering events by path prefix.

  async def list_events(prefix: str | None = None) -> list[dict[str, str]]:
      """List events, optionally filtered by path prefix (e.g. 'event:/bumpers/')."""

• fmod_mcp/studio_client.py:117-192 (helper)

  StudioClient.eval() — the low-level helper that sends JavaScript to FMOD Studio and parses the response, used by list_events.

  async def eval(self, js: str, timeout: float = 30.0) -> Any:
      """Evaluate a JavaScript snippet in Studio and return the parsed value.
  
      The caller is responsible for making sure ``js`` uses ``return`` to
      surface its value (the snippet is wrapped in an IIFE).
      """
      req_id = uuid.uuid4().hex
      # Substitute the request id into the template BEFORE injecting user JS,
      # so a snippet that happens to contain "__ID__" or "__JS__" cannot
      # corrupt the surrounding wrapper.
      wrapped = _WRAPPER_TEMPLATE.replace("__ID__", req_id).replace("__JS__", js)
      # Studio's terminal evaluates per-line: strip `//` comments and collapse
      # internal newlines so the whole command arrives as one line.
      wrapped = _flatten_js(wrapped)
      async with self._lock:
          # connect() inside the lock so concurrent callers don't open
          # multiple sockets or fight over the same StreamReader.
          await self.connect()
          self._log_command(req_id, js)
          assert self._writer is not None
          self._writer.write(wrapped.encode("utf-8"))
          await self._writer.drain()
          return await self._await_response(req_id, timeout)
  
  async def _await_response(self, req_id: str, timeout: float) -> Any:
      assert self._reader is not None
      loop = asyncio.get_event_loop()
      deadline = loop.time() + timeout
      while True:
          found = self._consume_response(req_id)
          if found is not None:
              kind, payload = found
              if kind == "OK":
                  return payload
              message = payload.get("message", "FMOD Studio error") if isinstance(payload, dict) else str(payload)
              stack = payload.get("stack", "") if isinstance(payload, dict) else ""
              raise StudioError(f"{message}\n{stack}".rstrip())
          remaining = deadline - loop.time()
          if remaining <= 0:
              raise TimeoutError(
                  f"FMOD Studio did not respond to request {req_id} within {timeout}s"
              )
          try:
              chunk = await asyncio.wait_for(self._reader.read(4096), timeout=remaining)
          except asyncio.TimeoutError as exc:
              raise TimeoutError(
                  f"FMOD Studio did not respond to request {req_id} within {timeout}s"
              ) from exc
          if not chunk:
              raise ConnectionError("FMOD Studio closed the TCP connection")
          self._buffer += chunk.decode("utf-8", errors="replace")
  
  def _consume_response(self, req_id: str) -> tuple[str, Any] | None:
      for kind, sentinel in (("OK", _SENTINEL_OK), ("ERR", _SENTINEL_ERR)):
          tag = f"{sentinel}:{req_id}:"
          idx = self._buffer.find(tag)
          if idx == -1:
              continue
          end = self._buffer.find("\n", idx + len(tag))
          if end == -1:
              return None
          payload_str = self._buffer[idx + len(tag) : end]
          self._buffer = self._buffer[end + 1 :]
          try:
              return kind, json.loads(payload_str)
          except json.JSONDecodeError as exc:
              raise StudioError(f"Malformed JSON from Studio: {payload_str!r}") from exc
      return None
  
  def _log_command(self, req_id: str, js: str) -> None:
      try:
          with self._log_path.open("a", encoding="utf-8") as fh:
              fh.write(f"# {req_id}\n{js.rstrip()}\n\n")
      except OSError as exc:
          logger.warning("Could not write to command log %s: %s", self._log_path, exc)

• tests/test_server_registration.py:6-33 (registration)

  Test fixture listing 'list_events' in EXPECTED_TOOLS to verify it's registered.

  EXPECTED_TOOLS = {
      # discovery
      "ping",
      "list_banks",
      "list_events",
      "list_buses",
      "get_event",
      # audio + events
      "import_audio",
      "create_event",
      "add_single_sound",
      "set_event_property",
      "assign_to_bank",
      "assign_to_bus",
      # effects
      "list_effect_types",
      "add_effect",
      "list_effects",
      "get_effect",
      "set_effect_param",
      "remove_effect",
      "bypass_effect",
      # project
      "save_project",
      "build_banks",
      # escape
      "run_js",
  }