diff --git a/examples/online_serving/qwen3_tts/openai_speech_client.py b/examples/online_serving/qwen3_tts/openai_speech_client.py
index 3b83bc4b6..4741a4715 100644
--- a/examples/online_serving/qwen3_tts/openai_speech_client.py
+++ b/examples/online_serving/qwen3_tts/openai_speech_client.py
@@ -21,10 +21,16 @@ Examples:
         --task-type Base \
         --ref-audio "https://example.com/reference.wav" \
         --ref-text "This is the reference transcript"
+
+    # Base task with pre-computed speaker embedding
+    python openai_speech_client.py --text "Hello world" \
+        --task-type Base \
+        --speaker-embedding embedding.json
 """
 
 import argparse
 import base64
+import json
 import os
 
 import httpx
@@ -91,6 +97,9 @@ def run_tts_generation(args) -> None:
         payload["ref_text"] = args.ref_text
     if args.x_vector_only:
         payload["x_vector_only_mode"] = True
+    if args.speaker_embedding:
+        with open(args.speaker_embedding) as f:
+            payload["speaker_embedding"] = json.load(f)
 
     print(f"Model: {args.model}")
     print(f"Task type: {args.task_type or 'CustomVoice'}")
@@ -214,6 +223,12 @@ def parse_args():
         action="store_true",
         help="Use x-vector only mode for voice cloning (no ICL)",
     )
+    parser.add_argument(
+        "--speaker-embedding",
+        type=str,
+        default=None,
+        help="Path to JSON file containing a pre-computed speaker embedding vector (1024-dim for 0.6B, 2048-dim for 1.7B)",
+    )
 
     # Generation parameters
     parser.add_argument(
diff --git a/examples/online_serving/qwen3_tts/speaker_embedding_interpolation.py b/examples/online_serving/qwen3_tts/speaker_embedding_interpolation.py
new file mode 100644
index 000000000..e6786f886
--- /dev/null
+++ b/examples/online_serving/qwen3_tts/speaker_embedding_interpolation.py
@@ -0,0 +1,378 @@
+"""Speaker embedding extraction and interpolation for Qwen3-TTS.
+
+Extracts speaker embeddings from reference audio files using the ECAPA-TDNN
+speaker encoder from a Qwen3-TTS checkpoint, then interpolates between them
+using SLERP and sends the result to the /v1/audio/speech API.
+
+Requirements:
+    pip install torch librosa soundfile numpy httpx
+
+Examples:
+    # Extract and save an embedding
+    python speaker_embedding_interpolation.py extract \
+        --model Qwen/Qwen3-TTS-12Hz-1.7B-Base \
+        --audio voice_a.wav \
+        --output voice_a_embedding.json
+
+    # Interpolate between two embeddings and generate speech
+    python speaker_embedding_interpolation.py interpolate \
+        --embedding-a voice_a_embedding.json \
+        --embedding-b voice_b_embedding.json \
+        --ratio 0.5 \
+        --text "Hello, this is a blended voice." \
+        --output blended.wav
+
+    # Full pipeline: extract from two audio files + interpolate + generate
+    python speaker_embedding_interpolation.py pipeline \
+        --model Qwen/Qwen3-TTS-12Hz-1.7B-Base \
+        --audio-a voice_a.wav \
+        --audio-b voice_b.wav \
+        --ratios 0.0 0.25 0.5 0.75 1.0 \
+        --text "Hello, this is a blended voice." \
+        --output-dir ./interpolated/
+"""
+
+import argparse
+import json
+import os
+import sys
+
+import httpx
+import numpy as np
+import torch
+
+DEFAULT_API_BASE = "http://localhost:8000"
+DEFAULT_API_KEY = "EMPTY"
+
+# ──────────────────────────────────────────────
+# Speaker embedding extraction (offline)
+# ──────────────────────────────────────────────
+
+
+def load_speaker_encoder(model_path: str, device: str = "cpu") -> torch.nn.Module:
+    """Load just the ECAPA-TDNN speaker encoder from a Qwen3-TTS checkpoint.
+
+    This avoids loading the full TTS model by only instantiating the speaker
+    encoder sub-module from the checkpoint weights.
+    """
+    from transformers import AutoConfig
+
+    # Register the config class so AutoConfig can resolve it
+    sys.path.insert(0, os.path.dirname(__file__))
+    try:
+        from vllm_omni.model_executor.models.qwen3_tts.configuration_qwen3_tts import (
+            Qwen3TTSConfig,
+        )
+        from vllm_omni.model_executor.models.qwen3_tts.qwen3_tts_talker import (
+            Qwen3TTSSpeakerEncoder,
+        )
+
+        AutoConfig.register("qwen3_tts", Qwen3TTSConfig)
+    except ImportError:
+        # If running outside the vllm-omni tree, try importing from the HF hub
+        config = AutoConfig.from_pretrained(model_path, trust_remote_code=True)
+        # Dynamically import from the downloaded model files
+        import importlib
+
+        from huggingface_hub import snapshot_download
+
+        model_dir = snapshot_download(model_path)
+        spec = importlib.util.spec_from_file_location(
+            "modeling_qwen3_tts",
+            os.path.join(model_dir, "modeling_qwen3_tts.py"),
+        )
+        mod = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(mod)
+        Qwen3TTSSpeakerEncoder = mod.Qwen3TTSSpeakerEncoder
+
+        spec2 = importlib.util.spec_from_file_location(
+            "configuration_qwen3_tts",
+            os.path.join(model_dir, "configuration_qwen3_tts.py"),
+        )
+        mod2 = importlib.util.module_from_spec(spec2)
+        spec2.loader.exec_module(mod2)
+        Qwen3TTSConfig = mod2.Qwen3TTSConfig
+        config_obj = Qwen3TTSConfig.from_pretrained(model_path)
+        encoder = Qwen3TTSSpeakerEncoder(config_obj.speaker_encoder_config)
+        # Load weights
+        _load_speaker_encoder_weights(encoder, model_path)
+        encoder = encoder.to(device).eval()
+        return encoder
+
+    config = AutoConfig.from_pretrained(model_path, trust_remote_code=True)
+    encoder = Qwen3TTSSpeakerEncoder(config.speaker_encoder_config)
+
+    _load_speaker_encoder_weights(encoder, model_path)
+    encoder = encoder.to(device).eval()
+    return encoder
+
+
+def _load_speaker_encoder_weights(encoder: torch.nn.Module, model_path: str) -> None:
+    """Load only the speaker_encoder.* weights from the checkpoint."""
+    from huggingface_hub import snapshot_download
+    from safetensors.torch import load_file
+
+    model_dir = snapshot_download(model_path)
+
+    prefix = "speaker_encoder."
+    state_dict = {}
+
+    # Try safetensors first, then pytorch bin
+    safetensor_files = sorted(f for f in os.listdir(model_dir) if f.endswith(".safetensors"))
+    if safetensor_files:
+        for fname in safetensor_files:
+            shard = load_file(os.path.join(model_dir, fname))
+            for k, v in shard.items():
+                if k.startswith(prefix):
+                    state_dict[k[len(prefix) :]] = v
+    else:
+        bin_files = sorted(f for f in os.listdir(model_dir) if f.endswith(".bin"))
+        for fname in bin_files:
+            shard = torch.load(os.path.join(model_dir, fname), map_location="cpu", weights_only=True)
+            for k, v in shard.items():
+                if k.startswith(prefix):
+                    state_dict[k[len(prefix) :]] = v
+
+    if not state_dict:
+        raise RuntimeError(
+            f"No speaker_encoder weights found in {model_path}. Make sure this is a Qwen3-TTS-*-Base checkpoint."
+        )
+
+    encoder.load_state_dict(state_dict)
+
+
+def compute_mel_spectrogram(audio: np.ndarray, sr: int = 24000) -> torch.Tensor:
+    """Compute 128-bin mel spectrogram matching Qwen3-TTS's extraction pipeline."""
+    import librosa
+
+    # Resample to 24kHz if needed
+    if sr != 24000:
+        audio = librosa.resample(audio.astype(np.float32), orig_sr=sr, target_sr=24000)
+
+    y = torch.from_numpy(audio).unsqueeze(0).float()
+
+    from librosa.filters import mel as librosa_mel_fn
+
+    mel_basis = torch.from_numpy(librosa_mel_fn(sr=24000, n_fft=1024, n_mels=128, fmin=0, fmax=12000)).float()
+
+    n_fft = 1024
+    hop_size = 256
+    win_size = 1024
+    padding = (n_fft - hop_size) // 2
+    y = torch.nn.functional.pad(y.unsqueeze(1), (padding, padding), mode="reflect").squeeze(1)
+
+    hann_window = torch.hann_window(win_size)
+    spec = torch.stft(
+        y,
+        n_fft,
+        hop_length=hop_size,
+        win_length=win_size,
+        window=hann_window,
+        center=False,
+        return_complex=True,
+    )
+    spec = torch.abs(spec)
+    mel = torch.matmul(mel_basis, spec)
+    mel = torch.log(torch.clamp(mel, min=1e-5))
+    return mel.transpose(1, 2)  # (1, T, 128)
+
+
+@torch.inference_mode()
+def extract_embedding(encoder: torch.nn.Module, audio_path: str, device: str = "cpu") -> np.ndarray:
+    """Extract a 1024-dim speaker embedding from an audio file."""
+    import librosa
+
+    audio, sr = librosa.load(audio_path, sr=None, mono=True)
+    mel = compute_mel_spectrogram(audio, sr).to(device)
+    embedding = encoder(mel.to(next(encoder.parameters()).dtype))[0]
+    return embedding.float().cpu().numpy()
+
+
+# ──────────────────────────────────────────────
+# Interpolation
+# ──────────────────────────────────────────────
+
+
+def slerp(v0: np.ndarray, v1: np.ndarray, t: float) -> np.ndarray:
+    """Spherical linear interpolation between two vectors."""
+    v0_norm = v0 / (np.linalg.norm(v0) + 1e-8)
+    v1_norm = v1 / (np.linalg.norm(v1) + 1e-8)
+
+    dot = np.clip(np.dot(v0_norm, v1_norm), -1.0, 1.0)
+    omega = np.arccos(dot)
+
+    if np.abs(omega) < 1e-6:
+        # Vectors are nearly parallel, fall back to lerp
+        return (1.0 - t) * v0 + t * v1
+
+    sin_omega = np.sin(omega)
+    return (np.sin((1.0 - t) * omega) / sin_omega) * v0 + (np.sin(t * omega) / sin_omega) * v1
+
+
+# ──────────────────────────────────────────────
+# API client
+# ──────────────────────────────────────────────
+
+
+def generate_speech(
+    text: str,
+    speaker_embedding: list[float],
+    output_path: str,
+    api_base: str = DEFAULT_API_BASE,
+    api_key: str = DEFAULT_API_KEY,
+    model: str = "Qwen/Qwen3-TTS-12Hz-1.7B-Base",
+) -> None:
+    """Send a speaker_embedding to the TTS API and save the output."""
+    payload = {
+        "model": model,
+        "input": text,
+        "task_type": "Base",
+        "speaker_embedding": speaker_embedding,
+        "response_format": "wav",
+    }
+    headers = {
+        "Content-Type": "application/json",
+        "Authorization": f"Bearer {api_key}",
+    }
+    with httpx.Client(timeout=300.0) as client:
+        resp = client.post(f"{api_base}/v1/audio/speech", json=payload, headers=headers)
+
+    if resp.status_code != 200:
+        print(f"Error {resp.status_code}: {resp.text}")
+        sys.exit(1)
+
+    with open(output_path, "wb") as f:
+        f.write(resp.content)
+    print(f"Saved: {output_path}")
+
+
+# ──────────────────────────────────────────────
+# CLI commands
+# ──────────────────────────────────────────────
+
+
+def cmd_extract(args):
+    """Extract a speaker embedding from audio and save as JSON."""
+    print(f"Loading speaker encoder from {args.model}...")
+    encoder = load_speaker_encoder(args.model, device=args.device)
+
+    print(f"Extracting embedding from {args.audio}...")
+    emb = extract_embedding(encoder, args.audio, device=args.device)
+    print(f"Embedding shape: {emb.shape}")
+
+    output_path = args.output or os.path.splitext(args.audio)[0] + "_embedding.json"
+    with open(output_path, "w") as f:
+        json.dump(emb.tolist(), f)
+    print(f"Saved embedding to {output_path}")
+
+
+def cmd_interpolate(args):
+    """Interpolate between two embeddings and generate speech."""
+    with open(args.embedding_a) as f:
+        emb_a = np.array(json.load(f), dtype=np.float32)
+    with open(args.embedding_b) as f:
+        emb_b = np.array(json.load(f), dtype=np.float32)
+
+    print(f"Embedding A: {args.embedding_a} (dim={emb_a.shape[0]})")
+    print(f"Embedding B: {args.embedding_b} (dim={emb_b.shape[0]})")
+    print(f"SLERP ratio: {args.ratio}")
+
+    blended = slerp(emb_a, emb_b, args.ratio)
+    output_path = args.output or f"interpolated_t{args.ratio:.2f}.wav"
+
+    print(f"Generating speech: {args.text!r}")
+    generate_speech(
+        text=args.text,
+        speaker_embedding=blended.tolist(),
+        output_path=output_path,
+        api_base=args.api_base,
+        api_key=args.api_key,
+        model=args.model,
+    )
+
+
+def cmd_pipeline(args):
+    """Full pipeline: extract two embeddings, SLERP at multiple ratios, generate."""
+    print(f"Loading speaker encoder from {args.model}...")
+    encoder = load_speaker_encoder(args.model, device=args.device)
+
+    print(f"Extracting embedding A from {args.audio_a}...")
+    emb_a = extract_embedding(encoder, args.audio_a, device=args.device)
+    print(f"Extracting embedding B from {args.audio_b}...")
+    emb_b = extract_embedding(encoder, args.audio_b, device=args.device)
+
+    # Save extracted embeddings
+    os.makedirs(args.output_dir, exist_ok=True)
+    for label, emb, audio_path in [("a", emb_a, args.audio_a), ("b", emb_b, args.audio_b)]:
+        emb_path = os.path.join(args.output_dir, f"embedding_{label}.json")
+        with open(emb_path, "w") as f:
+            json.dump(emb.tolist(), f)
+        print(f"Saved embedding {label} to {emb_path}")
+
+    # Generate at each ratio
+    for t in args.ratios:
+        blended = slerp(emb_a, emb_b, t)
+        out_path = os.path.join(args.output_dir, f"t{t:.2f}.wav")
+        print(f"\n--- SLERP t={t:.2f} ---")
+        print(f"Generating: {args.text!r}")
+        generate_speech(
+            text=args.text,
+            speaker_embedding=blended.tolist(),
+            output_path=out_path,
+            api_base=args.api_base,
+            api_key=args.api_key,
+            model=args.model,
+        )
+
+    print(f"\nAll outputs saved to {args.output_dir}/")
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="Speaker embedding extraction and interpolation for Qwen3-TTS",
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+    )
+    parser.add_argument("--api-base", default=DEFAULT_API_BASE, help="TTS API base URL")
+    parser.add_argument("--api-key", default=DEFAULT_API_KEY, help="API key")
+    parser.add_argument(
+        "--model",
+        default="Qwen/Qwen3-TTS-12Hz-1.7B-Base",
+        help="Model name (used for both weight loading and API requests)",
+    )
+    parser.add_argument("--device", default="cpu", help="Device for embedding extraction (cpu/cuda)")
+
+    sub = parser.add_subparsers(dest="command", required=True)
+
+    # extract
+    p_ext = sub.add_parser("extract", help="Extract speaker embedding from audio")
+    p_ext.add_argument("--audio", required=True, help="Input audio file")
+    p_ext.add_argument("--output", "-o", help="Output JSON path (default: <audio>_embedding.json)")
+
+    # interpolate
+    p_interp = sub.add_parser("interpolate", help="Interpolate between two embeddings and generate speech")
+    p_interp.add_argument("--embedding-a", required=True, help="JSON file with embedding A")
+    p_interp.add_argument("--embedding-b", required=True, help="JSON file with embedding B")
+    p_interp.add_argument("--ratio", "-r", type=float, default=0.5, help="SLERP ratio (0=A, 1=B)")
+    p_interp.add_argument("--text", required=True, help="Text to synthesize")
+    p_interp.add_argument("--output", "-o", help="Output wav path")
+
+    # pipeline
+    p_pipe = sub.add_parser("pipeline", help="Full pipeline: extract + interpolate + generate")
+    p_pipe.add_argument("--audio-a", required=True, help="Audio file for voice A")
+    p_pipe.add_argument("--audio-b", required=True, help="Audio file for voice B")
+    p_pipe.add_argument(
+        "--ratios",
+        nargs="+",
+        type=float,
+        default=[0.0, 0.25, 0.5, 0.75, 1.0],
+        help="SLERP ratios to generate (default: 0.0 0.25 0.5 0.75 1.0)",
+    )
+    p_pipe.add_argument("--text", required=True, help="Text to synthesize")
+    p_pipe.add_argument("--output-dir", default="./interpolated", help="Output directory")
+
+    args = parser.parse_args()
+    {"extract": cmd_extract, "interpolate": cmd_interpolate, "pipeline": cmd_pipeline}[args.command](args)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/tests/e2e/online_serving/test_qwen3_tts_speaker_embedding.py b/tests/e2e/online_serving/test_qwen3_tts_speaker_embedding.py
new file mode 100644
index 000000000..64e13e155
--- /dev/null
+++ b/tests/e2e/online_serving/test_qwen3_tts_speaker_embedding.py
@@ -0,0 +1,249 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""
+E2E tests for speaker_embedding API parameter.
+
+Tests Base models (0.6B-Base and 1.7B-Base) which support voice cloning
+via pre-computed ECAPA-TDNN embeddings, bypassing ref_audio extraction.
+"""
+
+import os
+
+os.environ["VLLM_WORKER_MULTIPROC_METHOD"] = "spawn"
+os.environ["VLLM_TEST_CLEAN_GPU_MEMORY"] = "0"
+
+import struct
+from pathlib import Path
+
+import httpx
+import pytest
+
+from tests.conftest import OmniServer
+from tests.utils import hardware_test
+
+MODEL_BASE = "Qwen/Qwen3-TTS-12Hz-0.6B-Base"
+MODEL_BASE_1_7B = "Qwen/Qwen3-TTS-12Hz-1.7B-Base"
+
+# A synthetic 1024-dim speaker embedding (all 0.1 — not a real voice, but
+# exercises the full code path through the talker's _build_prompt_embeds).
+DUMMY_EMBEDDING_1024 = [0.1] * 1024
+DUMMY_EMBEDDING_2048 = [0.1] * 2048
+
+SYN_TEXT = "Hello."
+MIN_AUDIO_BYTES = 2000
+# Limit generation to keep tests fast (dummy embeddings produce nonsensical
+# output that may never hit a natural stop token).
+MAX_NEW_TOKENS = 256
+
+
+def get_stage_config():
+    return str(
+        Path(__file__).parent.parent.parent.parent / "vllm_omni" / "model_executor" / "stage_configs" / "qwen3_tts.yaml"
+    )
+
+
+def _server_args():
+    return [
+        "--stage-configs-path",
+        get_stage_config(),
+        "--stage-init-timeout",
+        "120",
+        "--trust-remote-code",
+        "--enforce-eager",
+        "--disable-log-stats",
+    ]
+
+
+def verify_wav_audio(content: bytes) -> bool:
+    if len(content) < 44:
+        return False
+    return content[:4] == b"RIFF" and content[8:12] == b"WAVE"
+
+
+def assert_not_silence(pcm_bytes: bytes):
+    """Assert PCM16 samples are not all identical (e.g. all-silence)."""
+    samples = struct.unpack(f"<{len(pcm_bytes) // 2}h", pcm_bytes)
+    unique = set(samples)
+    assert len(unique) > 1, f"All-silence detected: {len(samples)} samples, unique values: {unique}"
+
+
+# ── 0.6B-Base model tests ──
+
+
+@pytest.fixture(scope="module")
+def base_server():
+    """Start vLLM-Omni server with 0.6B-Base model."""
+    with OmniServer(MODEL_BASE, _server_args()) as server:
+        yield server
+
+
+class TestSpeakerEmbeddingBase:
+    """Speaker embedding tests against the 0.6B-Base model (supports Base task)."""
+
+    @pytest.mark.core_model
+    @pytest.mark.omni
+    @hardware_test(res={"cuda": "L4"}, num_cards=1)
+    def test_speaker_embedding_produces_audio(self, base_server) -> None:
+        """speaker_embedding with Base task produces valid WAV audio."""
+        url = f"http://{base_server.host}:{base_server.port}/v1/audio/speech"
+        payload = {
+            "model": MODEL_BASE,
+            "input": SYN_TEXT,
+            "task_type": "Base",
+            "speaker_embedding": DUMMY_EMBEDDING_1024,
+            "x_vector_only_mode": True,
+            "response_format": "wav",
+            "max_new_tokens": MAX_NEW_TOKENS,
+        }
+        with httpx.Client(timeout=120.0) as client:
+            response = client.post(url, json=payload)
+
+        assert response.status_code == 200, f"Request failed: {response.text}"
+        assert response.headers.get("content-type") == "audio/wav"
+        assert verify_wav_audio(response.content), "Response is not valid WAV"
+        assert len(response.content) > MIN_AUDIO_BYTES, f"Audio too small: {len(response.content)} bytes"
+
+    @pytest.mark.core_model
+    @pytest.mark.omni
+    @hardware_test(res={"cuda": "L4"}, num_cards=1)
+    def test_speaker_embedding_pcm_not_silence(self, base_server) -> None:
+        """speaker_embedding PCM output contains real audio, not all-silence."""
+        url = f"http://{base_server.host}:{base_server.port}/v1/audio/speech"
+        payload = {
+            "model": MODEL_BASE,
+            "input": SYN_TEXT,
+            "task_type": "Base",
+            "speaker_embedding": DUMMY_EMBEDDING_1024,
+            "x_vector_only_mode": True,
+            "response_format": "pcm",
+            "max_new_tokens": MAX_NEW_TOKENS,
+        }
+        with httpx.Client(timeout=120.0) as client:
+            response = client.post(url, json=payload)
+
+        assert response.status_code == 200, f"Request failed: {response.text}"
+        assert len(response.content) > MIN_AUDIO_BYTES
+        assert_not_silence(response.content)
+
+    @pytest.mark.core_model
+    @pytest.mark.omni
+    @hardware_test(res={"cuda": "L4"}, num_cards=1)
+    def test_speaker_embedding_streaming(self, base_server) -> None:
+        """speaker_embedding works with streaming PCM output."""
+        url = f"http://{base_server.host}:{base_server.port}/v1/audio/speech"
+        payload = {
+            "model": MODEL_BASE,
+            "input": SYN_TEXT,
+            "task_type": "Base",
+            "speaker_embedding": DUMMY_EMBEDDING_1024,
+            "x_vector_only_mode": True,
+            "response_format": "pcm",
+            "stream": True,
+            "max_new_tokens": MAX_NEW_TOKENS,
+        }
+        with httpx.Client(timeout=120.0) as client:
+            response = client.post(url, json=payload)
+
+        assert response.status_code == 200, f"Request failed: {response.text}"
+        assert "audio/pcm" in response.headers.get("content-type", "")
+        assert len(response.content) > MIN_AUDIO_BYTES
+        assert_not_silence(response.content)
+
+    @pytest.mark.core_model
+    @pytest.mark.omni
+    @hardware_test(res={"cuda": "L4"}, num_cards=1)
+    def test_speaker_embedding_mutually_exclusive_with_ref_audio(self, base_server) -> None:
+        """Sending both speaker_embedding and ref_audio returns 400."""
+        url = f"http://{base_server.host}:{base_server.port}/v1/audio/speech"
+        payload = {
+            "model": MODEL_BASE,
+            "input": SYN_TEXT,
+            "task_type": "Base",
+            "speaker_embedding": DUMMY_EMBEDDING_1024,
+            "ref_audio": "https://example.com/audio.wav",
+            "response_format": "wav",
+        }
+        with httpx.Client(timeout=30.0) as client:
+            response = client.post(url, json=payload)
+
+        assert response.status_code == 400, f"Expected 400, got {response.status_code}: {response.text}"
+        assert "mutually exclusive" in response.text
+
+    @pytest.mark.core_model
+    @pytest.mark.omni
+    @hardware_test(res={"cuda": "L4"}, num_cards=1)
+    def test_speaker_embedding_empty_rejected(self, base_server) -> None:
+        """Empty speaker_embedding list returns 400."""
+        url = f"http://{base_server.host}:{base_server.port}/v1/audio/speech"
+        payload = {
+            "model": MODEL_BASE,
+            "input": SYN_TEXT,
+            "task_type": "Base",
+            "speaker_embedding": [],
+            "response_format": "wav",
+        }
+        with httpx.Client(timeout=30.0) as client:
+            response = client.post(url, json=payload)
+
+        assert response.status_code == 400, f"Expected 400, got {response.status_code}: {response.text}"
+        assert "non-empty" in response.text
+
+
+# ── 1.7B-Base model tests (2048-dim embeddings) ──
+
+
+@pytest.fixture(scope="module")
+def base_1_7b_server():
+    """Start vLLM-Omni server with 1.7B-Base model."""
+    with OmniServer(MODEL_BASE_1_7B, _server_args()) as server:
+        yield server
+
+
+class TestSpeakerEmbedding1_7B:
+    """Speaker embedding tests against the 1.7B-Base model (2048-dim embeddings)."""
+
+    @pytest.mark.core_model
+    @pytest.mark.omni
+    @hardware_test(res={"cuda": "L4"}, num_cards=1)
+    def test_2048_dim_embedding_produces_audio(self, base_1_7b_server) -> None:
+        """2048-dim speaker_embedding with 1.7B-Base model produces valid WAV audio."""
+        url = f"http://{base_1_7b_server.host}:{base_1_7b_server.port}/v1/audio/speech"
+        payload = {
+            "model": MODEL_BASE_1_7B,
+            "input": SYN_TEXT,
+            "task_type": "Base",
+            "speaker_embedding": DUMMY_EMBEDDING_2048,
+            "x_vector_only_mode": True,
+            "response_format": "wav",
+            "max_new_tokens": MAX_NEW_TOKENS,
+        }
+        with httpx.Client(timeout=120.0) as client:
+            response = client.post(url, json=payload)
+
+        assert response.status_code == 200, f"Request failed: {response.text}"
+        assert response.headers.get("content-type") == "audio/wav"
+        assert verify_wav_audio(response.content), "Response is not valid WAV"
+        assert len(response.content) > MIN_AUDIO_BYTES, f"Audio too small: {len(response.content)} bytes"
+
+    @pytest.mark.core_model
+    @pytest.mark.omni
+    @hardware_test(res={"cuda": "L4"}, num_cards=1)
+    def test_1024_dim_on_1_7b_model_rejected_or_errors(self, base_1_7b_server) -> None:
+        """1024-dim embedding on a 1.7B model (expects 2048) should fail gracefully."""
+        url = f"http://{base_1_7b_server.host}:{base_1_7b_server.port}/v1/audio/speech"
+        payload = {
+            "model": MODEL_BASE_1_7B,
+            "input": SYN_TEXT,
+            "task_type": "Base",
+            "speaker_embedding": DUMMY_EMBEDDING_1024,
+            "x_vector_only_mode": True,
+            "response_format": "wav",
+            "max_new_tokens": MAX_NEW_TOKENS,
+        }
+        with httpx.Client(timeout=120.0) as client:
+            response = client.post(url, json=payload)
+
+        # Wrong dimensions should produce an error, not silent garbage.
+        assert response.status_code != 200 or len(response.content) < MIN_AUDIO_BYTES, (
+            "Expected failure or degenerate output with wrong embedding dimensions"
+        )
diff --git a/tests/entrypoints/openai_api/test_serving_speech.py b/tests/entrypoints/openai_api/test_serving_speech.py
index ce32eb8b8..c84b501c2 100644
--- a/tests/entrypoints/openai_api/test_serving_speech.py
+++ b/tests/entrypoints/openai_api/test_serving_speech.py
@@ -235,6 +235,8 @@ def test_app(mocker: MockerFixture):
                         "created_at": info.get("created_at", 0),
                         "file_size": info.get("file_size", 0),
                         "mime_type": info.get("mime_type", ""),
+                        "embedding_source": info.get("embedding_source", "audio"),
+                        "embedding_dim": info.get("embedding_dim"),
                     }
                 )
         return {"voices": speakers, "uploaded_voices": uploaded_voices}
@@ -243,9 +245,21 @@ def test_app(mocker: MockerFixture):
     app.add_api_route("/v1/audio/speech/batch", speech_server.create_speech_batch, methods=["POST"])
 
     # Add upload_voice endpoint
-    async def upload_voice(audio_sample: UploadFile = File(...), consent: str = Form(...), name: str = Form(...)):
+    async def upload_voice(
+        audio_sample: UploadFile | None = File(None),
+        speaker_embedding: str | None = Form(None),
+        consent: str = Form(...),
+        name: str = Form(...),
+    ):
         try:
-            result = await speech_server.upload_voice(audio_sample, consent, name)
+            if speaker_embedding is not None and audio_sample is not None:
+                raise ValueError("'audio_sample' and 'speaker_embedding' are mutually exclusive")
+            if speaker_embedding is not None:
+                result = await speech_server.upload_voice_embedding(speaker_embedding, consent, name)
+            elif audio_sample is not None:
+                result = await speech_server.upload_voice(audio_sample, consent, name)
+            else:
+                raise ValueError("Either 'audio_sample' or 'speaker_embedding' must be provided")
             return {"success": True, "voice": result}
         except ValueError as e:
             raise HTTPException(status_code=400, detail=str(e))
@@ -446,13 +460,13 @@ class TestSpeechAPI:
         response = client.post("/v1/audio/voices", files=files, data=data)
         assert response.status_code == 422  # Validation error
 
-        # Missing file
+        # Missing both audio_sample and speaker_embedding
         data = {
             "consent": "user_consent_123",
             "name": "test_voice6",
         }
         response = client.post("/v1/audio/voices", data=data)
-        assert response.status_code == 422  # Validation error
+        assert response.status_code == 400
 
     def test_delete_voice_success(self, client):
         """Test successful voice deletion."""
@@ -489,6 +503,89 @@ class TestSpeechAPI:
         result = response.json()
         assert "not found" in result["detail"]
 
+    # ── speaker_embedding upload via voices endpoint ──
+
+    def test_upload_voice_embedding_success(self, client):
+        """Upload a voice via speaker_embedding JSON."""
+        import json
+
+        emb = [0.1] * 1024
+        data = {
+            "speaker_embedding": json.dumps(emb),
+            "consent": "consent_emb_1",
+            "name": "emb_voice",
+        }
+        response = client.post("/v1/audio/voices", data=data)
+        assert response.status_code == 200, f"Upload failed: {response.text}"
+        result = response.json()
+        assert result["success"] is True
+        voice = result["voice"]
+        assert voice["name"] == "emb_voice"
+        assert voice["embedding_source"] == "direct"
+        assert voice["embedding_dim"] == 1024
+        # Clean up
+        client.delete("/v1/audio/voices/emb_voice")
+
+    def test_upload_voice_embedding_appears_in_listing(self, client):
+        """Embedding-uploaded voice appears in list with correct source."""
+        import json
+
+        emb = [0.2] * 2048
+        data = {
+            "speaker_embedding": json.dumps(emb),
+            "consent": "consent_list",
+            "name": "listed_emb_voice",
+        }
+        response = client.post("/v1/audio/voices", data=data)
+        assert response.status_code == 200
+
+        listing = client.get("/v1/audio/voices").json()
+        uploaded = {v["name"]: v for v in listing["uploaded_voices"]}
+        assert "listed_emb_voice" in uploaded
+        assert uploaded["listed_emb_voice"]["embedding_source"] == "direct"
+        assert uploaded["listed_emb_voice"]["embedding_dim"] == 2048
+        # Clean up
+        client.delete("/v1/audio/voices/listed_emb_voice")
+
+    def test_upload_voice_embedding_and_audio_mutually_exclusive(self, client):
+        """Providing both audio_sample and speaker_embedding returns 400."""
+        import json
+
+        emb = [0.1] * 1024
+        files = {"audio_sample": ("test.wav", b"fake", "audio/wav")}
+        data = {
+            "speaker_embedding": json.dumps(emb),
+            "consent": "consent_mx",
+            "name": "mx_voice",
+        }
+        response = client.post("/v1/audio/voices", files=files, data=data)
+        assert response.status_code == 400
+        assert "mutually exclusive" in response.json()["detail"]
+
+    def test_upload_voice_embedding_invalid_json(self, client):
+        """Invalid JSON in speaker_embedding returns 400."""
+        data = {
+            "speaker_embedding": "not valid json [[[",
+            "consent": "consent_bad",
+            "name": "bad_json_voice",
+        }
+        response = client.post("/v1/audio/voices", data=data)
+        assert response.status_code == 400
+        assert "JSON" in response.json()["detail"]
+
+    def test_upload_voice_embedding_nan_rejected(self, client):
+        """NaN values in speaker_embedding return 400."""
+        import json
+
+        data = {
+            "speaker_embedding": json.dumps([0.1] * 1023 + [float("nan")]),
+            "consent": "consent_nan",
+            "name": "nan_voice",
+        }
+        response = client.post("/v1/audio/voices", data=data)
+        assert response.status_code == 400
+        assert "finite" in response.json()["detail"]
+
 
 class TestTTSMethods:
     """Unit tests for TTS validation and parameter building."""
@@ -610,6 +707,96 @@ class TestTTSMethods:
         result = speech_server._validate_tts_request(req)
         assert "does not support CustomVoice" in result
 
+    # ── speaker_embedding validation ──
+
+    def test_speaker_embedding_valid_base_task(self, speech_server):
+        """speaker_embedding with Base task, x_vector_only_mode, and no ref_audio is accepted."""
+        emb = [0.1] * 1024
+        req = OpenAICreateSpeechRequest(input="Hello", task_type="Base", speaker_embedding=emb, x_vector_only_mode=True)
+        assert speech_server._validate_tts_request(req) is None
+
+    def test_speaker_embedding_auto_sets_x_vector_only_mode(self, speech_server):
+        """speaker_embedding auto-implies x_vector_only_mode, so validation passes."""
+        emb = [0.1] * 1024
+        req = OpenAICreateSpeechRequest(input="Hello", task_type="Base", speaker_embedding=emb)
+        result = speech_server._validate_tts_request(req)
+        assert result is None
+        assert req.x_vector_only_mode is True
+
+    def test_speaker_embedding_wrong_task_type(self, speech_server):
+        """speaker_embedding is only valid for Base task."""
+        emb = [0.1] * 1024
+        req = OpenAICreateSpeechRequest(
+            input="Hello", task_type="VoiceDesign", speaker_embedding=emb, instructions="warm"
+        )
+        result = speech_server._validate_tts_request(req)
+        assert "only valid for Base task" in result
+
+    def test_speaker_embedding_mutually_exclusive_with_ref_audio(self, speech_server):
+        """speaker_embedding and ref_audio cannot both be provided (pydantic validation)."""
+        emb = [0.1] * 1024
+        with pytest.raises(ValidationError, match="mutually exclusive"):
+            OpenAICreateSpeechRequest(
+                input="Hello", task_type="Base", speaker_embedding=emb, ref_audio="data:audio/wav;base64,abc"
+            )
+
+    def test_speaker_embedding_nan_rejected(self, speech_server):
+        """NaN values in speaker_embedding are rejected at parse level."""
+        with pytest.raises(ValidationError, match="finite"):
+            OpenAICreateSpeechRequest(input="Hello", task_type="Base", speaker_embedding=[0.1] * 1023 + [float("nan")])
+
+    def test_speaker_embedding_inf_rejected(self, speech_server):
+        """Inf values in speaker_embedding are rejected at parse level."""
+        with pytest.raises(ValidationError, match="finite"):
+            OpenAICreateSpeechRequest(input="Hello", task_type="Base", speaker_embedding=[float("inf")] + [0.1] * 1023)
+
+    def test_speaker_embedding_empty_list_rejected(self, speech_server):
+        """Empty speaker_embedding list is rejected."""
+        req = OpenAICreateSpeechRequest(input="Hello", task_type="Base", speaker_embedding=[])
+        result = speech_server._validate_tts_request(req)
+        assert "non-empty" in result
+
+    def test_speaker_embedding_wrong_dims_accepted(self, speech_server):
+        """Non-standard dimensions pass validation (warning only, not an error)."""
+        emb = [0.1] * 512  # not 1024 or 2048
+        req = OpenAICreateSpeechRequest(input="Hello", task_type="Base", speaker_embedding=emb, x_vector_only_mode=True)
+        result = speech_server._validate_tts_request(req)
+        assert result is None
+
+    def test_speaker_embedding_2048_dims_accepted(self, speech_server):
+        """2048-dim embedding (1.7B model) is accepted without warning."""
+        emb = [0.1] * 2048
+        req = OpenAICreateSpeechRequest(input="Hello", task_type="Base", speaker_embedding=emb, x_vector_only_mode=True)
+        assert speech_server._validate_tts_request(req) is None
+
+    def test_base_task_requires_ref_audio_or_speaker_embedding(self, speech_server):
+        """Base task without ref_audio or speaker_embedding is rejected."""
+        req = OpenAICreateSpeechRequest(input="Hello", task_type="Base")
+        result = speech_server._validate_tts_request(req)
+        assert "ref_audio" in result and "speaker_embedding" in result
+
+    # ── speaker_embedding in _build_tts_params ──
+
+    def test_build_tts_params_with_speaker_embedding(self, speech_server):
+        """speaker_embedding produces voice_clone_prompt and x_vector_only_mode."""
+        emb = [0.1] * 1024
+        req = OpenAICreateSpeechRequest(input="Hello", task_type="Base", speaker_embedding=emb)
+        params = speech_server._build_tts_params(req)
+
+        assert "voice_clone_prompt" in params
+        vcp = params["voice_clone_prompt"][0]
+        assert "ref_spk_embedding" in vcp
+        # Stored as plain list (not tensor) so it survives msgspec IPC serialization
+        assert isinstance(vcp["ref_spk_embedding"], list)
+        assert len(vcp["ref_spk_embedding"]) == 1024
+        assert params["x_vector_only_mode"] == [True]
+
+    def test_build_tts_params_without_speaker_embedding(self, speech_server):
+        """Without speaker_embedding, voice_clone_prompt is not set."""
+        req = OpenAICreateSpeechRequest(input="Hello", voice="Ryan", language="English")
+        params = speech_server._build_tts_params(req)
+        assert "voice_clone_prompt" not in params
+
     def test_build_tts_params(self, speech_server):
         """Test TTS parameter building."""
         req = OpenAICreateSpeechRequest(input="Hello", voice="Ryan", language="English")
diff --git a/tests/examples/test_slerp_interpolation.py b/tests/examples/test_slerp_interpolation.py
new file mode 100644
index 000000000..fe7db74cc
--- /dev/null
+++ b/tests/examples/test_slerp_interpolation.py
@@ -0,0 +1,55 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""Unit tests for the SLERP interpolation math in speaker_embedding_interpolation.py."""
+
+import sys
+from pathlib import Path
+
+import numpy as np
+import pytest
+
+# Import the slerp function from the example script.
+_examples_dir = str(Path(__file__).parent.parent.parent / "examples" / "online_serving" / "qwen3_tts")
+sys.path.insert(0, _examples_dir)
+from speaker_embedding_interpolation import slerp  # noqa: E402
+
+pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
+
+
+class TestSlerp:
+    def test_endpoints(self):
+        """t=0 returns v0, t=1 returns v1."""
+        v0 = np.random.randn(1024).astype(np.float64)
+        v1 = np.random.randn(1024).astype(np.float64)
+        np.testing.assert_allclose(slerp(v0, v1, 0.0), v0, atol=1e-6)
+        np.testing.assert_allclose(slerp(v0, v1, 1.0), v1, atol=1e-6)
+
+    def test_midpoint_unit_norm(self):
+        """Midpoint of two unit vectors should also be approximately unit norm."""
+        v0 = np.random.randn(1024)
+        v0 /= np.linalg.norm(v0)
+        v1 = np.random.randn(1024)
+        v1 /= np.linalg.norm(v1)
+        mid = slerp(v0, v1, 0.5)
+        assert abs(np.linalg.norm(mid) - 1.0) < 0.05
+
+    def test_parallel_vectors_fallback(self):
+        """Parallel vectors (omega~0) fall back to lerp without error."""
+        v0 = np.ones(1024)
+        v1 = np.ones(1024) * 1.001  # nearly parallel
+        result = slerp(v0, v1, 0.5)
+        expected = 0.5 * v0 + 0.5 * v1
+        np.testing.assert_allclose(result, expected, atol=1e-4)
+
+    def test_antiparallel_vectors(self):
+        """Antiparallel vectors (omega~pi) should not produce NaN."""
+        v0 = np.ones(1024)
+        v1 = -np.ones(1024)
+        result = slerp(v0, v1, 0.5)
+        assert not np.any(np.isnan(result))
+
+    def test_output_shape_matches_input(self):
+        """Output shape must match input shape."""
+        v0 = np.random.randn(2048)
+        v1 = np.random.randn(2048)
+        assert slerp(v0, v1, 0.3).shape == (2048,)
diff --git a/vllm_omni/entrypoints/openai/api_server.py b/vllm_omni/entrypoints/openai/api_server.py
index 454cb261e..717bc08ce 100644
--- a/vllm_omni/entrypoints/openai/api_server.py
+++ b/vllm_omni/entrypoints/openai/api_server.py
@@ -1024,6 +1024,8 @@ async def list_voices(raw_request: Request):
                     "created_at": info.get("created_at", 0),
                     "file_size": info.get("file_size", 0),
                     "mime_type": info.get("mime_type", ""),
