k8s-pilot

Instructions

Implementation Reference

• tools/namespace.py:337-427 (handler)

  The handler function for the 'set_namespace_resource_quota' tool. It creates or updates a ResourceQuota object in the specified namespace with the given CPU, memory, and pod limits using the Kubernetes CoreV1Api. The @mcp.tool() decorator registers it as an MCP tool.

  @mcp.tool()
  @use_current_context
  @check_readonly_permission
  def set_namespace_resource_quota(context_name: str, namespace: str,
                                   cpu_limit: Optional[str] = None,
                                   memory_limit: Optional[str] = None,
                                   pod_count: Optional[int] = None):
      """
      Set or update resource quotas for a namespace.
  
      Args:
          context_name: The Kubernetes context name
          namespace: The name of the namespace
          cpu_limit: Optional CPU limit (e.g., "2", "500m")
          memory_limit: Optional memory limit (e.g., "1Gi", "500Mi")
          pod_count: Optional maximum number of pods
  
      Returns:
          JSON string containing the resource quota status
      """
      from kubernetes.client.models.v1_resource_quota import V1ResourceQuota
      from kubernetes.client.models.v1_resource_quota_spec import V1ResourceQuotaSpec
  
      core_v1: CoreV1Api = get_api_clients(context_name)["core"]
  
      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)}"})
  
          # If no limits are provided, return error
          if not any([cpu_limit, memory_limit, pod_count]):
              return json.dumps({"error": "At least one resource limit must be specified"})
  
          # Prepare the resource quota
          quota_name = f"{namespace}-resource-quota"
          hard_quotas = {}
  
          if cpu_limit:
              hard_quotas["limits.cpu"] = cpu_limit
  
          if memory_limit:
              hard_quotas["limits.memory"] = memory_limit
  
          if pod_count:
              hard_quotas["pods"] = str(pod_count)
  
          # Check if quota already exists
          try:
              existing_quota = core_v1.read_namespaced_resource_quota(quota_name, namespace)
              # Update existing quota
              body = {
                  "spec": {
                      "hard": hard_quotas
                  }
              }
              updated_quota = core_v1.patch_namespaced_resource_quota(quota_name, namespace, body)
  
              result = {
                  "name": updated_quota.metadata.name,
                  "namespace": namespace,
                  "quotas": updated_quota.spec.hard,
                  "message": "Resource quota updated successfully"
              }
          except ApiException as e:
              if e.status == 404:
                  # Create new quota
                  quota_spec = V1ResourceQuotaSpec(hard=hard_quotas)
                  quota_metadata = V1ObjectMeta(name=quota_name)
                  quota_body = V1ResourceQuota(metadata=quota_metadata, spec=quota_spec)
  
                  created_quota = core_v1.create_namespaced_resource_quota(namespace, quota_body)
  
                  result = {
                      "name": created_quota.metadata.name,
                      "namespace": namespace,
                      "quotas": created_quota.spec.hard,
                      "message": "Resource quota created successfully"
                  }
              else:
                  return json.dumps({"error": f"API error: {str(e)}"})
  
          return json.dumps(result)
      except ApiException as e:
          return json.dumps({"error": f"Failed to set resource quota: {str(e)}"})