diff --git a/.github/workflows/test-conformance-hcp.yaml b/.github/workflows/test-conformance-hcp.yaml
new file mode 100644
index 00000000..ec303517
--- /dev/null
+++ b/.github/workflows/test-conformance-hcp.yaml
@@ -0,0 +1,407 @@
+name: Conformance Tests - HCP Mode
+
+on:
+  schedule:
+    - cron: "0 1 * * *"
+  workflow_dispatch:
+    inputs:
+      k3k_version:
+        description: 'K3k version to test (e.g. v1.1.0). Leave empty to build from source.'
+        required: false
+        type: string
+      k8s_version:
+        description: 'Kubernetes version to test'
+        required: false
+        type: choice
+        options:
+          - ""
+          - "v1.34.6"
+          - "v1.35.3"
+
+permissions:
+    contents: read
+
+env:
+  K8S_VERSIONS: "v1.34.6,v1.35.3"
+  HELM_VERSION: v4.1.3
+  HELM_CHECKSUM_AMD64: 02ce9722d541238f81459938b84cf47df2fdf1187493b4bfb2346754d82a4700
+
+jobs:
+  setup:
+    runs-on: ubuntu-latest
+    outputs:
+      k8s_versions: ${{ steps.set-matrix.outputs.k8s_versions }}
+    steps:
+      - id: set-matrix
+        run: |
+          if [[ -z "${{ inputs.k8s_version }}" ]]; then
+            JSON_ARRAY=$(jq -nc '"${{ env.K8S_VERSIONS }}" | split(",")')
+            echo "k8s_versions=${JSON_ARRAY}" >> "$GITHUB_OUTPUT"
+          else
+            echo "k8s_versions=[\"${{ inputs.k8s_version }}\"]" >> "$GITHUB_OUTPUT"
+          fi
+
+  conformance:
+    needs: setup
+    runs-on: ubuntu-latest
+
+    strategy:
+      fail-fast: false
+      matrix:
+        k8s_version: ${{ fromJSON(needs.setup.outputs.k8s_versions) }}
+
+    env:
+      KUBERNETES_VERSION: ${{ matrix.k8s_version }}
+
+    steps:
+    - name: Checkout code
+      uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6
+      with:
+        fetch-depth: 0
+        fetch-tags: true
+
+    - uses: actions/setup-go@4a3601121dd01d1626a1e23e37211e3254c1c06c # v6
+      with:
+        go-version-file: go.mod
+
+    - name: Install helm
+      env:
+        FILENAME: helm.tar.gz
+      run: |
+        curl -sSfL -o ${{ env.FILENAME }} https://get.helm.sh/helm-${{ env.HELM_VERSION }}-linux-amd64.tar.gz
+        echo "${{ env.HELM_CHECKSUM_AMD64 }}  ${{ env.FILENAME }}" | sha256sum --check
+        tar -xvzf ${{ env.FILENAME }} linux-amd64/helm
+        sudo install -m 755 linux-amd64/helm /usr/local/bin/helm
+
+        rm -fr "${{ env.FILENAME }}" linux-amd64/helm
+
+    - name: Install hydrophone
+      run: go install sigs.k8s.io/hydrophone@3de3e886a2f6f09635d8b981c195490af1584d97 #v0.7.0
+
+    - name: Install k3s
+      env:
+        KUBECONFIG: /etc/rancher/k3s/k3s.yaml
+        K3S_HOST_VERSION: ${{ env.KUBERNETES_VERSION }}+k3s1
+      run: |
+        curl -sfL https://get.k3s.io | INSTALL_K3S_VERSION=${K3S_HOST_VERSION} INSTALL_K3S_EXEC="--write-kubeconfig-mode=777" sh -s -
+
+        kubectl cluster-info
+        kubectl get nodes
+
+        echo "KUBECONFIG=${KUBECONFIG}" >> $GITHUB_ENV
+
+    - name: Setup K3k (from source)
+      if: inputs.k3k_version == ''
+      run: |
+        export REPO=ttl.sh/$(uuidgen)
+        export VERSION=1h
+
+        make build
+        make package
+        make push
+        make install
+
+        # add k3kcli to $PATH
+        echo "${{ github.workspace }}/bin" >> $GITHUB_PATH
+
+    - name: Setup K3k (from release)
+      if: inputs.k3k_version != ''
+      run: |
+        K3K_VERSION="${{ inputs.k3k_version }}"
+        CHART_VERSION="${K3K_VERSION#v}"
+
+        helm repo add k3k https://rancher.github.io/k3k
+        helm repo update
+        helm install --namespace k3k-system --create-namespace --version "${CHART_VERSION}" k3k k3k/k3k
+        
+        wget -qO k3kcli "https://github.com/rancher/k3k/releases/download/${{ inputs.k3k_version }}/k3kcli-linux-amd64"
+        sudo mv k3kcli /usr/local/bin/k3kcli
+        sudo chmod +x /usr/local/bin/k3kcli
+
+    - name: Wait for K3k controller
+      run: |
+        echo "Wait for K3k controller deployment to be available"
+        kubectl wait -n k3k-system deployment -l "app.kubernetes.io/name=k3k" --for=condition=Available --timeout=5m
+
+    - name: Check k3kcli
+      run: k3kcli -v
+
+    - name: Create virtual cluster
+      run: |
+        kubectl create namespace k3k-mycluster
+
+        cat <<EOF | kubectl apply -f -
+        apiVersion: k3k.io/v1beta1
+        kind: Cluster
+        metadata:
+          name: mycluster
+          namespace: k3k-mycluster
+        spec:
+          mode: hcp
+          tlsSANs:
+            - "127.0.0.1"
+            - "192.168.100.1"
+          expose:
+            nodePort:
+              serverPort: 30001
+        EOF
+
+        echo "Wait for bootstrap secret to be available"
+        kubectl wait -n k3k-mycluster --for=create secret k3k-mycluster-bootstrap --timeout=5m
+
+        k3kcli kubeconfig generate --name mycluster
+
+        export KUBECONFIG=${{ github.workspace }}/k3k-mycluster-mycluster-kubeconfig.yaml
+
+        kubectl cluster-info
+        kubectl get nodes
+        kubectl get pods -A
+
+    - name: Install Virtualization Dependencies
+      run: |
+        sudo apt-get update
+        sudo apt-get install -y qemu-kvm qemu-utils cloud-image-utils wget
+        sudo usermod -aG kvm $USER
+
+        kvm-ok
+
+    - name: Set up bridge network for VMs
+      run: |
+        # Create a bridge so both VMs share an L2 segment with unique routable IPs.
+        # Required because QEMU `-net user` gives every VM the same 10.0.2.15 NAT
+        # address, which breaks flannel VXLAN between workers.
+        sudo ip link add name k3kbr0 type bridge
+        sudo ip addr add 192.168.100.1/24 dev k3kbr0
+        sudo ip link set k3kbr0 up
+
+        # NAT outbound so VMs can reach the internet (image pulls etc).
+        sudo sysctl -w net.ipv4.ip_forward=1
+        sudo iptables -t nat -A POSTROUTING -s 192.168.100.0/24 ! -o k3kbr0 -j MASQUERADE
+        sudo iptables -A FORWARD -i k3kbr0 -j ACCEPT
+        sudo iptables -A FORWARD -o k3kbr0 -j ACCEPT
+
+        # One TAP per VM, attached to the bridge.
+        sudo ip tuntap add tap-w1 mode tap
+        sudo ip link set tap-w1 master k3kbr0
+        sudo ip link set tap-w1 up
+
+        sudo ip tuntap add tap-w2 mode tap
+        sudo ip link set tap-w2 master k3kbr0
+        sudo ip link set tap-w2 up
+
+    - name: Download Base Cloud Image
+      run: |
+        # PINNED TO UBUNTU 24.04 LTS (noble).
+        #
+        # Newer Ubuntu releases (tested: 26.04 "resolute") ship a stricter
+        # `cri-containerd.apparmor.d` profile that denies inter-thread signal
+        # delivery. The BIND ISC library used by `nslookup` relies on those
+        # signals during shutdown (`isc_app_ctxshutdown()` calls `kill()`),
+        # so when AppArmor denies them nslookup exits 139 with
+        # "kill: Permission denied". The conformance tests
+        #   `[sig-network] Services should be able to change the type from
+        #   {NodePort,ClusterIP} to ExternalName`
+        # both run `nslookup` from an exec pod and fail in that case.
+        #
+        # Before bumping past 24.04, verify those two conformance tests still
+        # pass — or that the containerd AppArmor profile on the newer release
+        # has been relaxed to allow intra-pod signals.
+        wget -q https://cloud-images.ubuntu.com/noble/current/noble-server-cloudimg-amd64.img -O ubuntu-cloudimg.img
+
+    - name: Generate SSH Key and Cloud-Init Seeds
+      run: |
+        ssh-keygen -t rsa -b 4096 -f ./id_rsa -N ""
+        PUBKEY="$(cat ./id_rsa.pub)"
+
+        # Per-VM cloud-init: each worker gets a unique hostname and a static IP
+        # on the bridge subnet via netplan.
+        for i in 1 2; do
+          IP="192.168.100.1${i}"  # 192.168.100.11 / 192.168.100.12
+          cat <<EOF > user-data-${i}
+        #cloud-config
+        hostname: worker-${i}
+        preserve_hostname: false
+        manage_etc_hosts: true
+        # Stop cloud-init from generating its own DHCP netplan that would
+        # conflict with the static one we write in write_files below.
+        network:
+          config: disabled
+        users:
+          - name: ubuntu
+            ssh_authorized_keys:
+              - ${PUBKEY}
+            sudo: ['ALL=(ALL) NOPASSWD:ALL']
+            shell: /bin/bash
+        write_files:
+          - path: /etc/netplan/50-static.yaml
+            permissions: '0600'
+            content: |
+              network:
+                version: 2
+                ethernets:
+                  ens3:
+                    dhcp4: false
+                    addresses: [${IP}/24]
+                    routes:
+                      - to: default
+                        via: 192.168.100.1
+                    nameservers:
+                      addresses: [8.8.8.8, 1.1.1.1]
+        runcmd:
+          - netplan apply
+        EOF
+          cloud-localds seed-${i}.img user-data-${i}
+        done
+
+    - name: Create Worker Disks
+      run: |
+        qemu-img create -f qcow2 -b ubuntu-cloudimg.img -F qcow2 worker-1.qcow2 20G
+        qemu-img create -f qcow2 -b ubuntu-cloudimg.img -F qcow2 worker-2.qcow2 20G
+
+    - name: Launch Worker VMs
+      run: |
+        # Launch Worker 1 — attached to tap-w1 on k3kbr0.
+        sudo qemu-system-x86_64 \
+          -m 2048 -smp 2 -cpu host -enable-kvm -nographic \
+          -drive file=worker-1.qcow2,if=virtio \
+          -drive file=seed-1.img,format=raw,if=virtio \
+          -netdev tap,id=net0,ifname=tap-w1,script=no,downscript=no \
+          -device virtio-net-pci,netdev=net0,mac=52:54:00:12:34:56 \
+          &
+
+        # Wait a moment before launching the second VM
+        sleep 5
+
+        # Launch Worker 2 — attached to tap-w2 on k3kbr0.
+        sudo qemu-system-x86_64 \
+          -m 2048 -smp 2 -cpu host -enable-kvm -nographic \
+          -drive file=worker-2.qcow2,if=virtio \
+          -drive file=seed-2.img,format=raw,if=virtio \
+          -netdev tap,id=net0,ifname=tap-w2,script=no,downscript=no \
+          -device virtio-net-pci,netdev=net0,mac=52:54:00:12:34:57 \
+          &
+
+    - name: Wait for SSH Availability
+      run: |
+        echo "Waiting for Worker 1 (192.168.100.11) to respond..."
+        timeout 180s bash -c '
+          until ssh -i ./id_rsa -o StrictHostKeyChecking=no -o ConnectTimeout=2 ubuntu@192.168.100.11 true 2>/dev/null; do sleep 3; done
+        '
+
+        echo "Waiting for Worker 2 (192.168.100.12) to respond..."
+        timeout 180s bash -c '
+          until ssh -i ./id_rsa -o StrictHostKeyChecking=no -o ConnectTimeout=2 ubuntu@192.168.100.12 true 2>/dev/null; do sleep 3; done
+        '
+
+        echo "Both VMs are up and running!"
+
+        echo "Testing connectivity from VMs to K3k API server..."
+        ssh -i ./id_rsa -o StrictHostKeyChecking=no ubuntu@192.168.100.11 \
+          "curl -kv --max-time 10 https://192.168.100.1:30001/readyz || echo 'Worker 1 connectivity test failed'"
+
+        ssh -i ./id_rsa -o StrictHostKeyChecking=no ubuntu@192.168.100.12 \
+          "curl -kv --max-time 10 https://192.168.100.1:30001/readyz || echo 'Worker 2 connectivity test failed'"
+
+    - name: Verify Worker VM Configuration
+      run: |
+        for IP in 192.168.100.11 192.168.100.12; do
+          echo "=== Worker at ${IP} ==="
+          ssh -i ./id_rsa -o StrictHostKeyChecking=no ubuntu@${IP} \
+            "echo 'Hostname:' \$(hostname) && \
+             echo 'IP Address:' \$(ip -4 addr show ens3 | grep -oP '(?<=inet\s)\d+(\.\d+){3}') && \
+             echo 'Gateway:' \$(ip route | grep default)"
+          echo ""
+        done
+
+    - name: Join Workers to K3k Control Plane
+      env:
+        K3S_WORKER_VERSION: ${{ env.KUBERNETES_VERSION }}+k3s1
+      run: |
+        K3S_TOKEN=$(kubectl get secret -n k3k-mycluster k3k-mycluster-token -o jsonpath='{.data.token}' | base64 -d)
+
+        echo "Registering Worker 1 (k3s ${K3S_WORKER_VERSION})..."
+        ssh -i ./id_rsa -o StrictHostKeyChecking=no ubuntu@192.168.100.11 \
+          "curl -sfL https://get.k3s.io | INSTALL_K3S_VERSION=${K3S_WORKER_VERSION} K3S_URL=https://192.168.100.1:30001 K3S_TOKEN=${K3S_TOKEN} sh -"
+
+        echo "Registering Worker 2 (k3s ${K3S_WORKER_VERSION})..."
+        ssh -i ./id_rsa -o StrictHostKeyChecking=no ubuntu@192.168.100.12 \
+          "curl -sfL https://get.k3s.io | INSTALL_K3S_VERSION=${K3S_WORKER_VERSION} K3S_URL=https://192.168.100.1:30001 K3S_TOKEN=${K3S_TOKEN} sh -"
+    
+    - name: Verify Cluster Nodes
+      env:
+        KUBECONFIG: ${{ github.workspace }}/k3k-mycluster-mycluster-kubeconfig.yaml
+      run: |
+        echo "Monitoring K3k Virtual Cluster for Node registration..."
+
+        kubectl get pod -A
+        kubectl get nodes
+
+        timeout 180s bash -c '
+          until [ $(kubectl get nodes --no-headers 2>/dev/null | grep -c "Ready") -eq 2 ]; do
+            echo "Waiting for both worker nodes to show Ready..."
+            kubectl get nodes || true
+            sleep 5
+          done
+        '
+
+    - name: Run conformance tests
+      run: |
+        hydrophone --conformance --parallel 4 \
+          --kubeconfig ${{ github.workspace }}/k3k-mycluster-mycluster-kubeconfig.yaml \
+          --output-dir /tmp
+
+    - name: Collect logs
+      if: always()
+      env:
+        KUBECONFIG: /etc/rancher/k3s/k3s.yaml
+      run: |
+        journalctl -u k3s -o cat --no-pager > /tmp/k3s.log
+        kubectl logs -n k3k-system -l "app.kubernetes.io/name=k3k" --tail=-1 > /tmp/k3k.log
+
+    - name: Archive K3s logs
+      uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7
+      if: always()
+      with:
+        name: k3s-${{ matrix.k8s_version }}-logs
+        path: /tmp/k3s.log
+
+    - name: Archive K3k logs
+      uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7
+      if: always()
+      with:
+        name: k3k-${{ matrix.k8s_version }}-logs
+        path: /tmp/k3k.log
+
+    - name: Archive conformance logs
+      uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7
+      if: always()
+      with:
+        name: conformance-${{ matrix.k8s_version }}-logs
+        path: /tmp/e2e.log
+
+    - name: Archive results
+      uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7
+      if: always()
+      with:
+        name: conformance-${{ matrix.k8s_version }}-results
+        path: /tmp/junit_01.xml
+
+    - name: Job Summary
+      if: always()
+      run: |
+        echo '## 📊 Conformance Tests Results (${{ matrix.k8s_version }})' >> $GITHUB_STEP_SUMMARY
+        echo '| Passed | Failed | Pending | Skipped |'              >> $GITHUB_STEP_SUMMARY
+        echo '|---|---|---|---|'                                    >> $GITHUB_STEP_SUMMARY
+
+        RESULTS=$(tail -10 /tmp/e2e.log | grep -E "Passed .* Failed .* Pending .* Skipped" | cut -d '-' -f 3)
+        RESULTS=$(echo $RESULTS | grep -oE '[0-9]+' | xargs | sed 's/ / | /g')
+        echo "| $RESULTS |" >> $GITHUB_STEP_SUMMARY
+
+        # only include failed tests section if there are any
+        if grep -q '\[FAIL\]' /tmp/e2e.log; then
+          echo ''                       >> $GITHUB_STEP_SUMMARY
+          echo '### Failed Tests'       >> $GITHUB_STEP_SUMMARY
+          echo '```'                    >> $GITHUB_STEP_SUMMARY
+          grep '\[FAIL\]' /tmp/e2e.log  >> $GITHUB_STEP_SUMMARY
+          echo '```'                    >> $GITHUB_STEP_SUMMARY
+        fi
diff --git a/.gitignore b/.gitignore
index b01c05eb..cce85605 100644
--- a/.gitignore
+++ b/.gitignore
@@ -11,3 +11,8 @@ __debug*
 cover.out
 covcounters.**
 covmeta.**
