get_node_info

Instructions

Implementation Reference

• src/cloudnet_mcp/server.py:1-55 (helper)

  CloudNetClient class that handles HTTP communication with the CloudNet API, including authentication and request retry logic. Used by all tool handlers including get_node_info.

  import asyncio
  import os
  import httpx
  from pydantic import BaseModel, Field
  from typing import Optional, Dict, Any, List
  import mcp.types as types
  from mcp.server import Server
  from mcp.server.stdio import stdio_server
  
  app = Server("cloudnet-mcp")
  
  CLOUDNET_URL = os.environ.get("CLOUDNET_URL", "http://127.0.0.1:2812/api/v3")
  CLOUDNET_USER = os.environ.get("CLOUDNET_USER", "admin")
  CLOUDNET_PASSWORD = os.environ.get("CLOUDNET_PASSWORD", "admin")
  
  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()

• src/cloudnet_mcp/server.py:60-81 (registration)

  Registration of the get_node_info tool through the @app.list_tools() decorator. Defines the tool's name, description, and inputSchema requiring a 'node_id' string parameter.

  @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": {},
              },
          ),
          types.Tool(
              name="get_node_info",
              description="Get detailed information about a specific node",
              inputSchema={
                  "type": "object",
                  "properties": {
                      "node_id": {"type": "string", "description": "The ID of the node"}
                  },
                  "required": ["node_id"],
              },
          ),

• src/cloudnet_mcp/server.py:164-179 (handler)

  Handler for get_node_info tool in the @app.call_tool() decorated function. Extracts 'node_id' from arguments, validates it, and makes a GET request to 'node/{node_id}' via CloudNetClient, returning the response as text.

  @app.call_tool()
  async def call_tool(
      name: str, arguments: dict[str, Any] | None
  ) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
      if arguments is None:
          arguments = {}
  
      if name == "get_nodes":
          data = await client.request("GET", "node")
          return [types.TextContent(type="text", text=str(data))]
      elif name == "get_node_info":
          node_id = arguments.get("node_id")
          if not node_id:
              raise ValueError("node_id is required")
          data = await client.request("GET", f"node/{node_id}")
          return [types.TextContent(type="text", text=str(data))]