Skip to main content
Glama
BenedatLLC

Kubernetes Tools MCP Server

by BenedatLLC

get_service_summaries

Retrieve service summaries from Kubernetes clusters to monitor network endpoints, including cluster IPs, external IPs, and exposed ports across namespaces.

Instructions

Retrieves a list of ServiceSummary objects for services in a given namespace or all namespaces. Similar to kubectl get services.

Parameters
----------
namespace : Optional[str], default=None
    The specific namespace to list services from. If None, lists services from all namespaces.

Returns
-------
list of ServiceSummary
    A list of ServiceSummary objects, each providing a summary of a service's status with the following fields:

    name : str
        Name of the service.
    namespace : str
        Namespace in which the service is running.
    type : str
        Type of the service (ClusterIP, NodePort, LoadBalancer, ExternalName).
    cluster_ip : Optional[str]
        Cluster IP address assigned to the service (None for ExternalName services).
    external_ip : Optional[str]
        External IP address if applicable (for LoadBalancer services).
    ports : list[PortInfo]
        List of ports (and their protocols) exposed by the service.
    age : datetime.timedelta
        Age of the service (current time minus creation timestamp).

Raises
------
K8sConfigError
    If unable to initialize the K8S API.
K8sApiError
    If the API call to list services fails.

Input Schema

TableJSON Schema
NameRequiredDescriptionDefault
namespaceNo

Implementation Reference

  • The core handler function for the 'get_service_summaries' tool. It queries the Kubernetes API for services in the specified namespace (or all), constructs ServiceSummary objects, and returns the list. Uses global K8S client and handles exceptions.
    def get_service_summaries(namespace: Optional[str] = None) -> list[ServiceSummary]:
        """Retrieves a list of ServiceSummary objects for services in a given namespace or all namespaces.
        Similar to `kubectl get services`.
    
        Parameters
        ----------
        namespace : Optional[str], default=None
            The specific namespace to list services from. If None, lists services from all namespaces.
    
        Returns
        -------
        list of ServiceSummary
            A list of ServiceSummary objects, each providing a summary of a service's status with the following fields:
    
            name : str
                Name of the service.
            namespace : str
                Namespace in which the service is running.
            type : str
                Type of the service (ClusterIP, NodePort, LoadBalancer, ExternalName).
            cluster_ip : Optional[str]
                Cluster IP address assigned to the service (None for ExternalName services).
            external_ip : Optional[str]
                External IP address if applicable (for LoadBalancer services).
            ports : list[PortInfo]
                List of ports (and their protocols) exposed by the service.
            age : datetime.timedelta
                Age of the service (current time minus creation timestamp).
    
        Raises
        ------
        K8sConfigError
            If unable to initialize the K8S API.
        K8sApiError
            If the API call to list services fails.
        """
        global K8S
        
        # Load Kubernetes configuration and initialize client only once
        if K8S is None:
            K8S = _get_api_client()
    
        logging.info(f"get_service_summaries(namespace={namespace})")
        service_summaries: list[ServiceSummary] = []
        
        try:
            if namespace:
                # List services in a specific namespace
                services = K8S.list_namespaced_service(namespace=namespace).items
            else:
                # List services across all namespaces
                services = K8S.list_service_for_all_namespaces().items
        except client.ApiException as e:
            raise K8sApiError(f"Error fetching services: {e}") from e
    
        current_time_utc = datetime.datetime.now(datetime.timezone.utc)
    
        for service in services:
            service_name = service.metadata.name
            service_namespace = service.metadata.namespace
            service_type = service.spec.type if service.spec.type else "ClusterIP"
            
            # Get cluster IP (None for ExternalName services)
            cluster_ip = service.spec.cluster_ip if service.spec.cluster_ip != "None" else None
            
            # Get external IP for LoadBalancer services
            external_ip = None
            if service.status and service.status.load_balancer and service.status.load_balancer.ingress:
                # Take the first ingress IP or hostname
                ingress = service.status.load_balancer.ingress[0]
                external_ip = ingress.ip or ingress.hostname
            
            # Extract port information
            ports = []
            if service.spec.ports:
                for port in service.spec.ports:
                    ports.append(PortInfo(
                        port=port.port,
                        protocol=port.protocol if port.protocol else "TCP"
                    ))
            
            # Calculate age
            age = datetime.timedelta(0)  # Default to 0 if creation_timestamp is missing
            if service.metadata.creation_timestamp:
                age = current_time_utc - service.metadata.creation_timestamp
    
            service_summary = ServiceSummary(
                name=service_name,
                namespace=service_namespace,
                type=service_type,
                cluster_ip=cluster_ip,
                external_ip=external_ip,
                ports=ports,
                age=age
            )
            service_summaries.append(service_summary)
        
        return service_summaries
  • Pydantic BaseModel classes defining the structure of the output: PortInfo for service ports and ServiceSummary for each service summary returned by the tool.
    class PortInfo(BaseModel):
        """A representation of a port, to be used in various specs."""
        port: int
        protocol: str
    
    class ServiceSummary(BaseModel):
        """A summary of a service's status like returned by `kubectl get servicess`"""
        name: str
        namespace: str
        type: str
        cluster_ip: Optional[str] = None
        external_ip: Optional[str] = None
        ports: list[PortInfo]
        age: datetime.timedelta
  • The TOOLS list in k8s_tools.py registers get_service_summaries among other tools. This list is imported by mcp_server.py and each function is wrapped into an MCP Tool using Tool.from_function.
    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
    ]
  • In the MCP server entrypoint, the TOOLS list is imported and each tool function (including get_service_summaries) is converted to an MCP Tool object using Tool.from_function, which uses the function name as the tool name.
        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]
  • Helper function that calls get_service_summaries and formats the output like 'kubectl get services' for printing to stdout.
    def print_service_summaries(namespace: Optional[str] = None) -> None:
        """
        Calls get_service_summaries and prints the output to stdout, using
        the same format as `kubectl get services`.
        """
        service_summaries = get_service_summaries(namespace)
        print(f"{'NAME':<32} {'NAMESPACE':<20} {'TYPE':<15} {'CLUSTER-IP':<16} {'EXTERNAL-IP':<16} {'PORT(S)':<20} {'AGE':<12}")
        for service in service_summaries:
            service_type = service.type
            cluster_ip = service.cluster_ip if service.cluster_ip else "<none>"
            external_ip = service.external_ip if service.external_ip else "<none>"
            
            # Format ports as "port/protocol,port/protocol"
            ports_str = ",".join([f"{port.port}/{port.protocol}" for port in service.ports]) if service.ports else "<none>"
            
            age = _format_timedelta(service.age)
            print(f"{service.name:<32} {service.namespace:<20} {service_type:<15} {cluster_ip:<16} {external_ip:<16} {age:<12} {ports_str:<20}")

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