Provider API

Use Provider with a ProviderType value and explicit credentials/model. OpenAIProvider adds OpenAI authentication/discovery; FoundryProvider accepts a caller-owned sync or async Azure credential or an API key. See Providers, Foundry authentication, and Multi-Provider Setup.

base_url is an API prefix, not a complete generation endpoint. A shared interface does not imply equal media, reasoning, retry, or batch capabilities across providers and HTTP contracts. This reference follows the current source.

The library constructors Provider, OpenAIProvider, and FoundryProvider accept timeout=120.0 for the total duration of each model HTTP request, including its streaming response. Named ngn connections expose this existing parameter as request_timeout, with the same default for API-key and ChatGPT/Codex authentication. Catalog requests retain their 30-second cap, and shell, whole-run, authentication, and Live session budgets remain separate. Retry policy is unchanged.

nagents.ProviderType

Bases: Enum

Supported provider types.

Source code in src/nagents/provider/base.py

class ProviderType(Enum):
    """Supported provider types."""

    OPENAI_COMPATIBLE = "openai_compatible"  # OpenAI, Gemini-via-compat, Ollama, etc.
    OPENROUTER = "openrouter"  # OpenRouter (OpenAI-compatible with reasoning support)
    LITELLM = "litellm"  # Self-hosted API-key gateway; requires an explicit base URL
    GEMINI_NATIVE = "gemini_native"  # Native Gemini REST API
    ANTHROPIC = "anthropic"  # Anthropic Claude API
    AZURE_OPENAI_COMPATIBLE_V1 = "azure_openai_compatible_v1"  # Azure OpenAI Service
    AZURE_OPENAI_COMPATIBLE = (
        "azure_openai_compatible"  # Azure OpenAI Service (new name for v1, will be default in future)
    )

nagents.Provider

Unified LLM provider for all supported backends.

Example

provider = Provider( provider_type=ProviderType.OPENAI_COMPATIBLE, api_key="sk-...", model="gpt-4o", )

async for event in provider.generate(messages, tools): print(event)

Source code in src/nagents/provider/base.py

