query_system_stats
Retrieve time-series statistics for CPU, memory, disk, and network usage from a specified system by providing its ID and optional time range.
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 |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- src/beszel_mcp/server.py:159-196 (handler)The main async handler function implementing the query_system_stats tool logic, including registration via @mcp.tool(), input parameters defining the schema, docstring description, client authentication, filter building, and querying of system_stats collection.
@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", ) - PocketBaseClient helper method invoked by the tool handler to perform the actual paginated query on the stats collection using PocketBase API.
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, ) - PocketBaseClient helper method used by the tool handler to construct time range filters for the query.
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 ""