Kubernetes Tools MCP Server

Instructions

Get the status for all containers in a specified Kubernetes pod.

Parameters
----------
pod_name : str
    Name of the pod to retrieve container statuses for.
namespace : str, optional
    Namespace of the pod (default is "default").

Returns
-------
list of ContainerStatus
    List of container status objects for the specified pod. Each ContainerStatus has the following fields:

    pod_name : str
        Name of the pod.
    namespace : str
        Namespace of the pod.
    container_name : str
        Name of the container.
    image : str
        Image name.
    ready : bool
        Whether the container is currently passing its readiness check.
        The value will change as readiness probes keep executing.
    restart_count : int
        Number of times the container has restarted.
    started : Optional[bool]
        Started indicates whether the container has finished its postStart
        lifecycle hook and passed its startup probe.
    stop_signal : Optional[str]
        Stop signal for the container.
    state : Optional[ContainerState]
        Current state of the container.
    last_state : Optional[ContainerState]
        Last state of the container.
    volume_mounts : list[VolumeMountStatus]
        Status of volume mounts for the container
    resource_requests : dict[str, str]
        Describes the minimum amount of compute resources required. If Requests
        is omitted for a container, it defaults to Limits if that is explicitly specified,
        otherwise to an implementation-defined value. Requests cannot exceed Limits. 
    resource_limits : dict[str, str]
        Describes the maximum amount of compute resources allowed.
    allocated_resources : dict[str, str]
        Compute resources allocated for this container by the node.

Raises
------
K8sConfigError
    If unable to initialize the K8S API.
K8sApiError
    If the API call to read the pod fails.

Implementation Reference

• src/k8stools/k8s_tools.py:588-696 (handler)

  Core handler function that executes the tool: fetches pod via Kubernetes CoreV1Api, processes each container_status into structured ContainerStatus using helper converters, handles resources, and returns the list.

  def get_pod_container_statuses(pod_name: str, namespace: str = "default") -> list[ContainerStatus]:
      """
      Get the status for all containers in a specified Kubernetes pod.
  
      Parameters
      ----------
      pod_name : str
          Name of the pod to retrieve container statuses for.
      namespace : str, optional
          Namespace of the pod (default is "default").
  
      Returns
      -------
      list of ContainerStatus
          List of container status objects for the specified pod. Each ContainerStatus has the following fields:
  
          pod_name : str
              Name of the pod.
          namespace : str
              Namespace of the pod.
          container_name : str
              Name of the container.
          image : str
              Image name.
          ready : bool
              Whether the container is currently passing its readiness check.
              The value will change as readiness probes keep executing.
          restart_count : int
              Number of times the container has restarted.
          started : Optional[bool]
              Started indicates whether the container has finished its postStart
              lifecycle hook and passed its startup probe.
          stop_signal : Optional[str]
              Stop signal for the container.
          state : Optional[ContainerState]
              Current state of the container.
          last_state : Optional[ContainerState]
              Last state of the container.
          volume_mounts : list[VolumeMountStatus]
              Status of volume mounts for the container
          resource_requests : dict[str, str]
              Describes the minimum amount of compute resources required. If Requests
              is omitted for a container, it defaults to Limits if that is explicitly specified,
              otherwise to an implementation-defined value. Requests cannot exceed Limits. 
          resource_limits : dict[str, str]
              Describes the maximum amount of compute resources allowed.
          allocated_resources : dict[str, str]
              Compute resources allocated for this container by the node.
  
      Raises
      ------
      K8sConfigError
          If unable to initialize the K8S API.
      K8sApiError
          If the API call to read the pod fails.
      """   
      global K8S
      if K8S is None:
          K8S = _get_api_client()
      logging.info(f"get_pod_container_statuses(pod_name={pod_name}, namespace={namespace})")
      pod = K8S.read_namespaced_pod(name=pod_name, namespace=namespace)
      # Only proceed if pod is a V1Pod instance
      if not isinstance(pod, client.V1Pod):
          raise K8sApiError(f"Unexpected type for pod: {type(pod)}")
      result:list[ContainerStatus] = []
      if not pod.status or not pod.status.container_statuses:
          return result
      for container_status in pod.status.container_statuses:
          container_name = container_status.name
          image = container_status.image
          ready = container_status.ready
          restart_count = container_status.restart_count
          started = container_status.started
          stop_signal = container_status.stop_signal
          state = _v1_container_state_to_container_state(container_status.state) \
                  if container_status.state is not None else None
          last_state = _v1_container_state_to_container_state(container_status.last_state) \
                  if container_status.last_state is not None else None
          volume_mounts = [_v1_volume_mount_status_to_mount_statis(volume_mount)
                           for volume_mount in container_status.volume_mounts] \
                  if container_status.volume_mounts is not None else []
          if container_status.resources:
              resource_requests = container_status.resources.requests \
                                  if container_status.resources.requests is not None else {}
              resource_limits = container_status.resources.limits \
                                if container_status.resources.limits is not None else {}
          else:
              resource_requests = {}
              resource_limits = {}
          allocated_resources = container_status.allocated_resources \
                                if container_status.allocated_resources is not None else {}
  
          result.append(ContainerStatus(
              pod_name=pod_name,
              namespace=namespace,
              container_name=container_name,
              image=image,
              ready=ready,
              restart_count=restart_count,
              started=started,
              stop_signal=stop_signal,
              state=state,
              last_state=last_state,
              volume_mounts=volume_mounts,
              resource_requests=resource_requests,
              resource_limits=resource_limits,
              allocated_resources=allocated_resources
          ))
      return result

