Kubernetes Tools MCP Server

Instructions

Implementation Reference

• src/k8stools/k8s_tools.py:133-249 (handler)

  The handler function that executes the get_node_summaries tool, querying the Kubernetes API for node information and returning a list of NodeSummary objects.
  def get_node_summaries() -> list[NodeSummary]: """Return a summary of the nodes for this Kubernetes cluster, similar to that returned by `kubectl get nodes -o wide`. Parameters ---------- None This function does not take any parameters. Returns ------- list of NodeSummary List of node summary objects. Each NodeSummary has the following fields: name : str Name of the node. status : str Status of the node (Ready, NotReady, etc.). roles : list[str] List of roles for the node (e.g., ['control-plane', 'master']). age : datetime.timedelta Age of the node (current time minus creation timestamp). version : str Kubernetes version running on the node. internal_ip : Optional[str] Internal IP address of the node. external_ip : Optional[str] External IP address of the node (if available). os_image : Optional[str] Operating system image running on the node. kernel_version : Optional[str] Kernel version of the node. container_runtime : Optional[str] Container runtime version on the node. Raises ------ K8sConfigError If unable to initialize the K8S API. K8sApiError If the API call to list nodes fails. """ global K8S if K8S is None: K8S = _get_api_client() logging.info(f"get_node_summaries()") try: nodes = K8S.list_node().items except client.ApiException as e: raise K8sApiError(f"Error fetching nodes: {e}") from e current_time_utc = datetime.datetime.now(datetime.timezone.utc) node_summaries: list[NodeSummary] = [] for node in nodes: node_name = node.metadata.name # Determine node status status = "Unknown" if node.status and node.status.conditions: for condition in node.status.conditions: if condition.type == "Ready": status = "Ready" if condition.status == "True" else "NotReady" break # Extract roles from labels roles = [] if node.metadata.labels: for label_key in node.metadata.labels: if label_key.startswith("node-role.kubernetes.io/"): role = label_key.replace("node-role.kubernetes.io/", "") if role: # Skip empty roles roles.append(role) # Also check for older master label elif label_key == "kubernetes.io/role" and node.metadata.labels[label_key]: roles.append(node.metadata.labels[label_key]) if not roles: roles = ["<none>"] # Calculate age age = datetime.timedelta(0) if node.metadata.creation_timestamp: age = current_time_utc - node.metadata.creation_timestamp # Extract version and system info version = node.status.node_info.kubelet_version if node.status and node.status.node_info else "Unknown" os_image = node.status.node_info.os_image if node.status and node.status.node_info else None kernel_version = node.status.node_info.kernel_version if node.status and node.status.node_info else None container_runtime = node.status.node_info.container_runtime_version if node.status and node.status.node_info else None # Extract IP addresses internal_ip = None external_ip = None if node.status and node.status.addresses: for address in node.status.addresses: if address.type == "InternalIP": internal_ip = address.address elif address.type == "ExternalIP": external_ip = address.address node_summary = NodeSummary( name=node_name, status=status, roles=roles, age=age, version=version, internal_ip=internal_ip, external_ip=external_ip, os_image=os_image, kernel_version=kernel_version, container_runtime=container_runtime ) node_summaries.append(node_summary) return node_summaries
• src/k8stools/k8s_tools.py:120-132 (schema)

  Pydantic BaseModel defining the structure of each node summary returned by the tool.
  class NodeSummary(BaseModel): """A summary of a node's status like returned by `kubectl get nodes -o wide`""" name: str status: str roles: list[str] age: datetime.timedelta version: str internal_ip: Optional[str] = None external_ip: Optional[str] = None os_image: Optional[str] = None kernel_version: Optional[str] = None container_runtime: Optional[str] = None
• src/k8stools/mcp_server.py:46-52 (registration)

  Registers the get_node_summaries tool (as part of TOOLS list) by wrapping functions with Tool.from_function and adding to FastMCP server instance.
  if not args.mock: from .k8s_tools import TOOLS else: from .mock_tools import TOOLS logging.warning(f"Using mock versions of the tools") wrapped_tools = [get_tool_for_function(fn) for fn in TOOLS]
• src/k8stools/k8s_tools.py:1101-1111 (registration)

  List of tools including get_node_summaries, imported by mcp_server.py for registration.
  TOOLS = [ get_namespaces, get_node_summaries, get_pod_summaries, get_pod_container_statuses, get_pod_events, get_pod_spec, get_logs_for_pod_and_container, get_deployment_summaries, get_service_summaries ]
• src/k8stools/mock_tools.py:194-198 (handler)

  Mock implementation of get_node_summaries for testing without a real Kubernetes cluster.
  def get_node_summaries() -> list[k8s_tools.NodeSummary]: """Mock implementation that returns static node data""" return _MOCK_DATA['nodes'] get_node_summaries.__doc__ = k8s_tools.get_node_summaries.__doc__