ping

Instructions

Implementation Reference

• fmod_mcp/tools/discovery.py:10-18 (handler)

  Core implementation: sends JavaScript to FMOD Studio via TCP to retrieve the studio version. Wraps the version lookup in try/catch for resilience, returns {ok, version} dict.

  async def ping(client: StudioClient) -> dict[str, Any]:
      """Sanity-check the TCP connection and fetch Studio version if available."""
      js = """
          var v = null;
          try { v = studio.version; } catch (e) {}
          try { if (v == null && studio.application && studio.application.version) v = studio.application.version; } catch (e) {}
          return { ok: true, version: v };
      """
      return await client.eval(js)

• fmod_mcp/server.py:35-38 (registration)

  Tool registration via @mcp.tool() decorator on FastMCP instance. Delegates to discovery.ping() with a lazily-initialized StudioClient singleton.

  @mcp.tool()
  async def ping() -> dict[str, Any]:
      """Sanity-check the FMOD Studio connection. Returns version if available."""
      return await discovery.ping(_studio())

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

  The StudioClient.eval() method is called by discovery.ping() to send/receive JS over TCP. Wraps the snippet in a sentinel-tagged IIFE for response framing.

  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)

• fmod_mcp/tools/discovery.py:1-3 (schema)

  Module docstring lists ping as a read-only discovery tool. No explicit schema beyond the return type hint dict[str, Any].

  """Read-only tools: ping, list_banks, list_events, list_buses, get_event."""
  from __future__ import annotations

• fmod_mcp/studio_client.py:64-195 (helper)

  The StudioClient class that manages the TCP connection to FMOD Studio and is used by discovery.ping().

  class StudioClient:
      """Single-connection, serialized TCP client for FMOD Studio scripting."""
  
      def __init__(
          self,
          host: str = DEFAULT_HOST,
          port: int = DEFAULT_PORT,
          log_path: Path | None = None,
      ) -> None:
          self.host = host
          self.port = port
          self._reader: asyncio.StreamReader | None = None
          self._writer: asyncio.StreamWriter | None = None
          self._lock = asyncio.Lock()
          self._buffer = ""
          self._log_path = log_path or (Path.home() / ".cache" / "fmod-mcp" / "commands.log")
          self._log_path.parent.mkdir(parents=True, exist_ok=True)
  
      async def connect(self) -> None:
          if self._writer is not None:
              return
          try:
              self._reader, self._writer = await asyncio.open_connection(self.host, self.port)
          except OSError as exc:
              raise ConnectionError(
                  f"Could not reach FMOD Studio at {self.host}:{self.port} — "
                  "is Studio running with the project open and TCP scripting enabled?"
              ) from exc
          await self._drain_banner()
  
      async def _drain_banner(self) -> None:
          """Studio prints a prompt/banner on connect; swallow it briefly."""
          assert self._reader is not None
          try:
              while True:
                  chunk = await asyncio.wait_for(self._reader.read(4096), timeout=0.2)
                  if not chunk:
                      break
          except asyncio.TimeoutError:
              pass
  
      async def close(self) -> None:
          if self._writer is None:
              return
          self._writer.close()
          try:
              await self._writer.wait_closed()
          except Exception:  # noqa: BLE001 — best-effort cleanup
              pass
          self._reader = None
          self._writer = None
          self._buffer = ""
  
      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)