class Provider:
    """
    Unified LLM provider for all supported backends.

    Example:
        provider = Provider(
            provider_type=ProviderType.OPENAI_COMPATIBLE,
            api_key="sk-...",
            model="gpt-4o",
        )

        async for event in provider.generate(messages, tools):
            print(event)
    """

    def __init__(
        self,
        provider_type: ProviderType,
        api_key: str = "",
        model: str = "",
        base_url: str | None = None,
        timeout: float = 120.0,
        api_version: str | None = None,
        retry_config: RetryConfig | None = None,
        realtime_config: "RealtimeConfig | None" = None,
        api: str = "auto",
        live_config: "LiveConfig | None" = None,
        *,
        bearer_token_provider: BearerTokenProvider | None = None,
    ):
        """
        Initialize the provider.

        Args:
            provider_type: The type of provider to use
            api_key: API key for authentication
            model: Model identifier to use (for Azure, this is the deployment name)
            base_url: Optional custom base URL (overrides defaults).
                      For Azure OpenAI: https://{resource}.cognitiveservices.azure.com
            timeout: Request timeout in seconds
            api_version: API version (required for Azure OpenAI, e.g., "2024-05-01-preview")
            retry_config: Retry configuration for rate limit and server error handling.
                          Defaults to RetryConfig() (3 retries, 5s base delay).
                          Set RetryConfig(max_retries=0) to disable retry.
            realtime_config: Optional :class:`~nagents.realtime.RealtimeConfig`
                for Realtime speech-to-speech sessions. When set, voice sessions
                reuse the provider's model and this configuration.
            api: HTTP contract: auto, chat_completions, responses, or messages.
                Auto preserves native providers and selects
                chat_completions for LiteLLM and OpenRouter. base_url is an API
                 prefix, not a full generation endpoint.
            live_config: GPT-Live voice/delegation settings for Agent.run() with audio.
            bearer_token_provider: Optional async callable returning a bearer token.
                Mutually exclusive with api_key; supports OpenAI-compatible v1
                chat/Responses and GPT-Live. Called per request attempt/connection.
                The caller owns the credential and its cache/refresh/close lifecycle.
        """
        self.provider_type = provider_type
        self.api_key = api_key
        self.bearer_token_provider = bearer_token_provider
        if bearer_token_provider is not None:
            if api_key:
                raise ValueError("Choose api_key or bearer_token_provider, not both")
            if not callable(bearer_token_provider):
                raise ValueError("bearer_token_provider must be an async callable")
            if provider_type not in {ProviderType.OPENAI_COMPATIBLE, ProviderType.AZURE_OPENAI_COMPATIBLE_V1}:
                raise ValueError("Bearer token authentication requires an OpenAI-compatible v1 provider")
            if api not in {"auto", "chat_completions", "responses"} or realtime_config is not None:
                raise ValueError("Bearer token authentication supports text chat/responses and GPT-Live, not Realtime")
        self.model = model
        self.api_version = api_version
        self.retry_config = retry_config or RetryConfig()
        self.realtime_config = realtime_config
        self.live_config = live_config
        if live_config is not None and realtime_config is not None:
            raise ValueError("Choose live_config or realtime_config, not both")
        if live_config is not None and provider_type not in {
            ProviderType.OPENAI_COMPATIBLE,
            ProviderType.AZURE_OPENAI_COMPATIBLE_V1,
        }:
            raise ValueError("GPT-Live requires an OpenAI-compatible v1 provider")
        if api not in {"auto", "chat_completions", "responses", "messages"}:
            raise ValueError("api must be auto, chat_completions, responses, or messages")
        if provider_type == ProviderType.LITELLM and not base_url:
            raise ValueError("LiteLLM requires an explicit base_url pointing to your gateway")
        default_api = "messages" if provider_type == ProviderType.ANTHROPIC else "chat_completions"
        if provider_type == ProviderType.GEMINI_NATIVE:
            default_api = "auto"
        self.api = default_api if api == "auto" else api
        if provider_type == ProviderType.AZURE_OPENAI_COMPATIBLE_V1 and self.api not in {
            "chat_completions",
            "responses",
        }:
            raise ValueError("Azure v1 supports chat_completions or responses")
        if (
            provider_type
            not in {
                ProviderType.LITELLM,
                ProviderType.OPENAI_COMPATIBLE,
                ProviderType.OPENROUTER,
                ProviderType.AZURE_OPENAI_COMPATIBLE_V1,
            }
            and self.api != default_api
        ):
            raise ValueError("This provider does not support the selected HTTP API contract")
        if provider_type == ProviderType.OPENAI_COMPATIBLE and self.api == "messages" and not base_url:
            raise ValueError("The messages API requires an explicit compatible base_url or an Anthropic provider")
        if base_url:
            parsed = urlsplit(base_url)
            if (
                parsed.scheme not in {"http", "https"}
                or not parsed.hostname
                or any(char.isspace() for char in base_url)
            ):
                raise ValueError("base_url must be an HTTP(S) API prefix")
            if parsed.username or parsed.password or parsed.query or parsed.fragment:
                raise ValueError("base_url must not contain credentials, query parameters, or fragments")
            if parsed.path.rstrip("/").endswith(("/chat/completions", "/responses", "/messages", "/completions")):
                raise ValueError("base_url must be an API prefix, not a generation endpoint")
        self._http = (
            GatewayHTTPClient(timeout=timeout)
            if bearer_token_provider is not None
            or provider_type in {ProviderType.LITELLM, ProviderType.OPENROUTER, ProviderType.AZURE_OPENAI_COMPATIBLE_V1}
            or api != "auto"
            else HTTPClient(timeout=timeout)
        )
        self._model_verified: bool | None = None  # None = not checked, True/False = result
        self._model_list_timeout = min(timeout, 30.0) if math.isfinite(timeout) and timeout > 0 else 30.0

        # Validate Azure-specific requirements
        if provider_type == ProviderType.AZURE_OPENAI_COMPATIBLE_V1 and not base_url:
            raise ValueError(
                "base_url is required for Azure OpenAI. Format: https://{resource}.cognitiveservices.azure.com/openai"
            )
        if provider_type == ProviderType.AZURE_OPENAI_COMPATIBLE:
            if not api_version:
                raise ValueError("api_version is required for Azure OpenAI")
            if not base_url:
                raise ValueError("base_url is required for Azure OpenAI")
        if base_url:
            self.base_url = base_url.rstrip("/")
        elif provider_type == ProviderType.OPENAI_COMPATIBLE:
            self.base_url = "https://api.openai.com/v1"
        elif provider_type == ProviderType.OPENROUTER:
            self.base_url = "https://openrouter.ai/api/v1"
        elif provider_type == ProviderType.ANTHROPIC:
            self.base_url = "https://api.anthropic.com/v1"
        elif provider_type == ProviderType.AZURE_OPENAI_COMPATIBLE_V1:
            # Azure requires base_url, this should not be reached due to validation above
            raise ValueError("base_url is required for Azure OpenAI")
        else:  # GEMINI_NATIVE
            self.base_url = "https://generativelanguage.googleapis.com/v1beta"
        if bearer_token_provider is not None or live_config is not None:
            validate_prefix(self.base_url)
        if not self.api_key and bearer_token_provider is None:
            raise ValueError("API key is required for provider initialization")

    async def auth_headers(self, url: str) -> dict[str, str]:
        """Fresh authentication per attempt; caller owns the credential lifetime."""
        validate_prefix(self.base_url)
        validate_endpoint(url, websocket=url.startswith(("ws:", "wss:")))
        if self.bearer_token_provider is not None:
            return await bearer_headers(self.bearer_token_provider)
        return {"Authorization": f"Bearer {self.api_key}"}

    def live_endpoint(self, *, websocket: bool = False) -> str:
        """Derive GPT-Live endpoints independently of Realtime/Voice Live."""
        validate_prefix(self.base_url)
        url = self._api_prefix() + "/live/sessions"
        return url.replace("https://", "wss://", 1).replace("http://", "ws://", 1) if websocket else url

    def _api_prefix(self) -> str:
        prefix = self.base_url.rstrip("/")
        if self.provider_type == ProviderType.AZURE_OPENAI_COMPATIBLE_V1 and not prefix.endswith("/v1"):
            prefix += "/v1"
        return prefix

    def set_http_logger(self, http_logger: "HTTPLogger | None") -> None:
        """
        Set the HTTP logger for debugging/auditing.

        Args:
            http_logger: The HTTP logger to use, or None to disable logging.
        """
        self._http.set_logger(http_logger)

    def set_session_id(self, session_id: str | None) -> None:
        """
        Set the current session ID for HTTP logging.

        Args:
            session_id: The session ID to include in log entries.
        """
        self._http.set_session_id(session_id)

    async def get_model_list(self) -> list[str]:
        """Fetch model IDs explicitly, without changing the model or verification cache.

        OpenAI-compatible, OpenRouter, and LiteLLM providers use GET on the
        configured API prefix plus /models. IDs retain upstream order and spelling;
        duplicates are removed. An empty catalog is valid, not a fallback on error.

        Raises:
            NotImplementedError: Discovery is not implemented for this provider.
            ModelListError: Credentials, configuration, transport, or catalog data
                are invalid. Upstream bodies and credentials are never included.

        Catalog membership does not guarantee generation access or capabilities.
        Each call makes a fresh bounded request, with no redirects or raw logging.
        """
        if self.provider_type not in {
            ProviderType.OPENAI_COMPATIBLE,
            ProviderType.OPENROUTER,
            ProviderType.LITELLM,
            ProviderType.AZURE_OPENAI_COMPATIBLE_V1,
        }:
            raise NotImplementedError(
                "Model discovery is not implemented for this provider; enter a model ID manually."
            )
        try:
            # base_url is mutable: validate the current prefix before sending a key.
            prefix = self.base_url
            parsed = urlsplit(prefix)
            if (
                parsed.scheme not in {"http", "https"}
                or not parsed.hostname
                or parsed.username
                or parsed.password
                or parsed.query
                or parsed.fragment
                or any(char.isspace() or unicodedata.category(char).startswith("C") for char in prefix)
                or parsed.path.rstrip("/").endswith(("/chat/completions", "/responses", "/messages", "/completions"))
            ):
                raise ValueError
            if not self.api_key.strip() and self.bearer_token_provider is None:
                raise ValueError
            headers = (
                await self.auth_headers(prefix)
                if self.bearer_token_provider
                else {"Authorization": f"Bearer {self.api_key}"}
            )
            if self.provider_type == ProviderType.OPENROUTER:
                headers["HTTP-Referer"] = "https://github.com/nagents"
            # A short-lived secure transport also isolates catalogs from generation
            # cookies, loggers, and the ordinary HTTPClient's redirect behavior.
            client = GatewayHTTPClient(timeout=self._model_list_timeout)
            async with client:
                response = await client.get_json(f"{self._api_prefix()}/models", headers)
        except Exception:
            raise ModelListError(
                "Model discovery failed; check provider configuration, credentials, and availability."
            ) from None
        if "error" in response or not isinstance(response.get("data"), list):
            raise ModelListError("Provider returned an invalid model catalog.")
        models: list[str] = []
        seen: set[str] = set()
        for item in list_data(response["data"]):
            model_id = item.get("id") if isinstance(item, dict) else None
            if (
                not isinstance(model_id, str)
                or not model_id.strip()
                or any(unicodedata.category(char).startswith("C") for char in model_id)
            ):
                raise ModelListError("Provider returned an invalid model ID.")
            if model_id not in seen:
                models.append(model_id)
                seen.add(model_id)
        return models

    async def verify_model(self, force: bool = False) -> bool:
        """Verify that the specified model exists in model list endpoint.

        Args:
            force: If True, re-verify even if already cached.

        Returns:
            True if the model is found, False otherwise.

        Note:
            Verification is not implemented here for Anthropic or Azure;
            those providers continue to return True without a catalog request.
        """
        # Return cached result if available and not forcing re-verification
        if not force and self._model_verified is not None:
            return self._model_verified

        # Preserve local verification for Anthropic and Azure.
        if self.provider_type == ProviderType.ANTHROPIC:
            logger.info(f"Skipping model verification for Anthropic (model: {self.model})")
            self._model_verified = True
            return True

        if self.provider_type == ProviderType.AZURE_OPENAI_COMPATIBLE_V1:
            logger.info(f"Skipping model verification for Azure OpenAI (deployment: {self.model})")
            self._model_verified = True
            return True
        if self.provider_type == ProviderType.AZURE_OPENAI_COMPATIBLE:
            logger.info(f"Skipping model verification for Azure OpenAI (deployment: {self.model})")
            self._model_verified = True
            return True
        try:
            if self.provider_type in (ProviderType.OPENAI_COMPATIBLE, ProviderType.OPENROUTER, ProviderType.LITELLM):
                models = await self.get_model_list()

                # Check for exact match first
                if self.model in models:
                    logger.info(f"Model '{self.model}' verified successfully")
                    self._model_verified = True
                    return True

                # Check for match with "models/" prefix (Gemini via OpenAI compat)
                prefixed_model = f"models/{self.model}"
                if prefixed_model in models:
                    logger.info(f"Model '{self.model}' verified successfully (with models/ prefix)")
                    self._model_verified = True
                    return True

                logger.warning("Selected model not found in provider catalog")
                self._model_verified = False
                return False
            else:  # GEMINI_NATIVE
                url = f"{self.base_url}/models?key={self.api_key}"
                response = await self._http.get_json(url)
                models = [m["name"] for m in response.get("models", [])]

                # Check for exact match first
                if self.model in models:
                    logger.info(f"Model '{self.model}' verified successfully")
                    self._model_verified = True
                    return True

                # Check for match with "models/" prefix
                prefixed_model = f"models/{self.model}"
                if prefixed_model in models:
                    logger.info(f"Model '{self.model}' verified successfully (with models/ prefix)")
                    self._model_verified = True
                    return True

                # Check for match without "models/" prefix (user might provide full name)
                if self.model.startswith("models/"):
                    stripped_model = self.model[7:]  # Remove "models/" prefix
                    if f"models/{stripped_model}" in models:
                        logger.info(f"Model '{self.model}' verified successfully")
                        self._model_verified = True
                        return True

                logger.warning(f"Model '{self.model}' not found in provider. Available models: {models[:10]}...")
                self._model_verified = False
                return False
        except HTTPError as e:
            logger.error("Model verification failed (HTTP %s)", e.status)
            self._model_verified = False
            return False
        except Exception:
            logger.error("Model verification failed")
            self._model_verified = False
            return False

    @property
    def is_model_verified(self) -> bool | None:
        """Check if the model has been verified.

        Returns:
            True if verified successfully, False if verification failed,
            None if verification has not been performed yet.
        """
        return self._model_verified

    @property
    def supported_media_formats(self) -> MediaCapabilities:
        """Get the media format capabilities for this provider+model.

        Returns model-specific overrides when available, otherwise
        falls back to provider-level defaults.

        Returns:
            MediaCapabilities describing supported audio, image, and document formats.
        """
        if self.api == "responses":
            # The Responses adapter supports user images, not audio or files.
            return get_media_capabilities("openai_compatible", "")
        if self.api == "messages":
            return get_media_capabilities("anthropic", self.model)
        provider = "openai_compatible" if self.provider_type == ProviderType.LITELLM else self.provider_type.value
        return get_media_capabilities(provider, self.model)

    async def generate(
        self,
        messages: list[Message],
        tools: list[ToolDefinition] | None = None,
        config: GenerationConfig | None = None,
        stream: bool = True,
        verify_model: bool = False,
    ) -> AsyncIterator[Event]:
        """
        Generate a response from the model.

        Args:
            messages: Conversation history
            tools: Optional list of tools the model can call
            config: Optional generation configuration
            stream: Whether to stream the response
            verify_model: If True, verify the model exists before generating

        Yields:
            Events as they occur (text chunks, tool calls, etc.)
        """
        started = monotonic()
        # Verify model if requested
        if verify_model:
            is_valid = await self.verify_model()
            if not is_valid:
                yield ErrorEvent(
                    message=f"Model '{self.model}' not found in provider",
                    code="MODEL_NOT_FOUND",
                    recoverable=False,
                )
                return

        try:
            async for event in self._with_retry(messages, tools, config, stream):
                yield event
        except HTTPError as e:
            logger.error("Generation failed (HTTP %s)", e.status)
            yield ErrorEvent(
                message=f"Provider request failed (HTTP {e.status}); check the API route, model access, credentials, or gateway availability.",
                code=str(e.status),
                recoverable=e.is_retryable(),
                extra=failure_extra(e, started),
            )
        except ProtocolError as e:
            yield ErrorEvent(
                message=str(e), code="PROVIDER_PROTOCOL_ERROR", recoverable=False, extra=failure_extra(e, started)
            )
        except Exception as e:
            logger.error("Provider request failed or returned invalid data")
            yield ErrorEvent(
                message="Provider request failed, timed out, or returned invalid data; check the API route and request settings.",
                code="PROVIDER_REQUEST_FAILED",
                recoverable=False,
                extra=failure_extra(e, started),
            )

    async def _dispatch(
        self,
        messages: list[Message],
        tools: list[ToolDefinition] | None,
        config: GenerationConfig | None,
        stream: bool,
    ) -> AsyncIterator[Event]:
        """Dispatch to the appropriate provider-specific generator."""
        if self.api == "responses":
            async for event in self._generate_responses(messages, tools, config, stream):
                yield event
        elif self.api == "messages":
            async for event in self._generate_anthropic(messages, tools, config, stream):
                yield event
        elif self.provider_type in {ProviderType.OPENAI_COMPATIBLE, ProviderType.LITELLM}:
            async for event in self._generate_openai(messages, tools, config, stream):
                yield event
        elif self.provider_type == ProviderType.OPENROUTER:
            async for event in self._generate_openrouter(messages, tools, config, stream):
                yield event
        elif (
            self.provider_type == ProviderType.AZURE_OPENAI_COMPATIBLE
            or self.provider_type == ProviderType.AZURE_OPENAI_COMPATIBLE_V1
        ):
            async for event in self._generate_azure_openai(messages, tools, config, stream):
                yield event
        elif self.provider_type == ProviderType.ANTHROPIC:
            async for event in self._generate_anthropic(messages, tools, config, stream):
                yield event
        else:
            async for event in self._generate_gemini(messages, tools, config, stream):
                yield event

    async def _with_retry(
        self,
        messages: list[Message],
        tools: list[ToolDefinition] | None,
        config: GenerationConfig | None,
        stream: bool,
    ) -> AsyncIterator[Event]:
        """Wrap dispatch with retry logic for rate limit and server errors.

        On retryable errors (429, 5xx):
        - Yields a RateLimitEvent so callers have visibility
        - Logs a WARNING with retry details
        - Sleeps for the calculated delay
        - Retries the request

        On retry exhaustion:
        - Logs an ERROR
        - Re-raises the HTTPError (caught by generate()'s outer handler)
        """
        max_retries = self.retry_config.max_retries

        for attempt in range(max_retries + 1):
            emitted = False
            pending: dict[tuple[str, int], ToolCallProgressEvent] = {}
            try:
                async for event in self._dispatch(messages, tools, config, stream):
                    if isinstance(event, ToolCallProgressEvent):
                        key = (event.generation_id, event.index)
                        if event.status == "streaming":
                            pending[key] = event
                        else:
                            pending.pop(key, None)
                    else:
                        emitted = True
                    yield event
                return  # Success — exit retry loop
            except Exception as e:
                # A preview never authorizes execution. Retire it before an
                # error or retry so the client cannot leave a stale skeleton.
                for snapshot in pending.values():
                    yield replace(snapshot, status="abandoned")
                if not isinstance(e, HTTPError):
                    raise
                is_last_attempt = attempt >= max_retries
                if emitted or not e.is_retryable() or is_last_attempt:
                    if attempt > 0:
                        logger.error("HTTP %s failed after %s retries", e.status, attempt)
                    raise  # Re-raise to generate()'s outer except

                delay = self._calculate_delay(e, attempt)
                error_label = "RateLimitReached" if e.is_rate_limited() else "ServerError"

                logger.warning(
                    f"Retrying HTTP {e.status} in {delay}s (attempt {attempt + 1}/{max_retries}): {error_label}"
                )

                yield RateLimitEvent(
                    attempt=attempt + 1,
                    max_retries=max_retries,
                    retry_after=delay,
                    status_code=e.status,
                    message=f"Retrying HTTP {e.status} in {delay}s "
                    f"(attempt {attempt + 1}/{max_retries}): {error_label}",
                    rate_limit_info={},
                )

                await asyncio.sleep(delay)

    def _calculate_delay(self, error: HTTPError, attempt: int) -> float:
        """Calculate retry delay, preferring Retry-After header when available.

        Args:
            error: The HTTP error with response headers
            attempt: Zero-based attempt index

        Returns:
            Delay in seconds before next retry
        """
        if self.retry_config.respect_retry_after:
            retry_after = error.get_retry_after()
            if retry_after is not None and retry_after >= 0:
                return min(retry_after, self.retry_config.max_delay)

        # Exponential backoff: 5s, 10s, 20s, ...
        delay: float = min(
            self.retry_config.base_delay * (2**attempt),
            self.retry_config.max_delay,
        )
        return delay

    async def _generate_openai(
        self,
        messages: list[Message],
        tools: list[ToolDefinition] | None,
        config: GenerationConfig | None,
        stream: bool,
    ) -> AsyncIterator[Event]:
        """Generate using OpenAI-compatible API."""
        url = f"{self.base_url}/chat/completions"
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
        }
        if self.bearer_token_provider is not None:
            headers.update(await self.auth_headers(url))

        body: dict[str, Any] = {
            "model": self.model,
            "messages": openai_adapter.format_messages(messages),
            "stream": stream,
        }

        # Request usage data in streaming responses
        if stream:
            body["stream_options"] = {"include_usage": True}

        if tools:
            body["tools"] = openai_adapter.format_tools(tools)

        if config:
            if config.temperature is not None:
                body["temperature"] = config.temperature
            if config.max_tokens is not None:
                body["max_completion_tokens"] = config.max_tokens
            if config.top_p is not None:
                body["top_p"] = config.top_p
            if config.stop:
                body["stop"] = config.stop

        if stream:
            async for event in self._stream_openai(url, body, headers):
                yield event
        else:
            async for event in self._non_stream_openai(url, body, headers):
                yield event

    async def _generate_openrouter(
        self,
        messages: list[Message],
        tools: list[ToolDefinition] | None,
        config: GenerationConfig | None,
        stream: bool,
    ) -> AsyncIterator[Event]:
        """Generate using OpenRouter API (OpenAI-compatible with reasoning support)."""
        url = f"{self.base_url}/chat/completions"
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
            "HTTP-Referer": "https://github.com/nagents",
        }

        body: dict[str, Any] = {
            "model": self.model,
            "messages": openai_adapter.format_messages(messages),
            "stream": stream,
        }

        if stream:
            body["stream_options"] = {"include_usage": True}

        if tools:
            body["tools"] = openai_adapter.format_tools(tools)

        if config:
            if config.temperature is not None:
                body["temperature"] = config.temperature
            if config.max_tokens is not None:
                body["max_completion_tokens"] = config.max_tokens
            if config.top_p is not None:
                body["top_p"] = config.top_p
            if config.stop:
                body["stop"] = config.stop
            if config.reasoning:
                body["reasoning"] = config.reasoning

        if stream:
            async for event in self._stream_openai(url, body, headers):
                yield event
        else:
            async for event in self._non_stream_openai(url, body, headers):
                yield event

    async def _generate_azure_openai(
        self,
        messages: list[Message],
        tools: list[ToolDefinition] | None,
        config: GenerationConfig | None,
        stream: bool,
    ) -> AsyncIterator[Event]:
        """Generate using Azure OpenAI Service API.

        Two variants are supported:

        AZURE_OPENAI_COMPATIBLE (deployment-based):
        - URL: https://{resource}.cognitiveservices.azure.com/openai/deployments/{deployment}/chat/completions
        - Auth: api-key header
        - Requires api-version query parameter

        AZURE_OPENAI_COMPATIBLE_V1 (OpenAI-compatible):
        - URL: https://{resource}.openai.azure.com/openai/v1/chat/completions
        - Auth: Authorization Bearer token
        - Model specified in request body (like standard OpenAI)
        """
        # Azure URL format: {base_url}/deployments/{deployment}/chat/completions?api-version={version}
        # Note: base_url should include /openai suffix (e.g., https://{resource}.cognitiveservices.azure.com/openai)
        if self.provider_type == ProviderType.AZURE_OPENAI_COMPATIBLE:
            url = f"{self.base_url}/deployments/{self.model}/chat/completions?api-version={self.api_version}"
            # Azure uses api-key header instead of Authorization Bearer
            headers = {
                "api-key": self.api_key,
                "Content-Type": "application/json",
            }
        elif self.provider_type == ProviderType.AZURE_OPENAI_COMPATIBLE_V1:
            # V1 Azure OpenAI format - standard OpenAI-compatible endpoint
            # URL: https://{resource}.openai.azure.com/openai/v1/chat/completions
            # Uses Bearer token auth and model in body (like standard OpenAI)
            prefix = self._api_prefix()
            url = f"{prefix}/chat/completions"
            headers = {
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json",
            }
        else:
            raise ValueError(f"Unexpected provider type for Azure generation: {self.provider_type}")

        if self.bearer_token_provider is not None:
            headers.update(await self.auth_headers(url))

        body: dict[str, Any] = {
            "messages": openai_adapter.format_messages(messages),
            "stream": stream,
        }

        # V1 API requires model in body (like standard OpenAI)
        if self.provider_type == ProviderType.AZURE_OPENAI_COMPATIBLE_V1:
            body["model"] = self.model

        # Request usage data in streaming responses
        if stream:
            body["stream_options"] = {"include_usage": True}

        if tools:
            body["tools"] = openai_adapter.format_tools(tools)

        if config:
            if config.temperature is not None:
                body["temperature"] = config.temperature
            if config.max_tokens is not None:
                body["max_tokens"] = config.max_tokens  # Azure uses max_tokens, not max_completion_tokens
            if config.top_p is not None:
                body["top_p"] = config.top_p
            if config.stop:
                body["stop"] = config.stop

        # Reuse the same streaming/non-streaming handlers as OpenAI
        if stream:
            async for event in self._stream_openai(url, body, headers):
                yield event
        else:
            async for event in self._non_stream_openai(url, body, headers):
                yield event

    async def _stream_openai(
        self,
        url: str,
        body: dict[str, Any],
        headers: dict[str, str],
    ) -> AsyncIterator[Event]:
        """Handle streaming OpenAI response."""
        full_text = ""
        tool_accumulator = openai_adapter.StreamingToolCallAccumulator()
        latest_usage = Usage()
        finish_reason = FinishReason.UNKNOWN
        extra: dict[str, Any] = {}
        completed = False
        reasoning_details: dict[int, dict[str, Any]] = {}
        terminal_reason = ""

        async for data in self._http.post_stream(url, body, headers):
            if data == "[DONE]":
                completed = True
                break

            chunk = load_object(data)
            if chunk.get("error"):
                raise ProtocolError("Provider stream reported a generation error; no tools were released.")

            # Capture extra metadata
            if "id" in chunk:
                extra["id"] = chunk["id"]
            if "model" in chunk:
                extra["model"] = chunk["model"]
            if "created" in chunk:
                extra["created"] = chunk["created"]

            # Handle usage (OpenAI sends this in a separate final chunk with empty choices)
            usage = chunk.get("usage")
            if usage:
                latest_usage = token_usage(usage, "chat_completions")

            choices = list_data(chunk.get("choices") or [])
            if not choices:
                continue

            if len(choices) != 1:
                raise ProtocolError("Provider returned multiple choices for a single-choice request.")
            choice = object_data(choices[0])
            index = choice.get("index", 0)
            if type(index) is not int or index != 0:
                raise ProtocolError("Provider returned an unexpected choice index; no tools were released.")
            delta = object_data(choice.get("delta") or {})
            if terminal_reason and any(
                delta.get(key)
                for key in ("content", "refusal", "tool_calls", "reasoning", "reasoning_content", "reasoning_details")
            ):
                raise ProtocolError("Provider sent generation data after the terminal choice; no tools were released.")

            # Parse finish_reason
            fr = choice.get("finish_reason")
            if fr:
                if terminal_reason and fr != terminal_reason:
                    raise ProtocolError("Provider returned inconsistent terminal finish reasons.")
                terminal_reason = string_data(fr)
                if fr == "stop":
                    finish_reason = FinishReason.STOP
                elif fr == "tool_calls":
                    finish_reason = FinishReason.TOOL_CALLS
                elif fr == "length":
                    finish_reason = FinishReason.LENGTH
                elif fr == "content_filter":
                    finish_reason = FinishReason.CONTENT_FILTER
                else:
                    raise ProtocolError(
                        "Provider stream returned an unsuccessful finish reason; no tools were released."
                    )

            # Handle reasoning content (e.g., from Kimi, DeepSeek, o1 models, Ollama)
            reasoning = delta.get("reasoning_content") or delta.get("reasoning")
            if reasoning:
                yield ReasoningChunkEvent(chunk=string_data(reasoning), usage=latest_usage)
            for raw_detail in list_data(delta.get("reasoning_details") or []):
                detail = object_data(raw_detail)
                index = detail.get("index", 0)
                if type(index) is not int or not 0 <= index < 1024:
                    raise ProtocolError("Provider returned an invalid reasoning index.")
                previous = reasoning_details.setdefault(index, {})
                for key, value in detail.items():
                    previous[key] = (
                        previous.get(key, "") + string_data(value) if key in {"text", "data", "summary"} else value
                    )

            # Handle text content
            content = delta.get("content") or delta.get("refusal")
            if content:
                text = string_data(content)
                full_text += text
                yield TextChunkEvent(chunk=text, usage=latest_usage, extra=extra)

            # Handle tool calls
            if "tool_calls" in delta:
                tool_accumulator.add_delta(delta)
                for preview in tool_accumulator.previews():
                    yield preview

        if not completed or finish_reason == FinishReason.UNKNOWN:
            raise ProtocolError("Provider stream ended before completion; no tools were released.")
        tool_calls = tool_accumulator.get_complete_tool_calls()
        if tool_calls and finish_reason not in {FinishReason.TOOL_CALLS, FinishReason.STOP}:
            raise ProtocolError("Provider returned incomplete tool calls; no tools were released.")
        if finish_reason == FinishReason.TOOL_CALLS and not tool_calls:
            raise ProtocolError("Provider finished with tools but returned no complete calls.")
        if tool_calls and reasoning_details:
            tool_calls[0].metadata["reasoning_details"] = json.dumps(
                [detail for _, detail in sorted(reasoning_details.items())]
            )
        yield TextDoneEvent(
            text=full_text,
            usage=latest_usage,
            finish_reason=FinishReason.TOOL_CALLS if tool_calls else finish_reason,
            extra=extra,
        )
        if tool_calls:
            for tc in tool_calls:
                call = ToolCallEvent(
                    id=tc.id,
                    name=tc.name,
                    arguments=tc.arguments,
                    usage=latest_usage,
                    finish_reason=FinishReason.TOOL_CALLS,
                    extra=extra,
                    metadata=tc.metadata,
                )
                yield tool_accumulator.ready(call)
                yield call

    async def _non_stream_openai(
        self,
        url: str,
        body: dict[str, Any],
        headers: dict[str, str],
    ) -> AsyncIterator[Event]:
        """Handle non-streaming OpenAI response."""
        response = await self._http.post_json(url, body, headers)
        if response.get("error"):
            raise ProtocolError("Provider reported a generation error; no tools were released.")

        choices = response.get("choices") or []
        if not choices:
            yield ErrorEvent(message="No choices in response")
            return

        choice = choices[0]
        message = choice.get("message") or {}
        if len(choices) != 1:
            raise ProtocolError("Provider returned multiple choices for a single-choice request.")
        index = choice.get("index", 0)
        if type(index) is not int or index != 0:
            raise ProtocolError("Provider returned an unexpected choice index; no tools were released.")

        # Parse finish_reason
        finish_reason = FinishReason.UNKNOWN
        fr = choice.get("finish_reason")
        if fr == "stop":
            finish_reason = FinishReason.STOP
        elif fr == "tool_calls":
            finish_reason = FinishReason.TOOL_CALLS
        elif fr == "length":
            finish_reason = FinishReason.LENGTH
        elif fr == "content_filter":
            finish_reason = FinishReason.CONTENT_FILTER
        else:
            raise ProtocolError("Provider returned an unsuccessful finish reason; no tools were released.")

        # Parse extra metadata
        extra: dict[str, Any] = {}
        if "id" in response:
            extra["id"] = response["id"]
        if "model" in response:
            extra["model"] = response["model"]
        if "created" in response:
            extra["created"] = response["created"]

        # Handle usage with detailed breakdown
        latest_usage = token_usage(response.get("usage"), "chat_completions")

        # Handle reasoning content (from chain-of-thought models, Ollama)
        reasoning = message.get("reasoning_content") or message.get("reasoning")
        if reasoning:
            yield ReasoningChunkEvent(chunk=reasoning, usage=latest_usage, extra=extra)

        # Handle tool calls
        tool_calls = openai_adapter.parse_tool_calls(choice)
        if tool_calls and finish_reason not in {FinishReason.STOP, FinishReason.TOOL_CALLS}:
            raise ProtocolError("Provider returned incomplete tool calls; no tools were released.")
        if finish_reason == FinishReason.TOOL_CALLS and not tool_calls:
            raise ProtocolError("Provider finished with tools but returned no complete calls.")
        if tool_calls and message.get("reasoning_details"):
            tool_calls[0].metadata["reasoning_details"] = json.dumps(list_data(message["reasoning_details"]))
        content = string_data(message.get("content") or message.get("refusal") or "")
        yield TextDoneEvent(
            text=content,
            usage=latest_usage,
            finish_reason=FinishReason.TOOL_CALLS if tool_calls else finish_reason,
            extra=extra,
        )
        if tool_calls:
            for tc in tool_calls:
                yield ToolCallEvent(
                    id=tc.id,
                    name=tc.name,
                    arguments=tc.arguments,
                    usage=latest_usage,
                    finish_reason=FinishReason.TOOL_CALLS,
                    extra=extra,
                    metadata=tc.metadata,
                )

    async def _generate_gemini(
        self,
        messages: list[Message],
        tools: list[ToolDefinition] | None,
        config: GenerationConfig | None,
        stream: bool,
    ) -> AsyncIterator[Event]:
        """Generate using native Gemini REST API."""
        # Import gemini adapter (lazy import to allow optional usage)
        from ..adapters import gemini as gemini_adapter

        endpoint = "streamGenerateContent" if stream else "generateContent"
        url = f"{self.base_url}/models/{self.model}:{endpoint}?key={self.api_key}"

        if stream:
            url += "&alt=sse"

        headers = {"Content-Type": "application/json"}

        body = gemini_adapter.format_request(messages, tools, config)

        if stream:
            async for event in self._stream_gemini(url, body, headers):
                yield event
        else:
            async for event in self._non_stream_gemini(url, body, headers):
                yield event

    async def _stream_gemini(
        self,
        url: str,
        body: dict[str, Any],
        headers: dict[str, str],
    ) -> AsyncIterator[Event]:
        """Handle streaming Gemini response."""
        from ..adapters import gemini as gemini_adapter

        full_text = ""
        all_tool_calls: list[ToolCall] = []
        progress = ToolCallProgressTracker()
        latest_usage = Usage()
        finish_reason = FinishReason.UNKNOWN
        extra: dict[str, Any] = {}

        async for data in self._http.post_stream(url, body, headers):
            chunk = load_object(data)
            text, tool_calls, usage = gemini_adapter.parse_response(chunk)
            reason = gemini_adapter.parse_finish_reason(chunk)
            if finish_reason != FinishReason.UNKNOWN:
                if text or tool_calls or reason not in {FinishReason.UNKNOWN, finish_reason}:
                    raise ProtocolError("Gemini sent generation data after completion; no tools were released.")
            elif reason != FinishReason.UNKNOWN:
                finish_reason = reason

            if usage is not None:
                latest_usage = token_usage(usage, "gemini")

            if text:
                full_text += text
                yield TextChunkEvent(chunk=text, usage=latest_usage, extra=extra)

            if tool_calls:
                for parsed_call in tool_calls:
                    for preview in progress.preview(
                        len(all_tool_calls),
                        parsed_call.id,
                        parsed_call.name,
                        json.dumps(parsed_call.arguments, allow_nan=False),
                    ):
                        yield preview
                    all_tool_calls.append(parsed_call)

        if finish_reason == FinishReason.UNKNOWN:
            raise ProtocolError("Gemini stream ended before completion; no tools were released.")
        if all_tool_calls and finish_reason != FinishReason.STOP:
            raise ProtocolError("Gemini returned incomplete tool calls; no tools were released.")
        yield TextDoneEvent(
            text=full_text,
            usage=latest_usage,
            finish_reason=FinishReason.TOOL_CALLS if all_tool_calls else finish_reason,
            extra=extra,
        )
        if all_tool_calls:
            for index, tc in enumerate(all_tool_calls):
                call = ToolCallEvent(
                    id=tc.id,
                    name=tc.name,
                    arguments=tc.arguments,
                    usage=latest_usage,
                    finish_reason=FinishReason.TOOL_CALLS,
                    extra=extra,
                    metadata=tc.metadata,
                )
                yield progress.ready(index, call)
                yield call

    async def _non_stream_gemini(
        self,
        url: str,
        body: dict[str, Any],
        headers: dict[str, str],
    ) -> AsyncIterator[Event]:
        """Handle non-streaming Gemini response."""
        from ..adapters import gemini as gemini_adapter

        response = await self._http.post_json(url, body, headers)
        text, tool_calls, usage = gemini_adapter.parse_response(response)

        latest_usage = token_usage(usage, "gemini")
        finish_reason = gemini_adapter.parse_finish_reason(response)
        if finish_reason == FinishReason.UNKNOWN:
            raise ProtocolError("Gemini response ended before completion; no tools were released.")
        if tool_calls and finish_reason != FinishReason.STOP:
            raise ProtocolError("Gemini returned incomplete tool calls; no tools were released.")

        extra: dict[str, Any] = {}
        if "model" in response:
            extra["model"] = response["model"]

        yield TextDoneEvent(
            text=text,
            usage=latest_usage,
            finish_reason=FinishReason.TOOL_CALLS if tool_calls else finish_reason,
            extra=extra,
        )
        if tool_calls:
            for tc in tool_calls:
                yield ToolCallEvent(
                    id=tc.id,
                    name=tc.name,
                    arguments=tc.arguments,
                    usage=latest_usage,
                    finish_reason=FinishReason.TOOL_CALLS,
                    extra=extra,
                    metadata=tc.metadata,
                )

    async def _generate_anthropic(
        self,
        messages: list[Message],
        tools: list[ToolDefinition] | None,
        config: GenerationConfig | None,
        stream: bool,
    ) -> AsyncIterator[Event]:
        """Generate using Anthropic Claude API."""
        from ..adapters import anthropic as anthropic_adapter

        url = f"{self.base_url}/messages"
        headers = {
            "x-api-key": self.api_key,
            "anthropic-version": "2023-06-01",
            "Content-Type": "application/json",
        }
        if self.provider_type != ProviderType.ANTHROPIC:
            prefix = self.base_url if self.base_url.endswith("/v1") else f"{self.base_url}/v1"
            url = f"{prefix}/messages"
            headers.pop("x-api-key")
            headers["Authorization"] = f"Bearer {self.api_key}"

        # Format messages (extracts system prompt separately)
        system_prompt, formatted_messages = anthropic_adapter.format_messages(messages)

        body: dict[str, Any] = {
            "model": self.model,
            "messages": formatted_messages,
            "max_tokens": config.max_tokens if config and config.max_tokens else 4096,
            "stream": stream,
        }

        if system_prompt:
            body["system"] = system_prompt

        if tools:
            body["tools"] = anthropic_adapter.format_tools(tools)

        if config:
            if config.temperature is not None:
                body["temperature"] = config.temperature
            if config.top_p is not None:
                body["top_p"] = config.top_p
            if config.stop:
                body["stop_sequences"] = config.stop

        if stream:
            body["stream"] = True
            async for event in self._stream_anthropic(url, body, headers):
                yield event
        else:
            async for event in self._non_stream_anthropic(url, body, headers):
                yield event

    async def _stream_anthropic(
        self,
        url: str,
        body: dict[str, Any],
        headers: dict[str, str],
    ) -> AsyncIterator[Event]:
        """Handle streaming Anthropic response."""
        from ..adapters import anthropic as anthropic_adapter

        full_text = ""
        tool_accumulator = anthropic_adapter.StreamingToolCallAccumulator()
        current_block_index = 0
        latest_usage = Usage()
        usage_data: dict[str, Any] = {}
        finish_reason = FinishReason.UNKNOWN
        completed = False
        open_blocks: dict[int, str] = {}
        seen_blocks: set[int] = set()

        async for data in self._http.post_stream(url, body, headers):
            chunk = load_object(data)

            event_type = chunk.get("type")
            if event_type == "error" or chunk.get("error"):
                raise ProtocolError("Messages stream reported a generation error; no tools were released.")
            if event_type == "message_stop":
                completed = True
                break

            if event_type == "content_block_start":
                # New content block starting
                index = chunk.get("index", 0)
                if type(index) is not int or not 0 <= index < 1024 or index in seen_blocks:
                    raise ProtocolError("Messages returned a duplicate or invalid content block index.")
                current_block_index = index
                seen_blocks.add(index)
                content_block = object_data(chunk.get("content_block", {}))
                block_type = string_data(content_block.get("type", ""))
                open_blocks[index] = block_type

                if block_type == "tool_use":
                    # Start of a tool call
                    tool_accumulator.start_tool_call(
                        index=current_block_index,
                        tool_id=string_data(content_block.get("id", "")),
                        name=string_data(content_block.get("name", "")),
                        initial_input=content_block.get("input"),
                    )
                    for preview in tool_accumulator.previews():
                        yield preview
                elif block_type == "text" and content_block.get("text"):
                    text = string_data(content_block["text"])
                    full_text += text
                    yield TextChunkEvent(chunk=text, usage=latest_usage)

            elif event_type == "content_block_stop":
                index = chunk.get("index")
                if type(index) is not int or index not in open_blocks:
                    raise ProtocolError("Messages stopped an unknown content block.")
                del open_blocks[index]

            elif event_type == "content_block_delta":
                index = chunk.get("index", 0)
                if type(index) is not int or index not in open_blocks:
                    raise ProtocolError("Messages returned a delta outside an open content block.")
                current_block_index = index
                delta = object_data(chunk.get("delta", {}))
                delta_type = string_data(delta.get("type", ""))
                expected_block = {
                    "text_delta": "text",
                    "input_json_delta": "tool_use",
                    "thinking_delta": "thinking",
                    "signature_delta": "thinking",
                    "citations_delta": "text",
                }.get(delta_type)
                if (expected_block is not None and open_blocks[index] != expected_block) or (
                    open_blocks[index] == "tool_use" and delta_type != "input_json_delta"
                ):
                    raise ProtocolError(
                        "Messages returned a delta for the wrong content block; no tools were released."
                    )

                if delta_type == "text_delta":
                    # Text content
                    text = string_data(delta.get("text", ""))
                    if text:
                        full_text += text
                        yield TextChunkEvent(chunk=text, usage=latest_usage)

                elif delta_type == "input_json_delta":
                    # Tool call argument fragment
                    partial_json = string_data(delta.get("partial_json", ""))
                    tool_accumulator.add_input_delta(current_block_index, partial_json)
                    for preview in tool_accumulator.previews():
                        yield preview

            elif event_type == "message_delta":
                # Message-level update (contains usage info at end)
                usage = chunk.get("usage")
                if usage:
                    usage_data.update(object_data(usage))
                    latest_usage = token_usage(usage_data, "messages")
                stop_reason = object_data(chunk.get("delta") or {}).get("stop_reason")
                if stop_reason:
                    finish_reason = {
                        "end_turn": FinishReason.STOP,
                        "stop_sequence": FinishReason.STOP,
                        "tool_use": FinishReason.TOOL_CALLS,
                        "max_tokens": FinishReason.LENGTH,
                        "refusal": FinishReason.CONTENT_FILTER,
                    }.get(string_data(stop_reason), FinishReason.UNKNOWN)

            elif event_type == "message_start":
                # Initial message info (may contain usage)
                message = object_data(chunk.get("message", {}))
                usage = message.get("usage")
                if usage:
                    usage_data.update(object_data(usage))
                    latest_usage = token_usage(usage_data, "messages")

        if not completed or open_blocks or finish_reason == FinishReason.UNKNOWN:
            raise ProtocolError("Messages stream ended before completion; no tools were released.")

        # Emit accumulated tool calls
        tool_calls = tool_accumulator.get_complete_tool_calls()
        if tool_calls and finish_reason != FinishReason.TOOL_CALLS:
            raise ProtocolError("Messages returned incomplete tool calls; no tools were released.")
        if finish_reason == FinishReason.TOOL_CALLS and not tool_calls:
            raise ProtocolError("Messages finished with tools but returned no complete calls.")
        yield TextDoneEvent(text=full_text, usage=latest_usage, finish_reason=finish_reason)
        if tool_calls:
            for tc in tool_calls:
                call = ToolCallEvent(
                    id=tc.id,
                    name=tc.name,
                    arguments=tc.arguments,
                    usage=latest_usage,
                    finish_reason=FinishReason.TOOL_CALLS,
                    metadata=tc.metadata,
                )
                yield tool_accumulator.ready(call)
                yield call

    async def _non_stream_anthropic(
        self,
        url: str,
        body: dict[str, Any],
        headers: dict[str, str],
    ) -> AsyncIterator[Event]:
        """Handle non-streaming Anthropic response."""
        from ..adapters import anthropic as anthropic_adapter

        response = await self._http.post_json(url, body, headers)
        if response.get("error") or response.get("type") == "error":
            raise ProtocolError("Messages reported a generation error; no tools were released.")
        text, tool_calls, usage = anthropic_adapter.parse_response(response)

        # Parse finish_reason
        finish_reason = FinishReason.UNKNOWN
        stop_reason = response.get("stop_reason")
        if stop_reason in {"end_turn", "stop_sequence"}:
            finish_reason = FinishReason.STOP
        elif stop_reason == "tool_use":
            finish_reason = FinishReason.TOOL_CALLS
        elif stop_reason == "max_tokens":
            finish_reason = FinishReason.LENGTH
        elif stop_reason == "refusal":
            finish_reason = FinishReason.CONTENT_FILTER
        else:
            raise ProtocolError("Messages returned an unsuccessful stop reason; no tools were released.")
        if tool_calls and finish_reason != FinishReason.TOOL_CALLS:
            raise ProtocolError("Messages returned incomplete tool calls; no tools were released.")
        if finish_reason == FinishReason.TOOL_CALLS and not tool_calls:
            raise ProtocolError("Messages finished with tools but returned no complete calls.")

        # Parse extra metadata
        extra: dict[str, Any] = {}
        if "id" in response:
            extra["id"] = response["id"]
        if "model" in response:
            extra["model"] = response["model"]

        # Parse usage
        latest_usage = token_usage(usage, "messages")

        yield TextDoneEvent(text=text, usage=latest_usage, finish_reason=finish_reason, extra=extra)
        if tool_calls:
            for tc in tool_calls:
                yield ToolCallEvent(
                    id=tc.id,
                    name=tc.name,
                    arguments=tc.arguments,
                    usage=latest_usage,
                    finish_reason=FinishReason.TOOL_CALLS,
                    extra=extra,
                    metadata=tc.metadata,
                )

    async def _generate_responses(
        self,
        messages: list[Message],
        tools: list[ToolDefinition] | None,
        config: GenerationConfig | None,
        stream: bool,
    ) -> AsyncIterator[Event]:
        from ..adapters.responses import ResponseAccumulator
        from ..adapters.responses import format_request

        body = format_request(self.model, messages, tools, config, stream)
        url = f"{self._api_prefix()}/responses"
        headers = {"Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json"}
        if self.bearer_token_provider is not None:
            headers.update(await self.auth_headers(url))
        if self.provider_type == ProviderType.OPENROUTER:
            headers["HTTP-Referer"] = "https://github.com/nagents"
        accumulator = ResponseAccumulator(streaming=stream)
        if stream:
            async with aclosing(
                cast("AsyncGenerator[str, None]", self._http.post_stream(url, body, headers))
            ) as chunks:
                async for data in chunks:
                    if data == "[DONE]":
                        break
                    events = accumulator.add(load_object(data))
                    if accumulator.completed:
                        break
                    for event in events:
                        yield event
            if not accumulator.completed:
                raise ProtocolError("Responses stream ended before completion; no tools were released.")
        else:
            events = accumulator.finish(await self._http.post_json(url, body, headers))
        for event in events:
            yield event

    async def close(self) -> None:
        """Close the provider and release resources."""
        await self._http.close()

    async def __aenter__(self) -> "Provider":
        return self

    async def __aexit__(self, *args: object) -> None:
        await self.close()