+                    "embedding_source": info.get("embedding_source", "audio"),
+                    "embedding_dim": info.get("embedding_dim"),
                 }
             )
 
@@ -1040,18 +1042,26 @@ async def list_voices(raw_request: Request):
 )
 async def upload_voice(
     raw_request: Request,
-    audio_sample: UploadFile = File(...),
+    audio_sample: UploadFile | None = File(None),
+    speaker_embedding: str | None = Form(None),
     consent: str = Form(...),
     name: str = Form(...),
 ):
-    """Upload a new voice sample for voice cloning.
+    """Upload a new voice for voice cloning.
 
-    Uploads an audio file that can be used as a reference for voice cloning
-    in Base task TTS requests. The voice can then be referenced by name
-    in subsequent TTS requests.
+    Accepts either an audio file or a pre-computed speaker embedding vector.
+    These are mutually exclusive: provide one or the other.
+
+    When using ``audio_sample``, the server stores the audio and extracts the
+    speaker embedding on first use (Base task models only).
+
+    When using ``speaker_embedding``, pass a JSON-encoded list of floats
+    (1024-dim for 0.6B, 2048-dim for 1.7B). The voice is stored as a
+    safetensors file and is immediately ready for use.
 
     Args:
-        audio_sample: Audio file (max 10MB)
+        audio_sample: Audio file (max 10MB). Mutually exclusive with speaker_embedding.
+        speaker_embedding: JSON-encoded float list. Mutually exclusive with audio_sample.
         consent: Consent recording ID
         name: Name for the new voice
         raw_request: Raw FastAPI request
