Kubernetes Tools MCP Server

Instructions

Implementation Reference

• src/k8stools/k8s_tools.py:296-406 (handler)

  The handler function that implements the logic for the 'get_pod_summaries' tool. It initializes the Kubernetes client if necessary, lists pods (namespaced or all), processes each pod to extract summary information, and returns a list of PodSummary objects.
  def get_pod_summaries(namespace: Optional[str] = None) -> list[PodSummary]: """ Retrieves a list of PodSummary objects for pods in a given namespace or all namespaces. Parameters ---------- namespace : Optional[str], default=None The specific namespace to list pods from. If None, lists pods from all namespaces. Returns ------- list of PodSummary A list of PodSummary objects, each providing a summary of a pod's status with the following fields: name : str Name of the pod. namespace : str Namespace in which the pod is running. total_containers : int Total number of containers in the pod. ready_containers : int Number of containers currently in ready state. restarts : int Total number of restarts for all containers in the pod. last_restart : Optional[datetime.timedelta] Time since the container last restart (None if never restarted). age : datetime.timedelta Age of the pod (current time minus creation timestamp). ip : Optional[str] Pod IP address (None if not assigned). node : Optional[str] Name of the node where the pod is running (None if not scheduled). Raises ------ K8sConfigError If unable to initialize the K8S API. K8sApiError If the API call to list pods fails. """ global K8S # Load Kubernetes configuration and initialize client only once if K8S is None: K8S = _get_api_client() logging.info(f"get_pod_summaries(namespace={namespace})") pod_summaries: list[PodSummary] = [] try: if namespace: # List pods in a specific namespace pods = K8S.list_namespaced_pod(namespace=namespace).items else: # List pods across all namespaces pods = K8S.list_pod_for_all_namespaces().items except client.ApiException as e: raise K8sApiError(f"Error fetching pods: {e}") from e current_time_utc = datetime.datetime.now(datetime.timezone.utc) for pod in pods: pod_name = pod.metadata.name pod_namespace = pod.metadata.namespace total_containers = len(pod.spec.containers) ready_containers = 0 total_restarts = 0 latest_restart_time: Optional[datetime.datetime] = None if pod.status and pod.status.container_statuses: for container_status in pod.status.container_statuses: if container_status.ready: ready_containers += 1 total_restarts += container_status.restart_count # Check for last restart time if container_status.last_state and container_status.last_state.terminated: terminated_at = container_status.last_state.terminated.finished_at if terminated_at: if latest_restart_time is None or terminated_at > latest_restart_time: latest_restart_time = terminated_at # Calculate age age = datetime.timedelta(0) # Default to 0 if creation_timestamp is missing if pod.metadata.creation_timestamp: age = current_time_utc - pod.metadata.creation_timestamp # Calculate last_restart timedelta if a latest_restart_time was found last_restart_timedelta: Optional[datetime.timedelta] = None if latest_restart_time: last_restart_timedelta = current_time_utc - latest_restart_time # Extract IP and node information pod_ip = pod.status.pod_ip if pod.status and pod.status.pod_ip else None node_name = pod.spec.node_name if pod.spec and pod.spec.node_name else None pod_summary = PodSummary( name=pod_name, namespace=pod_namespace, total_containers=total_containers, ready_containers=ready_containers, restarts=total_restarts, last_restart=last_restart_timedelta, age=age, ip=pod_ip, node=node_name ) pod_summaries.append(pod_summary) return pod_summaries
• src/k8stools/k8s_tools.py:282-293 (schema)

  Pydantic model defining the output schema for PodSummary objects returned by get_pod_summaries.
  class PodSummary(BaseModel): """A summary of a pod's status like returned by `kubectl get pods -o wide`""" name: str namespace: str total_containers: int ready_containers: int restarts: int last_restart: Optional[datetime.timedelta] age: datetime.timedelta ip: Optional[str] = None node: Optional[str] = None
• src/k8stools/k8s_tools.py:1101-1111 (registration)

  List of tool handler functions including get_pod_summaries, imported by mcp_server.py and wrapped into MCP Tool objects.
  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:46-66 (registration)

  MCP server registration where TOOLS from k8s_tools.py are wrapped using Tool.from_function and passed to FastMCP server instance, registering get_pod_summaries as an MCP tool.
  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] mcp = FastMCP( name="k8stools-"+args.transport, tools=wrapped_tools, streamable_http_path="/mcp", stateless_http=(args.transport == 'streamable-http'), host=args.host, port=args.port, log_level=args.log_level, debug=args.debug ) logging.debug(f"Settings are: {mcp.settings}") logging.info(f"Starting with {len(wrapped_tools)} tools on transport {args.transport}") # this starts the uvicorn server mcp.run(transport=args.transport)