Beszel MCP Server

Instructions

Implementation Reference

• src/beszel_mcp/server.py:69-96 (handler)

  The @mcp.tool()-decorated async handler function implementing the list_containers tool. It fetches a paginated list of containers from the Beszel PocketBase 'containers' collection using the PocketBaseClient.

  @mcp.tool()
  async def list_containers(
      page: int = 1,
      per_page: int = 50,
      filter: Optional[str] = None,
      sort: Optional[str] = None,
  ) -> dict:
      """List all monitored containers in Beszel.
      
      Args:
          page: Page number (default: 1)
          per_page: Number of results per page (default: 50)
          filter: PocketBase filter string
          sort: Sort order
      
      Returns:
          Dictionary containing paginated list of containers running on monitored systems
      """
      client = get_client()
      await ensure_authenticated(client)
      
      return await client.get_list(
          collection="containers",
          page=page,
          per_page=per_page,
          filter=filter,
          sort=sort,
      )

• src/beszel_mcp/pocketbase_client.py:51-97 (helper)

  The PocketBaseClient.get_list method called by the list_containers handler to retrieve paginated records from the PocketBase API.

  async def get_list(
      self,
      collection: str,
      page: int = 1,
      per_page: int = 50,
      filter: Optional[str] = None,
      sort: Optional[str] = None,
      expand: Optional[str] = None,
  ) -> dict[str, Any]:
      """Get a paginated list of records from a collection.
      
      Args:
          collection: The collection name
          page: Page number (default: 1)
          per_page: Number of records per page (default: 50)
          filter: PocketBase filter string
          sort: Sort order (e.g., "-created")
          expand: Fields to expand (e.g., "relField1,relField2")
          
      Returns:
          Dictionary containing paginated results
      """
      params = {
          "page": page,
          "perPage": per_page,
      }
      
      if filter:
          params["filter"] = filter
      if sort:
          params["sort"] = sort
      if expand:
          params["expand"] = expand
  
      try:
          response = await self.client.get(
              f"{self.base_url}/api/collections/{collection}/records",
              params=params,
              headers=self._get_headers(),
          )
          response.raise_for_status()
          return response.json()
      except httpx.HTTPStatusError as e:
          raise Exception(f"Failed to get records from {collection}: {e.response.text}")
      except Exception as e:
          raise Exception(f"Failed to get records from {collection}: {e}")

• src/beszel_mcp/server.py:16-30 (helper)

  Utility function to get or initialize the global PocketBaseClient instance used in the list_containers handler.

  def get_client() -> PocketBaseClient:
      """Get or create the PocketBase client."""
      global pb_client
      
      if pb_client is None:
          base_url = os.environ.get("BESZEL_URL")
          if not base_url:
              raise ValueError("BESZEL_URL environment variable is required")
          
          email = os.environ.get("BESZEL_EMAIL")
          password = os.environ.get("BESZEL_PASSWORD")
          
          pb_client = PocketBaseClient(base_url, email, password)
      
      return pb_client

• src/beszel_mcp/server.py:33-37 (helper)

  Utility function to ensure the PocketBase client is authenticated before use in list_containers.

  async def ensure_authenticated(client: PocketBaseClient) -> None:
      """Ensure the client is authenticated."""
      if client.email and client.password and not client.token:
          await client.authenticate()