@@ -1064,8 +1074,18 @@ async def upload_voice(
         return base(raw_request).create_error_response(message="The model does not support Speech API")
 
     try:
-        # Upload the voice
-        result = await handler.upload_voice(audio_sample, consent, name)
+        if speaker_embedding is not None and audio_sample is not None:
+            return base(raw_request).create_error_response(
+                message="'audio_sample' and 'speaker_embedding' are mutually exclusive"
+            )
+        if speaker_embedding is not None:
+            result = await handler.upload_voice_embedding(speaker_embedding, consent, name)
+        elif audio_sample is not None:
+            result = await handler.upload_voice(audio_sample, consent, name)
+        else:
+            return base(raw_request).create_error_response(
+                message="Either 'audio_sample' or 'speaker_embedding' must be provided"
+            )
 
         return JSONResponse(content={"success": True, "voice": result})
 
diff --git a/vllm_omni/entrypoints/openai/protocol/audio.py b/vllm_omni/entrypoints/openai/protocol/audio.py
index e8b4f65d5..89d2dc02f 100644
--- a/vllm_omni/entrypoints/openai/protocol/audio.py
+++ b/vllm_omni/entrypoints/openai/protocol/audio.py
@@ -1,8 +1,11 @@
+import math
 from typing import Literal
 
 import numpy as np
 from pydantic import BaseModel, Field, field_validator, model_validator
 
+_MAX_EMBEDDING_DIM = 8192
+
 
 class OpenAICreateSpeechRequest(BaseModel):
     input: str
@@ -41,7 +44,7 @@ class OpenAICreateSpeechRequest(BaseModel):
     )
     ref_audio: str | None = Field(
         default=None,
-        description="Reference audio for voice cloning (Base task). URL, base64, or file path.",
+        description="Reference audio for voice cloning (Base task). URL, base64, or file URI.",
     )
     ref_text: str | None = Field(
         default=None,
@@ -51,6 +54,13 @@ class OpenAICreateSpeechRequest(BaseModel):
         default=None,
         description="Use speaker embedding only without in-context learning (Base task)",
     )