__init__(provider_type, api_key='', model='', base_url=None, timeout=120.0, api_version=None, retry_config=None, realtime_config=None, api='auto', live_config=None, *, bearer_token_provider=None)

Initialize the provider.

Parameters:

Name	Type	Description	Default
provider_type	ProviderType	The type of provider to use	required
api_key	str	API key for authentication	''
model	str	Model identifier to use (for Azure, this is the deployment name)	''
base_url	str | None	Optional custom base URL (overrides defaults). For Azure OpenAI: https://{resource}.cognitiveservices.azure.com	None
timeout	float	Request timeout in seconds	120.0
api_version	str | None	API version (required for Azure OpenAI, e.g., "2024-05-01-preview")	None
retry_config	RetryConfig | None	Retry configuration for rate limit and server error handling. Defaults to RetryConfig() (3 retries, 5s base delay). Set RetryConfig(max_retries=0) to disable retry.	None
realtime_config	RealtimeConfig | None	Optional :class:~nagents.realtime.RealtimeConfig for Realtime speech-to-speech sessions. When set, voice sessions reuse the provider's model and this configuration.	None
api	str	HTTP contract: auto, chat_completions, responses, or messages. Auto preserves native providers and selects chat_completions for LiteLLM and OpenRouter. base_url is an API prefix, not a full generation endpoint.	'auto'
live_config	LiveConfig | None	GPT-Live voice/delegation settings for Agent.run() with audio.	None
bearer_token_provider	BearerTokenProvider | None	Optional async callable returning a bearer token. Mutually exclusive with api_key; supports OpenAI-compatible v1 chat/Responses and GPT-Live. Called per request attempt/connection. The caller owns the credential and its cache/refresh/close lifecycle.	None

