Kubernetes Tools MCP Server

Instructions

Implementation Reference

• src/k8stools/k8s_tools.py:865-947 (handler)

  Main tool handler: Queries Kubernetes AppsV1Api for deployments (namespaced or all), extracts summary fields like replicas and age, constructs DeploymentSummary pydantic models, returns list. Handles API init and errors.
  def get_deployment_summaries(namespace: Optional[str] = None) -> list[DeploymentSummary]: """ Retrieves a list of DeploymentSummary objects for deployments in a given namespace or all namespaces. Similar to `kubectl get deployements`. Parameters ---------- namespace : Optional[str], default=None The specific namespace to list deployments from. If None, lists deployments from all namespaces. Returns ------- list of DeploymentSummary A list of DeploymentSummary objects, each providing a summary of a deployment's status with the following fields: name : str Name of the deployment. namespace : str Namespace in which the deployment is running. total_replicas : int Total number of replicas desired for this deployment. ready_replicas : int Number of replicas that are currently ready. up_to_date_replicas : int Number of replicas that are up to date. available_replicas : int Number of replicas that are available. age : datetime.timedelta Age of the deployment (current time minus creation timestamp). Raises ------ K8sConfigError If unable to initialize the K8S API. K8sApiError If the API call to list deployments fails. """ global APPS_V1_API # Load Kubernetes configuration and initialize client only once if APPS_V1_API is None: APPS_V1_API = _get_apps_v1_api_client() logging.info(f"get_deployment_summaries(namespace={namespace})") deployment_summaries: list[DeploymentSummary] = [] try: if namespace: deployments = APPS_V1_API.list_namespaced_deployment(namespace=namespace) else: deployments = APPS_V1_API.list_deployment_for_all_namespaces() except client.ApiException as e: raise K8sApiError(f"Error fetching deployments: {e}") from e current_time_utc = datetime.datetime.now(datetime.timezone.utc) for deployment in deployments.items: deployment_name = deployment.metadata.name deployment_namespace = deployment.metadata.namespace # Extract replica counts from deployment status total_replicas = deployment.spec.replicas if deployment.spec.replicas is not None else 0 ready_replicas = deployment.status.ready_replicas if deployment.status.ready_replicas is not None else 0 up_to_date_replicas = deployment.status.updated_replicas if deployment.status.updated_replicas is not None else 0 available_replicas = deployment.status.available_replicas if deployment.status.available_replicas is not None else 0 # Calculate age age = datetime.timedelta(0) # Default to 0 if creation_timestamp is missing if deployment.metadata.creation_timestamp: age = current_time_utc - deployment.metadata.creation_timestamp deployment_summary = DeploymentSummary( name=deployment_name, namespace=deployment_namespace, total_replicas=total_replicas, ready_replicas=ready_replicas, up_to_date_relicas=up_to_date_replicas, available_replicas=available_replicas, age=age ) deployment_summaries.append(deployment_summary) return deployment_summaries
• src/k8stools/k8s_tools.py:855-863 (schema)

  Pydantic model defining the output schema for deployment summaries, matching kubectl get deployments output fields.
  class DeploymentSummary(BaseModel): """A summary of a deployment's status like returned by `kubectl get deployments`""" name: str namespace: str total_replicas: int ready_replicas: int up_to_date_relicas: int available_replicas: int age: datetime.timedelta
• src/k8stools/k8s_tools.py:1101-1111 (registration)

  Exported TOOLS list including get_deployment_summaries, imported by mcp_server for tool registration.
  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)

  Server code that imports TOOLS list and wraps each function (including get_deployment_summaries) into FastMCP Tool objects using Tool.from_function with structured_output=True.
  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/mock_tools.py:326-335 (helper)

  Mock implementation of the tool for testing, returns static data filtered by namespace, copies docstring from real handler.
  def get_deployment_summaries(namespace: Optional[str] = None) -> list[k8s_tools.DeploymentSummary]: """Mock implementation that returns static deployment data, filtered by namespace if specified""" deployments = _MOCK_DATA['deployments'] if namespace is not None: return [deployment for deployment in deployments if deployment.namespace == namespace] return deployments get_deployment_summaries.__doc__ = k8s_tools.get_deployment_summaries.__doc__