Kubernetes Tools MCP Server

Instructions

Implementation Reference

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

  The core handler function that retrieves the Kubernetes pod specification using the CoreV1Api client and returns it 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 that registers get_pod_spec among other Kubernetes tools for use in the MCP server.
  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)

  Imports the TOOLS list (including get_pod_spec) and wraps each function into an MCP Tool object using Tool.from_function for registration in FastMCP.
  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:287-306 (helper)

  Mock implementation of get_pod_spec for testing without a real Kubernetes cluster.
  def get_pod_spec(pod_name: str, namespace: str = "default") -> dict[str, Any]: """Mock implementation that returns static pod spec data for the specified pod""" # For the specific ad pod, return cached data if pod_name == "ad-647b4947cc-s5mpm" and namespace == "default": return _MOCK_DATA['ad_pod_spec'] # For other pods, return generic mock spec return { "containers": [{ "name": pod_name.split('-')[0], "image": "nginx:latest", "ports": [{"containerPort": 80, "protocol": "TCP"}], "resources": {} }], "restart_policy": "Always", "node_name": "minikube" } get_pod_spec.__doc__ = k8s_tools.get_pod_spec.__doc__