query_system_stats
Retrieve system performance metrics including CPU, memory, disk, and network usage statistics for monitoring and analysis purposes.
Instructions
Query statistics for a specific system.
Args: system_id: The system ID to query statistics for start_time: Start time in ISO 8601 format (e.g., '2024-01-01T00:00:00Z') end_time: End time in ISO 8601 format page: Page number (default: 1) per_page: Number of results per page (default: 100)
Returns: Dictionary containing time-series data for CPU, memory, disk, and network usage
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| system_id | Yes | ||
| start_time | No | ||
| end_time | No | ||
| page | No | ||
| per_page | No |
Implementation Reference
- src/beszel_mcp/server.py:159-195 (handler)The main execution logic for the 'query_system_stats' tool. This async function handles input parameters, authenticates the client, constructs a PocketBase filter for the system ID and time range, and queries the 'system_stats' collection. The @mcp.tool() decorator also registers it with the FastMCP server.@mcp.tool() async def query_system_stats( system_id: str, start_time: Optional[str] = None, end_time: Optional[str] = None, page: int = 1, per_page: int = 100, ) -> dict: """Query statistics for a specific system. Args: system_id: The system ID to query statistics for start_time: Start time in ISO 8601 format (e.g., '2024-01-01T00:00:00Z') end_time: End time in ISO 8601 format page: Page number (default: 1) per_page: Number of results per page (default: 100) Returns: Dictionary containing time-series data for CPU, memory, disk, and network usage """ client = get_client() await ensure_authenticated(client) # Build filter for system and time range filters = [f"system = '{system_id}'"] time_filter = client.build_time_filter("created", start_time, end_time) if time_filter: filters.append(time_filter) return await client.query_stats( collection="system_stats", filter=" && ".join(filters), page=page, per_page=per_page, sort="-created", )
- Supporting method on PocketBaseClient that delegates to get_list for querying statistics collections with custom filters, used directly by the tool handler.async def query_stats( self, collection: str, filter: str, page: int = 1, per_page: int = 100, sort: str = "-created", ) -> dict[str, Any]: """Query statistics records with filtering. Args: collection: The stats collection name (system_stats or container_stats) filter: PocketBase filter string page: Page number per_page: Number of records per page sort: Sort order Returns: Dictionary containing paginated statistics """ return await self.get_list( collection=collection, page=page, per_page=per_page, filter=filter, sort=sort, )
- Utility method to construct PocketBase filter strings for time ranges, invoked in the tool handler to filter stats by start_time and end_time.def build_time_filter( self, field: str, start_time: Optional[str] = None, end_time: Optional[str] = None, ) -> str: """Build a time-based filter string. Args: field: The field name (e.g., "created") start_time: Start time in ISO 8601 format end_time: End time in ISO 8601 format Returns: PocketBase filter string """ filters = [] if start_time: filters.append(f"{field} >= '{start_time}'") if end_time: filters.append(f"{field} <= '{end_time}'") return " && ".join(filters) if filters else ""