get_nodes
Retrieve a list of all nodes in the CloudNet cluster to monitor cluster health and node status.
Instructions
List all nodes in the CloudNet cluster
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- src/cloudnet_mcp/server.py:60-70 (registration)Registers the 'get_nodes' tool with its name, description, and empty inputSchema (no parameters) in the list_tools handler.
@app.list_tools() async def list_tools() -> list[types.Tool]: return [ types.Tool( name="get_nodes", description="List all nodes in the CloudNet cluster", inputSchema={ "type": "object", "properties": {}, }, ), - src/cloudnet_mcp/server.py:171-173 (handler)Handler for 'get_nodes': makes a GET request to 'node' endpoint on the CloudNet API and returns the result as text.
if name == "get_nodes": data = await client.request("GET", "node") return [types.TextContent(type="text", text=str(data))] - src/cloudnet_mcp/server.py:16-55 (helper)CloudNetClient class: an async HTTP client helper with auto-authentication, used by all tool handlers including get_nodes.
class CloudNetClient: def __init__(self, base_url: str, user: str, password: str): self.base_url = base_url.rstrip("/") self.user = user self.password = password self.token = None self.client = httpx.AsyncClient() async def _authenticate(self): resp = await self.client.post( f"{self.base_url}/auth", auth=(self.user, self.password) ) resp.raise_for_status() data = resp.json() self.token = data.get("token") self.client.headers.update({"Authorization": f"Bearer {self.token}"}) async def request(self, method: str, endpoint: str, **kwargs): if not self.token: await self._authenticate() path = endpoint.lstrip("/") url = f"{self.base_url}/{path}" try: resp = await self.client.request(method, url, **kwargs) if resp.status_code == 401: # Token might be expired, re-authenticate and retry await self._authenticate() resp = await self.client.request(method, url, **kwargs) resp.raise_for_status() if resp.status_code == 204: return {"status": "success"} return resp.json() except httpx.HTTPError as e: return {"error": str(e)} async def close(self): await self.client.aclose()