Kubernetes Tools MCP Server

Instructions

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.

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)