+.claude/
+.claude-flow/
+.swarm
+.mcp.json
+CLAUDE.md
\ No newline at end of file
diff --git a/charts/k3k/templates/crds/k3k.io_clusters.yaml b/charts/k3k/templates/crds/k3k.io_clusters.yaml
index 35de1c3e..03aa27ef 100644
--- a/charts/k3k/templates/crds/k3k.io_clusters.yaml
+++ b/charts/k3k/templates/crds/k3k.io_clusters.yaml
@@ -1289,17 +1289,14 @@ spec:
                   are mirrored into the virtual cluster.
                 type: boolean
               mode:
-                allOf:
-                - enum:
-                  - shared
-                  - virtual
-                - enum:
-                  - shared
-                  - virtual
                 default: shared
                 description: |-
-                  Mode specifies the cluster provisioning mode: "shared" or "virtual".
+                  Mode specifies the cluster provisioning mode: "shared", "virtual" or "hcp".
                   Defaults to "shared". This field is immutable.
+                enum:
+                - shared
+                - virtual
+                - hcp
                 type: string
                 x-kubernetes-validations:
                 - message: mode is immutable
@@ -1419,7 +1416,7 @@ spec:
                     role:
                       description: |-
                         Role is the type of the k3k pod that will be used to mount the secret.
-                        This can be 'server', 'agent', or 'all' (for both).
+                        This can be `server`, `agent`, or `all` (for both).
                       enum:
                       - server
                       - agent
diff --git a/charts/k3k/templates/crds/k3k.io_virtualclusterpolicies.yaml b/charts/k3k/templates/crds/k3k.io_virtualclusterpolicies.yaml
index a056411b..e4996125 100644
--- a/charts/k3k/templates/crds/k3k.io_virtualclusterpolicies.yaml
+++ b/charts/k3k/templates/crds/k3k.io_virtualclusterpolicies.yaml
@@ -55,6 +55,7 @@ spec:
                 enum:
                 - shared
                 - virtual
+                - hcp
                 type: string
                 x-kubernetes-validations:
                 - message: mode is immutable
diff --git a/cli/cmds/cluster_create.go b/cli/cmds/cluster_create.go
index cb6af241..9978ce2f 100644
--- a/cli/cmds/cluster_create.go
+++ b/cli/cmds/cluster_create.go
@@ -5,7 +5,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"net/url"
 	"os"
 	"strings"
 	"text/template"
@@ -84,10 +83,14 @@ func createAction(appCtx *AppContext, config *CreateConfig) func(cmd *cobra.Comm
 			return errors.New("invalid cluster name")
 		}
 
