k8s-pilot

Instructions

Implementation Reference

• tools/namespace.py:261-335 (handler)

  The handler function for the 'list_namespace_resources' tool, decorated with @mcp.tool(). It lists various Kubernetes resources (pods, services, deployments, statefulsets, daemonsets, configmaps, secrets, PVCs) in the given namespace, provides counts and some details, and returns a JSON summary.

  @mcp.tool()
  @use_current_context
  def list_namespace_resources(context_name: str, namespace: str):
      """
      List resources (pods, services, deployments, etc.) in a namespace.
  
      Args:
          context_name: The Kubernetes context name
          namespace: The name of the namespace
  
      Returns:
          JSON string containing a summary of resources in the namespace
      """
      from kubernetes.client import AppsV1Api
  
      core_v1: CoreV1Api = get_api_clients(context_name)["core"]
      apps_v1 = get_api_clients(context_name).get("apps", AppsV1Api(get_api_clients(context_name)["api_client"]))
  
      try:
          # Check if namespace exists
          try:
              core_v1.read_namespace(namespace)
          except ApiException as e:
              if e.status == 404:
                  return json.dumps({"error": f"Namespace '{namespace}' not found"})
              else:
                  return json.dumps({"error": f"API error: {str(e)}"})
  
          # Get pods
          pods = core_v1.list_namespaced_pod(namespace)
  
          # Get services
          services = core_v1.list_namespaced_service(namespace)
  
          # Get deployments
          deployments = apps_v1.list_namespaced_deployment(namespace)
  
          # Get stateful sets
          stateful_sets = apps_v1.list_namespaced_stateful_set(namespace)
  
          # Get daemon sets
          daemon_sets = apps_v1.list_namespaced_daemon_set(namespace)
  
          # Get config maps
          config_maps = core_v1.list_namespaced_config_map(namespace)
  
          # Get secrets
          secrets = core_v1.list_namespaced_secret(namespace)
  
          # Get persistent volume claims
          pvcs = core_v1.list_namespaced_persistent_volume_claim(namespace)
  
          result = {
              "namespace": namespace,
              "resource_counts": {
                  "pods": len(pods.items),
                  "services": len(services.items),
                  "deployments": len(deployments.items),
                  "statefulSets": len(stateful_sets.items),
                  "daemonSets": len(daemon_sets.items),
                  "configMaps": len(config_maps.items),
                  "secrets": len(secrets.items),
                  "persistentVolumeClaims": len(pvcs.items)
              },
              "pods": [{"name": pod.metadata.name, "status": pod.status.phase} for pod in pods.items],
              "services": [{"name": svc.metadata.name, "type": svc.spec.type, "cluster_ip": svc.spec.cluster_ip} for svc
                           in services.items],
              "deployments": [{"name": deploy.metadata.name, "replicas": deploy.spec.replicas} for deploy in
                              deployments.items]
          }
  
          return json.dumps(result)
      except ApiException as e:
          return json.dumps({"error": f"Failed to list namespace resources: {str(e)}"})