get_pod_container_statuses
Retrieve container statuses for a specific Kubernetes pod to monitor health, readiness, and resource allocation.
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.Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| pod_name | Yes | ||
| namespace | No | default |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
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