query_container_stats
Retrieve time-series statistics for container CPU, memory, and network usage by specifying container ID and optional time range parameters.
Instructions
Query statistics for a specific container.
Args: container_id: The container 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 container CPU, memory, and network usage
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| container_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:198-234 (handler)The main asynchronous handler function for the query_container_stats tool, registered via @mcp.tool() decorator. It handles input parameters, authenticates with PocketBase, constructs filters for the specific container and time range, and delegates to PocketBaseClient.query_stats to retrieve paginated time-series statistics from the 'container_stats' collection.
@mcp.tool() async def query_container_stats( container_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 container. Args: container_id: The container 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 container CPU, memory, and network usage """ client = get_client() await ensure_authenticated(client) # Build filter for container and time range filters = [f"container = '{container_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="container_stats", filter=" && ".join(filters), page=page, per_page=per_page, sort="-created", ) - Helper method on PocketBaseClient class invoked by the tool handler. It performs the actual API query to PocketBase by calling get_list with the provided filter, pagination, and sorting parameters on the specified stats collection (e.g., 'container_stats').
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 on PocketBaseClient used by the tool handler to construct PocketBase-compatible filter strings for time-range queries on stats records.
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 ""