Kubernetes Tools MCP Server

Instructions

Implementation Reference

• src/k8stools/k8s_tools.py:980-1077 (handler)

  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
• src/k8stools/k8s_tools.py:965-979 (schema)

  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
• src/k8stools/k8s_tools.py:1101-1111 (registration)

  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 ]
• src/k8stools/mcp_server.py:47-51 (registration)

  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]
• src/k8stools/k8s_tools.py:1080-1097 (helper)

  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}")