Skip to main content
Glama
BenedatLLC

Kubernetes Tools MCP Server

by BenedatLLC

get_node_summaries

Retrieve node summaries for a Kubernetes cluster, including status, roles, IP addresses, and system details to monitor cluster health and configuration.

Instructions

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.

Input Schema

TableJSON Schema
NameRequiredDescriptionDefault

No arguments

Implementation Reference

  • 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
  • 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
  • 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]
  • 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
    ]
  • 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__

Latest Blog Posts

MCP directory API

We provide all the information about MCP servers via our MCP API.

curl -X GET 'https://glama.ai/api/mcp/v1/servers/BenedatLLC/k8stools'

If you have feedback or need assistance with the MCP directory API, please join our Discord server