Source code in src/nagents/provider/base.py

def __init__(
    self,
    provider_type: ProviderType,
    api_key: str = "",
    model: str = "",
    base_url: str | None = None,
    timeout: float = 120.0,
    api_version: str | None = None,
    retry_config: RetryConfig | None = None,
    realtime_config: "RealtimeConfig | None" = None,
    api: str = "auto",
    live_config: "LiveConfig | None" = None,
    *,
    bearer_token_provider: BearerTokenProvider | None = None,
):
    """
    Initialize the provider.

    Args:
        provider_type: The type of provider to use
        api_key: API key for authentication
        model: Model identifier to use (for Azure, this is the deployment name)
        base_url: Optional custom base URL (overrides defaults).
                  For Azure OpenAI: https://{resource}.cognitiveservices.azure.com
        timeout: Request timeout in seconds
        api_version: API version (required for Azure OpenAI, e.g., "2024-05-01-preview")
        retry_config: Retry configuration for rate limit and server error handling.
                      Defaults to RetryConfig() (3 retries, 5s base delay).
                      Set RetryConfig(max_retries=0) to disable retry.
        realtime_config: Optional :class:`~nagents.realtime.RealtimeConfig`
            for Realtime speech-to-speech sessions. When set, voice sessions
            reuse the provider's model and this configuration.
        api: HTTP contract: auto, chat_completions, responses, or messages.
            Auto preserves native providers and selects
            chat_completions for LiteLLM and OpenRouter. base_url is an API
             prefix, not a full generation endpoint.
        live_config: GPT-Live voice/delegation settings for Agent.run() with audio.
        bearer_token_provider: Optional async callable returning a bearer token.
            Mutually exclusive with api_key; supports OpenAI-compatible v1
            chat/Responses and GPT-Live. Called per request attempt/connection.
            The caller owns the credential and its cache/refresh/close lifecycle.
    """
    self.provider_type = provider_type
    self.api_key = api_key
    self.bearer_token_provider = bearer_token_provider
    if bearer_token_provider is not None:
        if api_key:
            raise ValueError("Choose api_key or bearer_token_provider, not both")
        if not callable(bearer_token_provider):
            raise ValueError("bearer_token_provider must be an async callable")
        if provider_type not in {ProviderType.OPENAI_COMPATIBLE, ProviderType.AZURE_OPENAI_COMPATIBLE_V1}:
            raise ValueError("Bearer token authentication requires an OpenAI-compatible v1 provider")
        if api not in {"auto", "chat_completions", "responses"} or realtime_config is not None:
            raise ValueError("Bearer token authentication supports text chat/responses and GPT-Live, not Realtime")
    self.model = model
    self.api_version = api_version
    self.retry_config = retry_config or RetryConfig()
    self.realtime_config = realtime_config
    self.live_config = live_config
    if live_config is not None and realtime_config is not None:
        raise ValueError("Choose live_config or realtime_config, not both")
    if live_config is not None and provider_type not in {
        ProviderType.OPENAI_COMPATIBLE,
        ProviderType.AZURE_OPENAI_COMPATIBLE_V1,
    }:
        raise ValueError("GPT-Live requires an OpenAI-compatible v1 provider")
    if api not in {"auto", "chat_completions", "responses", "messages"}:
        raise ValueError("api must be auto, chat_completions, responses, or messages")
    if provider_type == ProviderType.LITELLM and not base_url:
        raise ValueError("LiteLLM requires an explicit base_url pointing to your gateway")
    default_api = "messages" if provider_type == ProviderType.ANTHROPIC else "chat_completions"
    if provider_type == ProviderType.GEMINI_NATIVE:
        default_api = "auto"
    self.api = default_api if api == "auto" else api
    if provider_type == ProviderType.AZURE_OPENAI_COMPATIBLE_V1 and self.api not in {
        "chat_completions",
        "responses",
    }:
        raise ValueError("Azure v1 supports chat_completions or responses")
    if (
        provider_type
        not in {
            ProviderType.LITELLM,
            ProviderType.OPENAI_COMPATIBLE,
            ProviderType.OPENROUTER,
            ProviderType.AZURE_OPENAI_COMPATIBLE_V1,
        }
        and self.api != default_api
    ):
        raise ValueError("This provider does not support the selected HTTP API contract")
    if provider_type == ProviderType.OPENAI_COMPATIBLE and self.api == "messages" and not base_url:
        raise ValueError("The messages API requires an explicit compatible base_url or an Anthropic provider")
    if base_url:
        parsed = urlsplit(base_url)
        if (
            parsed.scheme not in {"http", "https"}
            or not parsed.hostname
            or any(char.isspace() for char in base_url)
        ):
            raise ValueError("base_url must be an HTTP(S) API prefix")
        if parsed.username or parsed.password or parsed.query or parsed.fragment:
            raise ValueError("base_url must not contain credentials, query parameters, or fragments")
        if parsed.path.rstrip("/").endswith(("/chat/completions", "/responses", "/messages", "/completions")):
            raise ValueError("base_url must be an API prefix, not a generation endpoint")
    self._http = (
        GatewayHTTPClient(timeout=timeout)
        if bearer_token_provider is not None
        or provider_type in {ProviderType.LITELLM, ProviderType.OPENROUTER, ProviderType.AZURE_OPENAI_COMPATIBLE_V1}
        or api != "auto"
        else HTTPClient(timeout=timeout)
    )
    self._model_verified: bool | None = None  # None = not checked, True/False = result
    self._model_list_timeout = min(timeout, 30.0) if math.isfinite(timeout) and timeout > 0 else 30.0

    # Validate Azure-specific requirements
    if provider_type == ProviderType.AZURE_OPENAI_COMPATIBLE_V1 and not base_url:
        raise ValueError(
            "base_url is required for Azure OpenAI. Format: https://{resource}.cognitiveservices.azure.com/openai"
        )
    if provider_type == ProviderType.AZURE_OPENAI_COMPATIBLE:
        if not api_version:
            raise ValueError("api_version is required for Azure OpenAI")
        if not base_url:
            raise ValueError("base_url is required for Azure OpenAI")
    if base_url:
        self.base_url = base_url.rstrip("/")
    elif provider_type == ProviderType.OPENAI_COMPATIBLE:
        self.base_url = "https://api.openai.com/v1"
    elif provider_type == ProviderType.OPENROUTER:
        self.base_url = "https://openrouter.ai/api/v1"
    elif provider_type == ProviderType.ANTHROPIC:
        self.base_url = "https://api.anthropic.com/v1"
    elif provider_type == ProviderType.AZURE_OPENAI_COMPATIBLE_V1:
        # Azure requires base_url, this should not be reached due to validation above
        raise ValueError("base_url is required for Azure OpenAI")
    else:  # GEMINI_NATIVE
        self.base_url = "https://generativelanguage.googleapis.com/v1beta"
    if bearer_token_provider is not None or live_config is not None:
        validate_prefix(self.base_url)
    if not self.api_key and bearer_token_provider is None:
        raise ValueError("API key is required for provider initialization")

generate(messages, tools=None, config=None, stream=True, verify_model=False) async

Generate a response from the model.

Parameters:

Name	Type	Description	Default
messages	list[Message]	Conversation history	required
tools	list[ToolDefinition] | None	Optional list of tools the model can call	None
config	GenerationConfig | None	Optional generation configuration	None
stream	bool	Whether to stream the response	True
verify_model	bool	If True, verify the model exists before generating	False

Yields:

Type	Description
AsyncIterator[Event]	Events as they occur (text chunks, tool calls, etc.)

Source code in src/nagents/provider/base.py

async def generate(
    self,
    messages: list[Message],
    tools: list[ToolDefinition] | None = None,
    config: GenerationConfig | None = None,
    stream: bool = True,
    verify_model: bool = False,
) -> AsyncIterator[Event]:
    """
    Generate a response from the model.

    Args:
        messages: Conversation history
        tools: Optional list of tools the model can call
        config: Optional generation configuration
        stream: Whether to stream the response
        verify_model: If True, verify the model exists before generating

    Yields:
        Events as they occur (text chunks, tool calls, etc.)
    """
    started = monotonic()
    # Verify model if requested
    if verify_model:
        is_valid = await self.verify_model()
        if not is_valid:
            yield ErrorEvent(
                message=f"Model '{self.model}' not found in provider",
                code="MODEL_NOT_FOUND",
                recoverable=False,
            )
            return

    try:
        async for event in self._with_retry(messages, tools, config, stream):
            yield event
    except HTTPError as e:
        logger.error("Generation failed (HTTP %s)", e.status)
        yield ErrorEvent(
            message=f"Provider request failed (HTTP {e.status}); check the API route, model access, credentials, or gateway availability.",
            code=str(e.status),
            recoverable=e.is_retryable(),
            extra=failure_extra(e, started),
        )
    except ProtocolError as e:
        yield ErrorEvent(
            message=str(e), code="PROVIDER_PROTOCOL_ERROR", recoverable=False, extra=failure_extra(e, started)
        )
    except Exception as e:
        logger.error("Provider request failed or returned invalid data")
        yield ErrorEvent(
            message="Provider request failed, timed out, or returned invalid data; check the API route and request settings.",
            code="PROVIDER_REQUEST_FAILED",
            recoverable=False,
            extra=failure_extra(e, started),
        )

get_model_list() async

Fetch model IDs explicitly, without changing the model or verification cache.

OpenAI-compatible, OpenRouter, and LiteLLM providers use GET on the configured API prefix plus /models. IDs retain upstream order and spelling; duplicates are removed. An empty catalog is valid, not a fallback on error.

Raises:

Type	Description
NotImplementedError	Discovery is not implemented for this provider.
ModelListError	Credentials, configuration, transport, or catalog data are invalid. Upstream bodies and credentials are never included.

Catalog membership does not guarantee generation access or capabilities. Each call makes a fresh bounded request, with no redirects or raw logging.

Source code in src/nagents/provider/base.py

async def get_model_list(self) -> list[str]:
    """Fetch model IDs explicitly, without changing the model or verification cache.

    OpenAI-compatible, OpenRouter, and LiteLLM providers use GET on the
    configured API prefix plus /models. IDs retain upstream order and spelling;
    duplicates are removed. An empty catalog is valid, not a fallback on error.

    Raises:
        NotImplementedError: Discovery is not implemented for this provider.
        ModelListError: Credentials, configuration, transport, or catalog data
            are invalid. Upstream bodies and credentials are never included.

    Catalog membership does not guarantee generation access or capabilities.
    Each call makes a fresh bounded request, with no redirects or raw logging.
    """
    if self.provider_type not in {
        ProviderType.OPENAI_COMPATIBLE,
        ProviderType.OPENROUTER,
        ProviderType.LITELLM,
        ProviderType.AZURE_OPENAI_COMPATIBLE_V1,
    }:
        raise NotImplementedError(
            "Model discovery is not implemented for this provider; enter a model ID manually."
        )
    try:
        # base_url is mutable: validate the current prefix before sending a key.
        prefix = self.base_url
        parsed = urlsplit(prefix)
        if (
            parsed.scheme not in {"http", "https"}
            or not parsed.hostname
            or parsed.username
            or parsed.password
            or parsed.query
            or parsed.fragment
            or any(char.isspace() or unicodedata.category(char).startswith("C") for char in prefix)
            or parsed.path.rstrip("/").endswith(("/chat/completions", "/responses", "/messages", "/completions"))
        ):
            raise ValueError
        if not self.api_key.strip() and self.bearer_token_provider is None:
            raise ValueError
        headers = (
            await self.auth_headers(prefix)
            if self.bearer_token_provider
            else {"Authorization": f"Bearer {self.api_key}"}
        )
        if self.provider_type == ProviderType.OPENROUTER:
            headers["HTTP-Referer"] = "https://github.com/nagents"
        # A short-lived secure transport also isolates catalogs from generation
        # cookies, loggers, and the ordinary HTTPClient's redirect behavior.
        client = GatewayHTTPClient(timeout=self._model_list_timeout)
        async with client:
            response = await client.get_json(f"{self._api_prefix()}/models", headers)
    except Exception:
        raise ModelListError(
            "Model discovery failed; check provider configuration, credentials, and availability."
        ) from None
    if "error" in response or not isinstance(response.get("data"), list):
        raise ModelListError("Provider returned an invalid model catalog.")
    models: list[str] = []
    seen: set[str] = set()
    for item in list_data(response["data"]):
        model_id = item.get("id") if isinstance(item, dict) else None
        if (
            not isinstance(model_id, str)
            or not model_id.strip()
            or any(unicodedata.category(char).startswith("C") for char in model_id)
        ):
            raise ModelListError("Provider returned an invalid model ID.")
        if model_id not in seen:
            models.append(model_id)
            seen.add(model_id)
    return models

