Beszel MCP Server

Instructions

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",
      )

• src/beszel_mcp/pocketbase_client.py:131-157 (helper)

  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,
      )

• src/beszel_mcp/pocketbase_client.py:163-187 (helper)

  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 ""