• src/k8stools/k8s_tools.py:568-586 (schema)

  Pydantic model defining the output type list[ContainerStatus] with fields matching Kubernetes V1ContainerStatus, including nested ContainerState union and VolumeMountStatus.

  class ContainerStatus(BaseModel):
      """Provides information about a container running in a specific pod. This corresponds to
      kubernetes.client.models.v1_container_tatus.V1ContainerStatus.
      """
      pod_name: str
      namespace: str
      container_name: str
      image: str
      ready: bool
      restart_count: int
      started: Optional[bool]
      stop_signal: Optional[str]
      state: Optional[ContainerState]
      last_state: Optional[ContainerState]
      volume_mounts: list[VolumeMountStatus]
      resource_requests: dict[str, str]
      resource_limits: dict[str, str]
      allocated_resources: dict[str, str]

• src/k8stools/k8s_tools.py:515-536 (schema)

  Supporting Pydantic models for ContainerState variants (Running/Waiting/Terminated) used in ContainerStatus.state and .last_state.

  class ContainerStateRunning(BaseModel):
      state_name: Literal['Running'] = 'Running'
      started_at: datetime.datetime
  
  # see kubernetes.client.models.v1_container_state_waiting.V1ContainerStateWaiting
  class ContainerStateWaiting(BaseModel):
      state_name: Literal['Waiting'] = 'Waiting'
      reason: str
      message: Optional[str] = None
  
  # see kubernetes.client.models.v1_container_state_terminated.V1ContainerStateTerminated
  class ContainerStateTerminated(BaseModel):
      state_name: Literal['Terminated'] = 'Terminated'
      exit_code: Optional[int] = None
      finished_at: Optional[datetime.datetime] = None
      reason: Optional[str] = None
      message: Optional[str] = None
      started_at: Optional[datetime.datetime] = None
  
  # see kubernetes.client.models.v1_container_state.V1ContainerState
  ContainerState = Union[ContainerStateRunning, ContainerStateWaiting, ContainerStateTerminated]

• src/k8stools/k8s_tools.py:1101-1111 (registration)

  The TOOLS list registers get_pod_container_statuses alongside other functions; this list is imported by mcp_server.py.

  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-52 (registration)

  MCP server imports TOOLS from k8s_tools.py (real impl) or mock_tools.py, wraps each into FastMCP Tool (with structured_output=True), and passes 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]

• src/k8stools/k8s_tools.py:537-553 (helper)

  Helper to convert Kubernetes client V1ContainerState to custom Pydantic ContainerState union; used in handler for both current state and last_state.

  def _v1_container_state_to_container_state(container_state:client.V1ContainerState) -> Optional[ContainerState]:
      if container_state.running:
          return ContainerStateRunning(started_at=container_state.running.started_at)
      elif container_state.waiting:
          return ContainerStateWaiting(reason=container_state.waiting.reason,
                                       message=container_state.waiting.message)
      elif container_state.terminated:
          cst = container_state.terminated
          return ContainerStateTerminated(exit_code=cst.exit_code,
                                          reason=cst.reason,
                                          finished_at=cst.finished_at,
                                          message=cst.message,
                                          started_at=cst.started_at)
      else:
          # All states are None - this is valid (e.g., for last_state when container has never been in a previous state)
          return None