verify_model(force=False) async

Verify that the specified model exists in model list endpoint.

Parameters:

Name	Type	Description	Default
force	bool	If True, re-verify even if already cached.	False

Returns:

Type	Description
bool	True if the model is found, False otherwise.

Note

Verification is not implemented here for Anthropic or Azure; those providers continue to return True without a catalog request.

Source code in src/nagents/provider/base.py

async def verify_model(self, force: bool = False) -> bool:
    """Verify that the specified model exists in model list endpoint.

    Args:
        force: If True, re-verify even if already cached.

    Returns:
        True if the model is found, False otherwise.

    Note:
        Verification is not implemented here for Anthropic or Azure;
        those providers continue to return True without a catalog request.
    """
    # Return cached result if available and not forcing re-verification
    if not force and self._model_verified is not None:
        return self._model_verified

    # Preserve local verification for Anthropic and Azure.
    if self.provider_type == ProviderType.ANTHROPIC:
        logger.info(f"Skipping model verification for Anthropic (model: {self.model})")
        self._model_verified = True
        return True

    if self.provider_type == ProviderType.AZURE_OPENAI_COMPATIBLE_V1:
        logger.info(f"Skipping model verification for Azure OpenAI (deployment: {self.model})")
        self._model_verified = True
        return True
    if self.provider_type == ProviderType.AZURE_OPENAI_COMPATIBLE:
        logger.info(f"Skipping model verification for Azure OpenAI (deployment: {self.model})")
        self._model_verified = True
        return True
    try:
        if self.provider_type in (ProviderType.OPENAI_COMPATIBLE, ProviderType.OPENROUTER, ProviderType.LITELLM):
            models = await self.get_model_list()

            # Check for exact match first
            if self.model in models:
                logger.info(f"Model '{self.model}' verified successfully")
                self._model_verified = True
                return True

            # Check for match with "models/" prefix (Gemini via OpenAI compat)
            prefixed_model = f"models/{self.model}"
            if prefixed_model in models:
                logger.info(f"Model '{self.model}' verified successfully (with models/ prefix)")
                self._model_verified = True
                return True

            logger.warning("Selected model not found in provider catalog")
            self._model_verified = False
            return False
        else:  # GEMINI_NATIVE
            url = f"{self.base_url}/models?key={self.api_key}"
            response = await self._http.get_json(url)
            models = [m["name"] for m in response.get("models", [])]

            # Check for exact match first
            if self.model in models:
                logger.info(f"Model '{self.model}' verified successfully")
                self._model_verified = True
                return True

            # Check for match with "models/" prefix
            prefixed_model = f"models/{self.model}"
            if prefixed_model in models:
                logger.info(f"Model '{self.model}' verified successfully (with models/ prefix)")
                self._model_verified = True
                return True

            # Check for match without "models/" prefix (user might provide full name)
            if self.model.startswith("models/"):
                stripped_model = self.model[7:]  # Remove "models/" prefix
                if f"models/{stripped_model}" in models:
                    logger.info(f"Model '{self.model}' verified successfully")
                    self._model_verified = True
                    return True

            logger.warning(f"Model '{self.model}' not found in provider. Available models: {models[:10]}...")
            self._model_verified = False
            return False
    except HTTPError as e:
        logger.error("Model verification failed (HTTP %s)", e.status)
        self._model_verified = False
        return False
    except Exception:
        logger.error("Model verification failed")
        self._model_verified = False
        return False

close() async

Close the provider and release resources.

Source code in src/nagents/provider/base.py

async def close(self) -> None:
    """Close the provider and release resources."""
    await self._http.close()

Foundry credentials

FoundryProvider reuses the Provider text/GPT-Live transports. credential and api_key are mutually exclusive; scope defaults to https://ai.azure.com/.default. It never creates, selects, caches, or closes an Azure credential. Applications choose a sync or async Azure Identity credential (or a structural equivalent) themselves. Async is preferred; synchronous token calls are offloaded to a worker thread to keep the voice event loop responsive. Azure Identity is a developer-testing dependency only, not a project runtime dependency or extra.

Foundry API-key GPT-Live WebSockets use the api-key header, including attach/fork. Entra WebSockets and REST requests retain Bearer authentication, as do REST API keys. OpenAI WebSocket authentication is unchanged. Redirects and query-string credentials are not allowed.

nagents.FoundryProvider

Bases: Provider

Text and GPT-Live over Foundry's OpenAI-compatible v1 API.

Pass a sync or async DefaultAzureCredential, ManagedIdentityCredential, WorkloadIdentityCredential, ClientSecretCredential, or a structural equivalent. Keep it alive until every sharing provider/agent closes, then close it yourself. API-key authentication is mutually exclusive with credential authentication.

Source code in src/nagents/provider/foundry.py

class FoundryProvider(Provider):
    """Text and GPT-Live over Foundry's OpenAI-compatible v1 API.

    Pass a sync or async DefaultAzureCredential, ManagedIdentityCredential,
    WorkloadIdentityCredential, ClientSecretCredential, or a structural equivalent.
    Keep it alive until every sharing provider/agent closes, then close it yourself.
    API-key authentication is mutually exclusive with credential authentication.
    """

    def __init__(
        self,
        *,
        base_url: str,
        model: str,
        credential: AsyncTokenCredential | TokenCredential | None = None,
        api_key: str = "",
        scope: str = "https://ai.azure.com/.default",
        api: str = "responses",
        timeout: float = 120.0,
        retry_config: RetryConfig | None = None,
        live_config: LiveConfig | None = None,
    ) -> None:
        if credential is not None and api_key:
            raise ValueError("Choose credential or api_key, not both")
        if credential is None and not api_key:
            raise ValueError("FoundryProvider requires credential or api_key")
        if not scope or any(char.isspace() for char in scope):
            raise ValueError("scope must be a non-empty token scope without whitespace")
        validate_prefix(base_url)

        async def token() -> str:
            assert credential is not None
            result: AccessToken | Awaitable[AccessToken]
            if inspect.iscoroutinefunction(credential.get_token):
                result = credential.get_token(scope)
            else:
                result = await asyncio.to_thread(credential.get_token, scope)
            # Also supports decorated async methods that expose a synchronous
            # wrapper returning an awaitable. Only invocation goes to the worker;
            # the returned coroutine always executes on the calling event loop.
            if isinstance(result, Awaitable):
                result = await result
            return result.token

        super().__init__(
            ProviderType.AZURE_OPENAI_COMPATIBLE_V1,
            api_key=api_key,
            model=model,
            base_url=base_url,
            api=api,
            timeout=timeout,
            retry_config=retry_config,
            live_config=live_config,
            bearer_token_provider=token if credential is not None else None,
        )

    async def auth_headers(self, url: str) -> dict[str, str]:
        """Foundry Live WS keys use api-key; REST and Entra retain Bearer auth."""
        headers = await super().auth_headers(url)
        if self.bearer_token_provider is None and url.startswith(("ws://", "wss://")):
            return {"api-key": self.api_key}
        return headers

__init__(*, base_url, model, credential=None, api_key='', scope='https://ai.azure.com/.default', api='responses', timeout=120.0, retry_config=None, live_config=None)

Source code in src/nagents/provider/foundry.py

def __init__(
    self,
    *,
    base_url: str,
    model: str,
    credential: AsyncTokenCredential | TokenCredential | None = None,
    api_key: str = "",
    scope: str = "https://ai.azure.com/.default",
    api: str = "responses",
    timeout: float = 120.0,
    retry_config: RetryConfig | None = None,
    live_config: LiveConfig | None = None,
) -> None:
    if credential is not None and api_key:
        raise ValueError("Choose credential or api_key, not both")
    if credential is None and not api_key:
        raise ValueError("FoundryProvider requires credential or api_key")
    if not scope or any(char.isspace() for char in scope):
        raise ValueError("scope must be a non-empty token scope without whitespace")
    validate_prefix(base_url)

    async def token() -> str:
        assert credential is not None
        result: AccessToken | Awaitable[AccessToken]
        if inspect.iscoroutinefunction(credential.get_token):
            result = credential.get_token(scope)
        else:
            result = await asyncio.to_thread(credential.get_token, scope)
        # Also supports decorated async methods that expose a synchronous
        # wrapper returning an awaitable. Only invocation goes to the worker;
        # the returned coroutine always executes on the calling event loop.
        if isinstance(result, Awaitable):
            result = await result
        return result.token

    super().__init__(
        ProviderType.AZURE_OPENAI_COMPATIBLE_V1,
        api_key=api_key,
        model=model,
        base_url=base_url,
        api=api,
        timeout=timeout,
        retry_config=retry_config,
        live_config=live_config,
        bearer_token_provider=token if credential is not None else None,
    )

nagents.provider.AsyncTokenCredential

Bases: Protocol

Structural subset of azure.core.credentials_async.AsyncTokenCredential.

No close method is required because FoundryProvider never takes ownership of the credential.

Source code in src/nagents/provider/foundry.py

class AsyncTokenCredential(Protocol):
    """Structural subset of azure.core.credentials_async.AsyncTokenCredential.

    No close method is required because
    FoundryProvider never takes ownership of the credential.
    """

    def get_token(self, *scopes: str) -> Awaitable[AccessToken]: ...

nagents.provider.TokenCredential

Bases: Protocol

Structural subset of azure.core.credentials.TokenCredential.

Synchronous get_token calls run in a worker thread, never on the voice loop. The caller owns the credential and its close lifecycle.

Source code in src/nagents/provider/foundry.py

class TokenCredential(Protocol):
    """Structural subset of azure.core.credentials.TokenCredential.

    Synchronous get_token calls run in a worker thread, never on the voice loop.
    The caller owns the credential and its close lifecycle.
    """

    def get_token(self, *scopes: str) -> AccessToken: ...

GPT-Live Voice

Set live_config=LiveConfig(...) on the voice provider, configure the existing AudioDuplex on the agent, and consume agent.run() as with Realtime. The library owns the connection, audio streaming and delegation protocol.

• LiveConfig(delegation="responses", backend_model=..., web_search=True) uses OpenAI-hosted backend inference.
• LiveConfig(delegation="client") uses the ordinary Nagents Agent passed as delegation_agent=... to the voice agent. That backend owns its model, tools, permissions and task history. Audio continues while it works.

The voice provider selects model="gpt-live-1"; backend model selection is independent. The runnable example is examples/live/, with separate hosted, client-agent, and optional commentary examples.

The client backend is an ordinary Agent. Its emitted events are forwarded immediately with their original types and extra["delegation_id"] plus extra["source"] == "live_backend". A tagged backend DoneEvent does not end the voice conversation.

await voice.add_comment(text), add_thinking(text), and add_instructions(text) send native Live updates after startup. They are callable from application code or other tools, and can optionally be registered as model-callable backend tools. Delegated calls automatically carry their original task ID; application calls outside delegated work use a null ID. The returned event ID correlates with the server acknowledgment, rather than certifying speech playback.

voice.live exposes concrete commands and retained session status:

• wait(event_id) correlates acknowledgments and command errors.
• mute_input() / unmute_input() affect model input, independently of playback.
• update_backend(...) sends a sparse supported Responses configuration update.
• submit_text(...) / submit_image(...) route application input to the backend.
• invalidate_tasks() marks an application intent change so obsolete results are not delivered; business tools must still enforce their own revisions.
• close() requests graceful finalization; status retains duration, terminal reason and whether final usage was confirmed.

In hosted mode, register functions on the voice agent. The native runtime collects complete hosted function batches and uses its tool executor, submitting every result before continuing. A plain async service can instead be configured with LiveConfig(delegation="client", client_request_handler=...).

For correlated application work, client_request_handler receives a ClientDelegationRequest (from nagents.live) contains the original identifier, notice-time transcript, optional native text, and offset_ms. The immutable snapshot never absorbs later speech while waiting for a backend. A synchronous client_request_observer can report receipt immediately, including when all workers are busy; keep this callback short and nonblocking.

Independent callbacks run with client_concurrency=4 (configurable from 1 to 32), with at most 32 queued requests. A supplied Agent.delegation_agent keeps one lane because it owns one conversation. Cancellation joins all workers before disconnecting.

The typed handler accepts voice notices only. Route application text or images directly to your backend; submit_text and submit_image reject this handler configuration instead of discarding the supplied content. A delegation agent continues to accept these multimodal input commands.

A handler may return "" when the application owns ongoing results. The ID remains valid until the connection closes: send quiet progress with voice.live.append("thinking", text, identifier) and speakable results with voice.live.append("commentary", text, identifier). Use an empty identifier for session-wide work. Keep each public append below 500 tokens; the transport's automatic results use conservative 480-byte UTF-8 chunks. An append receipt does not confirm audio playback. See the OpenAI delegation guide.

The web ChatGPT login adapter uses its distinct native protocol: quiet progress maps to channel: commentary, spoken results to channel: speakable, and an empty ID uses session.context.append. Native instruction appends are explicitly unsupported; they are not converted into spoken instructions. Native content is also split below its 500-byte limit.

LiveAPI implements WebRTC setup, SIP controls, recording download and stored forks. agent.live_configuration(media=True) supplies the session configuration and registered hosted tool schemas without a WebSocket audio format. Attach an Agent to the returned ID with LiveConfig(attach_to=...); it does not start the session twice. Observers use handle_delegations=False to avoid duplicate tool execution. Live lifecycle/control traffic is LiveEvent, separate from Realtime.

nagents.live.audio provides paced inputs and application-owned playback gates. PacedAudioInput preserves source chunks and schedules their duration using the source's audio format. Its clock starts when the first nonempty chunk is available; waiting for source setup does not give buffered audio a head start. The default silence_tail=True continues with silence after the source ends. Verified-clip completion requires an output adapter with an actual drain() implementation. Muting a browser or SIP player's output remains the responsibility of the media path that owns playback.

nagents.LiveConfig dataclass

Provider voice settings. Client mode uses Agent.delegation_agent.

Hosted mode runs the chosen Responses backend at OpenAI. Client mode runs the supplied Nagents agent, retaining its provider, tools and permissions.

Source code in src/nagents/live/runtime.py

@dataclass(frozen=True)
class LiveConfig:
    """Provider voice settings. Client mode uses Agent.delegation_agent.

    Hosted mode runs the chosen Responses backend at OpenAI. Client mode runs
    the supplied Nagents agent, retaining its provider, tools and permissions.
    """

    delegation: Literal["responses", "client"] = "responses"
    backend_model: str = "gpt-6-astra"
    backend_instructions: str = "Return concise verified facts and status."
    web_search: bool = False
    voice: str | dict[str, str] = "marin"
    history: tuple[dict[str, object], ...] = ()
    # Standalone callers may delegate startup history. Web voice seeds are
    # background only: its main-assistant handler must wait for fresh speech.
    history_in_client_context: bool = True
    store: bool | None = None
    attach_to: str = ""
    fork_from: str = ""
    backend_timeout: float = 120
    close_timeout: float = 15
    close_session_on_exit: bool = True
    backend_options: dict[str, object] = field(default_factory=dict)
    client_request_handler: ClientDelegationHandler | None = None
    client_request_observer: ClientDelegationObserver | None = None
    client_concurrency: int = 4
    event_queue_size: int = 1024
    handle_delegations: bool = True

    def __post_init__(self) -> None:
        if self.delegation not in {"responses", "client"}:
            raise ValueError("Live delegation must be responses or client")
        if self.attach_to and self.fork_from:
            raise ValueError("Choose attachment or a stored fork, not both")
        if type(self.client_concurrency) is not int or not 1 <= self.client_concurrency <= 32:
            raise ValueError("Live client concurrency must be between 1 and 32")
        if len(self.history) > 128 or self.backend_timeout <= 0 or self.close_timeout <= 0 or self.event_queue_size < 1:
            raise ValueError("Invalid Live history length or timeout")
        for message in self.history:
            parts = message.get("content")
            if (
                message.get("role") not in {"developer", "user", "assistant"}
                or not isinstance(parts, list)
                or len(parts) != 1
            ):
                raise ValueError("Live startup history requires one text part per developer/user/assistant message")
            part = parts[0]
            if (
                not isinstance(part, dict)
                or not isinstance(part.get("text"), str)
                or part.get("type") not in {"input_text", "output_text", "text"}
            ):
                raise ValueError("Live startup history is text-only")

    def session(self, model: str = "gpt-live-1", instructions: str = "", *, media: bool = False) -> Event:
        """Create the supported session body for WebSocket, WebRTC or SIP."""
        if self.backend_options.keys() - {
            "model",
            "instructions",
            "tool_choice",
            "parallel_tool_calls",
            "max_output_tokens",
            "service_tier",
            "reasoning",
            "text",
        }:
            raise ValueError("Unsupported Live Responses backend option")
        delegation = ResponsesDelegation(self.backend_model, self.backend_instructions, self.web_search).to_dict()
        object_value(delegation["responses"]).update(self.backend_options)
        result: Event = {
            "model": model,
            "instructions": instructions,
            "input": list(self.history),
            "audio": {"output": {"voice": self.voice}},
            "delegation": delegation if self.delegation == "responses" else {"type": "client"},
        }
        if not media:
            object_value(result["audio"])["format"] = {"type": "audio/pcm", "rate": 24000}
        if self.store is not None:
            result["store"] = self.store
        return result

