Kubernetes Tools MCP Server

Instructions

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