TrueNAS Core MCP Server

Instructions

Implementation Reference

• truenas_mcp_server/tools/storage.py:186-236 (handler)

  The handler function that implements the core logic of the 'list_datasets' tool. It retrieves datasets from the TrueNAS API endpoint '/pool/dataset', processes and formats the data (including size formatting, property extraction), organizes datasets by pool, and returns a structured response with success flag and metadata statistics.

  @tool_handler
  async def list_datasets(self) -> Dict[str, Any]:
      """
      List all datasets
      
      Returns:
          Dictionary containing list of datasets
      """
      await self.ensure_initialized()
      
      datasets = await self.client.get("/pool/dataset")
      
      dataset_list = []
      for ds in datasets:
          # Calculate usage
          used = ds.get("used", {}).get("parsed") if isinstance(ds.get("used"), dict) else ds.get("used", 0)
          available = ds.get("available", {}).get("parsed") if isinstance(ds.get("available"), dict) else ds.get("available", 0)
          
          dataset_info = {
              "name": ds.get("name"),
              "pool": ds.get("pool"),
              "type": ds.get("type"),
              "mountpoint": ds.get("mountpoint"),
              "compression": ds.get("compression", {}).get("value") if isinstance(ds.get("compression"), dict) else ds.get("compression"),
              "deduplication": ds.get("deduplication", {}).get("value") if isinstance(ds.get("deduplication"), dict) else ds.get("deduplication"),
              "encrypted": ds.get("encrypted"),
              "used": self.format_size(used) if isinstance(used, (int, float)) else str(used),
              "available": self.format_size(available) if isinstance(available, (int, float)) else str(available),
              "quota": ds.get("quota", {}).get("value") if isinstance(ds.get("quota"), dict) else ds.get("quota"),
              "children": ds.get("children", [])
          }
          dataset_list.append(dataset_info)
      
      # Organize by pool
      pools_datasets = {}
      for ds in dataset_list:
          pool = ds["pool"]
          if pool not in pools_datasets:
              pools_datasets[pool] = []
          pools_datasets[pool].append(ds)
      
      return {
          "success": True,
          "datasets": dataset_list,
          "metadata": {
              "total_datasets": len(dataset_list),
              "by_pool": {pool: len(datasets) for pool, datasets in pools_datasets.items()},
              "encrypted_datasets": sum(1 for ds in dataset_list if ds.get("encrypted")),
              "compressed_datasets": sum(1 for ds in dataset_list if ds.get("compression") and ds.get("compression") != "off")
          }
      }

• truenas_mcp_server/tools/storage.py:12-33 (registration)

  The get_tool_definitions method in StorageTools class that registers all storage tools, including 'list_datasets' with description 'List all datasets' and empty input schema ({}). This method is called during server initialization to register tools with the MCP server.

  def get_tool_definitions(self) -> list:
      """Get tool definitions for storage management"""
      return [
          ("list_pools", self.list_pools, "List all storage pools", {}),
          ("get_pool_status", self.get_pool_status, "Get detailed status of a specific pool",
           {"pool_name": {"type": "string", "required": True}}),
          ("list_datasets", self.list_datasets, "List all datasets", {}),
          ("get_dataset", self.get_dataset, "Get detailed information about a dataset",
           {"dataset": {"type": "string", "required": True}}),
          ("create_dataset", self.create_dataset, "Create a new dataset",
           {"pool": {"type": "string", "required": True},
            "name": {"type": "string", "required": True},
            "compression": {"type": "string", "required": False},
            "quota": {"type": "string", "required": False},
            "recordsize": {"type": "string", "required": False}}),
          ("delete_dataset", self.delete_dataset, "Delete a dataset",
           {"dataset": {"type": "string", "required": True},
            "recursive": {"type": "boolean", "required": False}}),
          ("update_dataset", self.update_dataset, "Update dataset properties",
           {"dataset": {"type": "string", "required": True},
            "properties": {"type": "object", "required": True}}),
      ]

• truenas_mcp_server/server.py:84-93 (registration)

  The _register_tool_methods method in TrueNASMCPServer that iterates over tool definitions from StorageTools (among others) and registers each tool with the MCP server using FastMCP.tool(), effectively registering 'list_datasets'.

  def _register_tool_methods(self, tool_instance):
      """Register individual tool methods from a tool instance"""
      # Get all methods that should be exposed as MCP tools
      tool_methods = tool_instance.get_tool_definitions()
      
      for method_name, method_func, method_description, method_params in tool_methods:
          # Register with MCP
          self.mcp.tool(name=method_name, description=method_description)(method_func)
          logger.debug(f"Registered tool: {method_name}")