session(model='gpt-live-1', instructions='', *, media=False)

Create the supported session body for WebSocket, WebRTC or SIP.

Source code in src/nagents/live/runtime.py

def session(self, model: str = "gpt-live-1", instructions: str = "", *, media: bool = False) -> Event:
    """Create the supported session body for WebSocket, WebRTC or SIP."""
    if self.backend_options.keys() - {
        "model",
        "instructions",
        "tool_choice",
        "parallel_tool_calls",
        "max_output_tokens",
        "service_tier",
        "reasoning",
        "text",
    }:
        raise ValueError("Unsupported Live Responses backend option")
    delegation = ResponsesDelegation(self.backend_model, self.backend_instructions, self.web_search).to_dict()
    object_value(delegation["responses"]).update(self.backend_options)
    result: Event = {
        "model": model,
        "instructions": instructions,
        "input": list(self.history),
        "audio": {"output": {"voice": self.voice}},
        "delegation": delegation if self.delegation == "responses" else {"type": "client"},
    }
    if not media:
        object_value(result["audio"])["format"] = {"type": "audio/pcm", "rate": 24000}
    if self.store is not None:
        result["store"] = self.store
    return result

nagents.live.LiveAPI

Source code in src/nagents/live/api.py

class LiveAPI:
    def __init__(self, provider: Provider) -> None:
        if provider.live_config is None:
            raise ValueError("Use a voice provider with live_config for Live API authentication")
        self.provider = provider
        self.base_url = provider.live_endpoint()

    def url(self, session_id: str = "", operation: str = "") -> str:
        if operation and not session_id:
            raise ValueError("A session ID is required")
        return self.base_url + (f"/{quote(session_id, safe='')}/{operation}" if session_id else "")

    async def _post(self, url: str, payload: dict[str, object] | None) -> dict[str, object]:
        validate_endpoint(url)
        headers = await self.provider.auth_headers(url)
        async with (
            aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=60)) as http,
            http.post(url, json=payload, headers=headers, allow_redirects=False) as response,
        ):
            if not 200 <= response.status < 300:
                raise RuntimeError(f"Live HTTP operation failed ({response.status})")
            raw = bytearray()
            async for chunk in response.content.iter_chunked(65536):
                raw.extend(chunk)
                if len(raw) > 1024 * 1024:
                    raise ValueError("Live HTTP response is too large")
            value: object = json.loads(raw) if raw.strip() else {}
            if not isinstance(value, dict):
                raise ValueError("Invalid Live HTTP response")
            return cast("dict[str, object]", value)

    async def create_webrtc(self, sdp: str, session: dict[str, object]) -> dict[str, object]:
        if not sdp.strip():
            raise ValueError("An SDP offer is required")
        return await self._post(self.url(), {"session": session, "transport": {"type": "webrtc", "sdp": sdp}})

    async def fork_webrtc(self, source_id: str, sdp: str, *, store: bool = False) -> dict[str, object]:
        return await self._post(
            self.url(source_id, "fork"), {"session": {"store": store}, "transport": {"type": "webrtc", "sdp": sdp}}
        )

    async def accept(self, session_id: str, session: dict[str, object]) -> None:
        await self._post(self.url(session_id, "accept"), {"session": {**session, "type": "live"}})

    async def reject(self, session_id: str, status_code: int = 486) -> None:
        if type(status_code) is not int or not 300 <= status_code <= 699:
            raise ValueError("SIP rejection status must be 300-699")
        await self._post(self.url(session_id, "reject"), {"status_code": status_code})

    async def refer(self, session_id: str, target_uri: str) -> None:
        if not target_uri.startswith(("sip:", "tel:")):
            raise ValueError("A SIP or telephone target URI is required")
        await self._post(self.url(session_id, "refer"), {"target_uri": target_uri})

    async def hangup(self, session_id: str) -> None:
        await self._post(self.url(session_id, "hangup"), None)

    async def recording(self, session_id: str) -> AsyncGenerator[bytes, None]:
        url = self.url(session_id, "content")
        headers = await self.provider.auth_headers(url)
        async with (
            aiohttp.ClientSession() as http,
            http.get(
                url,
                headers=headers,
                allow_redirects=False,
            ) as response,
        ):
            if response.status != 200:
                raise RuntimeError(f"Recording unavailable ({response.status})")
            async for chunk in response.content.iter_chunked(65536):
                yield chunk

    async def download_recording(self, session_id: str, path: Path) -> None:
        with path.open("wb") as stream:
            async for chunk in self.recording(session_id):
                stream.write(chunk)

Explicit Model Discovery

async def Provider.get_model_list(self) -> list[str] provides explicit model discovery and is included in v0.15.0. It fetches IDs on each call; there is no catalog cache, implicit generation, or change to the selected model.

import asyncio
import os

from nagents import ModelListError, Provider, ProviderType


async def main() -> None:
    async with Provider(
        provider_type=ProviderType.OPENAI_COMPATIBLE,
        api_key=os.environ["OPENAI_API_KEY"],  # Your own OpenAI Platform API key.
        model="gpt-6-luna",
    ) as provider:
        try:
            model_ids = await provider.get_model_list()
        except (ModelListError, NotImplementedError):
            print("Catalog unavailable; keep or enter your model ID manually.")
        else:
            print(model_ids)
        # provider.model remains "gpt-6-luna" in either case.


asyncio.run(main())

For OPENAI_COMPATIBLE, OPENROUTER, and LITELLM, discovery uses GET {base_url}/models with the provider's API key as Bearer authentication. OpenRouter retains its existing identification header. Custom API prefixes are preserved; no /v1 is inserted and no pagination query is invented. LiteLLM requires an explicit gateway prefix. Only the conventional {"data": [{"id": "model-id"}]} envelope is accepted. A valid data: [] returns []; malformed envelopes, error envelopes, and invalid IDs raise ModelListError instead. IDs preserve spelling and response order, with duplicates removed. Blank IDs and control characters are rejected; provider metadata is not returned.

Requests use a bounded response body (32 MiB), a timeout of at most 30 seconds (or a shorter positive provider timeout), no redirects, no environment proxies, no cookies, and no raw HTTP logging. ModelListError messages do not include upstream response bodies, headers, URLs, or credentials. Unsupported discovery raises NotImplementedError; this currently includes native Gemini, Anthropic, and Azure provider types. This is a library implementation limit, not a claim that those services lack catalog APIs. Offline Harness demo mode rejects discovery without network requests.

get_model_list() is independent of is_model_verified. Existing verify_model(force=False) retains its cached boolean behavior and models/ prefix matching; force=True refreshes verification. Native Gemini verification and the local Anthropic/Azure verification behavior are unchanged. A catalog ID is not a guarantee of tool/media support, capabilities, or account entitlement.

OpenAI Provider

OpenAIProvider supports both API-key requests and ChatGPT subscription authentication. An explicit api_key bypasses local discovery:

from nagents import OpenAIProvider

provider = OpenAIProvider(api_key="your-api-key", model="gpt-5.6-terra", api="responses")

API-key requests default to Responses; select api="chat_completions" for Chat Completions. A custom base_url requires an explicit key. uses_chatgpt_auth reports the selected authentication mode, independently of the provider class. retry_config applies to API-key requests; the subscription transport does not automatically retry or replay requests.

Local configuration and ChatGPT authentication

OpenAIProvider() discovers local Codex configuration and file credentials from CODEX_HOME or ~/.codex, including the selected model, profile and wire API. Pass home=, profile=, or model= only to override discovery. API-key settings use the configured compatible API; a saved ChatGPT login uses the dedicated Codex OAuth transport. See local discovery.

OpenAIProvider(model="gpt-live-1", live_config=LiveConfig(...)) uses the same discovery for the public Live WebSocket API. This path requires an API key and may use its OPENAI_API_KEY environment fallback even when normal text inference uses saved ChatGPT authentication. OAuth tokens are never sent as Live API keys. The web voice interface uses the normal provider registry and defaults, with an optional explicit voice connection. Its browser exchanges PCM audio only with the ngn server over a same-origin WebSocket. The server owns the native Codex media/control path for ChatGPT login, or the provider's Live WebSocket for supported API-key/Entra connections. The selected provider must support Live.

For the CLI's own login workflow and limitations, see ngn and release availability.

An explicit credential callback can instead use your saved ngn login:

ngn login --device-auth
ngn login --status

import asyncio

from nagents import OpenAIProvider, ModelListError
from nagents.harness.auth import OpenAIAuth


async def codex_models() -> None:
    auth = OpenAIAuth()
    try:
        async with OpenAIProvider(credentials=auth.credentials) as provider:
            try:
                print(await provider.get_model_list())
            except (NotImplementedError, ModelListError):
                print("Keep or enter your Codex model ID manually.")
    finally:
        await auth.close()


asyncio.run(codex_models())

For ChatGPT/OAuth authentication, Codex discovery uses the fixed https://chatgpt.com/backend-api/codex/models route. Its client_version query and version header use the latest stable Codex release version discovered from OpenAI's public @openai/codex package metadata, not the ngn package version. This matters because upstream catalog visibility can depend on client version. Requests still identify the application honestly as originator: ngn and User-Agent: ngn/<package-version>.

