save_project

Instructions

Implementation Reference

• fmod_mcp/tools/project.py:9-15 (handler)

  Core handler that executes 'save_project'. Sends JavaScript `studio.project.save()` to FMOD Studio via TCP client.

  async def save_project(client: StudioClient) -> dict[str, Any]:
      """Save the currently open FMOD Studio project."""
      js = """
          studio.project.save();
          return { ok: true, saved: true };
      """
      return await client.eval(js, timeout=120.0)

• fmod_mcp/server.py:184-187 (registration)

  Registers 'save_project' as an MCP tool on the FastMCP server using the @mcp.tool() decorator.

  @mcp.tool()
  async def save_project() -> dict[str, Any]:
      """Save the open FMOD Studio project. Must be called before build_banks."""
      return await project.save_project(_studio())

• tests/test_server_registration.py:29-42 (schema)

  Registration test confirms 'save_project' is listed in the expected tools set.

      "save_project",
      "build_banks",
      # escape
      "run_js",
  }
  
  
  def test_all_tools_registered():
      tools = set(mcp._tool_manager._tools.keys())
      missing = EXPECTED_TOOLS - tools
      extra = tools - EXPECTED_TOOLS
      assert not missing, f"Missing tools: {missing}"
      assert not extra, f"Unexpected extra tools: {extra}"

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

  StudioClient.eval() is the helper used by save_project to execute JavaScript in FMOD Studio. It wraps JS in an IIFE with sentinel-based response detection over TCP.

  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_tools.py:176-180 (schema)

  Unit test for save_project, verifying it calls 'studio.project.save' and returns saved=True.

  async def test_save_project(client: StudioClient, mock_studio: MockStudio):
      mock_studio.responder = responder_sequence([("OK", {"ok": True, "saved": True})])
      result = await project.save_project(client)
      assert result["saved"] is True
      assert "studio.project.save" in _last_sent_js(mock_studio)