livekit · russellmartin-livekit · Sep 14, 2026 · Sep 15, 2026 · Sep 15, 2026 · Sep 16, 2026
@@ -5,10 +5,16 @@
     GPTLiveVoices,
     ResponsesDelegationOptions,
 )
+from .inference_gpt_live_model import (
+    InferenceGPTLiveModel,
+    InferenceResponsesDelegationOptions,
+)
 from .inference_realtime_model import InferenceRealtimeModel
 from .realtime_model import RealtimeModel, RealtimeSession
 
 __all__ = [
+    "InferenceGPTLiveModel",
+    "InferenceResponsesDelegationOptions",
     "InferenceRealtimeModel",
     "RealtimeSession",
     "RealtimeModel",

@@ -76,12 +76,7 @@
 lk_oai_debug = int(os.getenv("LK_OPENAI_DEBUG", 0))
 
 
-class ResponsesDelegationOptions(TypedDict, total=False):
-    """The backend Responses model delegated work runs on, under ``delegation="responses"``.
-
-    A key left unset is not sent, and the service's own default applies.
-    """
-
+class _ResponsesDelegationOptionsBase(TypedDict, total=False):
     model: str
     """Responses model slug; ``gpt-5.6-luna`` when unset."""
     instructions: str
@@ -92,11 +87,16 @@ class ResponsesDelegationOptions(TypedDict, total=False):
     """Responses reasoning settings, for example ``{"effort": "medium"}``."""
     text: ResponseTextConfigParam
     """Responses text settings, for example ``{"verbosity": "low"}``."""
-    service_tier: Literal["auto", "default", "flex", "priority"]
     max_output_tokens: int
     """Upper bound on the tokens one backend response may generate; at least 16."""
 
 
+class ResponsesDelegationOptions(_ResponsesDelegationOptionsBase, total=False):
+    """Backend Responses options for direct OpenAI GPT-Live sessions."""
+
+    service_tier: Literal["auto", "default", "flex", "priority"]
+
+
 @dataclass
 class GPTLiveDelegation:
     """Work the model handed to the application, under client delegation.
@@ -443,15 +443,7 @@ def _reset_for_reconnect(self) -> None:
         self._session_id = None
 
     async def _create_ws_conn(self) -> aiohttp.ClientWebSocketResponse:
-        headers = {
-            "User-Agent": "LiveKit Agents",
-            "Authorization": f"Bearer {self._opts.api_key}",
-        }
-        parsed = urlparse(self._opts.base_url.replace("http", "ws", 1))
-        path = parsed.path.rstrip("/")
-        if not path.endswith("/live/sessions"):
-            path = f"{path}/live/sessions"
-        url = urlunparse((parsed.scheme, parsed.netloc, path, "", "", ""))
+        url, headers = self._create_ws_url_and_headers()
         if lk_oai_debug:
             logger.debug("connecting to GPT-Live API", extra={"lk.pii.url": url})
 
@@ -468,6 +460,18 @@ async def _create_ws_conn(self) -> aiohttp.ClientWebSocketResponse:
                 f"{self._live_model._provider_label} connection error"
             ) from None
 
+    def _create_ws_url_and_headers(self) -> tuple[str, dict[str, str]]:
+        headers = {
+            "User-Agent": "LiveKit Agents",
+            "Authorization": f"Bearer {self._opts.api_key}",
+        }
+        parsed = urlparse(self._opts.base_url.replace("http", "ws", 1))
+        path = parsed.path.rstrip("/")
+        if not path.endswith("/live/sessions"):
+            path = f"{path}/live/sessions"
+        url = urlunparse((parsed.scheme, parsed.netloc, path, "", "", ""))
+        return url, headers
+
     async def _run_ws(self, ws_conn: aiohttp.ClientWebSocketResponse) -> None:
         closing = False
 
@@ -865,7 +869,7 @@ def _handle_error(self, error: types.ErrorBody) -> None:
             "gpt-live returned an error",
             extra={"lk.pii.error": error.model_dump(exclude_none=True)},
         )
-        recoverable = (error.code or error.type or "") not in _FATAL_ERROR_CODES
+        recoverable = not self._is_fatal_error(error)
         api_error = APIError(
             message="GPT-Live returned an error",
             retryable=recoverable,
@@ -874,6 +878,9 @@ def _handle_error(self, error: types.ErrorBody) -> None:
             raise api_error
         self._emit_error(api_error, recoverable=True)
 
+    def _is_fatal_error(self, error: types.ErrorBody) -> bool:
+        return (error.code or error.type or "") in _FATAL_ERROR_CODES
+
     def _emit_error(self, error: Exception, recoverable: bool) -> None:
         self.emit(
             "error",

@@ -0,0 +1,170 @@
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+from typing import Any, Literal, cast
+from urllib.parse import urlencode, urlparse, urlunparse
+
+import aiohttp
+
+from livekit.agents.inference._utils import (
+    HEADER_INFERENCE_PROVIDER,
+    create_access_token,
+    get_default_inference_url,
+    get_inference_headers,
+)
+from livekit.agents.types import (
+    DEFAULT_API_CONNECT_OPTIONS,
+    NOT_GIVEN,
+    APIConnectOptions,
+    NotGivenOr,
+)
+from livekit.agents.utils import is_given
+
+from . import gpt_live_types as types
+from .gpt_live_model import (
+    DEFAULT_VOICE,
+    GPTLiveModel,
+    GPTLiveSession,
+    GPTLiveVoices,
+    ResponsesDelegationOptions,
+    _ResponsesDelegationOptionsBase,
+)
+
+InferenceClass = Literal["priority", "standard", "low"]
+
+
+class InferenceResponsesDelegationOptions(_ResponsesDelegationOptionsBase, total=False):
+    """Responses options supported through LiveKit Inference."""
+
+    service_tier: Literal["default"]
+
+
+_GATEWAY_FATAL_ERROR_CODES = frozenset(
+    {
+        "invalid_event",
+        "session_start_required",
+        "invalid_session",
+        "invalid_model",
+        "invalid_delegated_model",
+        "unsupported_delegated_model",
+        "unsupported_server_tool",
+        "unsupported_service_tier",
+    }
+)
+
+
+@dataclass
+class _InferenceGPTLiveOptions:
+    provider: str | None
+    api_key: str
+    api_secret: str
+    inference_class: InferenceClass | None
+
+
+class InferenceGPTLiveModel(GPTLiveModel):
+    """Native GPT-Live through LiveKit Inference.
+
+    Responses delegation supports function tools and the default service tier.
+    Separately priced OpenAI-hosted tools and service tiers are not available.
+    """
+
+    def __init__(
+        self,
+        model: str,
+        *,
+        provider: str | None = None,
+        voice: GPTLiveVoices | str | dict[str, Any] = DEFAULT_VOICE,
+        delegation: types.DelegationTarget = "responses",
+        responses_options: NotGivenOr[InferenceResponsesDelegationOptions] = NOT_GIVEN,
+        base_url: str | None = None,
+        api_key: str | None = None,
+        api_secret: str | None = None,
+        inference_class: InferenceClass | None = None,
+        http_session: aiohttp.ClientSession | None = None,
+        max_session_duration: NotGivenOr[float | None] = NOT_GIVEN,
+        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
+    ) -> None:
+        if "/" not in model:
+            raise ValueError("model must be provider-prefixed, for example 'openai/gpt-live-1'")
+        if (
+            is_given(responses_options)
+            and "service_tier" in responses_options
+            and responses_options["service_tier"] != "default"
+        ):
+            raise ValueError("LiveKit Inference GPT-Live supports only service_tier='default'")
+
+        resolved_api_key = api_key or os.getenv(
+            "LIVEKIT_INFERENCE_API_KEY", os.getenv("LIVEKIT_API_KEY", "")
+        )
+        if not resolved_api_key:
+            raise ValueError(
+                "api_key is required, either as argument or set LIVEKIT_API_KEY environmental variable"
+            )
+
+        resolved_api_secret = api_secret or os.getenv(
+            "LIVEKIT_INFERENCE_API_SECRET", os.getenv("LIVEKIT_API_SECRET", "")
+        )
+        if not resolved_api_secret:
+            raise ValueError(
+                "api_secret is required, either as argument or set LIVEKIT_API_SECRET environmental variable"
+            )
+
+        super().__init__(
+            model=model,
+            voice=voice,
+            delegation=delegation,
+            responses_options=cast(NotGivenOr[ResponsesDelegationOptions], responses_options),
+            api_key="livekit-inference",
+            base_url=base_url or get_default_inference_url(),
+            http_session=http_session,
+            max_session_duration=max_session_duration,
+            conn_options=conn_options,
+        )
+        self._inference_opts = _InferenceGPTLiveOptions(
+            provider=provider,
+            api_key=resolved_api_key,
+            api_secret=resolved_api_secret,
+            inference_class=inference_class,
+        )
+        self._provider_label = "LiveKit Inference GPT-Live"
+
+    @property
+    def provider(self) -> str:
+        return "livekit"
+
+    def session(self) -> InferenceGPTLiveSession:
+        return InferenceGPTLiveSession(self)
+
+
+class InferenceGPTLiveSession(GPTLiveSession):
+    def __init__(self, duplex_model: InferenceGPTLiveModel) -> None:
+        self._inference_model = duplex_model
+        super().__init__(duplex_model)
+
+    def _create_ws_url_and_headers(self) -> tuple[str, dict[str, str]]:
+        url, _ = super()._create_ws_url_and_headers()
+        parsed = urlparse(url)
+        url = urlunparse(
+            (
+                parsed.scheme,
+                parsed.netloc,
+                parsed.path,
+                "",
+                urlencode({"model": self._opts.model}),
+                "",
+            )
+        )
+
+        opts = self._inference_model._inference_opts
+        headers = get_inference_headers(inference_class=opts.inference_class)
+        headers["Authorization"] = f"Bearer {create_access_token(opts.api_key, opts.api_secret)}"
+        if opts.provider:
+            headers[HEADER_INFERENCE_PROVIDER] = opts.provider
+        return url, headers
+
+    def _is_fatal_error(self, error: types.ErrorBody) -> bool:
+        code = error.code or error.type or ""
+        return (
+            not self._session_started_fut.done() and code in _GATEWAY_FATAL_ERROR_CODES
+        ) or super()._is_fatal_error(error)