Every explicit catalog lookup revalidates that public release metadata, using If-None-Match when the registry supplies an ETag; overlapping lookups share one check. There is no positive freshness delay that hides a newly published stable version. Metadata requests have a separate maximum two-second deadline (or the connection's shorter catalog deadline), a 64 KiB response limit, no redirects, cookies, proxy/netrc settings, provider credentials, account headers, or request tracing. Only the exact package name and a plain stable major.minor.patch version are accepted. ngn does not install or execute Codex to discover models.

If metadata is unavailable, invalid, or older than the known version, ngn uses the last valid version in this process and waits five minutes before checking metadata again. A process with no successful metadata lookup uses the verified 0.162.0 fallback. This is an offline compatibility fallback, not a cached model list: the authenticated catalog itself is fetched on every request. The cache contains only public version/ETag data, stays in memory, and writes no settings or credential files. Logs report the selected version, metadata source, and fallback status without upstream response data. ngn serve routes these bounded diagnostics through its application logger, so they remain visible in normal server/container logs without enabling verbose logging for other library or MCP operations.

Each fetch obtains one current OpenAIAuth.credentials snapshot and uses its access token, optional account ID, and optional residency together. It never copies the OAuth token into Provider.api_key, derives a catalog route from a custom URL, or falls back to the OpenAI API-key service. Changing an OAuth provider's base_url rejects discovery before requesting credentials. Login, refresh, and generation behavior are unchanged; no additional account-routing features are inferred from geolocation or other metadata.

The response must contain a models array of objects with valid slug and visibility fields. Only visibility: "list" contributes an ID; "hide" and "none" are excluded. Missing, null, or unknown visibility is an error, matching Codex's required enum rather than guessing a default. Models with supported_in_api: false remain eligible for OAuth discovery. IDs retain upstream order (no priority sorting), with duplicates removed; descriptions, instructions, account metadata, and other extra fields are not returned.

The authenticated catalog uses the same bounded, redirect-free, cookie-free, non-logging transport limits described above. There is no model-list cache or pagination. For ChatGPT authentication, OpenAIProvider.verify_model() remains local and does not prove account entitlement or request a catalog. This is an official Codex client contract, not a stable public OpenAI REST API guarantee. ModelListError and manual model entry remain important if that contract or account access changes.

Verified upstream sources: catalog route and client-version query, model fields and picker visibility, and Codex's public release metadata checks.

nagents.CodexCredentials dataclass

Source code in src/nagents/provider/openai.py

@dataclass
class CodexCredentials:
    access_token: str = field(repr=False)
    account_id: str = field(default="", repr=False)
    residency: str = field(default="", repr=False)

nagents.OpenAIProvider

Bases: Provider

OpenAI inference with explicit credentials or automatic local discovery.

An explicit credential callback selects the existing OAuth transport. On that route, verify_model is local and no automatic retry/replay follows failures. API-key configurations use the selected standard Provider API contract.

Source code in src/nagents/provider/openai.py

class OpenAIProvider(Provider):
    """OpenAI inference with explicit credentials or automatic local discovery.

    An explicit credential callback selects the existing OAuth transport. On that
    route, verify_model is local and no automatic retry/replay follows failures.
    API-key configurations use the selected standard Provider API contract.
    """

    def __init__(
        self,
        credentials: Callable[[], Awaitable[CodexCredentials]] | None = None,
        model: str = "",
        timeout: float = 120.0,
        *,
        home: str | Path = "",
        profile: str = "",
        live_config: LiveConfig | None = None,
        api_key: str = "",
        base_url: str = "",
        api: str = "auto",
        realtime_config: RealtimeConfig | None = None,
        retry_config: RetryConfig | None = None,
        bearer_token_provider: BearerTokenProvider | None = None,
    ) -> None:
        """Explicit API keys/token providers bypass discovery; otherwise use CODEX_HOME/~/.codex.

        Local model/profile/wire_api and auth.json select API-key or ChatGPT
        authentication. Explicit credential callbacks select ChatGPT. A custom
        base_url requires an explicit key or bearer_token_provider; saved OAuth never goes to that URL.
        API-key requests default to Responses unless another API is selected.
        """
        self._local_api = False
        self._credentials: Callable[[], Awaitable[CodexCredentials]]
        if sum((credentials is not None, bool(api_key), bearer_token_provider is not None)) > 1:
            raise ValueError("Choose api_key, bearer_token_provider, or ChatGPT credentials, not multiple auth sources")
        if base_url and not api_key and bearer_token_provider is None:
            raise ValueError("An explicit base_url requires an explicit api_key or bearer_token_provider")
        if not api_key and credentials is None and bearer_token_provider is None:
            local = _load_config(
                home, profile=profile, model=model, for_live=live_config is not None or realtime_config is not None
            )
            model = local.model
            credentials = local.credentials
            if not local.oauth:
                api_key = local.api_key
                base_url = local.base_url
                api = local.api if api == "auto" else api
        if api_key or bearer_token_provider is not None:
            super().__init__(
                ProviderType.OPENAI_COMPATIBLE,
                api_key,
                model or DEFAULT_CODEX_MODEL,
                base_url=base_url or None,
                api="responses" if api == "auto" else api,
                timeout=timeout,
                live_config=live_config,
                realtime_config=realtime_config,
                retry_config=retry_config,
                bearer_token_provider=bearer_token_provider,
            )
            self._local_api = True
            return
        assert credentials is not None
        if api not in {"auto", "responses"}:
            raise ValueError("ChatGPT authentication uses the Responses protocol")
        if live_config is not None or realtime_config is not None:
            raise ValueError(
                "Selected ChatGPT subscription authentication. The public GPT-Live WebSocket API and Realtime "
                "require OpenAI API-key authentication; this saved login can still run the delegated backend. "
                "Use ngn serve browser voice for GPT-Live with ChatGPT login."
            )
        super().__init__(
            ProviderType.OPENAI_COMPATIBLE,
            "oauth-not-an-api-key",
            model or DEFAULT_CODEX_MODEL,
            timeout=timeout,
            api="responses",
        )
        self.base_url = CODEX_ENDPOINT
        self._credentials = credentials
        self._timeout = timeout

    @property
    def uses_chatgpt_auth(self) -> bool:
        """Whether this instance uses the subscription transport, not an API key."""
        return not self._local_api

    async def verify_model(self, force: bool = False) -> bool:
        if self._local_api:
            return await super().verify_model(force)
        self._model_verified = True
        return True

    async def get_model_list(self) -> list[str]:
        """Fetch model IDs from the API-key catalog or the subscription catalog.

        With ChatGPT authentication this follows the Codex client's catalog contract, not the public
        OpenAI API-key catalog. It is fresh, read-only, and independent of model
        verification. Failures raise ModelListError without upstream data; there
        is no API-key fallback or entitlement guarantee.
        """
        if self._local_api:
            return await super().get_model_list()
        if self.base_url != CODEX_ENDPOINT:
            raise ModelListError("Codex model discovery requires its fixed endpoint; custom URLs are unsupported.")
        try:
            credentials = await self._credentials()
            # Capture all routing fields before any further await, even when the
            # callback returns a mutable credentials object shared with its caller.
            access_token, account_id, residency = (
                credentials.access_token,
                credentials.account_id,
                credentials.residency,
            )
            if not access_token or not re.fullmatch(r"[\x21-\x7e]{1,65536}", access_token):
                raise ValueError
            if account_id and not re.fullmatch(r"[A-Za-z0-9_.:-]{1,256}", account_id):
                raise ValueError
            if residency and not re.fullmatch(r"[A-Za-z0-9_-]{1,64}", residency):
                raise ValueError
        except CodexConfigError as error:
            raise ModelListError(str(error)) from None
        except Exception:
            raise ModelListError("ChatGPT credentials are unavailable; sign in again with /login.") from None
        client_version = await catalog_version(self._model_list_timeout)
        headers = {
            "Authorization": f"Bearer {access_token}",
            "Accept": "application/json",
            "version": client_version,
            "originator": "ngn",
            "User-Agent": USER_AGENT,
        }
        if account_id:
            headers["ChatGPT-Account-Id"] = account_id
        if residency and residency != "no_constraint":
            headers["x-openai-internal-codex-residency"] = residency
        try:
            client = GatewayHTTPClient(timeout=self._model_list_timeout)
            async with client:
                response = await client.get_json(f"{CODEX_MODELS_ENDPOINT}?client_version={client_version}", headers)
        except Exception:
            raise ModelListError(
                "Codex model discovery failed; check service availability and your ChatGPT login."
            ) from None
        if "error" in response or not isinstance(response.get("models"), list):
            raise ModelListError("Codex returned an invalid model catalog.")
        models: list[str] = []
        seen: set[str] = set()
        for item in list_data(response["models"]):
            if not isinstance(item, dict):
                raise ModelListError("Codex returned an invalid model entry.")
            model_id, visibility = item.get("slug"), item.get("visibility")
            if (
                not isinstance(model_id, str)
                or not model_id.strip()
                or any(unicodedata.category(char).startswith("C") for char in model_id)
            ):
                raise ModelListError("Codex returned an invalid model ID.")
            # Visibility is required by the pinned upstream ModelInfo struct.
            if not isinstance(visibility, str) or visibility not in {"list", "hide", "none"}:
                raise ModelListError("Codex returned invalid model visibility.")
            # supported_in_api is not an OAuth availability filter.
            if visibility == "list" and model_id not in seen:
                models.append(model_id)
                seen.add(model_id)
        return models

    async def generate(
        self,
        messages: list[Message],
        tools: list[ToolDefinition] | None = None,
        config: GenerationConfig | None = None,
        stream: bool = True,
        verify_model: bool = False,
    ) -> AsyncGenerator[Event, None]:
        if self._local_api:
            async for generated in super().generate(messages, tools, config, stream, verify_model):
                yield generated
            return
        started = monotonic()
        phase: FailurePhase = "request"
        try:
            body = _request_body(self.model, messages, tools, config)
        except (TypeError, ValueError) as error:
            message = str(error) if isinstance(error, _ProtocolError) else "Codex request data is invalid."
            yield ErrorEvent(message=message, code="CODEX_REQUEST_INVALID")
            return
        try:
            credentials = await self._credentials()
            if not credentials.access_token or not re.fullmatch(r"[\x21-\x7e]{1,65536}", credentials.access_token):
                raise ValueError
            if credentials.account_id and not re.fullmatch(r"[A-Za-z0-9_.:-]{1,256}", credentials.account_id):
                raise ValueError
            if credentials.residency and not re.fullmatch(r"[A-Za-z0-9_-]{1,64}", credentials.residency):
                raise ValueError
        except CodexConfigError as error:
            yield ErrorEvent(message=str(error), code="CODEX_AUTH")
            return
        except Exception:
            yield ErrorEvent(
                message="ChatGPT credentials are unavailable; sign in again with /login.", code="CODEX_AUTH"
            )
            return
        headers = {
            "Authorization": f"Bearer {credentials.access_token}",
            "Accept": "text/event-stream",
            "originator": "ngn",
            "User-Agent": USER_AGENT,
        }
        if credentials.account_id:
            headers["ChatGPT-Account-Id"] = credentials.account_id
        if credentials.residency and credentials.residency != "no_constraint":
            headers["x-openai-internal-codex-residency"] = credentials.residency
        progress = ToolCallProgressTracker()
        try:
            async with (
                aiohttp.ClientSession(
                    timeout=aiohttp.ClientTimeout(total=self._timeout),
                    trust_env=False,
                    cookie_jar=aiohttp.DummyCookieJar(),
                    trace_configs=[trace_config()],
                ) as client,
                client.post(CODEX_ENDPOINT, json=body, headers=headers, allow_redirects=False) as response,
            ):
                phase = "response"
                if response.status != 200:
                    messages_by_status = {
                        401: "ChatGPT login expired or was rejected; sign in again with /login.",
                        403: "Codex access denied; check your ChatGPT plan, workspace permissions, and model access.",
                        429: "ChatGPT Codex usage limit reached; wait before retrying or check your plan limits.",
                    }
                    yield ErrorEvent(
                        message=messages_by_status.get(
                            response.status,
                            f"Codex request failed (HTTP {response.status}); check service availability and model access.",
                        ),
                        code=f"CODEX_HTTP_{response.status}",
                    )
                    return
                # Some Codex responses omit Content-Type; _sse still validates framing
                # and no calls are released without a completed, validated response.
                if response.headers.get("Content-Type") and response.content_type != "text/event-stream":
                    raise _ProtocolError("Codex did not return an event stream; retry the request.")
                items: dict[int, dict[str, object]] = {}
                finished: set[int] = set()
                arguments: dict[int, str] = {}
                completed_arguments: dict[int, ToolArguments] = {}
                texts: dict[tuple[int, int], str] = {}
                texts_done: set[tuple[int, int]] = set()
                completed = False
                usage = Usage()
                async for event in _sse(response):
                    kind = _string(event.get("type"))
                    if kind in {"error", "response.failed", "response.incomplete", "response.cancelled"}:
                        raise _ProtocolError(
                            "Codex response failed, was incomplete, or was cancelled; no tool calls were released. Retry the request."
                        )
                    if kind in {"response.output_item.added", "response.output_item.done"}:
                        index = _index(event.get("output_index"))
                        item = _object(event.get("item"))
                        previous = items.get(index, {})
                        for key in ("id", "call_id", "name", "type"):
                            if key in previous and key in item and previous[key] != item[key]:
                                raise _ProtocolError("Codex returned inconsistent output item identities.")
                        items[index] = {**previous, **item}
                        merged_item = items[index]
                        if (
                            merged_item.get("type") == "function_call"
                            and "arguments" in merged_item
                            and (kind == "response.output_item.done" or merged_item["arguments"] != "")
                        ):
                            _remember_completed_arguments(completed_arguments, index, merged_item["arguments"])
                        if kind == "response.output_item.done":
                            finished.add(index)
                        if stream and merged_item.get("type") == "function_call":
                            for preview in progress.preview(
                                index,
                                _string(merged_item.get("call_id", "")),
                                _string(merged_item.get("name", "")),
                                _string(arguments.get(index, merged_item.get("arguments", ""))),
                            ):
                                yield preview
                    elif kind in {"response.function_call_arguments.delta", "response.function_call_arguments.done"}:
                        index = _index(event.get("output_index"))
                        item = items.setdefault(index, {})
                        if "item_id" in event:
                            item_id = _string(event["item_id"])
                            if not item_id or ("id" in item and item_id != item["id"]):
                                raise _ProtocolError("Codex returned inconsistent argument item identities.")
                            item["id"] = item_id
                        if kind.endswith(".delta"):
                            arguments[index] = arguments.get(index, "") + _string(event.get("delta"))
                        else:
                            item["arguments"] = _string(event.get("arguments"))
                            _remember_completed_arguments(completed_arguments, index, item["arguments"])
                        if stream:
                            for preview in progress.preview(
                                index,
                                _string(item.get("call_id", "")),
                                _string(item.get("name", "")),
                                _string(arguments.get(index, item.get("arguments", ""))),
                            ):
                                yield preview
                    elif kind in {"response.output_text.delta", "response.output_text.done"}:
                        text_key = (_index(event.get("output_index", 0)), _index(event.get("content_index", 0)))
                        index = text_key[0]
                        if "item_id" in event:
                            item = items.setdefault(index, {})
                            item_id = _string(event["item_id"])
                            if not item_id or ("id" in item and item_id != item["id"]):
                                raise _ProtocolError("Codex returned inconsistent text item identities.")
                            item["id"] = item_id
                        if text_key in texts_done or index in finished:
                            raise _ProtocolError("Codex sent text after a content part finished.")
                        if kind.endswith(".delta"):
                            delta = _string(event.get("delta"))
                            texts[text_key] = texts.get(text_key, "") + delta
                            if stream:
                                yield TextChunkEvent(chunk=delta)
                        else:
                            text = _string(event.get("text"))
                            if text_key in texts and texts[text_key] != text:
                                raise _ProtocolError("Codex returned inconsistent completed text.")
                            texts[text_key] = text
                            texts_done.add(text_key)
                    elif kind == "response.reasoning_summary_text.delta":
                        delta = _string(event.get("delta"))
                        if stream:
                            yield ReasoningChunkEvent(chunk=delta)
                    elif kind == "response.completed":
                        result = _object(event.get("response"))
                        if result.get("status", "completed") != "completed" or result.get("error"):
                            raise _ProtocolError("Codex response was not completed; no tool calls were released.")
                        output = result.get("output")
                        if not isinstance(output, list) or len(output) > _MAX_ITEMS:
                            raise _ProtocolError("Codex completed response has malformed output.")
                        observed = set(items) | set(arguments) | {index for index, _ in texts}
                        # Codex may send output only in output_item.done, leaving the
                        # terminal response's output array empty to avoid duplication.
                        if not output and observed - finished:
                            raise _ProtocolError(
                                "Codex completed before streamed output finished; no tool calls were released."
                            )
                        if output and observed - set(range(len(output))):
                            raise _ProtocolError(
                                "Codex completed response omitted streamed output; no tool calls were released."
                            )
                        for index, raw in enumerate(output):
                            item = _object(raw)
                            previous = items.get(index, {})
                            for identity in ("id", "call_id", "name", "type"):
                                if identity in previous and identity in item and previous[identity] != item[identity]:
                                    raise _ProtocolError("Codex returned inconsistent completed output identities.")
                            items[index] = {**previous, **item}
                            finished.add(index)
                        usage = _usage(result.get("usage"))
                        completed = True
                        break
                if not completed:
                    raise _ProtocolError("Codex stream ended before response.completed; no tool calls were released.")
                calls: list[ToolCallEvent] = []
                call_indexes: list[int] = []
                call_ids: set[str] = set()
                final_texts: dict[tuple[int, int], str] = {}
                for index, item in sorted(items.items()):
                    if item.get("type") == "function_call":
                        call_id, name = _string(item.get("call_id")), _string(item.get("name"))
                        if (
                            not call_id
                            or not name
                            or call_id in call_ids
                            or index not in finished
                            or item.get("status") not in (None, "completed")
                        ):
                            raise _ProtocolError("Codex returned an invalid or unfinished tool call.")
                        raw_arguments = _string(item.get("arguments", arguments.get(index, "")))
                        try:
                            parsed_args = tool_arguments(raw_arguments)
                            if arguments.get(index) and not json_values_equal(
                                tool_arguments(arguments[index]), parsed_args
                            ):
                                raise _ProtocolError("Codex returned inconsistent tool arguments.")
                            if index in completed_arguments and not json_values_equal(
                                completed_arguments[index], parsed_args
                            ):
                                raise _ProtocolError("Codex returned inconsistent completed tool arguments.")
                        except (ValueError, UnicodeError, RecursionError):
                            raise _ProtocolError(
                                "Codex returned malformed tool arguments; no tool calls were released."
                            ) from None
                        call_ids.add(call_id)
                        calls.append(ToolCallEvent(id=call_id, name=name, arguments=parsed_args, usage=usage))
                        call_indexes.append(index)
                    elif index in arguments or index in completed_arguments:
                        raise _ProtocolError("Codex returned tool arguments without a complete tool call.")
                    elif item.get("type") == "message":
                        if item.get("status") not in (None, "completed"):
                            raise _ProtocolError("Codex returned an unfinished message.")
                        content = item.get("content")
                        if not isinstance(content, list):
                            raise _ProtocolError("Codex returned malformed message content.")
                        for part_index, raw_part in enumerate(content):
                            part = _object(raw_part)
                            if part.get("type") == "output_text":
                                final_texts[(index, part_index)] = _string(part.get("text"))
                            elif part.get("type") == "refusal":
                                final_texts[(index, part_index)] = _string(part.get("refusal"))
                for text_key, text in texts.items():
                    if final_texts.get(text_key) != text:
                        raise _ProtocolError("Codex terminal output did not match streamed text.")
            # Finish network cleanup before releasing executable calls.
            yield TextDoneEvent(
                text="".join(text for _, text in sorted(final_texts.items())),
                usage=usage,
                finish_reason=FinishReason.TOOL_CALLS if calls else FinishReason.STOP,
            )
            for index, call in zip(call_indexes, calls, strict=True):
                if stream:
                    yield progress.ready(index, call)
                yield call
        except (_ProtocolError, ProtocolError) as error:
            for preview in progress.abandon():
                yield preview
            yield ErrorEvent(
                message=str(error),
                code="CODEX_STREAM_INVALID",
                extra=failure_extra(error, started, phase=phase, protocol=True),
            )
        except (aiohttp.ClientError, TimeoutError, ValueError) as error:
            for preview in progress.abandon():
                yield preview
            yield ErrorEvent(
                message="Codex connection failed or timed out; retry the request.",
                code="CODEX_CONNECTION",
                extra=failure_extra(error, started, phase=phase),
            )

__init__(credentials=None, model='', timeout=120.0, *, home='', profile='', live_config=None, api_key='', base_url='', api='auto', realtime_config=None, retry_config=None, bearer_token_provider=None)

Explicit API keys/token providers bypass discovery; otherwise use CODEX_HOME/~/.codex.

Local model/profile/wire_api and auth.json select API-key or ChatGPT authentication. Explicit credential callbacks select ChatGPT. A custom base_url requires an explicit key or bearer_token_provider; saved OAuth never goes to that URL. API-key requests default to Responses unless another API is selected.

Source code in src/nagents/provider/openai.py

def __init__(
    self,
    credentials: Callable[[], Awaitable[CodexCredentials]] | None = None,
    model: str = "",
    timeout: float = 120.0,
    *,
    home: str | Path = "",
    profile: str = "",
    live_config: LiveConfig | None = None,
    api_key: str = "",
    base_url: str = "",
    api: str = "auto",
    realtime_config: RealtimeConfig | None = None,
    retry_config: RetryConfig | None = None,
    bearer_token_provider: BearerTokenProvider | None = None,
) -> None:
    """Explicit API keys/token providers bypass discovery; otherwise use CODEX_HOME/~/.codex.

    Local model/profile/wire_api and auth.json select API-key or ChatGPT
    authentication. Explicit credential callbacks select ChatGPT. A custom
    base_url requires an explicit key or bearer_token_provider; saved OAuth never goes to that URL.
    API-key requests default to Responses unless another API is selected.
    """
    self._local_api = False
    self._credentials: Callable[[], Awaitable[CodexCredentials]]
    if sum((credentials is not None, bool(api_key), bearer_token_provider is not None)) > 1:
        raise ValueError("Choose api_key, bearer_token_provider, or ChatGPT credentials, not multiple auth sources")
    if base_url and not api_key and bearer_token_provider is None:
        raise ValueError("An explicit base_url requires an explicit api_key or bearer_token_provider")
    if not api_key and credentials is None and bearer_token_provider is None:
        local = _load_config(
            home, profile=profile, model=model, for_live=live_config is not None or realtime_config is not None
        )
        model = local.model
        credentials = local.credentials
        if not local.oauth:
            api_key = local.api_key
            base_url = local.base_url
            api = local.api if api == "auto" else api
    if api_key or bearer_token_provider is not None:
        super().__init__(
            ProviderType.OPENAI_COMPATIBLE,
            api_key,
            model or DEFAULT_CODEX_MODEL,
            base_url=base_url or None,
            api="responses" if api == "auto" else api,
            timeout=timeout,
            live_config=live_config,
            realtime_config=realtime_config,
            retry_config=retry_config,
            bearer_token_provider=bearer_token_provider,
        )
        self._local_api = True
        return
    assert credentials is not None
    if api not in {"auto", "responses"}:
        raise ValueError("ChatGPT authentication uses the Responses protocol")
    if live_config is not None or realtime_config is not None:
        raise ValueError(
            "Selected ChatGPT subscription authentication. The public GPT-Live WebSocket API and Realtime "
            "require OpenAI API-key authentication; this saved login can still run the delegated backend. "
            "Use ngn serve browser voice for GPT-Live with ChatGPT login."
        )
    super().__init__(
        ProviderType.OPENAI_COMPATIBLE,
        "oauth-not-an-api-key",
        model or DEFAULT_CODEX_MODEL,
        timeout=timeout,
        api="responses",
    )
    self.base_url = CODEX_ENDPOINT
    self._credentials = credentials
    self._timeout = timeout

generate(messages, tools=None, config=None, stream=True, verify_model=False) async

Source code in src/nagents/provider/openai.py

async def generate(
    self,
    messages: list[Message],
    tools: list[ToolDefinition] | None = None,
    config: GenerationConfig | None = None,
    stream: bool = True,
    verify_model: bool = False,
) -> AsyncGenerator[Event, None]:
    if self._local_api:
        async for generated in super().generate(messages, tools, config, stream, verify_model):
            yield generated
        return
    started = monotonic()
    phase: FailurePhase = "request"
    try:
        body = _request_body(self.model, messages, tools, config)
    except (TypeError, ValueError) as error:
        message = str(error) if isinstance(error, _ProtocolError) else "Codex request data is invalid."
        yield ErrorEvent(message=message, code="CODEX_REQUEST_INVALID")
        return
    try:
        credentials = await self._credentials()
        if not credentials.access_token or not re.fullmatch(r"[\x21-\x7e]{1,65536}", credentials.access_token):
            raise ValueError
        if credentials.account_id and not re.fullmatch(r"[A-Za-z0-9_.:-]{1,256}", credentials.account_id):
            raise ValueError
        if credentials.residency and not re.fullmatch(r"[A-Za-z0-9_-]{1,64}", credentials.residency):
            raise ValueError
    except CodexConfigError as error:
        yield ErrorEvent(message=str(error), code="CODEX_AUTH")
        return
    except Exception:
        yield ErrorEvent(
            message="ChatGPT credentials are unavailable; sign in again with /login.", code="CODEX_AUTH"
        )
        return
    headers = {
        "Authorization": f"Bearer {credentials.access_token}",
        "Accept": "text/event-stream",
        "originator": "ngn",
        "User-Agent": USER_AGENT,
    }
    if credentials.account_id:
        headers["ChatGPT-Account-Id"] = credentials.account_id
    if credentials.residency and credentials.residency != "no_constraint":
        headers["x-openai-internal-codex-residency"] = credentials.residency
    progress = ToolCallProgressTracker()
    try:
        async with (
            aiohttp.ClientSession(
                timeout=aiohttp.ClientTimeout(total=self._timeout),
                trust_env=False,
                cookie_jar=aiohttp.DummyCookieJar(),
                trace_configs=[trace_config()],
            ) as client,
            client.post(CODEX_ENDPOINT, json=body, headers=headers, allow_redirects=False) as response,
        ):
            phase = "response"
            if response.status != 200:
                messages_by_status = {
                    401: "ChatGPT login expired or was rejected; sign in again with /login.",
                    403: "Codex access denied; check your ChatGPT plan, workspace permissions, and model access.",
                    429: "ChatGPT Codex usage limit reached; wait before retrying or check your plan limits.",
                }
                yield ErrorEvent(
                    message=messages_by_status.get(
                        response.status,
                        f"Codex request failed (HTTP {response.status}); check service availability and model access.",
                    ),
                    code=f"CODEX_HTTP_{response.status}",
                )
                return
            # Some Codex responses omit Content-Type; _sse still validates framing
            # and no calls are released without a completed, validated response.
            if response.headers.get("Content-Type") and response.content_type != "text/event-stream":
                raise _ProtocolError("Codex did not return an event stream; retry the request.")
            items: dict[int, dict[str, object]] = {}
            finished: set[int] = set()
            arguments: dict[int, str] = {}
            completed_arguments: dict[int, ToolArguments] = {}
            texts: dict[tuple[int, int], str] = {}
            texts_done: set[tuple[int, int]] = set()
            completed = False
            usage = Usage()
            async for event in _sse(response):
                kind = _string(event.get("type"))
                if kind in {"error", "response.failed", "response.incomplete", "response.cancelled"}:
                    raise _ProtocolError(
                        "Codex response failed, was incomplete, or was cancelled; no tool calls were released. Retry the request."
                    )
                if kind in {"response.output_item.added", "response.output_item.done"}:
                    index = _index(event.get("output_index"))
                    item = _object(event.get("item"))
                    previous = items.get(index, {})
                    for key in ("id", "call_id", "name", "type"):
                        if key in previous and key in item and previous[key] != item[key]:
                            raise _ProtocolError("Codex returned inconsistent output item identities.")
                    items[index] = {**previous, **item}
                    merged_item = items[index]
                    if (
                        merged_item.get("type") == "function_call"
                        and "arguments" in merged_item
                        and (kind == "response.output_item.done" or merged_item["arguments"] != "")
                    ):
                        _remember_completed_arguments(completed_arguments, index, merged_item["arguments"])
                    if kind == "response.output_item.done":
                        finished.add(index)
                    if stream and merged_item.get("type") == "function_call":
                        for preview in progress.preview(
                            index,
                            _string(merged_item.get("call_id", "")),
                            _string(merged_item.get("name", "")),
                            _string(arguments.get(index, merged_item.get("arguments", ""))),
                        ):
                            yield preview
                elif kind in {"response.function_call_arguments.delta", "response.function_call_arguments.done"}:
                    index = _index(event.get("output_index"))
                    item = items.setdefault(index, {})
                    if "item_id" in event:
                        item_id = _string(event["item_id"])
                        if not item_id or ("id" in item and item_id != item["id"]):
                            raise _ProtocolError("Codex returned inconsistent argument item identities.")
                        item["id"] = item_id
                    if kind.endswith(".delta"):
                        arguments[index] = arguments.get(index, "") + _string(event.get("delta"))
                    else:
                        item["arguments"] = _string(event.get("arguments"))
                        _remember_completed_arguments(completed_arguments, index, item["arguments"])
                    if stream:
                        for preview in progress.preview(
                            index,
                            _string(item.get("call_id", "")),
                            _string(item.get("name", "")),
                            _string(arguments.get(index, item.get("arguments", ""))),
                        ):
                            yield preview
                elif kind in {"response.output_text.delta", "response.output_text.done"}:
                    text_key = (_index(event.get("output_index", 0)), _index(event.get("content_index", 0)))
                    index = text_key[0]
                    if "item_id" in event:
                        item = items.setdefault(index, {})
                        item_id = _string(event["item_id"])
                        if not item_id or ("id" in item and item_id != item["id"]):
                            raise _ProtocolError("Codex returned inconsistent text item identities.")
                        item["id"] = item_id
                    if text_key in texts_done or index in finished:
                        raise _ProtocolError("Codex sent text after a content part finished.")
                    if kind.endswith(".delta"):
                        delta = _string(event.get("delta"))
                        texts[text_key] = texts.get(text_key, "") + delta
                        if stream:
                            yield TextChunkEvent(chunk=delta)
                    else:
                        text = _string(event.get("text"))
                        if text_key in texts and texts[text_key] != text:
                            raise _ProtocolError("Codex returned inconsistent completed text.")
                        texts[text_key] = text
                        texts_done.add(text_key)
                elif kind == "response.reasoning_summary_text.delta":
                    delta = _string(event.get("delta"))
                    if stream:
                        yield ReasoningChunkEvent(chunk=delta)
                elif kind == "response.completed":
                    result = _object(event.get("response"))
                    if result.get("status", "completed") != "completed" or result.get("error"):
                        raise _ProtocolError("Codex response was not completed; no tool calls were released.")
                    output = result.get("output")
                    if not isinstance(output, list) or len(output) > _MAX_ITEMS:
                        raise _ProtocolError("Codex completed response has malformed output.")
                    observed = set(items) | set(arguments) | {index for index, _ in texts}
                    # Codex may send output only in output_item.done, leaving the
                    # terminal response's output array empty to avoid duplication.
                    if not output and observed - finished:
                        raise _ProtocolError(
                            "Codex completed before streamed output finished; no tool calls were released."
                        )
                    if output and observed - set(range(len(output))):
                        raise _ProtocolError(
                            "Codex completed response omitted streamed output; no tool calls were released."
                        )
                    for index, raw in enumerate(output):
                        item = _object(raw)
                        previous = items.get(index, {})
                        for identity in ("id", "call_id", "name", "type"):
                            if identity in previous and identity in item and previous[identity] != item[identity]:
                                raise _ProtocolError("Codex returned inconsistent completed output identities.")
                        items[index] = {**previous, **item}
                        finished.add(index)
                    usage = _usage(result.get("usage"))
                    completed = True
                    break
            if not completed:
                raise _ProtocolError("Codex stream ended before response.completed; no tool calls were released.")
            calls: list[ToolCallEvent] = []
            call_indexes: list[int] = []
            call_ids: set[str] = set()
            final_texts: dict[tuple[int, int], str] = {}
            for index, item in sorted(items.items()):
                if item.get("type") == "function_call":
                    call_id, name = _string(item.get("call_id")), _string(item.get("name"))
                    if (
                        not call_id
                        or not name
                        or call_id in call_ids
                        or index not in finished
                        or item.get("status") not in (None, "completed")
                    ):
                        raise _ProtocolError("Codex returned an invalid or unfinished tool call.")
                    raw_arguments = _string(item.get("arguments", arguments.get(index, "")))
                    try:
                        parsed_args = tool_arguments(raw_arguments)
                        if arguments.get(index) and not json_values_equal(
                            tool_arguments(arguments[index]), parsed_args
                        ):
                            raise _ProtocolError("Codex returned inconsistent tool arguments.")
                        if index in completed_arguments and not json_values_equal(
                            completed_arguments[index], parsed_args
                        ):
                            raise _ProtocolError("Codex returned inconsistent completed tool arguments.")
                    except (ValueError, UnicodeError, RecursionError):
                        raise _ProtocolError(
                            "Codex returned malformed tool arguments; no tool calls were released."
                        ) from None
                    call_ids.add(call_id)
                    calls.append(ToolCallEvent(id=call_id, name=name, arguments=parsed_args, usage=usage))
                    call_indexes.append(index)
                elif index in arguments or index in completed_arguments:
                    raise _ProtocolError("Codex returned tool arguments without a complete tool call.")
                elif item.get("type") == "message":
                    if item.get("status") not in (None, "completed"):
                        raise _ProtocolError("Codex returned an unfinished message.")
                    content = item.get("content")
                    if not isinstance(content, list):
                        raise _ProtocolError("Codex returned malformed message content.")
                    for part_index, raw_part in enumerate(content):
                        part = _object(raw_part)
                        if part.get("type") == "output_text":
                            final_texts[(index, part_index)] = _string(part.get("text"))
                        elif part.get("type") == "refusal":
                            final_texts[(index, part_index)] = _string(part.get("refusal"))
            for text_key, text in texts.items():
                if final_texts.get(text_key) != text:
                    raise _ProtocolError("Codex terminal output did not match streamed text.")
        # Finish network cleanup before releasing executable calls.
        yield TextDoneEvent(
            text="".join(text for _, text in sorted(final_texts.items())),
            usage=usage,
            finish_reason=FinishReason.TOOL_CALLS if calls else FinishReason.STOP,
        )
        for index, call in zip(call_indexes, calls, strict=True):
            if stream:
                yield progress.ready(index, call)
            yield call
    except (_ProtocolError, ProtocolError) as error:
        for preview in progress.abandon():
            yield preview
        yield ErrorEvent(
            message=str(error),
            code="CODEX_STREAM_INVALID",
            extra=failure_extra(error, started, phase=phase, protocol=True),
        )
    except (aiohttp.ClientError, TimeoutError, ValueError) as error:
        for preview in progress.abandon():
            yield preview
        yield ErrorEvent(
            message="Codex connection failed or timed out; retry the request.",
            code="CODEX_CONNECTION",
            extra=failure_extra(error, started, phase=phase),
        )

get_model_list() async

Fetch model IDs from the API-key catalog or the subscription catalog.

With ChatGPT authentication this follows the Codex client's catalog contract, not the public OpenAI API-key catalog. It is fresh, read-only, and independent of model verification. Failures raise ModelListError without upstream data; there is no API-key fallback or entitlement guarantee.

Source code in src/nagents/provider/openai.py

async def get_model_list(self) -> list[str]:
    """Fetch model IDs from the API-key catalog or the subscription catalog.

    With ChatGPT authentication this follows the Codex client's catalog contract, not the public
    OpenAI API-key catalog. It is fresh, read-only, and independent of model
    verification. Failures raise ModelListError without upstream data; there
    is no API-key fallback or entitlement guarantee.
    """
    if self._local_api:
        return await super().get_model_list()
    if self.base_url != CODEX_ENDPOINT:
        raise ModelListError("Codex model discovery requires its fixed endpoint; custom URLs are unsupported.")
    try:
        credentials = await self._credentials()
        # Capture all routing fields before any further await, even when the
        # callback returns a mutable credentials object shared with its caller.
        access_token, account_id, residency = (
            credentials.access_token,
            credentials.account_id,
            credentials.residency,
        )
        if not access_token or not re.fullmatch(r"[\x21-\x7e]{1,65536}", access_token):
            raise ValueError
        if account_id and not re.fullmatch(r"[A-Za-z0-9_.:-]{1,256}", account_id):
            raise ValueError
        if residency and not re.fullmatch(r"[A-Za-z0-9_-]{1,64}", residency):
            raise ValueError
    except CodexConfigError as error:
        raise ModelListError(str(error)) from None
    except Exception:
        raise ModelListError("ChatGPT credentials are unavailable; sign in again with /login.") from None
    client_version = await catalog_version(self._model_list_timeout)
    headers = {
        "Authorization": f"Bearer {access_token}",
        "Accept": "application/json",
        "version": client_version,
        "originator": "ngn",
        "User-Agent": USER_AGENT,
    }
    if account_id:
        headers["ChatGPT-Account-Id"] = account_id
    if residency and residency != "no_constraint":
        headers["x-openai-internal-codex-residency"] = residency
    try:
        client = GatewayHTTPClient(timeout=self._model_list_timeout)
        async with client:
            response = await client.get_json(f"{CODEX_MODELS_ENDPOINT}?client_version={client_version}", headers)
    except Exception:
        raise ModelListError(
            "Codex model discovery failed; check service availability and your ChatGPT login."
        ) from None
    if "error" in response or not isinstance(response.get("models"), list):
        raise ModelListError("Codex returned an invalid model catalog.")
    models: list[str] = []
    seen: set[str] = set()
    for item in list_data(response["models"]):
        if not isinstance(item, dict):
            raise ModelListError("Codex returned an invalid model entry.")
        model_id, visibility = item.get("slug"), item.get("visibility")
        if (
            not isinstance(model_id, str)
            or not model_id.strip()
            or any(unicodedata.category(char).startswith("C") for char in model_id)
        ):
            raise ModelListError("Codex returned an invalid model ID.")
        # Visibility is required by the pinned upstream ModelInfo struct.
        if not isinstance(visibility, str) or visibility not in {"list", "hide", "none"}:
            raise ModelListError("Codex returned invalid model visibility.")
        # supported_in_api is not an OAuth availability filter.
        if visibility == "list" and model_id not in seen:
            models.append(model_id)
            seen.add(model_id)
    return models