diff --git a/examples/kubeconfig/README.md b/examples/kubeconfig/README.md
new file mode 100644
index 0000000..f90a2b2
--- /dev/null
+++ b/examples/kubeconfig/README.md
@@ -0,0 +1,113 @@
+# Kubeconfig Provider Example
+
+This example demonstrates how to use the kubeconfig provider to manage multiple Kubernetes clusters using kubeconfig secrets.
+
+## Overview
+
+The kubeconfig provider allows you to:
+1. Discover and connect to multiple Kubernetes clusters using kubeconfig secrets
+2. Run controllers that can operate across all discovered clusters
+3. Manage cluster access through RBAC rules and service accounts
+
+## Directory Structure
+
+```
+examples/kubeconfig/
+├── controllers/                    # Example controller that simply lists pods
+│   ├── pod_lister.go
+├── scripts/                 # Utility scripts
+│   └── create-kubeconfig-secret.sh
+└── main.go                 # Example operator implementation
+```
+
+## Usage
+
+### 1. Setting Up Cluster Access
+
+Before creating a kubeconfig secret, ensure that:
+1. The remote cluster has a service account with the necessary RBAC permissions for your operator
+2. The service account exists in the namespace where you want to create the kubeconfig secret
+
+Use the `create-kubeconfig-secret.sh` script to create a kubeconfig secret for each cluster you want to manage:
+
+```bash
+./scripts/create-kubeconfig-secret.sh \
+  --name cluster1 \
+  -n default \
+  -c prod-cluster \
+  -a my-service-account
+```
+
+The script will:
+- Use the specified service account from the remote cluster
+- Generate a kubeconfig using the service account's token
+- Store the kubeconfig in a secret in your local cluster
+
+Command line options:
+- `-c, --context`: Kubeconfig context to use (required)
+- `--name`: Name for the secret (defaults to context name)
+- `-n, --namespace`: Namespace to create the secret in (default: "default")
+- `-a, --service-account`: Service account name to use from the remote cluster (default: "default")
+
+### 2. Customizing RBAC Rules
+
+The service account in the remote cluster must have the necessary RBAC permissions for your operator to function. Edit the RBAC templates in the `rbac/` directory to define the permissions your operator needs:
+
+```yaml
+# rbac/clusterrole.yaml
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRole
+metadata:
+  name: ${SECRET_NAME}-role
+rules:
+# Add permissions for your operator <--------------------------------
+- apiGroups: [""]
+  resources: ["pods"]
+  verbs: ["list", "get", "watch"]  # watch is needed for controllers that observe resources
+```
+
+Important RBAC considerations:
+- Use `watch` verb if your controller needs to observe resource changes
+- Use `list` and `get` for reading resources
+- Use `create`, `update`, `patch`, `delete` for modifying resources
+- Consider using `Role` instead of `ClusterRole` if you only need namespace-scoped permissions
+
+### 3. Implementing Your Operator
+
+Add your controllers to `main.go`:
+
+```go
+func main() {
+    // Import your controllers here <--------------------------------
+	"sigs.k8s.io/multicluster-runtime/examples/kubeconfig/controllers"
+
+    //...
+
+    // Run your controllers here <--------------------------------
+	podWatcher := controllers.NewPodWatcher(mgr)
+	if err := mgr.Add(podWatcher); err != nil {
+		entryLog.Error(err, "Unable to add pod watcher")
+		os.Exit(1)
+	}
+}
+```
+
+Your controllers can then use the manager to access any cluster and view the resources that the RBAC permissions allow.
+
+## How It Works
+
+1. The kubeconfig provider watches for secrets with a specific label in a namespace
+2. When a new secret is found, it:
+   - Extracts the kubeconfig data
+   - Creates a new controller-runtime cluster
+   - Makes the cluster available to your controllers
+3. Your controllers can access any cluster through the manager
+4. RBAC rules ensure your operator has the necessary permissions in each cluster
+
+## Labels and Configuration
+
+The provider uses the following labels and keys by default:
+- Label: `sigs.k8s.io/multicluster-runtime-kubeconfig: "true"`
+- Secret data key: `kubeconfig`
+
+You can customize these in the provider options when creating it. 
\ No newline at end of file
diff --git a/examples/kubeconfig/controllers/pod_lister.go b/examples/kubeconfig/controllers/pod_lister.go
new file mode 100644
index 0000000..fb3a1ce
--- /dev/null
+++ b/examples/kubeconfig/controllers/pod_lister.go
@@ -0,0 +1,102 @@
+/*
+Copyright 2025 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package controllers
+
+import (
+	"context"
+	"time"
+
+	"github.com/go-logr/logr"
+
+	corev1 "k8s.io/api/core/v1"
+
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/cluster"
+	ctrllog "sigs.k8s.io/controller-runtime/pkg/log"
+
+	mcmanager "sigs.k8s.io/multicluster-runtime/pkg/manager"
+)
+
+// PodWatcher is a simple controller that watches pods across multiple clusters
+type PodWatcher struct {
+	Manager mcmanager.Manager
+	Log     logr.Logger
+}
+
+// NewPodWatcher creates a new PodWatcher
+func NewPodWatcher(mgr mcmanager.Manager) *PodWatcher {
+	return &PodWatcher{
+		Manager: mgr,
+		Log:     ctrllog.Log.WithName("pod-watcher"),
+	}
+}
+
+// Start implements Runnable
+func (p *PodWatcher) Start(ctx context.Context) error {
+	// Nothing to do here - we'll handle everything in Engage
+	return nil
+}
+
+// Engage implements multicluster.Aware and gets called when a new cluster is engaged
+func (p *PodWatcher) Engage(ctx context.Context, clusterName string, cl cluster.Cluster) error {
+	log := p.Log.WithValues("cluster", clusterName)
+	log.Info("Engaging cluster")
+
+	// Start a goroutine to periodically list pods
+	go func() {
+		ticker := time.NewTicker(30 * time.Second)
+		defer ticker.Stop()
+
+		// Initial list
+		if err := p.listPods(ctx, cl, clusterName, log); err != nil {
+			log.Error(err, "Failed to list pods")
+		}
+
+		for {
+			select {
+			case <-ctx.Done():
+				log.Info("Context done, stopping pod watcher")
+				return
+			case <-ticker.C:
+				if err := p.listPods(ctx, cl, clusterName, log); err != nil {
+					log.Error(err, "Failed to list pods")
+				}
+			}
+		}
+	}()
+
+	return nil
+}
+
+// listPods lists pods in the default namespace
+func (p *PodWatcher) listPods(ctx context.Context, cl cluster.Cluster, clusterName string, log logr.Logger) error {
+	var pods corev1.PodList
+	if err := cl.GetClient().List(ctx, &pods, &client.ListOptions{
+		Namespace: "default",
+	}); err != nil {
+		return err
+	}
+
+	log.Info("Pods in default namespace", "count", len(pods.Items))
+	for _, pod := range pods.Items {
+		log.Info("Pod",
+			"name", pod.Name,
+			"status", pod.Status.Phase)
+	}
+
+	return nil
+}
diff --git a/examples/kubeconfig/main.go b/examples/kubeconfig/main.go
new file mode 100644
index 0000000..b9a8ae4
--- /dev/null
+++ b/examples/kubeconfig/main.go
@@ -0,0 +1,114 @@
+/*
+Copyright 2025 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package main
+
+import (
+	"flag"
+	"os"
+
+	// Import all Kubernetes client auth plugins (e.g. Azure, GCP, OIDC, etc.)
+	// to ensure that exec-entrypoint and run can make use of them.
+	"k8s.io/client-go/kubernetes/scheme"
+	_ "k8s.io/client-go/plugin/pkg/client/auth"
+
+	ctrl "sigs.k8s.io/controller-runtime"
+	ctrllog "sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/controller-runtime/pkg/log/zap"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+
+	// Import your controllers here <--------------------------------
+	"sigs.k8s.io/multicluster-runtime/examples/kubeconfig/controllers"
+
+	mcmanager "sigs.k8s.io/multicluster-runtime/pkg/manager"
+	kubeconfigprovider "sigs.k8s.io/multicluster-runtime/providers/kubeconfig"
+)
+
+func main() {
+	var namespace string
+	var kubeconfigSecretLabel string
+	var kubeconfigSecretKey string
+
+	flag.StringVar(&namespace, "namespace", "default", "Namespace where kubeconfig secrets are stored")
+	flag.StringVar(&kubeconfigSecretLabel, "kubeconfig-label", "sigs.k8s.io/multicluster-runtime-kubeconfig",
+		"Label used to identify secrets containing kubeconfig data")
+	flag.StringVar(&kubeconfigSecretKey, "kubeconfig-key", "kubeconfig", "Key in the secret data that contains the kubeconfig")
+	opts := zap.Options{
+		Development: true,
+	}
+	opts.BindFlags(flag.CommandLine)
+	flag.Parse()
+
+	ctrllog.SetLogger(zap.New(zap.UseFlagOptions(&opts)))
+	entryLog := ctrllog.Log.WithName("entrypoint")
+	ctx := ctrl.SetupSignalHandler()
+
+	entryLog.Info("Starting application", "namespace", namespace, "kubeconfigSecretLabel", kubeconfigSecretLabel)
+
+	// Create the kubeconfig provider with options
+	providerOpts := kubeconfigprovider.Options{
+		Namespace:             namespace,
+		KubeconfigSecretLabel: kubeconfigSecretLabel,
+		KubeconfigSecretKey:   kubeconfigSecretKey,
+	}
+
+	// Create the provider first, then the manager with the provider
+	entryLog.Info("Creating provider")
+	provider := kubeconfigprovider.New(providerOpts)
+
+	// Create the multicluster manager with the provider
+	entryLog.Info("Creating manager")
+
+	// Modify manager options to avoid waiting for cache sync
+	managerOpts := manager.Options{
+		// Don't block main thread on leader election
+		LeaderElection: false,
+		// Add the scheme
+		Scheme: scheme.Scheme,
+	}
+
+	mgr, err := mcmanager.New(ctrl.GetConfigOrDie(), provider, managerOpts)
+	if err != nil {
+		entryLog.Error(err, "Unable to create manager")
+		os.Exit(1)
+	}
+
+	// Add our controllers
+	entryLog.Info("Adding controllers")
+
+	// Run your controllers here <--------------------------------
+	podWatcher := controllers.NewPodWatcher(mgr)
+	if err := mgr.Add(podWatcher); err != nil {
+		entryLog.Error(err, "Unable to add pod watcher")
+		os.Exit(1)
+	}
+
+	// Start provider in a goroutine
+	entryLog.Info("Starting provider")
+	go func() {
+		err := provider.Run(ctx, mgr)
+		if err != nil && ctx.Err() == nil {
+			entryLog.Error(err, "Provider exited with error")
+		}
+	}()
+
+	// Start the manager
+	entryLog.Info("Starting manager")
+	if err := mgr.Start(ctx); err != nil {
+		entryLog.Error(err, "Error running manager")
+		os.Exit(1)
+	}
+}
diff --git a/examples/kubeconfig/scripts/create-kubeconfig-secret.sh b/examples/kubeconfig/scripts/create-kubeconfig-secret.sh
new file mode 100755
index 0000000..6d13102
--- /dev/null
+++ b/examples/kubeconfig/scripts/create-kubeconfig-secret.sh
@@ -0,0 +1,151 @@
+#!/bin/bash
+
+# Script to create a kubeconfig secret for the pod lister controller
+
+set -e
+
+# Default values
+NAMESPACE="default"
+SERVICE_ACCOUNT="default"
+KUBECONFIG_CONTEXT=""
+SECRET_NAME=""
+
+# Function to display usage information
+function show_help {
+  echo "Usage: $0 [options]"
+  echo "  -c, --context CONTEXT    Kubeconfig context to use (required)"
+  echo "  --name NAME              Name for the secret (defaults to context name)"
+  echo "  -n, --namespace NS       Namespace to create the secret in (default: ${NAMESPACE})"
+  echo "  -a, --service-account SA Service account name to use (default: ${SERVICE_ACCOUNT})"
+  echo "  -h, --help               Show this help message"
+  echo ""
+  echo "Example: $0 -c prod-cluster"
+}
+
+# Parse command line options
+while [[ $# -gt 0 ]]; do
+  key="$1"
+  case $key in
+    --name)
+      SECRET_NAME="$2"
+      shift 2
+      ;;
+    -n|--namespace)
+      NAMESPACE="$2"
+      shift 2
+      ;;
+    -c|--context)
+      KUBECONFIG_CONTEXT="$2"
+      shift 2
+      ;;
+    -a|--service-account)
+      SERVICE_ACCOUNT="$2"
+      shift 2
+      ;;
+    -h|--help)
+      show_help
+      exit 0
+      ;;
+    *)
+      echo "Unknown option: $1"
+      show_help
+      exit 1
+      ;;
+  esac
+done
+
+# Validate required arguments
+if [ -z "$KUBECONFIG_CONTEXT" ]; then
+  echo "ERROR: Kubeconfig context is required (-c, --context)"
+  show_help
+  exit 1
+fi
+
+# Set secret name to context if not specified
+if [ -z "$SECRET_NAME" ]; then
+  SECRET_NAME="$KUBECONFIG_CONTEXT"
+fi
+
+# Get the cluster CA certificate from the remote cluster
+CLUSTER_CA=$(kubectl --context=${KUBECONFIG_CONTEXT} config view --raw --minify --flatten -o jsonpath='{.clusters[].cluster.certificate-authority-data}')
+if [ -z "$CLUSTER_CA" ]; then
+  echo "ERROR: Could not get cluster CA certificate"
+  exit 1
+fi
+
+# Get the cluster server URL from the remote cluster
+CLUSTER_SERVER=$(kubectl --context=${KUBECONFIG_CONTEXT} config view --raw --minify --flatten -o jsonpath='{.clusters[].cluster.server}')
+if [ -z "$CLUSTER_SERVER" ]; then
+  echo "ERROR: Could not get cluster server URL"
+  exit 1
+fi
+
+# Get the service account token from the remote cluster
+SA_TOKEN=$(kubectl --context=${KUBECONFIG_CONTEXT} -n ${NAMESPACE} create token ${SERVICE_ACCOUNT} --duration=8760h)
+if [ -z "$SA_TOKEN" ]; then
+  echo "ERROR: Could not create service account token"
+  exit 1
+fi
+
+# Create a new kubeconfig using the service account token
+NEW_KUBECONFIG=$(cat <<EOF
+apiVersion: v1
+kind: Config
+clusters:
+- name: ${SECRET_NAME}
+  cluster:
+    server: ${CLUSTER_SERVER}
+    certificate-authority-data: ${CLUSTER_CA}
+contexts:
+- name: ${SECRET_NAME}
+  context:
+    cluster: ${SECRET_NAME}
+    user: ${SERVICE_ACCOUNT}
+current-context: ${SECRET_NAME}
+users:
+- name: ${SERVICE_ACCOUNT}
+  user:
+    token: ${SA_TOKEN}
+EOF
+)
+
+# Save kubeconfig temporarily for testing
+TEMP_KUBECONFIG=$(mktemp)
+echo "$NEW_KUBECONFIG" > "$TEMP_KUBECONFIG"
+
+# Verify the kubeconfig works
+echo "Verifying kubeconfig..."
+if ! kubectl --kubeconfig="$TEMP_KUBECONFIG" get pods -A &>/dev/null; then
+  rm "$TEMP_KUBECONFIG"
+  echo "ERROR: Failed to verify kubeconfig - unable to list pods."
+  echo "- Ensure that the service account '${NAMESPACE}/${SERVICE_ACCOUNT}' on cluster '${KUBECONFIG_CONTEXT}' has the necessary permissions to list pods."
+  echo "- You may specify a namespace using the -n flag."
+  echo "- You may specify a service account using the -a flag."
+  exit 1
+fi
+echo "Kubeconfig verified successfully!"
+
+# Encode the verified kubeconfig
+KUBECONFIG_B64=$(cat "$TEMP_KUBECONFIG" | base64 -w0)
+rm "$TEMP_KUBECONFIG"
+
+# Generate and apply the secret
+SECRET_YAML=$(cat <<EOF
+apiVersion: v1
+kind: Secret
+metadata:
+  name: ${SECRET_NAME}
+  namespace: ${NAMESPACE}
+  labels:
+    sigs.k8s.io/multicluster-runtime-kubeconfig: "true"
+type: Opaque
+data:
+  kubeconfig: ${KUBECONFIG_B64}
+EOF
+)
+
+echo "Creating kubeconfig secret..."
+echo "$SECRET_YAML" | kubectl apply -f -
+
+echo "Secret '${SECRET_NAME}' created in namespace '${NAMESPACE}'"
+echo "The operator should now be able to discover and connect to this cluster" 
\ No newline at end of file
diff --git a/providers/kubeconfig/provider.go b/providers/kubeconfig/provider.go
new file mode 100644
index 0000000..1863bbb
--- /dev/null
+++ b/providers/kubeconfig/provider.go
@@ -0,0 +1,358 @@
+/*
+Copyright 2025 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Package kubeconfig provides a Kubernetes cluster provider that watches secrets
+// containing kubeconfig data and creates controller-runtime clusters for each.
+package kubeconfig
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"path/filepath"
+	"sync"
+
+	"github.com/go-logr/logr"
+
+	corev1 "k8s.io/api/core/v1"
+	toolscache "k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/tools/clientcmd"
+
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/cluster"
+	"sigs.k8s.io/controller-runtime/pkg/log"
+
+	mcmanager "sigs.k8s.io/multicluster-runtime/pkg/manager"
+	"sigs.k8s.io/multicluster-runtime/pkg/multicluster"
+)
+
+const (
+	// DefaultKubeconfigSecretLabel is the default label key to identify kubeconfig secrets
+	DefaultKubeconfigSecretLabel = "sigs.k8s.io/multicluster-runtime-kubeconfig"
+
+	// DefaultKubeconfigSecretKey is the default key in the secret data that contains the kubeconfig
+	DefaultKubeconfigSecretKey = "kubeconfig"
+)
+
+var _ multicluster.Provider = &Provider{}
+
+// New creates a new Kubeconfig Provider.
+func New(opts Options) *Provider {
+	// Set defaults
+	if opts.KubeconfigSecretLabel == "" {
+		opts.KubeconfigSecretLabel = DefaultKubeconfigSecretLabel
+	}
+	if opts.KubeconfigSecretKey == "" {
+		opts.KubeconfigSecretKey = DefaultKubeconfigSecretKey
+	}
+	if opts.LocalKubeconfigPath == "" {
+		opts.LocalKubeconfigPath = filepath.Join(os.Getenv("HOME"), ".kube", "config")
+	}
+
+	return &Provider{
+		opts:      opts,
+		log:       log.Log.WithName("kubeconfig-provider"),
+		client:    nil, // Will be set in Run
+		clusters:  map[string]cluster.Cluster{},
+		cancelFns: map[string]context.CancelFunc{},
+	}
+}
+
+// Options contains the configuration for the kubeconfig provider.
+type Options struct {
+	// Namespace is the namespace where kubeconfig secrets are stored.
+	Namespace string
+	// KubeconfigSecretLabel is the label used to identify secrets containing kubeconfig data.
+	KubeconfigSecretLabel string
+	// KubeconfigSecretKey is the key in the secret data that contains the kubeconfig.
+	KubeconfigSecretKey string
+	// LocalKubeconfigPath is the path to kubeconfig file for test secrets.
+	LocalKubeconfigPath string
+}
+
+type index struct {
+	object       client.Object
+	field        string
+	extractValue client.IndexerFunc
+}
+
+// Provider is a cluster provider that watches for secrets containing kubeconfig data
+// and engages clusters based on those kubeconfigs.
+type Provider struct {
+	opts      Options
+	log       logr.Logger
+	client    client.Client
+	lock      sync.RWMutex // protects everything below.
+	clusters  map[string]cluster.Cluster
+	cancelFns map[string]context.CancelFunc
+	indexers  []index
+}
+
+// Get returns the cluster with the given name, if it is known.
+func (p *Provider) Get(ctx context.Context, clusterName string) (cluster.Cluster, error) {
+	p.lock.RLock()
+	defer p.lock.RUnlock()
+
+	if cl, ok := p.clusters[clusterName]; ok {
+		return cl, nil
+	}
+
+	return nil, fmt.Errorf("cluster %s not found", clusterName)
+}
+
+// Run starts the provider and blocks, watching for kubeconfig secrets.
+func (p *Provider) Run(ctx context.Context, mgr mcmanager.Manager) error {
+	log := p.log
+	log.Info("Starting kubeconfig provider", "namespace", p.opts.Namespace, "label", p.opts.KubeconfigSecretLabel)
+
+	// If client isn't set yet, get it from the manager
+	if p.client == nil && mgr != nil {
+		log.Info("Setting client from manager")
+		p.client = mgr.GetLocalManager().GetClient()
+		if p.client == nil {
+			return fmt.Errorf("failed to get client from manager")
+		}
+	}
+
+	// Get the informer for secrets
+	secretInf, err := mgr.GetLocalManager().GetCache().GetInformer(ctx, &corev1.Secret{})
+	if err != nil {
+		return fmt.Errorf("failed to get secret informer: %w", err)
+	}
+
+	// Add event handlers for secrets
+	if _, err := secretInf.AddEventHandler(toolscache.FilteringResourceEventHandler{
+		FilterFunc: func(obj interface{}) bool {
+			secret, ok := obj.(*corev1.Secret)
+			if !ok {
+				return false
+			}
+			// Only process secrets in our namespace with our label
+			return secret.Namespace == p.opts.Namespace &&
+				secret.Labels[p.opts.KubeconfigSecretLabel] == "true"
+		},
+		Handler: toolscache.ResourceEventHandlerFuncs{
+			AddFunc: func(obj interface{}) {
+				secret := obj.(*corev1.Secret)
+				log.Info("Processing new secret", "name", secret.Name)
+				if err := p.handleSecret(ctx, secret, mgr); err != nil {
+					log.Error(err, "Failed to handle secret", "name", secret.Name)
+				}
+			},
+			UpdateFunc: func(oldObj, newObj interface{}) {
+				secret := newObj.(*corev1.Secret)
+				log.Info("Processing updated secret", "name", secret.Name)
+				if err := p.handleSecret(ctx, secret, mgr); err != nil {
+					log.Error(err, "Failed to handle secret", "name", secret.Name)
+				}
+			},
+			DeleteFunc: func(obj interface{}) {
+				secret := obj.(*corev1.Secret)
+				log.Info("Processing deleted secret", "name", secret.Name)
+				p.handleSecretDelete(secret)
+			},
+		},
+	}); err != nil {
+		return fmt.Errorf("failed to add event handlers: %w", err)
+	}
+
+	// Block until context is done
+	<-ctx.Done()
+	log.Info("Context cancelled, exiting provider")
+	return ctx.Err()
+}
+
+// handleSecret processes a secret containing kubeconfig data
+func (p *Provider) handleSecret(ctx context.Context, secret *corev1.Secret, mgr mcmanager.Manager) error {
+	if secret == nil {
+		return fmt.Errorf("received nil secret")
+	}
+
+	// Extract name to use as cluster name
+	clusterName := secret.Name
+	log := p.log.WithValues("cluster", clusterName, "secret", fmt.Sprintf("%s/%s", secret.Namespace, secret.Name))
+
+	// Check if this secret has kubeconfig data
+	kubeconfigData, ok := secret.Data[p.opts.KubeconfigSecretKey]
+	if !ok {
+		log.Info("Secret does not contain kubeconfig data", "key", p.opts.KubeconfigSecretKey)
+		return nil
+	}
+
+	// Parse the kubeconfig
+	restConfig, err := clientcmd.RESTConfigFromKubeConfig(kubeconfigData)
+	if err != nil {
+		return fmt.Errorf("failed to parse kubeconfig: %w", err)
+	}
+
+	// Check if cluster exists and remove it if it does
+	p.lock.RLock()
+	_, clusterExists := p.clusters[clusterName]
+	p.lock.RUnlock()
+
+	if clusterExists {
+		log.Info("Cluster already exists, updating it")
+		if err := p.removeCluster(clusterName); err != nil {
+			return fmt.Errorf("failed to remove existing cluster: %w", err)
+		}
+	}
+
+	// Create a new cluster
+	log.Info("Creating new cluster from kubeconfig")
+	cl, err := cluster.New(restConfig)
+	if err != nil {
+		return fmt.Errorf("failed to create cluster: %w", err)
+	}
+
+	// Apply any field indexers
+	p.lock.RLock()
+	for _, idx := range p.indexers {
+		if err := cl.GetFieldIndexer().IndexField(ctx, idx.object, idx.field, idx.extractValue); err != nil {
+			p.lock.RUnlock()
+			return fmt.Errorf("failed to index field %q: %w", idx.field, err)
+		}
+	}
+	p.lock.RUnlock()
+
+	// Create a context that will be canceled when this cluster is removed
+	clusterCtx, cancel := context.WithCancel(ctx)
+
+	// Start the cluster
+	go func() {
+		if err := cl.Start(clusterCtx); err != nil {
+			log.Error(err, "Failed to start cluster")
+		}
+	}()
+
+	// Wait for cache to be ready
+	log.Info("Waiting for cluster cache to be ready")
+	if !cl.GetCache().WaitForCacheSync(clusterCtx) {
+		cancel() // Cancel context before returning error
+		return fmt.Errorf("failed to wait for cache sync")
+	}
+	log.Info("Cluster cache is ready")
+
+	// Store the cluster
+	p.lock.Lock()
+	p.clusters[clusterName] = cl
+	p.cancelFns[clusterName] = cancel
+	p.lock.Unlock()
+
+	log.Info("Successfully added cluster")
+
+	// Engage the manager if provided
+	if mgr != nil {
+		if err := mgr.Engage(clusterCtx, clusterName, cl); err != nil {
+			log.Error(err, "Failed to engage manager, removing cluster")
+			p.lock.Lock()
+			delete(p.clusters, clusterName)
+			delete(p.cancelFns, clusterName)
+			p.lock.Unlock()
+			cancel() // Cancel the cluster context
+			return fmt.Errorf("failed to engage manager: %w", err)
+		}
+		log.Info("Successfully engaged manager")
+	}
+
+	return nil
+}
+
+// handleSecretDelete handles the deletion of a secret
+func (p *Provider) handleSecretDelete(secret *corev1.Secret) {
+	if secret == nil {
+		return
+	}
+
+	clusterName := secret.Name
+	log := p.log.WithValues("cluster", clusterName)
+
+	log.Info("Handling deleted secret")
+
+	// Remove the cluster
+	if err := p.removeCluster(clusterName); err != nil {
+		log.Error(err, "Failed to remove cluster")
+	}
+}
+
+// removeCluster removes a cluster by name
+func (p *Provider) removeCluster(clusterName string) error {
+	log := p.log.WithValues("cluster", clusterName)
+	log.Info("Removing cluster")
+
+	// Find the cluster and cancel function
+	p.lock.RLock()
+	_, exists := p.clusters[clusterName]
+	if !exists {
+		p.lock.RUnlock()
+		return fmt.Errorf("cluster %s not found", clusterName)
+	}
+
+	// Get the cancel function
+	cancelFn, exists := p.cancelFns[clusterName]
+	if !exists {
+		p.lock.RUnlock()
+		return fmt.Errorf("cancel function for cluster %s not found", clusterName)
+	}
+	p.lock.RUnlock()
+
+	// Cancel the context to trigger cleanup for this cluster
+	cancelFn()
+	log.Info("Cancelled cluster context")
+
+	// Clean up our maps
+	p.lock.Lock()
+	delete(p.clusters, clusterName)
+	delete(p.cancelFns, clusterName)
+	p.lock.Unlock()
+
+	log.Info("Successfully removed cluster")
+	return nil
+}
+
+// IndexField indexes a field on all clusters, existing and future.
+func (p *Provider) IndexField(ctx context.Context, obj client.Object, field string, extractValue client.IndexerFunc) error {
+	p.lock.Lock()
+	defer p.lock.Unlock()
+
+	// Save for future clusters
+	p.indexers = append(p.indexers, index{
+		object:       obj,
+		field:        field,
+		extractValue: extractValue,
+	})
+
+	// Apply to existing clusters
+	for name, cl := range p.clusters {
+		if err := cl.GetFieldIndexer().IndexField(ctx, obj, field, extractValue); err != nil {
+			return fmt.Errorf("failed to index field %q on cluster %q: %w", field, name, err)
+		}
+	}
+
+	return nil
+}
+
+// ListClusters returns a list of all discovered clusters.
+func (p *Provider) ListClusters() map[string]cluster.Cluster {
+	p.lock.RLock()
+	defer p.lock.RUnlock()
+
+	// Return a copy of the map to avoid race conditions
+	result := make(map[string]cluster.Cluster, len(p.clusters))
+	for k, v := range p.clusters {
+		result[k] = v
+	}
+	return result
+}