+    speaker_embedding: list[float] | None = Field(
+        default=None,
+        max_length=_MAX_EMBEDDING_DIM,
+        description="Pre-computed speaker embedding vector (1024-dim for 0.6B, "
+        "2048-dim for 1.7B). Skips speaker encoder extraction from ref_audio. "
+        "Implies x_vector_only_mode=True. Mutually exclusive with ref_audio.",
+    )
     max_new_tokens: int | None = Field(
         default=None,
         description="Maximum tokens to generate",
@@ -68,6 +78,20 @@ class OpenAICreateSpeechRequest(BaseModel):
             raise ValueError("'sse' is not a supported stream_format yet. Please use 'audio'.")
         return v
 
+    @field_validator("speaker_embedding")
+    @classmethod
+    def validate_speaker_embedding(cls, v: list[float] | None) -> list[float] | None:
+        if v is not None and not all(math.isfinite(x) for x in v):
+            raise ValueError("'speaker_embedding' values must be finite (no NaN or Inf)")
+        return v
+
+    @model_validator(mode="after")
+    def validate_embedding_constraints(self) -> "OpenAICreateSpeechRequest":
+        if self.speaker_embedding is not None:
+            if self.ref_audio is not None:
+                raise ValueError("'speaker_embedding' and 'ref_audio' are mutually exclusive")
+        return self
+
     @model_validator(mode="after")
     def validate_streaming_constraints(self) -> "OpenAICreateSpeechRequest":
         if self.stream:
@@ -177,6 +201,11 @@ class StreamingSpeechSessionConfig(BaseModel):
     ref_audio: str | None = None
     ref_text: str | None = None
     x_vector_only_mode: bool | None = None
+    speaker_embedding: list[float] | None = Field(
+        default=None,
+        max_length=_MAX_EMBEDDING_DIM,
+        description="Pre-computed speaker embedding vector. Mutually exclusive with ref_audio.",
+    )
     stream_audio: bool = Field(
         default=False,
         description=(
diff --git a/vllm_omni/entrypoints/openai/serving_speech.py b/vllm_omni/entrypoints/openai/serving_speech.py
index c92704963..8d87bff00 100644
--- a/vllm_omni/entrypoints/openai/serving_speech.py
+++ b/vllm_omni/entrypoints/openai/serving_speech.py
@@ -532,6 +532,7 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
             "cache_status": "pending",  # The initial cache state is pending.
             "cache_file": None,  # The initial cache file is empty.
             "cache_generated_at": None,  # The initial cache generation time is empty.
+            "embedding_source": "audio",
         }
 
         # Save metadata using metadata manager (concurrency safe)
@@ -559,6 +560,101 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
             "file_size": file_size,
         }
 
+    async def upload_voice_embedding(self, embedding_json: str, consent: str, name: str) -> dict:
+        """Upload a voice from a pre-computed speaker embedding.
+
+        Stores the embedding as a safetensors file and marks it immediately
+        ready (no audio processing needed).
+
+        Args:
+            embedding_json: JSON-encoded list of floats (1024 or 2048 dim).
+            consent: Consent recording ID.
+            name: Name for the new voice.
+
+        Returns:
+            dict with voice information.
+        """
+        try:
+            embedding = json.loads(embedding_json)
+        except (json.JSONDecodeError, TypeError) as exc:
+            raise ValueError(f"'speaker_embedding' must be valid JSON: {exc}") from exc
+
+        if not isinstance(embedding, list) or not embedding:
+            raise ValueError("'speaker_embedding' must be a non-empty list of numbers")
+
+        if not all(isinstance(x, (int, float)) for x in embedding):
+            raise ValueError("'speaker_embedding' must contain only numeric values")
+
+        if not all(math.isfinite(x) for x in embedding):
+            raise ValueError("'speaker_embedding' values must be finite (no NaN or Inf)")
+
+        emb_dim = len(embedding)
+        if emb_dim not in {1024, 2048}:
+            logger.warning(
+                "speaker_embedding has %d dimensions; expected 1024 (0.6B) or 2048 (1.7B)",
+                emb_dim,
+            )
+
+        voice_name_lower = name.lower()
+        if voice_name_lower in self.uploaded_speakers:
+            raise ValueError(f"Voice '{name}' already exists")
+
+        sanitized_name = _sanitize_filename(name)
+        sanitized_consent = _sanitize_filename(consent)
+        timestamp = int(time.time())
+
+        # Store as safetensors for efficient loading
+        try:
+            import torch
+            from safetensors.torch import save_file
+
+            tensor = torch.tensor(embedding, dtype=torch.float32)
+            filename = f"{sanitized_name}_{sanitized_consent}_{timestamp}.safetensors"
+            file_path = self.uploaded_speakers_dir / filename
+
+            if not _validate_path_within_directory(file_path, self.uploaded_speakers_dir):
+                raise ValueError("Invalid file path: potential path traversal attack detected")
+
+            save_file({"speaker_embedding": tensor}, str(file_path))
+        except ImportError:
+            raise ValueError("safetensors and torch are required for embedding upload")
+
+        speaker_data = {
+            "name": name,
+            "consent": consent,
+            "file_path": str(file_path),
+            "created_at": timestamp,
+            "mime_type": "application/x-safetensors",
+            "original_filename": filename,
+            "file_size": file_path.stat().st_size,
+            "cache_status": "ready",
+            "cache_file": str(file_path),
+            "cache_generated_at": timestamp,
+            "embedding_source": "direct",
+            "embedding_dim": emb_dim,
+        }
+
+        success = self.metadata_manager.create_speaker(voice_name_lower, speaker_data)
+        if not success:
+            try:
+                file_path.unlink()
+            except Exception:
+                pass
+            raise ValueError(f"Failed to create metadata for voice '{name}' (possibly already exists)")
+
+        self.uploaded_speakers[voice_name_lower] = speaker_data
+        self.supported_speakers.add(voice_name_lower)
+
+        logger.info(f"Uploaded voice '{name}' from speaker embedding ({emb_dim}-dim)")
+
+        return {
+            "name": name,
+            "consent": consent,
+            "created_at": timestamp,
+            "embedding_source": "direct",
+            "embedding_dim": emb_dim,
+        }
+
     async def delete_voice(self, name: str) -> bool:
         """
         Delete an uploaded voice.
@@ -669,23 +765,46 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
             if request.voice is not None and request.voice not in self.supported_speakers:
                 return f"Invalid voice '{request.voice}'. Supported: {', '.join(sorted(self.supported_speakers))}"
 
+        # Validate speaker_embedding constraints
+        if request.speaker_embedding is not None:
+            if task_type != "Base":
+                return "'speaker_embedding' is only valid for Base task"
+            if not request.speaker_embedding:
+                return "'speaker_embedding' must be a non-empty list of floats"
+            # speaker_embedding implies x_vector_only_mode — set it before
+            # Base task validation so callers don't need to pass it explicitly.
+            request.x_vector_only_mode = True
+            emb_len = len(request.speaker_embedding)
+            # ECAPA-TDNN produces 1024-dim (0.6B) or 2048-dim (1.7B)
+            expected_dims = {1024, 2048}
+            if emb_len not in expected_dims:
+                logger.warning(
+                    "speaker_embedding has %d dimensions; expected 1024 "
+                    "(0.6B model) or 2048 (1.7B model). Wrong dimensions "
+                    "will likely result in errors or degraded quality.",
+                    emb_len,
+                )
         # Validate Base task requirements
         if task_type == "Base":
             if request.voice is None:
-                if request.ref_audio is None:
-                    return "Base task requires 'ref_audio' for voice cloning"
-                fmt_err = self._validate_ref_audio_format(request.ref_audio)
-                if fmt_err:
-                    return fmt_err
-                # In-context voice cloning (default) requires non-empty ref_text.
-                # x_vector_only_mode skips in-context and only uses speaker embedding.
-                if not request.x_vector_only_mode:
+                # 1. Ensure a voice source is provided
+                if request.ref_audio is None and getattr(request, "speaker_embedding", None) is None:
+                    return "Base task requires 'ref_audio' or 'speaker_embedding' for voice cloning"
+                # 2. Validate ref_audio format if it exists (using the helper from main)
+                if request.ref_audio is not None:
+                    fmt_err = self._validate_ref_audio_format(request.ref_audio)
+                    if fmt_err:
+                        return fmt_err
+                # 3. Validate text requirements based on the mode
+                if not getattr(request, "x_vector_only_mode", False):
                     if not request.ref_text or not request.ref_text.strip():
                         return (
                             "Base task requires non-empty 'ref_text' (transcript of "
                             "the reference audio) unless 'x_vector_only_mode' is enabled"
                         )
             else:
+                # Handle the case where request.voice is NOT None
+                pass
                 # voice is not None
                 voice_lower = request.voice.lower()
                 if voice_lower in self.uploaded_speakers:
@@ -890,7 +1009,18 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
         # Voice clone: ref_audio resolved in create_speech(), not here.
         if request.ref_text is not None:
             params["ref_text"] = [request.ref_text]
-        if request.x_vector_only_mode is not None:
+        if request.speaker_embedding is not None:
+            # Store as plain float list (not tensor) so it survives msgspec
+            # serialization through the EngineCore IPC boundary.  The talker's
+            # _build_prompt_embeds converts it back to a tensor on the GPU.
+            params["voice_clone_prompt"] = [
+                {
+                    "ref_spk_embedding": list(request.speaker_embedding),
+                }
+            ]
+            # speaker_embedding implies x_vector_only_mode
+            params["x_vector_only_mode"] = [True]
+        elif request.x_vector_only_mode is not None:
             params["x_vector_only_mode"] = [request.x_vector_only_mode]
 
         # Generation parameters
diff --git a/vllm_omni/entrypoints/openai/serving_speech_stream.py b/vllm_omni/entrypoints/openai/serving_speech_stream.py
index d88c88b65..695df04c2 100644
--- a/vllm_omni/entrypoints/openai/serving_speech_stream.py
+++ b/vllm_omni/entrypoints/openai/serving_speech_stream.py
@@ -224,6 +224,7 @@ class OmniStreamingSpeechHandler:
             ref_audio=config.ref_audio,
             ref_text=config.ref_text,
             x_vector_only_mode=config.x_vector_only_mode,
+            speaker_embedding=config.speaker_embedding,
             stream=config.stream_audio,
         )
 
diff --git a/vllm_omni/model_executor/models/qwen3_tts/qwen3_tts_talker.py b/vllm_omni/model_executor/models/qwen3_tts/qwen3_tts_talker.py
index 39d9dba2d..ca9c8ca41 100644
--- a/vllm_omni/model_executor/models/qwen3_tts/qwen3_tts_talker.py
+++ b/vllm_omni/model_executor/models/qwen3_tts/qwen3_tts_talker.py
@@ -1357,11 +1357,17 @@ class Qwen3TTSTalkerForConditionalGeneration(nn.Module):
                 ref_code_prompt = ref_code_t
 
             # Speaker embedding: use prompt embed if provided; otherwise extract from audio.
+            # NOTE: Do NOT use _as_singleton here — the embedding may be a plain
+            # float list (from API via msgspec IPC) that _as_singleton would
+            # destructively unwrap to a single scalar.
             spk = None
             if voice_clone_prompt is not None:
-                spk = _as_singleton(voice_clone_prompt.get("ref_spk_embedding"))
+                spk = voice_clone_prompt.get("ref_spk_embedding")
             if isinstance(spk, torch.Tensor):
                 speaker_embed = spk.to(device=input_ids.device, dtype=torch.bfloat16).view(1, 1, -1)
+            elif isinstance(spk, (list, np.ndarray)):
+                # Plain list/array from API (survived msgspec IPC serialization).
+                speaker_embed = torch.tensor(spk, dtype=torch.bfloat16, device=input_ids.device).view(1, 1, -1)
             else:
                 ref_audio_list = info_dict.get("ref_audio")
                 if not isinstance(ref_audio_list, list) or not ref_audio_list: