Skip to main content
Glama
vespo92

TrueNAS Core MCP Server

get_pool_status

Retrieve the detailed status of a specific storage pool on the TrueNAS Core MCP Server, enabling efficient monitoring and management of storage resources.

Instructions

Get detailed status of a specific pool

Args:
    pool_name: Name of the pool

Input Schema

TableJSON Schema
NameRequiredDescriptionDefault
pool_nameYes

Implementation Reference

  • The main tool handler function that executes the get_pool_status logic. It retrieves pool information from the TrueNAS API (trying /pool/id/{pool_name} first, falling back to searching all pools), processes detailed topology with vdev and device status/error counts, formats capacity metrics, and returns a structured response.
    @tool_handler
    async def get_pool_status(self, pool_name: str) -> Dict[str, Any]:
        """
        Get detailed status of a specific pool
        
        Args:
            pool_name: Name of the pool
            
        Returns:
            Dictionary containing detailed pool status
        """
        await self.ensure_initialized()
        
        try:
            pool = await self.client.get(f"/pool/id/{pool_name}")
        except Exception:
            # Try getting all pools and finding by name
            pools = await self.client.get("/pool")
            pool = None
            for p in pools:
                if p.get("name") == pool_name:
                    pool = p
                    break
            
            if not pool:
                return {
                    "success": False,
                    "error": f"Pool '{pool_name}' not found"
                }
        
        # Extract detailed information
        size = pool.get("size", 0)
        allocated = pool.get("allocated", 0)
        free = pool.get("free", 0)
        
        # Process topology
        topology = pool.get("topology", {})
        vdev_details = []
        
        for vdev_type in ["data", "cache", "log", "spare"]:
            vdevs = topology.get(vdev_type, [])
            for vdev in vdevs:
                vdev_info = {
                    "type": vdev_type,
                    "name": vdev.get("name"),
                    "status": vdev.get("status"),
                    "devices": []
                }
                for device in vdev.get("children", []):
                    vdev_info["devices"].append({
                        "name": device.get("name"),
                        "status": device.get("status"),
                        "read_errors": device.get("read", 0),
                        "write_errors": device.get("write", 0),
                        "checksum_errors": device.get("checksum", 0)
                    })
                vdev_details.append(vdev_info)
        
        return {
            "success": True,
            "pool": {
                "name": pool.get("name"),
                "id": pool.get("id"),
                "guid": pool.get("guid"),
                "status": pool.get("status"),
                "healthy": pool.get("healthy"),
                "encrypted": pool.get("encrypt", 0) > 0,
                "autotrim": pool.get("autotrim", {}).get("value") if pool.get("autotrim") else None,
                "capacity": {
                    "size": self.format_size(size),
                    "size_bytes": size,
                    "allocated": self.format_size(allocated),
                    "allocated_bytes": allocated,
                    "free": self.format_size(free),
                    "free_bytes": free,
                    "usage_percent": round((allocated / size * 100) if size > 0 else 0, 2),
                    "fragmentation": pool.get("fragmentation")
                },
                "topology": {
                    "vdevs": vdev_details,
                    "summary": {
                        "data_vdevs": len(topology.get("data", [])),
                        "cache_vdevs": len(topology.get("cache", [])),
                        "log_vdevs": len(topology.get("log", [])),
                        "spare_vdevs": len(topology.get("spare", []))
                    }
                },
                "scan": pool.get("scan"),
                "properties": pool.get("properties", {})
            }
        }
  • Local tool registration within StorageTools.get_tool_definitions(). Registers the tool name, references the handler method, provides description, and defines the input schema requiring 'pool_name' as a string.
    ("get_pool_status", self.get_pool_status, "Get detailed status of a specific pool",
     {"pool_name": {"type": "string", "required": True}}),
  • Global MCP tool registration in the server. Calls get_tool_definitions() on tool instances (including StorageTools) and registers each tool with FastMCP using self.mcp.tool().
    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}")
  • Input schema definition for the get_pool_status tool, specifying the required 'pool_name' parameter as a string.
    {"pool_name": {"type": "string", "required": True}}),

Latest Blog Posts

MCP directory API

We provide all the information about MCP servers via our MCP API.

curl -X GET 'https://glama.ai/api/mcp/v1/servers/vespo92/TrueNasCoreMCP'

If you have feedback or need assistance with the MCP directory API, please join our Discord server