-		if config.mode == string(v1beta1.SharedClusterMode) && config.agents != 0 {
+		if config.agents > 0 && config.mode != string(v1beta1.VirtualClusterMode) {
 			return errors.New("invalid flag, --agents flag is only allowed in virtual mode")
 		}
 
+		if config.mode == string(v1beta1.HCPClusterMode) {
+			logrus.Warn("HCP (Hosted Control Plane) mode is experimental.")
+		}
+
 		namespace := appCtx.Namespace(name)
 
 		if err := createNamespace(ctx, client, namespace, config.policy); err != nil {
@@ -128,17 +131,12 @@ func createAction(appCtx *AppContext, config *CreateConfig) func(cmd *cobra.Comm
 		}
 
 		// add Host IP address as an extra TLS-SAN to expose the k3k cluster
-		url, err := url.Parse(appCtx.RestConfig.Host)
+		host, err := resolveServerHost(appCtx.RestConfig.Host, config.kubeconfigServerHost)
 		if err != nil {
 			return err
 		}
 
-		host := strings.Split(url.Host, ":")
-		if config.kubeconfigServerHost != "" {
-			host = []string{config.kubeconfigServerHost}
-		}
-
-		cluster.Spec.TLSSANs = []string{host[0]}
+		cluster.Spec.TLSSANs = []string{host}
 
 		if err := client.Create(ctx, cluster); err != nil {
 			if apierrors.IsAlreadyExists(err) {
@@ -179,16 +177,40 @@ func createAction(appCtx *AppContext, config *CreateConfig) func(cmd *cobra.Comm
 		var kubeconfig *clientcmdapi.Config
 
 		if err := retry.OnError(availableBackoff, apierrors.IsNotFound, func() error {
-			kubeconfig, err = cfg.Generate(ctx, client, cluster, host[0], 0)
+			kubeconfig, err = cfg.Generate(ctx, client, cluster, host, 0)
 			return err
 		}); err != nil {
 			return err
 		}
 
-		return writeKubeconfigFile(cluster, kubeconfig, "")
+		if err := writeKubeconfigFile(cluster, kubeconfig, ""); err != nil {
+			return err
+		}
+
+		if cluster.Spec.Mode == v1beta1.HCPClusterMode {
+			printHCPJoinInstructions(cluster, kubeconfig)
+		}
+
+		return nil
 	}
 }
 
+func printHCPJoinInstructions(cluster *v1beta1.Cluster, kc *clientcmdapi.Config) {
+	tokenSecretName := k3kcluster.TokenSecretName(cluster.Name)
+	serverURL := kc.Clusters["default"].Server
+
+	logrus.Infof(`To join an external worker node to this HCP cluster:
+
+	1. On this machine, fetch the cluster token:
+
+		kubectl get secret -n %s %s -o jsonpath='{.data.token}' | base64 -d
+
+	2. On the worker node, run (replace <TOKEN> with the value from step 1):
+
+		curl -sfL https://get.k3s.io | K3S_URL=%s K3S_TOKEN=<TOKEN> sh -
+`, cluster.Namespace, tokenSecretName, serverURL)
+}
+
 func newCluster(name, namespace string, config *CreateConfig) (*v1beta1.Cluster, error) {
 	var storageRequestSize *resource.Quantity
 	if config.storageRequestSize != "" {
diff --git a/cli/cmds/cluster_create_flags.go b/cli/cmds/cluster_create_flags.go
index 47d2585b..ec6adc28 100644
--- a/cli/cmds/cluster_create_flags.go
+++ b/cli/cmds/cluster_create_flags.go
@@ -27,7 +27,7 @@ func createFlags(cmd *cobra.Command, cfg *CreateConfig) {
 	cmd.Flags().StringArrayVar(&cfg.labels, "labels", []string{}, "Labels to add to the cluster object (e.g. key=value)")
 	cmd.Flags().StringArrayVar(&cfg.annotations, "annotations", []string{}, "Annotations to add to the cluster object (e.g. key=value)")
 	cmd.Flags().StringVar(&cfg.version, "version", "", "k3s version")
-	cmd.Flags().StringVar(&cfg.mode, "mode", "shared", "k3k mode type (shared, virtual)")
+	cmd.Flags().StringVar(&cfg.mode, "mode", "shared", "k3k mode type (shared, virtual, hcp)")
 	cmd.Flags().StringVar(&cfg.kubeconfigServerHost, "kubeconfig-server", "", "override the kubeconfig server host")
 	cmd.Flags().StringVar(&cfg.policy, "policy", "", "The policy to create the cluster in")
 	cmd.Flags().StringVar(&cfg.customCertsPath, "custom-certs", "", "The path for custom certificate directory")
@@ -54,10 +54,10 @@ func validateCreateConfig(cfg *CreateConfig) error {
 
 	if cfg.mode != "" {
 		switch cfg.mode {
-		case string(v1beta1.VirtualClusterMode), string(v1beta1.SharedClusterMode):
+		case string(v1beta1.VirtualClusterMode), string(v1beta1.SharedClusterMode), string(v1beta1.HCPClusterMode):
 			return nil
 		default:
-			return errors.New(`mode should be one of "shared" or "virtual"`)
+			return errors.New(`mode should be one of "shared", "virtual" or "hcp"`)
 		}
 	}
 
diff --git a/cli/cmds/kubeconfig.go b/cli/cmds/kubeconfig.go
index 39763fd4..8d6e3cac 100644
--- a/cli/cmds/kubeconfig.go
+++ b/cli/cmds/kubeconfig.go
@@ -5,7 +5,6 @@ import (
 	"net/url"
 	"os"
 	"path/filepath"
-	"strings"
 	"time"
 
 	"github.com/sirupsen/logrus"
@@ -89,14 +88,12 @@ func generate(appCtx *AppContext, cfg *GenerateKubeconfigConfig) func(cmd *cobra
 			return err
 		}
 
-		url, err := url.Parse(appCtx.RestConfig.Host)
+		host, err := resolveServerHost(appCtx.RestConfig.Host, cfg.kubeconfigServerHost)
 		if err != nil {
 			return err
 		}
 
-		host := strings.Split(url.Host, ":")
 		if cfg.kubeconfigServerHost != "" {
-			host = []string{cfg.kubeconfigServerHost}
 			cfg.altNames = append(cfg.altNames, cfg.kubeconfigServerHost)
 		}
 
@@ -118,16 +115,40 @@ func generate(appCtx *AppContext, cfg *GenerateKubeconfigConfig) func(cmd *cobra
 		var kubeconfig *clientcmdapi.Config
 
 		if err := retry.OnError(controller.Backoff, apierrors.IsNotFound, func() error {
-			kubeconfig, err = kubeCfg.Generate(ctx, client, &cluster, host[0], 0)
+			kubeconfig, err = kubeCfg.Generate(ctx, client, &cluster, host, 0)
 			return err
 		}); err != nil {
 			return err
 		}
 
-		return writeKubeconfigFile(&cluster, kubeconfig, cfg.configName)
+		if err := writeKubeconfigFile(&cluster, kubeconfig, cfg.configName); err != nil {
+			return err
+		}
+
+		if cluster.Spec.Mode == v1beta1.HCPClusterMode {
+			printHCPJoinInstructions(&cluster, kubeconfig)
+		}
+
+		return nil
 	}
 }
 
+// resolveServerHost returns the host that should be embedded in the kubeconfig
+// server URL and used as the TLS-SAN. If override is set it takes precedence;
+// otherwise the host is extracted from restConfigHost.
+func resolveServerHost(restConfigHost, override string) (string, error) {
+	if override != "" {
+		return override, nil
+	}
+
+	u, err := url.Parse(restConfigHost)
+	if err != nil {
+		return "", err
+	}
+
+	return u.Hostname(), nil
+}
+
 func writeKubeconfigFile(cluster *v1beta1.Cluster, kubeconfig *clientcmdapi.Config, configName string) error {
 	if configName == "" {
 		configName = cluster.Namespace + "-" + cluster.Name + "-kubeconfig.yaml"
diff --git a/cli/cmds/policy_create.go b/cli/cmds/policy_create.go
index 7d999d01..fd35e5e9 100644
--- a/cli/cmds/policy_create.go
+++ b/cli/cmds/policy_create.go
@@ -34,10 +34,10 @@ func NewPolicyCreateCmd(appCtx *AppContext) *cobra.Command {
 		Example: "k3kcli policy create [command options] NAME",
 		PreRunE: func(cmd *cobra.Command, args []string) error {
 			switch config.mode {
-			case string(v1beta1.VirtualClusterMode), string(v1beta1.SharedClusterMode):
+			case string(v1beta1.VirtualClusterMode), string(v1beta1.SharedClusterMode), string(v1beta1.HCPClusterMode):
 				return nil
 			default:
-				return errors.New(`mode should be one of "shared" or "virtual"`)
+				return errors.New(`mode should be one of "shared", "virtual" or "hcp"`)
 			}
 		},
 		RunE: policyCreateAction(appCtx, config),
diff --git a/docs/cli/k3kcli.adoc b/docs/cli/k3kcli.adoc
index b6fbebf5..e31ed8c7 100644
--- a/docs/cli/k3kcli.adoc
+++ b/docs/cli/k3kcli.adoc
@@ -54,7 +54,7 @@ k3kcli cluster create [command options] NAME
       --kubeconfig-server string      override the kubeconfig server host
       --labels stringArray            Labels to add to the cluster object (e.g. key=value)
       --mirror-host-nodes             Mirror Host Cluster Nodes
-      --mode string                   k3k mode type (shared, virtual) (default "shared")
+      --mode string                   k3k mode type (shared, virtual, hcp) (default "shared")
   -n, --namespace string              namespace of the k3k cluster
       --persistence-type string       persistence mode for the nodes (dynamic, ephemeral) (default "dynamic")
       --policy string                 The policy to create the cluster in
diff --git a/docs/cli/k3kcli_cluster_create.md b/docs/cli/k3kcli_cluster_create.md
index 7afccb21..dc487f24 100644
--- a/docs/cli/k3kcli_cluster_create.md
+++ b/docs/cli/k3kcli_cluster_create.md
@@ -25,7 +25,7 @@ k3kcli cluster create [command options] NAME
       --kubeconfig-server string      override the kubeconfig server host
       --labels stringArray            Labels to add to the cluster object (e.g. key=value)
       --mirror-host-nodes             Mirror Host Cluster Nodes
-      --mode string                   k3k mode type (shared, virtual) (default "shared")
+      --mode string                   k3k mode type (shared, virtual, hcp) (default "shared")
   -n, --namespace string              namespace of the k3k cluster
       --persistence-type string       persistence mode for the nodes (dynamic, ephemeral) (default "dynamic")
       --policy string                 The policy to create the cluster in
diff --git a/docs/crds/crds.adoc b/docs/crds/crds.adoc
index 6d01db49..3c75619b 100644
--- a/docs/crds/crds.adoc
+++ b/docs/crds/crds.adoc
@@ -133,8 +133,9 @@ _Underlying type:_ _string_
 
 ClusterMode is the possible provisioning mode of a Cluster.
 
-_Validation:_
-- Enum: [shared virtual]
+Supported values: `shared`, `virtual`, `hcp`.
+
+
 
 _Appears In:_
 
@@ -177,8 +178,8 @@ _Appears In:_
 | *`version`* __string__ | Version is the K3s version to use for the virtual nodes. +
 It should follow the K3s versioning convention (e.g., v1.28.2-k3s1). +
 If not specified, the Kubernetes version of the host node will be used. + |  | 
-| *`mode`* __xref:{anchor_prefix}-github-com-rancher-k3k-pkg-apis-k3k-io-v1beta1-clustermode[$$ClusterMode$$]__ | Mode specifies the cluster provisioning mode: "shared" or "virtual". +
-Defaults to "shared". This field is immutable. + | shared | Enum: [shared virtual] +
+| *`mode`* __xref:{anchor_prefix}-github-com-rancher-k3k-pkg-apis-k3k-io-v1beta1-clustermode[$$ClusterMode$$]__ | Mode specifies the cluster provisioning mode: "shared", "virtual" or "hcp". +
+Defaults to "shared". This field is immutable. + | shared | Enum: [shared virtual hcp] +
 
 | *`servers`* __integer__ | Servers specifies the number of K3s pods to run in server (control plane) mode. +
 Must be at least 1. Defaults to 1. + | 1 | 
@@ -547,8 +548,9 @@ _Underlying type:_ _string_
 
 PodSecurityAdmissionLevel is the policy level applied to the pods in the namespace.
 
-_Validation:_
-- Enum: [privileged baseline restricted]
+Supported values: `privileged`, `baseline`, `restricted`.
+
+
 
 _Appears In:_
 
@@ -618,7 +620,7 @@ secret contents will be mounted. + |  |
 When specified, only the specified key from the secret will be mounted as a file +
 at MountPath, keeping the parent directory writable. + |  | 
 | *`role`* __string__ | Role is the type of the k3k pod that will be used to mount the secret. +
-This can be 'server', 'agent', or 'all' (for both). + |  | Enum: [server agent all] +
+This can be `server`, `agent`, or `all` (for both). + |  | Enum: [server agent all] +
 
 |===
 
@@ -790,7 +792,7 @@ to set defaults and constraints (min/max) + |  |
 This includes both node affinity and pod affinity/anti-affinity rules. + |  | 
 | *`defaultAgentAffinity`* __link:https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.31/#affinity-v1-core[$$Affinity$$]__ | DefaultAgentAffinity specifies the affinity rules applied to agent pods of all clusters in the target Namespace. +
 This includes both node affinity and pod affinity/anti-affinity rules. + |  | 
-| *`allowedMode`* __xref:{anchor_prefix}-github-com-rancher-k3k-pkg-apis-k3k-io-v1beta1-clustermode[$$ClusterMode$$]__ | AllowedMode specifies the allowed cluster provisioning mode. Defaults to "shared". + | shared | Enum: [shared virtual] +
+| *`allowedMode`* __xref:{anchor_prefix}-github-com-rancher-k3k-pkg-apis-k3k-io-v1beta1-clustermode[$$ClusterMode$$]__ | AllowedMode specifies the allowed cluster provisioning mode. Defaults to "shared". + | shared | Enum: [shared virtual hcp] +
 
 | *`disableNetworkPolicy`* __boolean__ | DisableNetworkPolicy indicates whether to disable the creation of a default network policy for cluster isolation. + |  | 
 | *`podSecurityAdmissionLevel`* __xref:{anchor_prefix}-github-com-rancher-k3k-pkg-apis-k3k-io-v1beta1-podsecurityadmissionlevel[$$PodSecurityAdmissionLevel$$]__ | PodSecurityAdmissionLevel specifies the pod security admission level applied to the pods in the namespace. + |  | Enum: [privileged baseline restricted] +
diff --git a/docs/crds/crds.md b/docs/crds/crds.md
index 35bd7a27..ebeb197f 100644
--- a/docs/crds/crds.md
+++ b/docs/crds/crds.md
@@ -102,8 +102,9 @@ _Underlying type:_ _string_
 
 ClusterMode is the possible provisioning mode of a Cluster.
 
-_Validation:_
-- Enum: [shared virtual]
+Supported values: `shared`, `virtual`, `hcp`.
+
+
 
 _Appears in:_
 - [ClusterSpec](#clusterspec)
@@ -138,7 +139,7 @@ _Appears in:_
 | Field | Description | Default | Validation |
 | --- | --- | --- | --- |
 | `version` _string_ | Version is the K3s version to use for the virtual nodes.<br />It should follow the K3s versioning convention (e.g., v1.28.2-k3s1).<br />If not specified, the Kubernetes version of the host node will be used. |  |  |
-| `mode` _[ClusterMode](#clustermode)_ | Mode specifies the cluster provisioning mode: "shared" or "virtual".<br />Defaults to "shared". This field is immutable. | shared | Enum: [shared virtual] <br /> |
+| `mode` _[ClusterMode](#clustermode)_ | Mode specifies the cluster provisioning mode: "shared", "virtual" or "hcp".<br />Defaults to "shared". This field is immutable. | shared | Enum: [shared virtual hcp] <br /> |
 | `servers` _integer_ | Servers specifies the number of K3s pods to run in server (control plane) mode.<br />Must be at least 1. Defaults to 1. | 1 |  |
 | `agents` _integer_ | Agents specifies the number of K3s pods to run in agent (worker) mode.<br />Must be 0 or greater. Defaults to 0.<br />This field is ignored in "shared" mode. | 0 |  |
 | `clusterCIDR` _string_ | ClusterCIDR is the CIDR range for pod IPs.<br />Defaults to 10.42.0.0/16 in shared mode and 10.52.0.0/16 in virtual mode.<br />This field is immutable. |  |  |
@@ -409,8 +410,9 @@ _Underlying type:_ _string_
 
 PodSecurityAdmissionLevel is the policy level applied to the pods in the namespace.
 
-_Validation:_
-- Enum: [privileged baseline restricted]
+Supported values: `privileged`, `baseline`, `restricted`.
+
+
 
 _Appears in:_
 - [VirtualClusterPolicySpec](#virtualclusterpolicyspec)
@@ -454,7 +456,7 @@ _Appears in:_
 | `optional` _boolean_ | optional field specify whether the Secret or its keys must be defined |  |  |
 | `mountPath` _string_ | MountPath is the path within server and agent pods where the<br />secret contents will be mounted. |  |  |
 | `subPath` _string_ | SubPath is an optional path within the secret to mount instead of the root.<br />When specified, only the specified key from the secret will be mounted as a file<br />at MountPath, keeping the parent directory writable. |  |  |
-| `role` _string_ | Role is the type of the k3k pod that will be used to mount the secret.<br />This can be 'server', 'agent', or 'all' (for both). |  | Enum: [server agent all] <br /> |
+| `role` _string_ | Role is the type of the k3k pod that will be used to mount the secret.<br />This can be `server`, `agent`, or `all` (for both). |  | Enum: [server agent all] <br /> |
 
 
 #### SecretSyncConfig
@@ -590,7 +592,7 @@ _Appears in:_
 | `defaultPriorityClass` _string_ | DefaultPriorityClass specifies the priorityClassName applied to all pods of all clusters in the target Namespace. |  |  |
 | `defaultServerAffinity` _[Affinity](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.31/#affinity-v1-core)_ | DefaultServerAffinity specifies the affinity rules applied to server pods of all clusters in the target Namespace.<br />This includes both node affinity and pod affinity/anti-affinity rules. |  |  |
 | `defaultAgentAffinity` _[Affinity](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.31/#affinity-v1-core)_ | DefaultAgentAffinity specifies the affinity rules applied to agent pods of all clusters in the target Namespace.<br />This includes both node affinity and pod affinity/anti-affinity rules. |  |  |
-| `allowedMode` _[ClusterMode](#clustermode)_ | AllowedMode specifies the allowed cluster provisioning mode. Defaults to "shared". | shared | Enum: [shared virtual] <br /> |
+| `allowedMode` _[ClusterMode](#clustermode)_ | AllowedMode specifies the allowed cluster provisioning mode. Defaults to "shared". | shared | Enum: [shared virtual hcp] <br /> |
 | `disableNetworkPolicy` _boolean_ | DisableNetworkPolicy indicates whether to disable the creation of a default network policy for cluster isolation. |  |  |
 | `podSecurityAdmissionLevel` _[PodSecurityAdmissionLevel](#podsecurityadmissionlevel)_ | PodSecurityAdmissionLevel specifies the pod security admission level applied to the pods in the namespace. |  | Enum: [privileged baseline restricted] <br /> |
 | `sync` _[SyncConfig](#syncconfig)_ | Sync specifies the resources types that will be synced from virtual cluster to host cluster. | \{  \} |  |
diff --git a/examples/hcp-server.yaml b/examples/hcp-server.yaml
new file mode 100644
index 00000000..8b4ed923
--- /dev/null
+++ b/examples/hcp-server.yaml
@@ -0,0 +1,10 @@
+apiVersion: k3k.io/v1beta1
+kind: Cluster
+metadata:
+  name: hcp-server
+spec:
+  mode: hcp
+  servers: 1
+  version: v1.33.1-k3s1
+  expose:
+    nodePort: {}
diff --git a/pkg/apis/k3k.io/v1beta1/types.go b/pkg/apis/k3k.io/v1beta1/types.go
index 17738d42..199b1471 100644
--- a/pkg/apis/k3k.io/v1beta1/types.go
+++ b/pkg/apis/k3k.io/v1beta1/types.go
@@ -45,11 +45,11 @@ type ClusterSpec struct {
 	// +optional
 	Version string `json:"version,omitempty"`
 
-	// Mode specifies the cluster provisioning mode: "shared" or "virtual".
+	// Mode specifies the cluster provisioning mode: "shared", "virtual" or "hcp".
 	// Defaults to "shared". This field is immutable.
 	//
-	// +kubebuilder:default="shared"
-	// +kubebuilder:validation:Enum=shared;virtual
+	// +kubebuilder:default=shared
+	// +kubebuilder:validation:Enum=shared;virtual;hcp
 	// +kubebuilder:validation:XValidation:message="mode is immutable",rule="self == oldSelf"
 	// +optional
 	Mode ClusterMode `json:"mode,omitempty"`
@@ -253,10 +253,10 @@ type SecretMount struct {
 	// +optional
 	SubPath string `json:"subPath,omitempty"`
 	// Role is the type of the k3k pod that will be used to mount the secret.
-	// This can be 'server', 'agent', or 'all' (for both).
+	// This can be `server`, `agent`, or `all` (for both).
 	//
-	// +optional
 	// +kubebuilder:validation:Enum=server;agent;all
+	// +optional
 	Role string `json:"role,omitempty"`
 }
 
@@ -413,8 +413,7 @@ type StorageClassSyncConfig struct {
 
 // ClusterMode is the possible provisioning mode of a Cluster.
 //
-// +kubebuilder:validation:Enum=shared;virtual
-// +kubebuilder:default="shared"
+// Supported values: `shared`, `virtual`, `hcp`.
 type ClusterMode string
 
 const (
@@ -423,6 +422,11 @@ const (
 
 	// VirtualClusterMode represents a cluster that runs in a virtual environment.
 	VirtualClusterMode = ClusterMode("virtual")
+
+	// HCPClusterMode represents a Hosted Control Plane: an agentless K3s control
+	// plane managed by k3k inside the host cluster. End users join their own
+	// external nodes (BYO) using the standard K3s installer command.
+	HCPClusterMode = ClusterMode("hcp")
 )
 
 // PersistenceMode is the storage mode of a Cluster.
@@ -626,7 +630,7 @@ type ClusterStatus struct {
 
 	// Phase is a high-level summary of the cluster's current lifecycle state.
 	//
-	// +kubebuilder:default="Unknown"
+	// +kubebuilder:default=Unknown
 	// +kubebuilder:validation:Enum=Pending;Provisioning;Ready;Failed;Terminating;Unknown
 	// +optional
 	Phase ClusterPhase `json:"phase,omitempty"`
@@ -774,6 +778,7 @@ type VirtualClusterPolicySpec struct {
 	// AllowedMode specifies the allowed cluster provisioning mode. Defaults to "shared".
 	//
 	// +kubebuilder:default=shared
+	// +kubebuilder:validation:Enum=shared;virtual;hcp
 	// +kubebuilder:validation:XValidation:message="mode is immutable",rule="self == oldSelf"
 	// +optional
 	AllowedMode ClusterMode `json:"allowedMode,omitempty"`
@@ -785,6 +790,7 @@ type VirtualClusterPolicySpec struct {
 
 	// PodSecurityAdmissionLevel specifies the pod security admission level applied to the pods in the namespace.
 	//
+	// +kubebuilder:validation:Enum=privileged;baseline;restricted
 	// +optional
 	PodSecurityAdmissionLevel *PodSecurityAdmissionLevel `json:"podSecurityAdmissionLevel,omitempty"`
 
@@ -817,7 +823,7 @@ type VirtualClusterPolicySpec struct {
 
 // PodSecurityAdmissionLevel is the policy level applied to the pods in the namespace.
 //
-// +kubebuilder:validation:Enum=privileged;baseline;restricted
+// Supported values: `privileged`, `baseline`, `restricted`.
 type PodSecurityAdmissionLevel string
 
 const (
diff --git a/pkg/controller/cluster/cluster.go b/pkg/controller/cluster/cluster.go
index ba3946dc..6b5326bc 100644
--- a/pkg/controller/cluster/cluster.go
+++ b/pkg/controller/cluster/cluster.go
@@ -400,9 +400,10 @@ func (c *ClusterReconciler) reconcile(ctx context.Context, cluster *v1beta1.Clus
 			}
 		}
 
-		// in virtual mode assign a default serviceCIDR
-		if cluster.Spec.Mode == v1beta1.VirtualClusterMode {
-			log.V(1).Info("assign default service CIDR for virtual mode")
+		// virtual and hcp modes both run a self-contained K3s control plane and
+		// need their own pod/service CIDR independent of the host cluster.
+		if cluster.Spec.Mode == v1beta1.VirtualClusterMode || cluster.Spec.Mode == v1beta1.HCPClusterMode {
+			log.V(1).Info("assign default service CIDR", "mode", cluster.Spec.Mode)
 
 			cluster.Status.ServiceCIDR = defaultVirtualServiceCIDR
 		}
@@ -447,6 +448,27 @@ func (c *ClusterReconciler) reconcile(ctx context.Context, cluster *v1beta1.Clus
 		return err
 	}
 
+	// In hcp mode, validate that the cluster has an externally-routable
+	// API server endpoint (NodePort / LoadBalancer / Ingress) so external
+	// workers can join. The join command itself is printed by the CLI
+	// (`k3kcli cluster create` / `k3kcli kubeconfig generate`).
+	// We also own the default/kubernetes Endpoints/EndpointSlice inside the
+	// virtual cluster (the apiserver reconciler is disabled for HCP) so
+	// external-node pods can reach the in-cluster apiserver ClusterIP.
+	if cluster.Spec.Mode == v1beta1.HCPClusterMode {
+		if err := c.ensureHCPRegistration(ctx, cluster); err != nil {
+			return err
+		}
+
+		if err := c.ensureHCPKubernetesEndpointSlice(ctx, cluster); err != nil {
+			return err
+		}
+
+		if err := c.ensureHCPKubernetesEndpoints(ctx, cluster); err != nil {
+			return err
+		}
+	}
+
 	// Important: if you need to call the Server API of the Virtual Cluster
 	// this needs to be done AFTER he kubeconfig has been generated
 
@@ -896,6 +918,14 @@ func (c *ClusterReconciler) bindClusterRoles(ctx context.Context, cluster *v1bet
 }
 
 func (c *ClusterReconciler) ensureAgent(ctx context.Context, cluster *v1beta1.Cluster, serviceIP, token string) error {
+	// hcp mode is BYO-node by design: external (out-of-host-cluster) nodes
+	// join using the standard K3s installer command, which is printed by
+	// `k3kcli cluster create` / `k3kcli kubeconfig generate`. k3k therefore
+	// does not provision any agent pods on the host cluster.
+	if cluster.Spec.Mode == v1beta1.HCPClusterMode {
+		return nil
+	}
+
 	config := agent.NewConfig(cluster, c.Client, c.Scheme)
 
 	var agentEnsurer agent.ResourceEnsurer
diff --git a/pkg/controller/cluster/hcp.go b/pkg/controller/cluster/hcp.go
new file mode 100644
index 00000000..18d44fd0
--- /dev/null
+++ b/pkg/controller/cluster/hcp.go
@@ -0,0 +1,325 @@
+package cluster
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"net/url"
+	"strconv"
+
+	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
+
+	corev1 "k8s.io/api/core/v1"
+	discoveryv1 "k8s.io/api/discovery/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	ctrl "sigs.k8s.io/controller-runtime"
+
+	"github.com/rancher/k3k/pkg/apis/k3k.io/v1beta1"
+	"github.com/rancher/k3k/pkg/controller/cluster/server"
+)
+
+// ErrHCPNoExternalEndpoint is returned by ensureHCPRegistration when an
+// HCP-mode cluster has no externally-routable endpoint (no NodePort,
+// LoadBalancer or Ingress) so external worker nodes cannot reach the API
+// server. updateStatus translates it into a Ready=False condition with
+// reason HCPNoExternalEndpoint instead of failing the reconcile outright.
+var ErrHCPNoExternalEndpoint = errors.New("HCP cluster has no external endpoint")
+
+// ensureHCPRegistration verifies that an HCP-mode cluster exposes an
+// externally-routable API server endpoint so external worker nodes can join.
+// Join instructions (the `curl ... | sh -` line) are printed by the CLI
+// (`k3kcli cluster create` / `k3kcli kubeconfig generate`); the controller
+// does not persist them on the Cluster object.
+//
+// Returns ErrHCPNoExternalEndpoint when no NodePort, LoadBalancer or Ingress
+// is configured, which updateStatus surfaces as Ready=False with reason
+// HCPNoExternalEndpoint.
+func (c *ClusterReconciler) ensureHCPRegistration(ctx context.Context, cluster *v1beta1.Cluster) error {
+	log := ctrl.LoggerFrom(ctx)
+
+	_, external, err := server.ServerURL(ctx, c.Client, cluster, selectNonLoopbackSAN(cluster), 0)
+	if err != nil {
+		return err
+	}
+
+	if !external {
+		log.Info("HCP cluster has no externally-routable endpoint",
+			"cluster", cluster.Name, "namespace", cluster.Namespace)
+
+		return ErrHCPNoExternalEndpoint
+	}
+
+	return nil
+}
+
+// selectNonLoopbackSAN returns the first non-loopback address from the
+// cluster's TLS SANs, preferring spec.TLSSANs then falling back to status.TLSSANs.
+// Returns empty string if no non-loopback address is found.
+func selectNonLoopbackSAN(cluster *v1beta1.Cluster) string {
+	// Try spec.TLSSANs first (user-provided values)
+	for _, san := range cluster.Spec.TLSSANs {
+		if ip := net.ParseIP(san); ip != nil && ip.IsLoopback() {
+			continue
+		}
+
+		if san == "localhost" {
+			continue
+		}
+
+		return san // Found a non-loopback address
+	}
+
+	// Fall back to status.TLSSANs (computed values)
+	for _, san := range cluster.Status.TLSSANs {
+		if ip := net.ParseIP(san); ip != nil && ip.IsLoopback() {
+			continue
+		}
+
+		if san == "localhost" {
+			continue
+		}
+
+		return san
+	}
+
+	return ""
+}
+
+// ensureHCPKubernetesEndpointSlice maintains the default/kubernetes Service
+// EndpointSlice inside the virtual cluster, pointing it at the externally
+// reachable host:port (NodePort / LoadBalancer / Ingress) so that pods
+// scheduled on external worker nodes can reach the in-cluster apiserver
+// ClusterIP.
+//
+// Background: the kube-apiserver normally reconciles default/kubernetes
+// EndpointSlice to its own --advertise-address:--secure-port (the host-cluster
+// pod IP and 6443). External worker nodes have no route to the host-cluster
+// pod CIDR, so kube-proxy DNAT to that endpoint fails. We disable the
+// apiserver reconciler in HCP mode (see serverOptions) and own this
+// EndpointSlice object instead.
+func (c *ClusterReconciler) ensureHCPKubernetesEndpointSlice(ctx context.Context, cluster *v1beta1.Cluster) error {
+	log := ctrl.LoggerFrom(ctx)
+
+	rawURL, external, err := server.ServerURL(ctx, c.Client, cluster, selectNonLoopbackSAN(cluster), 0)
+	if err != nil {
+		return err
+	}
+
+	if !external {
+		// Defensive: reconcile would have already short-circuited with
+		// ErrHCPNoExternalEndpoint via ensureHCPRegistration before reaching
+		// here, but skip gracefully if invoked directly.
+		return nil
+	}
+
+	host, port, err := parseHCPHostPort(rawURL)
+	if err != nil {
+		return fmt.Errorf("parsing HCP server URL %q: %w", rawURL, err)
+	}
+
+	addr, err := hcpEndpointAddress(ctx, host)
+	if err != nil {
+		return err
+	}
+
+	var addressType discoveryv1.AddressType
+
+	if ip := net.ParseIP(addr.IP); ip != nil {
+		if ip.To4() != nil {
+			addressType = discoveryv1.AddressTypeIPv4
+		} else {
+			addressType = discoveryv1.AddressTypeIPv6
+		}
+	} else {
+		return fmt.Errorf("invalid IP address %q", addr.IP)
+	}
+
+	virtClient, err := newVirtualClient(ctx, c.Client, cluster.Name, cluster.Namespace)
+	if err != nil {
+		return fmt.Errorf("creating virtual cluster client: %w", err)
+	}
+
+	endpointSlice := &discoveryv1.EndpointSlice{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "kubernetes",
+			Namespace: metav1.NamespaceDefault,
+		},
+	}
+
+	_, err = controllerutil.CreateOrUpdate(ctx, virtClient, endpointSlice, func() error {
+		if endpointSlice.Labels == nil {
+			endpointSlice.Labels = make(map[string]string)
+		}
+
+		// Ensure the service-name label is set
+		endpointSlice.Labels[discoveryv1.LabelServiceName] = "kubernetes"
+		endpointSlice.AddressType = addressType
+
+		endpointSlice.Endpoints = []discoveryv1.Endpoint{
+			{Addresses: []string{addr.IP}},
+		}
+		portName := "https"
+
+		endpointSlice.Ports = []discoveryv1.EndpointPort{
+			{
+				Name:     &portName,
+				Port:     &port,
+				Protocol: new(corev1.ProtocolTCP),
+			},
+		}
+
+		return nil
+	})
+	if err != nil {
+		return fmt.Errorf("upserting default/kubernetes endpointslice in virtual cluster: %w", err)
+	}
+
+	log.V(1).Info("HCP kubernetes endpointslice reconciled",
+		"address", addr.IP, "host", host, "port", port)
+
+	return nil
+}
+
+func (c *ClusterReconciler) ensureHCPKubernetesEndpoints(ctx context.Context, cluster *v1beta1.Cluster) error {
+	log := ctrl.LoggerFrom(ctx)
+
+	rawURL, external, err := server.ServerURL(ctx, c.Client, cluster, selectNonLoopbackSAN(cluster), 0)
+	if err != nil {
+		return err
+	}
+
+	if !external {
+		// Defensive: reconcile would have already short-circuited with
+		// ErrHCPNoExternalEndpoint via ensureHCPRegistration before reaching
+		// here, but skip gracefully if invoked directly.
+		return nil
+	}
+
+	host, port, err := parseHCPHostPort(rawURL)
+	if err != nil {
+		return fmt.Errorf("parsing HCP server URL %q: %w", rawURL, err)
+	}
+
+	addr, err := hcpEndpointAddress(ctx, host)
+	if err != nil {
+		return err
+	}
+
+	virtClient, err := newVirtualClient(ctx, c.Client, cluster.Name, cluster.Namespace)
+	if err != nil {
+		return fmt.Errorf("creating virtual cluster client: %w", err)
+	}
+
+	//nolint:staticcheck // SA1019 corev1.Endpoints is deprecated in v1.33+, but needed in the Conformance tests
+	// We are already using the discoveryv1.EndpointSlice
+	endpoints := &corev1.Endpoints{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "kubernetes",
+			Namespace: metav1.NamespaceDefault,
+		},
+	}
+
+	_, err = controllerutil.CreateOrUpdate(ctx, virtClient, endpoints, func() error {
+		if endpoints.Labels == nil {
+			endpoints.Labels = make(map[string]string)
+		}
+
+		// Ensure the skip-mirror label is set
+		endpoints.Labels[discoveryv1.LabelSkipMirror] = "true"
+
+		//nolint:staticcheck // SA1019 corev1.EndpointSubset is deprecated in v1.33+, but needed in the Conformance tests
+		endpoints.Subsets = []corev1.EndpointSubset{
+			{
+				Addresses: []corev1.EndpointAddress{addr},
+				Ports: []corev1.EndpointPort{
+					{
+						Name:     "https",
+						Port:     port,
+						Protocol: corev1.ProtocolTCP,
+					},
+				},
+			},
+		}
+
+		return nil
+	})
+	if err != nil {
+		return fmt.Errorf("upserting default/kubernetes endpoints in virtual cluster: %w", err)
+	}
+
+	log.V(1).Info("HCP kubernetes endpoints reconciled", "address", addr.IP, "host", host, "port", port)
+
+	return nil
+}
+
+// parseHCPHostPort extracts the host and port from a server URL produced by
+// server.ServerURL. The port defaults to 443 when omitted.
+func parseHCPHostPort(rawURL string) (string, int32, error) {
+	u, err := url.Parse(rawURL)
+	if err != nil {
+		return "", 0, err
+	}
+
+	host := u.Hostname()
+	if host == "" {
+		return "", 0, fmt.Errorf("missing host in URL %q", rawURL)
+	}
+
+	portStr := u.Port()
+
+	var port int32 = 443
+
+	if portStr != "" {
+		p, err := strconv.Atoi(portStr)
+		if err != nil {
+			return "", 0, fmt.Errorf("invalid port in URL %q: %w", rawURL, err)
+		}
+
+		if p <= 0 || p > 65535 {
+			return "", 0, fmt.Errorf("port %d out of range in URL %q", p, rawURL)
+		}
+
+		port = int32(p)
+	}
+
+	return host, port, nil
+}
+
+// hcpEndpointAddress builds a corev1.EndpointAddress from the externally
+// reachable host. Endpoints require an IP; if the host is a DNS name we
+// resolve it. The Hostname field is intentionally left unset:
+// the kubernetes API validates it as a DNS-1123 label (no dots),
+// so an FQDN like "host.example.com" would be rejected.
+func hcpEndpointAddress(ctx context.Context, host string) (corev1.EndpointAddress, error) {
+	if ip := net.ParseIP(host); ip != nil {
+		if ip.IsLoopback() {
+			return corev1.EndpointAddress{}, fmt.Errorf("HCP endpoint host %q is a loopback address and cannot be used", host)
+		}
+
+		return corev1.EndpointAddress{IP: host}, nil
+	}
+
+	ipAddrs, err := net.DefaultResolver.LookupIPAddr(ctx, host)
+	if err != nil {
+		return corev1.EndpointAddress{}, fmt.Errorf("HCP endpoint host %q is not an IP and does not resolve: %w", host, err)
+	}
+
+	var filteredIPs []net.IP
+
+	for _, addr := range ipAddrs {
+		if !addr.IP.IsLoopback() {
+			filteredIPs = append(filteredIPs, addr.IP)
+		}
+	}
+
+	if len(filteredIPs) == 0 {
+		return corev1.EndpointAddress{}, fmt.Errorf("HCP endpoint host %q resolved to no non-loopback IPs", host)
+	}
+
+	if v4 := filteredIPs[0].To4(); v4 != nil {
+		return corev1.EndpointAddress{IP: v4.String()}, nil
+	}
+
+	return corev1.EndpointAddress{IP: filteredIPs[0].String()}, nil
+}
diff --git a/pkg/controller/cluster/hcp_test.go b/pkg/controller/cluster/hcp_test.go
new file mode 100644
index 00000000..f264d2b2
--- /dev/null
+++ b/pkg/controller/cluster/hcp_test.go
@@ -0,0 +1,279 @@
+package cluster
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"k8s.io/apimachinery/pkg/runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	"github.com/rancher/k3k/pkg/apis/k3k.io/v1beta1"
+	"github.com/rancher/k3k/pkg/controller"
+	"github.com/rancher/k3k/pkg/controller/cluster/server"
+)
+
+func Test_ensureHCPRegistration(t *testing.T) {
+	scheme := runtime.NewScheme()
+	require.NoError(t, corev1.AddToScheme(scheme))
+	require.NoError(t, v1beta1.AddToScheme(scheme))
+
+	cluster := &v1beta1.Cluster{
+		ObjectMeta: metav1.ObjectMeta{Name: "demo", Namespace: "team-a"},
+		Spec: v1beta1.ClusterSpec{
+			Mode:    v1beta1.HCPClusterMode,
+			Version: "v1.33.1-k3s1",
+			TLSSANs: []string{"hcp.example.com"},
+		},
+		Status: v1beta1.ClusterStatus{
+			TLSSANs: []string{"hcp.example.com"},
+		},
+	}
+
+	t.Run("clusterip-only service returns ErrHCPNoExternalEndpoint", func(t *testing.T) {
+		svc := &corev1.Service{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      server.ServiceName(cluster.Name),
+				Namespace: cluster.Namespace,
+			},
+			Spec: corev1.ServiceSpec{
+				Type:      corev1.ServiceTypeClusterIP,
+				ClusterIP: "10.43.0.50",
+				Ports: []corev1.ServicePort{
+					{Name: "k3s-server-port", Port: 443},
+				},
+			},
+		}
+
+		fakeClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(svc).Build()
+		r := &ClusterReconciler{Client: fakeClient}
+
+		c := cluster.DeepCopy()
+		err := r.ensureHCPRegistration(context.Background(), c)
+		require.Error(t, err)
+		assert.True(t, errors.Is(err, ErrHCPNoExternalEndpoint))
+	})
+
+	t.Run("nodeport service returns no error", func(t *testing.T) {
+		svc := &corev1.Service{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      server.ServiceName(cluster.Name),
+				Namespace: cluster.Namespace,
+			},
+			Spec: corev1.ServiceSpec{
+				Type:      corev1.ServiceTypeNodePort,
+				ClusterIP: "10.43.0.50",
+				Ports: []corev1.ServicePort{
+					{Name: "k3s-server-port", Port: 443, NodePort: 30443},
+				},
+			},
+		}
+
+		fakeClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(svc).Build()
+		r := &ClusterReconciler{Client: fakeClient}
+
+		c := cluster.DeepCopy()
+		assert.NoError(t, r.ensureHCPRegistration(context.Background(), c))
+	})
+}
+
+func Test_parseHCPHostPort(t *testing.T) {
+	tests := []struct {
+		name     string
+		url      string
+		wantHost string
+		wantPort int32
+		wantErr  bool
+	}{
+		{
+			name:     "ip with explicit port",
+			url:      "https://10.144.101.195:30337",
+			wantHost: "10.144.101.195",
+			wantPort: 30337,
+		},
+		{
+			name:     "hostname without port defaults to 443",
+			url:      "https://hcp.example.com",
+			wantHost: "hcp.example.com",
+			wantPort: 443,
+		},
+		{
+			name:     "hostname with explicit port",
+			url:      "https://hcp.example.com:6443",
+			wantHost: "hcp.example.com",
+			wantPort: 6443,
+		},
+		{
+			name:    "missing host",
+			url:     "https://",
+			wantErr: true,
+		},
+		{
+			name:    "non-numeric port",
+			url:     "https://host:abc",
+			wantErr: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			host, port, err := parseHCPHostPort(tt.url)
+			if tt.wantErr {
+				require.Error(t, err)
+				return
+			}
+
+			require.NoError(t, err)
+			assert.Equal(t, tt.wantHost, host)
+			assert.Equal(t, tt.wantPort, port)
+		})
+	}
+}
+
+func Test_selectNonLoopbackSAN(t *testing.T) {
+	tests := []struct {
+		name       string
+		specSANs   []string
+		statusSANs []string
+		want       string
+	}{
+		{
+			name:       "spec with loopback first, external second",
+			specSANs:   []string{"127.0.0.1", "10.0.0.100"},
+			statusSANs: []string{},
+			want:       "10.0.0.100",
+		},
+		{
+			name:       "spec with external first",
+			specSANs:   []string{"10.0.0.100", "127.0.0.1"},
+			statusSANs: []string{},
+			want:       "10.0.0.100",
+		},
+		{
+			name:       "spec empty, status with loopback first, external second",
+			specSANs:   []string{},
+			statusSANs: []string{"127.0.0.1", "10.43.0.50"},
+			want:       "10.43.0.50",
+		},
+		{
+			name:       "spec with only loopback",
+			specSANs:   []string{"127.0.0.1", "::1"},
+			statusSANs: []string{},
+			want:       "",
+		},
+		{
+			name:       "spec with localhost hostname",
+			specSANs:   []string{"localhost", "example.com"},
+			statusSANs: []string{},
+			want:       "example.com",
+		},
+		{
+			name:       "spec with external hostname",
+			specSANs:   []string{"hcp.example.com"},
+			statusSANs: []string{},
+			want:       "hcp.example.com",
+		},
+		{
+			name:       "empty spec and status",
+			specSANs:   []string{},
+			statusSANs: []string{},
+			want:       "",
+		},
+		{
+			name:       "ipv6 loopback filtered",
+			specSANs:   []string{"::1", "2001:db8::1"},
+			statusSANs: []string{},
+			want:       "2001:db8::1",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			cluster := &v1beta1.Cluster{
+				Spec: v1beta1.ClusterSpec{
+					TLSSANs: tt.specSANs,
+				},
+				Status: v1beta1.ClusterStatus{
+					TLSSANs: tt.statusSANs,
+				},
+			}
+			got := selectNonLoopbackSAN(cluster)
+			assert.Equal(t, tt.want, got)
+		})
+	}
+}
+
+func Test_hcpEndpointAddress(t *testing.T) {
+	tests := []struct {
+		name         string
+		input        string
+		wantIP       string
+		wantHostname string
+		wantErr      bool
+	}{
+		{
+			name:         "ipv4 literal is passed through",
+			input:        "10.144.101.195",
+			wantIP:       "10.144.101.195",
+			wantHostname: "",
+			wantErr:      false,
+		},
+		{
+			name:    "unresolvable hostname errors",
+			input:   "definitely-not-a-real-host.invalid",
+			wantErr: true,
+		},
+		{
+			name:    "ipv4 loopback literal is rejected",
+			input:   "127.0.0.1",
+			wantErr: true,
+		},
+		{
+			name:    "ipv6 loopback literal is rejected",
+			input:   "::1",
+			wantErr: true,
+		},
+		{
+			name:    "ipv4 loopback in range is rejected",
+			input:   "127.0.0.100",
+			wantErr: true,
+		},
+		{
+			name:         "valid ipv6 literal passes through",
+			input:        "2001:db8::1",
+			wantIP:       "2001:db8::1",
+			wantHostname: "",
+			wantErr:      false,
+		},
+		{
+			name:    "localhost hostname filters loopbacks",
+			input:   "localhost",
+			wantErr: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := hcpEndpointAddress(context.Background(), tt.input)
+			if tt.wantErr {
+				require.Error(t, err)
+				return
+			}
+
+			require.NoError(t, err)
+
+			assert.Equal(t, tt.wantIP, got.IP)
+			assert.Equal(t, tt.wantHostname, got.Hostname)
+		})
+	}
+}
+
+// Compile-time assertion: every reused exported name from the controller
+// package below this test file must remain stable. If `controller.K3SImage`
+// disappears (refactor), this guards the dependency.
+var _ = controller.K3SImage
diff --git a/pkg/controller/cluster/server/config.go b/pkg/controller/cluster/server/config.go
index 52136afd..f4635003 100644
--- a/pkg/controller/cluster/server/config.go
+++ b/pkg/controller/cluster/server/config.go
@@ -11,7 +11,6 @@ import (
 
 	"github.com/rancher/k3k/pkg/apis/k3k.io/v1beta1"
 	"github.com/rancher/k3k/pkg/controller"
-	"github.com/rancher/k3k/pkg/controller/cluster/agent"
 )
 
 // serverConfig are few options from k3s server options that will
@@ -23,6 +22,7 @@ type serverConfig struct {
 	DisableAgent       bool     `yaml:"disable-agent,omitempty"`
 	Disable            []string `yaml:"disable,omitempty"`
 	EgressSelectorMode string   `yaml:"egress-selector-mode,omitempty"`
+	KubeApiServerArg   []string `yaml:"kube-apiserver-arg,omitempty"`
 	Server             string   `yaml:"server,omitempty"`
 	ServiceCIDR        string   `yaml:"service-cidr,omitempty"`
 	TLSSAN             []string `yaml:"tls-san,omitempty"`
@@ -77,10 +77,29 @@ func buildServerConfig(cluster *v1beta1.Cluster, initServer bool, serviceIP, tok
 		serverConfig.Server = "https://" + serviceIP
 	}
 
-	if cluster.Spec.Mode != agent.VirtualNodeMode {
+	// shared and hcp modes both run K3s with --disable-agent (agentless server).
+	switch cluster.Spec.Mode {
+	case "", v1beta1.SharedClusterMode:
 		serverConfig.DisableAgent = true
 		serverConfig.EgressSelectorMode = "disabled"
 		serverConfig.Disable = []string{"servicelb", "traefik", "metrics-server", "local-storage"}
+	case v1beta1.HCPClusterMode:
+		serverConfig.DisableAgent = true
+		// Tunnel apiserver egress through the k3s-agent WebSocket: the
+		// apiserver has no route to the virtual cluster's pod CIDR and
+		// bypasses kube-proxy when dialing pod IPs (webhooks, log/exec).
+		// "cluster" is the only safe mode — "agent" lets pod dials go
+		// direct (no route, fails); "pod" only permits pod IPs the agent
+		// has already watched, so a newly-created pod's IP is rejected
+		// and tears down the remotedialer session, making kubelet streams
+		// flaky. See k3s pkg/agent/tunnel/tunnel.go.
+		serverConfig.EgressSelectorMode = "cluster"
+		// Disable the apiserver's built-in endpoint reconciler so K3k can
+		// own default/kubernetes Endpoints and point it at the externally
+		// reachable host:port (NodePort / LB / Ingress).
+		serverConfig.KubeApiServerArg = append(serverConfig.KubeApiServerArg, "endpoint-reconciler-type=none")
+	case v1beta1.VirtualClusterMode:
+		// no extra config for virtual mode
 	}
 
 	return serverConfig
diff --git a/pkg/controller/cluster/server/config_test.go b/pkg/controller/cluster/server/config_test.go
index fe873d50..ede7a79b 100644
--- a/pkg/controller/cluster/server/config_test.go
+++ b/pkg/controller/cluster/server/config_test.go
@@ -2,9 +2,10 @@ package server
 
 import (
 	"fmt"
-	"reflect"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
+
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"github.com/rancher/k3k/pkg/apis/k3k.io/v1beta1"
@@ -43,6 +44,9 @@ func Test_BuildServerConfig(t *testing.T) {
 						Name:      testClusterName,
 						Namespace: testClusterNamespace,
 					},
+					Spec: v1beta1.ClusterSpec{
+						Mode: v1beta1.SharedClusterMode,
+					},
 					Status: v1beta1.ClusterStatus{
 						ClusterCIDR: defaultSharedClusterCIDR,
 						ServiceCIDR: defaultSharedServiceCIDR,
@@ -71,6 +75,9 @@ func Test_BuildServerConfig(t *testing.T) {
 						Name:      testClusterName,
 						Namespace: testClusterNamespace,
 					},
+					Spec: v1beta1.ClusterSpec{
+						Mode: v1beta1.SharedClusterMode,
+					},
 					Status: v1beta1.ClusterStatus{
 						ClusterCIDR: defaultSharedClusterCIDR,
 						ServiceCIDR: defaultSharedServiceCIDR,
@@ -158,6 +165,7 @@ func Test_BuildServerConfig(t *testing.T) {
 						Namespace: testClusterNamespace,
 					},
 					Spec: v1beta1.ClusterSpec{
+						Mode:       v1beta1.SharedClusterMode,
 						ClusterDNS: testClusterDNS,
 					},
 					Status: v1beta1.ClusterStatus{
@@ -186,9 +194,7 @@ func Test_BuildServerConfig(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			serverConfig := buildServerConfig(tt.args.cluster, tt.args.initServer, tt.args.serviceIP, tt.args.token)
-			if !reflect.DeepEqual(tt.expectedData, serverConfig) {
-				t.Errorf("found %v, expected %v", serverConfig, tt.expectedData)
-			}
+			assert.Equal(t, tt.expectedData, serverConfig)
 		})
 	}
 }
diff --git a/pkg/controller/cluster/server/endpoint.go b/pkg/controller/cluster/server/endpoint.go
new file mode 100644
index 00000000..29ad8e77
--- /dev/null
+++ b/pkg/controller/cluster/server/endpoint.go
@@ -0,0 +1,124 @@
+package server
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"slices"
+	"strconv"
+
+	"github.com/sirupsen/logrus"
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	corev1 "k8s.io/api/core/v1"
+	networkingv1 "k8s.io/api/networking/v1"
+
+	"github.com/rancher/k3k/pkg/apis/k3k.io/v1beta1"
+)
+
+// ServerURL returns the URL at which the K3s API server of a virtual cluster
+// is reachable. The second return value reports whether that URL is routable
+// from outside the host cluster (true for NodePort/LoadBalancer/Ingress, false
+// for plain ClusterIP exposition).
+//
+// hostServerIP is used as the address when the underlying Service is a
+// NodePort. serverPort, when non-zero, overrides the port discovered from the
+// Service.
+func ServerURL(ctx context.Context, c client.Client, cluster *v1beta1.Cluster, hostServerIP string, serverPort int) (string, bool, error) {
+	key := types.NamespacedName{
+		Name:      ServiceName(cluster.Name),
+		Namespace: cluster.Namespace,
+	}
+
+	var k3kService corev1.Service
+	if err := c.Get(ctx, key, &k3kService); err != nil {
+		return "", false, err
+	}
+
+	ip := k3kService.Spec.ClusterIP
+	port := int32(httpsPort)
+	external := false
+
+	if len(k3kService.Spec.Ports) == 0 {
+		logrus.Warn("No ports exposed by the cluster service.")
+	}
+
+	switch k3kService.Spec.Type {
+	case corev1.ServiceTypeNodePort:
+		ip = hostServerIP
+		external = true
+
+		if len(k3kService.Spec.Ports) > 0 {
+			port = k3kService.Spec.Ports[0].NodePort
+		}
+	case corev1.ServiceTypeLoadBalancer:
+		if len(k3kService.Status.LoadBalancer.Ingress) > 0 {
+			ingress := k3kService.Status.LoadBalancer.Ingress[0]
+			switch {
+			case ingress.IP != "":
+				ip = ingress.IP
+				external = true
+			case ingress.Hostname != "":
+				ip = ingress.Hostname
+				external = true
+			}
+		}
+
+		if !external {
+			logrus.Warn("No usable ingress address found in LoadBalancer service.")
+		}
+
+		if len(k3kService.Spec.Ports) > 0 {
+			port = k3kService.Spec.Ports[0].Port
+		}
+	}
+
+	if serverPort != 0 {
+		port = int32(serverPort)
+	}
+
+	if !slices.Contains(cluster.Status.TLSSANs, ip) {
+		logrus.Warnf("IP %s not in tlsSANs.", ip)
+
+		if len(cluster.Spec.TLSSANs) > 0 {
+			logrus.Warnf("Using the first TLS SAN in the spec as a fallback: %s", cluster.Spec.TLSSANs[0])
+
+			ip = cluster.Spec.TLSSANs[0]
+		} else if len(cluster.Status.TLSSANs) > 0 {
+			logrus.Warnf("No explicit tlsSANs specified. Trying to use the first TLS SAN in the status: %s", cluster.Status.TLSSANs[0])
+
+			ip = cluster.Status.TLSSANs[0]
+		} else {
+			logrus.Warn("IP not found in tlsSANs. This could cause issue with the certificate validation.")
+		}
+	}
+
+	host := ip
+	if port != httpsPort {
+		host = net.JoinHostPort(ip, strconv.Itoa(int(port)))
+	}
+
+	url := "https://" + host
+
+	// if ingress is specified, use the ingress host
+	if cluster.Spec.Expose != nil && cluster.Spec.Expose.Ingress != nil {
+		var k3kIngress networkingv1.Ingress
+
+		ingressKey := types.NamespacedName{
+			Name:      IngressName(cluster.Name),
+			Namespace: cluster.Namespace,
+		}
+
+		if err := c.Get(ctx, ingressKey, &k3kIngress); err != nil {
+			return "", external, err
+		}
+
+		if len(k3kIngress.Spec.Rules) > 0 {
+			url = fmt.Sprintf("https://%s", k3kIngress.Spec.Rules[0].Host)
+			external = true
+		}
+	}
+
+	return url, external, nil
+}
diff --git a/pkg/controller/cluster/server/endpoint_test.go b/pkg/controller/cluster/server/endpoint_test.go
new file mode 100644
index 00000000..22075c14
--- /dev/null
+++ b/pkg/controller/cluster/server/endpoint_test.go
@@ -0,0 +1,80 @@
+package server
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"k8s.io/apimachinery/pkg/runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	"github.com/rancher/k3k/pkg/apis/k3k.io/v1beta1"
+)
+
+func Test_ServerURL_LoadBalancer(t *testing.T) {
+	scheme := runtime.NewScheme()
+	require.NoError(t, corev1.AddToScheme(scheme))
+	require.NoError(t, v1beta1.AddToScheme(scheme))
+
+	cluster := &v1beta1.Cluster{
+		ObjectMeta: metav1.ObjectMeta{Name: "demo", Namespace: "team-a"},
+		Status:     v1beta1.ClusterStatus{TLSSANs: []string{"203.0.113.10", "lb.example.com"}},
+	}
+
+	tests := []struct {
+		name         string
+		ingress      []corev1.LoadBalancerIngress
+		wantURL      string
+		wantExternal bool
+	}{
+		{
+			name:         "no ingress yet (LB still provisioning)",
+			ingress:      nil,
+			wantExternal: false,
+		},
+		{
+			name:         "ingress with IP",
+			ingress:      []corev1.LoadBalancerIngress{{IP: "203.0.113.10"}},
+			wantURL:      "https://203.0.113.10",
+			wantExternal: true,
+		},
+		{
+			name:         "ingress with hostname only",
+			ingress:      []corev1.LoadBalancerIngress{{Hostname: "lb.example.com"}},
+			wantURL:      "https://lb.example.com",
+			wantExternal: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			svc := &corev1.Service{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      ServiceName(cluster.Name),
+					Namespace: cluster.Namespace,
+				},
+				Spec: corev1.ServiceSpec{
+					Type:      corev1.ServiceTypeLoadBalancer,
+					ClusterIP: "10.43.0.50",
+					Ports:     []corev1.ServicePort{{Name: "k3s-server-port", Port: 443}},
+				},
+				Status: corev1.ServiceStatus{
+					LoadBalancer: corev1.LoadBalancerStatus{Ingress: tt.ingress},
+				},
+			}
+
+			c := fake.NewClientBuilder().WithScheme(scheme).WithObjects(svc).Build()
+			url, external, err := ServerURL(context.Background(), c, cluster, "", 0)
+			require.NoError(t, err)
+			assert.Equal(t, tt.wantExternal, external)
+
+			if tt.wantExternal {
+				assert.Equal(t, tt.wantURL, url)
+			}
+		})
+	}
+}
diff --git a/pkg/controller/cluster/server/server.go b/pkg/controller/cluster/server/server.go
index 330ceefa..8ea304cb 100644
--- a/pkg/controller/cluster/server/server.go
+++ b/pkg/controller/cluster/server/server.go
@@ -22,7 +22,6 @@ import (
 
 	"github.com/rancher/k3k/pkg/apis/k3k.io/v1beta1"
 	"github.com/rancher/k3k/pkg/controller"
-	"github.com/rancher/k3k/pkg/controller/cluster/agent"
 	"github.com/rancher/k3k/pkg/controller/cluster/mounts"
 )
 
@@ -250,8 +249,10 @@ func (s *Server) podSpec(ctx context.Context, image, name string, persistent boo
 			},
 		},
 	}
-	// start the pod unprivileged in shared mode
-	if s.mode == agent.VirtualNodeMode {
+	// virtual mode runs an embedded kubelet inside the server pod and therefore
+	// requires Privileged. shared and hcp modes are agentless (no kubelet) and
+	// run unprivileged.
+	if s.mode == string(v1beta1.VirtualClusterMode) {
 		podSpec.Containers[0].SecurityContext = &corev1.SecurityContext{
 			Privileged: ptr.To(true),
 		}
diff --git a/pkg/controller/cluster/server/template.go b/pkg/controller/cluster/server/template.go
index daf83699..0fd48d4e 100644
--- a/pkg/controller/cluster/server/template.go
+++ b/pkg/controller/cluster/server/template.go
@@ -19,7 +19,7 @@ safe_mode() {
         CURRENT_IP=$(cat /var/lib/rancher/k3s/k3k-node-ip)
     fi
 
-    if [ -z "$CURRENT_IP" ] || [ "$CURRENT_IP" = "$POD_IP" ] || [ {{.K3K_MODE}} != "virtual" ]; then
+    if [ -z "$CURRENT_IP" ] || [ "$CURRENT_IP" = "$POD_IP" ] || [ "{{.K3K_MODE}}" != "virtual" ]; then
         return
     fi
 
@@ -116,7 +116,8 @@ configure_cgroups() {
 	fi
 
 	# only configure the cgroups if the runtime used is the default and the mode is virtual
-	if [ -n "$runtime_class" ] || [ "{{.K3K_MODE}}" != "virtual" ]; then
+	# shared and hcp run agentless (no kubelet) and don't need cgroup overrides.
+	if [ -n "$runtime_class" ] || [ "{{.K3K_MODE}}" != "virtual" ]; then	
 		return
 	fi
 
diff --git a/pkg/controller/cluster/status.go b/pkg/controller/cluster/status.go
index 0a229f77..7c61ed3f 100644
--- a/pkg/controller/cluster/status.go
+++ b/pkg/controller/cluster/status.go
@@ -19,11 +19,12 @@ const (
 	ConditionReady = "Ready"
 
 	// Condition Reasons
-	ReasonValidationFailed   = "ValidationFailed"
-	ReasonProvisioning       = "Provisioning"
-	ReasonProvisioned        = "Provisioned"
-	ReasonProvisioningFailed = "ProvisioningFailed"
-	ReasonTerminating        = "Terminating"
+	ReasonValidationFailed      = "ValidationFailed"
+	ReasonProvisioning          = "Provisioning"
+	ReasonProvisioned           = "Provisioned"
+	ReasonProvisioningFailed    = "ProvisioningFailed"
+	ReasonTerminating           = "Terminating"
+	ReasonHCPNoExternalEndpoint = "HCPNoExternalEndpoint"
 )
 
 func (c *ClusterReconciler) updateStatus(ctx context.Context, cluster *v1beta1.Cluster, reconcileErr error) {
@@ -69,6 +70,18 @@ func (c *ClusterReconciler) updateStatus(ctx context.Context, cluster *v1beta1.C
 		return
 	}
 
+	if errors.Is(reconcileErr, ErrHCPNoExternalEndpoint) {
+		cluster.Status.Phase = v1beta1.ClusterPending
+		meta.SetStatusCondition(&cluster.Status.Conditions, metav1.Condition{
+			Type:    ConditionReady,
+			Status:  metav1.ConditionFalse,
+			Reason:  ReasonHCPNoExternalEndpoint,
+			Message: "HCP cluster has no external endpoint; set spec.expose.nodePort, spec.expose.loadBalancer or spec.expose.ingress so external nodes can reach the API server",
+		})
+
+		return
+	}
+
 	// If there's an error, but it's not a validation error, the cluster is in a failed state.
 	if reconcileErr != nil {
 		cluster.Status.Phase = v1beta1.ClusterFailed
diff --git a/pkg/controller/kubeconfig/kubeconfig.go b/pkg/controller/kubeconfig/kubeconfig.go
index 44860314..9bb4cf8b 100644
--- a/pkg/controller/kubeconfig/kubeconfig.go
+++ b/pkg/controller/kubeconfig/kubeconfig.go
@@ -3,18 +3,12 @@ package kubeconfig
 import (
 	"context"
 	"crypto/x509"
-	"fmt"
-	"slices"
 	"time"
 
-	"github.com/sirupsen/logrus"
-	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apiserver/pkg/authentication/user"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	certutil "github.com/rancher/dynamiclistener/cert"
-	corev1 "k8s.io/api/core/v1"
-	networkingv1 "k8s.io/api/networking/v1"
 	clientcmdapi "k8s.io/client-go/tools/clientcmd/api"
 
 	"github.com/rancher/k3k/pkg/apis/k3k.io/v1beta1"
@@ -60,7 +54,7 @@ func (k *KubeConfig) Generate(ctx context.Context, client client.Client, cluster
 		return nil, err
 	}
 
-	url, err := getURLFromService(ctx, client, cluster, hostServerIP, port)
+	url, _, err := server.ServerURL(ctx, client, cluster, hostServerIP, port)
 	if err != nil {
 		return nil, err
 	}
@@ -92,85 +86,3 @@ func NewConfig(url string, serverCA, clientCert, clientKey []byte) *clientcmdapi
 
 	return config
 }
-
-func getURLFromService(ctx context.Context, client client.Client, cluster *v1beta1.Cluster, hostServerIP string, serverPort int) (string, error) {
-	// get the server service to extract the right IP
-	key := types.NamespacedName{
-		Name:      server.ServiceName(cluster.Name),
-		Namespace: cluster.Namespace,
-	}
-
-	var k3kService corev1.Service
-	if err := client.Get(ctx, key, &k3kService); err != nil {
-		return "", err
-	}
-
-	ip := k3kService.Spec.ClusterIP
-	port := int32(443)
-
-	if len(k3kService.Spec.Ports) == 0 {
-		logrus.Warn("No ports exposed by the cluster service.")
-	}
-
-	switch k3kService.Spec.Type {
-	case corev1.ServiceTypeNodePort:
-		ip = hostServerIP
-
-		if len(k3kService.Spec.Ports) > 0 {
-			port = k3kService.Spec.Ports[0].NodePort
-		}
-	case corev1.ServiceTypeLoadBalancer:
-		if len(k3kService.Status.LoadBalancer.Ingress) > 0 {
-			ip = k3kService.Status.LoadBalancer.Ingress[0].IP
-		} else {
-			logrus.Warn("No ingress found in LoadBalancer service.")
-		}
-
-		if len(k3kService.Spec.Ports) > 0 {
-			port = k3kService.Spec.Ports[0].Port
-		}
-	}
-
-	if serverPort != 0 {
-		port = int32(serverPort)
-	}
-
-	if !slices.Contains(cluster.Status.TLSSANs, ip) {
-		logrus.Warnf("IP %s not in tlsSANs.", ip)
-
-		if len(cluster.Spec.TLSSANs) > 0 {
-			logrus.Warnf("Using the first TLS SAN in the spec as a fallback: %s", cluster.Spec.TLSSANs[0])
-
-			ip = cluster.Spec.TLSSANs[0]
-		} else if len(cluster.Status.TLSSANs) > 0 {
-			logrus.Warnf("No explicit tlsSANs specified. Trying to use the first TLS SAN in the status: %s", cluster.Status.TLSSANs[0])
-
-			ip = cluster.Status.TLSSANs[0]
-		} else {
-			logrus.Warn("IP not found in tlsSANs. This could cause issue with the certificate validation.")
-		}
-	}
-
-	url := "https://" + ip
-	if port != 443 {
-		url = fmt.Sprintf("%s:%d", url, port)
-	}
-
-	// if ingress is specified, use the ingress host
-	if cluster.Spec.Expose != nil && cluster.Spec.Expose.Ingress != nil {
-		var k3kIngress networkingv1.Ingress
-
-		ingressKey := types.NamespacedName{
-			Name:      server.IngressName(cluster.Name),
-			Namespace: cluster.Namespace,
-		}
-
-		if err := client.Get(ctx, ingressKey, &k3kIngress); err != nil {
-			return "", err
-		}
-
-		url = fmt.Sprintf("https://%s", k3kIngress.Spec.Rules[0].Host)
-	}
-
-	return url, nil
-}
diff --git a/tests/e2e/cluster_create_test.go b/tests/e2e/cluster_create_test.go
index 9e2d2d5a..6856a7fd 100644
--- a/tests/e2e/cluster_create_test.go
+++ b/tests/e2e/cluster_create_test.go
@@ -5,8 +5,11 @@ import (
 
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
+	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
+	"github.com/rancher/k3k/pkg/apis/k3k.io/v1beta1"
+	k3kcluster "github.com/rancher/k3k/pkg/controller/cluster"
 	fwk3k "github.com/rancher/k3k/tests/framework/k3k"
 
 	. "github.com/onsi/ginkgo/v2"
@@ -61,3 +64,93 @@ var _ = When("creating a shared mode cluster", Label(e2eTestLabel), Label(slowTe
 			Should(Succeed())
 	})
 })
+
+var _ = When("creating an HCP mode cluster", Label(e2eTestLabel), Label(slowTestsLabel), func() {
+	var virtualCluster *VirtualCluster
+
+	BeforeEach(func() {
+		namespace := fwk3k.CreateNamespace(k8s)
+
+		DeferCleanup(func() {
+			fwk3k.DeleteNamespaces(k8s, namespace.Name)
+		})
+
+		cluster := NewCluster(namespace.Name)
+		cluster.Spec.Mode = v1beta1.HCPClusterMode
+
+		CreateCluster(cluster)
+		client, restConfig := NewVirtualK8sClientAndConfig(cluster)
+
+		virtualCluster = &VirtualCluster{
+			Cluster:    cluster,
+			RestConfig: restConfig,
+			Client:     client,
+		}
+	})
+
+	It("is up and running", func() {
+		Eventually(func(g Gomega) {
+			ctx := GinkgoT().Context()
+			key := client.ObjectKeyFromObject(virtualCluster.Cluster)
+			g.Expect(k8sClient.Get(ctx, key, virtualCluster.Cluster)).To(Succeed())
+			g.Expect(virtualCluster.Cluster.Status.Phase).To(BeEquivalentTo(v1beta1.ClusterReady))
+		}).
+			WithTimeout(time.Minute).
+			WithPolling(time.Second).
+			Should(Succeed())
+	})
+
+	It("creates a populated token secret", func() {
+		ctx := GinkgoT().Context()
+
+		Eventually(func(g Gomega) {
+			var tokenSecret corev1.Secret
+
+			key := client.ObjectKey{
+				Name:      k3kcluster.TokenSecretName(virtualCluster.Cluster.Name),
+				Namespace: virtualCluster.Cluster.Namespace,
+			}
+
+			g.Expect(k8sClient.Get(ctx, key, &tokenSecret)).To(Succeed())
+			g.Expect(tokenSecret.Data["token"]).NotTo(BeEmpty())
+		}).
+			WithTimeout(time.Minute).
+			WithPolling(time.Second).
+			Should(Succeed())
+	})
+
+	It("reconciles default/kubernetes Endpoints and EndpointSlice", func() {
+		ctx := GinkgoT().Context()
+
+		Eventually(func(g Gomega) {
+			key := client.ObjectKeyFromObject(virtualCluster.Cluster)
+			g.Expect(k8sClient.Get(ctx, key, virtualCluster.Cluster)).To(Succeed())
+
+			endpoints, err := virtualCluster.Client.CoreV1().Endpoints("default").Get(ctx, "kubernetes", metav1.GetOptions{})
+			g.Expect(err).To(Not(HaveOccurred()))
+			g.Expect(endpoints.Subsets).To(HaveLen(1))
+			g.Expect(endpoints.Subsets[0].Addresses).To(HaveLen(1))
+			g.Expect(virtualCluster.Cluster.Status.TLSSANs).To(ContainElement(endpoints.Subsets[0].Addresses[0].IP))
+		}).
+			WithTimeout(time.Minute).
+			WithPolling(time.Second).
+			Should(Succeed())
+
+		Eventually(func(g Gomega) {
+			key := client.ObjectKeyFromObject(virtualCluster.Cluster)
+			g.Expect(k8sClient.Get(ctx, key, virtualCluster.Cluster)).To(Succeed())
+
+			slices, err := virtualCluster.Client.DiscoveryV1().EndpointSlices("default").List(ctx, metav1.ListOptions{
+				LabelSelector: "kubernetes.io/service-name=kubernetes",
+			})
+			g.Expect(err).To(Not(HaveOccurred()))
+			g.Expect(slices.Items).To(HaveLen(1))
+			g.Expect(slices.Items[0].Endpoints).To(HaveLen(1))
+			g.Expect(slices.Items[0].Endpoints[0].Addresses).To(HaveLen(1))
+			g.Expect(virtualCluster.Cluster.Status.TLSSANs).To(ContainElement(slices.Items[0].Endpoints[0].Addresses[0]))
+		}).
+			WithTimeout(time.Minute).
+			WithPolling(time.Second).
+			Should(Succeed())
+	})
+})
diff --git a/tests/e2e/common_test.go b/tests/e2e/common_test.go
index 67668724..ed44c396 100644
--- a/tests/e2e/common_test.go
+++ b/tests/e2e/common_test.go
@@ -288,7 +288,8 @@ func (c *VirtualCluster) NewNginxPod(namespace string) (*corev1.Pod, string) {
 
 	By(fmt.Sprintf("Nginx Pod is running (%s/%s)", nginxPod.Namespace, nginxPod.Name))
 
-	// only check the pod on the host cluster if the mode is shared mode
+	// only check the pod on the host cluster if the mode is shared mode.
+	// hcp is agentless and BYO-node, so no host-side pod mirror exists.
 	if c.Cluster.Spec.Mode != v1beta1.SharedClusterMode {
 		return nginxPod, ""
 	}