Kubernetes Tools MCP Server

Instructions

Retrieves the spec for a given pod in a specific namespace.

Args:
    pod_name (str): The name of the pod.
    namespace (str): The namespace the pod belongs to (defaults to "default").

Returns
-------
dict[str, Any]
    The pod's spec object, containing its desired state. It is converted
    from a V1PodSpec to a dictionary. Key fields include:

    containers : list of kubernetes.client.V1Container
        List of containers belonging to the pod. Each container defines its image,
        ports, environment variables, resource requests/limits, etc.
    init_containers : list of kubernetes.client.V1Container, optional
        List of initialization containers belonging to the pod.
    volumes : list of kubernetes.client.V1Volume, optional
        List of volumes mounted in the pod and the sources available for
        the containers.
    node_selector : dict, optional
        A selector which must be true for the pod to fit on a node.
        Keys and values are strings.
    restart_policy : str
        Restart policy for all containers within the pod.
        Common values are "Always", "OnFailure", "Never".
    service_account_name : str, optional
        Service account name in the namespace that the pod will use to
        access the Kubernetes API.
    dns_policy : str
        DNS policy for the pod. Common values are "ClusterFirst", "Default".
    priority_class_name : str, optional
        If specified, indicates the pod's priority_class via its name.
    node_name : str, optional
        NodeName is a request to schedule this pod onto a specific node.

Raises
------
K8SConfigError
    If unable to initialize the K8S API
K8sApiError
    If the pod is not found, configuration fails, or any other API error occurs.

Implementation Reference

• src/k8stools/k8s_tools.py:729-795 (handler)

  Core handler function for the 'get_pod_spec' tool. Fetches the pod object from Kubernetes API and returns its spec as a dictionary.

  def get_pod_spec(pod_name: str, namespace: str = "default") -> dict[str,Any]:
      """
      Retrieves the spec for a given pod in a specific namespace.
  
      Args:
          pod_name (str): The name of the pod.
          namespace (str): The namespace the pod belongs to (defaults to "default").
  
      Returns
      -------
      dict[str, Any]
          The pod's spec object, containing its desired state. It is converted
          from a V1PodSpec to a dictionary. Key fields include:
  
          containers : list of kubernetes.client.V1Container
              List of containers belonging to the pod. Each container defines its image,
              ports, environment variables, resource requests/limits, etc.
          init_containers : list of kubernetes.client.V1Container, optional
              List of initialization containers belonging to the pod.
          volumes : list of kubernetes.client.V1Volume, optional
              List of volumes mounted in the pod and the sources available for
              the containers.
          node_selector : dict, optional
              A selector which must be true for the pod to fit on a node.
              Keys and values are strings.
          restart_policy : str
              Restart policy for all containers within the pod.
              Common values are "Always", "OnFailure", "Never".
          service_account_name : str, optional
              Service account name in the namespace that the pod will use to
              access the Kubernetes API.
          dns_policy : str
              DNS policy for the pod. Common values are "ClusterFirst", "Default".
          priority_class_name : str, optional
              If specified, indicates the pod's priority_class via its name.
          node_name : str, optional
              NodeName is a request to schedule this pod onto a specific node.
  
      Raises
      ------
      K8SConfigError
          If unable to initialize the K8S API
      K8sApiError
          If the pod is not found, configuration fails, or any other API error occurs.
      """
      global K8S
      if K8S is None:
          K8S = _get_api_client()
          logging.info(f"get_pod_spec(pod_name={pod_name}, namespace={namespace})")
      try:
          # Get the pod object
          pod = K8S.read_namespaced_pod(name=pod_name, namespace=namespace)
          # Ensure pod is a V1Pod instance and has a spec
          if not isinstance(pod, client.V1Pod) or not hasattr(pod, "spec") or pod.spec is None:
              raise K8sApiError(f"Pod '{pod_name}' in namespace '{namespace}' did not return a valid spec.")
          return pod.spec.to_dict()
      except ApiException as e:
          if hasattr(e, "status") and e.status == 404:
              raise K8sApiError(
                  f"Pod '{pod_name}' not found in namespace '{namespace}'."
              ) from e
          else:
              raise K8sApiError(
                  f"Error getting pod '{pod_name}' in namespace '{namespace}': {e}"
              ) from e
      except Exception as e:
          raise K8sApiError(f"Unexpected error getting pod spec: {e}") from e

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

  The TOOLS list in k8s_tools.py registers get_pod_spec among the available tools, which is imported into mcp_server.py for MCP tool setup.

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

  MCP server registers the tools by wrapping functions from TOOLS (including get_pod_spec) into FastMCP Tool objects using Tool.from_function.

      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]