Kubernetes Tools MCP Server

Instructions

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.

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__