litellm/litellm/proxy/response_api_endpoints/endpoints.py

import asyncio
import contextlib
import json
from collections.abc import AsyncGenerator, AsyncIterator, Awaitable, Mapping, Sequence
from enum import Enum
from functools import partial
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, NamedTuple, Protocol, TypeAlias, cast, get_args

import fastapi
from fastapi import APIRouter, Depends, HTTPException, Request, Response
from fastapi.responses import JSONResponse
from openai.types.responses import ResponseItemList
from openai.types.responses.response_create_params import ResponseInputParam
from pydantic import ConfigDict, TypeAdapter, ValidationError
from starlette.websockets import WebSocket, WebSocketDisconnect
from typing_extensions import ReadOnly, TypedDict

from litellm._logging import verbose_proxy_logger
from litellm.constants import EMPTY_MAPPING
from litellm.integrations.custom_guardrail import ModifyResponseException
from litellm.llms.openai.responses.guardrail_translation.handler import (
    OpenAIResponsesHandler,
    build_blocked_response,
)
from litellm.proxy._types import *
from litellm.proxy.auth.user_api_key_auth import (
    UserAPIKeyAuth,
    user_api_key_auth,
    user_api_key_auth_websocket,
)
from litellm.proxy.common_request_processing import ProxyBaseLLMRequestProcessing, create_response
from litellm.proxy.common_utils.http_parsing_utils import (  # noqa: F401  # legacy module exports
    _read_request_body,  # pyright: ignore[reportPrivateUsage,reportUnusedImport]  # backwards-compatible package export
    _safe_set_request_parsed_body,  # pyright: ignore[reportPrivateUsage,reportUnusedImport]  # backwards-compatible package export
    read_request_body,
    safe_set_request_parsed_body,
)
from litellm.proxy.common_utils.token_count_settings import count_source_fields
from litellm.proxy.route_llm_request import raise_if_required_body_param_missing
from litellm.types.llms.base import LiteLLMBaseModel
from litellm.types.llms.openai import (
    REASONING_EFFORT,
    ResponsesAPIOptionalRequestParams,
    ResponsesAPIResponse,
)
from litellm.types.responses.main import DeleteResponseResult
from litellm.types.token_count import TokenCountSourceFields
from litellm.types.utils import TokenCountResponse

if TYPE_CHECKING:
    from litellm.router import Router

router: Final = APIRouter()
_RESPONSES_WS_CONFIG_VALUE_ADAPTER: Final[TypeAdapter[object | None]] = TypeAdapter(object | None)

_ResponseDocSchemas: TypeAlias = dict[int | str, dict[str, object]]  # fastapi's responses kwarg

RESPONSES_API_RESPONSE_SCHEMAS: Final[_ResponseDocSchemas] = {200: {"model": ResponsesAPIResponse}}
RESPONSES_API_CREATE_RESPONSE_SCHEMAS: Final[_ResponseDocSchemas] = {
    200: {
        "model": ResponsesAPIResponse,
        "content": {
            "text/event-stream": {"schema": {"type": "string", "description": "Server sent events when stream=true"}}
        },
    }
}
DELETE_RESPONSE_SCHEMAS: Final[_ResponseDocSchemas] = {200: {"model": DeleteResponseResult}}
RESPONSE_ITEM_LIST_SCHEMAS: Final[_ResponseDocSchemas] = {200: {"model": ResponseItemList}}

_user_api_key_auth_dep: Final = Depends(user_api_key_auth)
_RESPONSES_TAGS: Final[list[str | Enum]] = ["responses"]  # mutable-ok: fastapi's route signature requires list tags

_TOOL_PAYLOAD_KEYS: Final[Mapping[str, tuple[str, ...]]] = MappingProxyType(
    {
        "custom": ("name", "description", "format"),
        "function": ("name", "description", "parameters", "strict"),
    }
)
_EMPTY_TOOL_PAYLOAD: Final[Mapping[str, object]] = MappingProxyType({})


def _convert_tool_payload_value(key: str, value: object, *, to_chat: bool) -> object:
    if key != "format" or not isinstance(value, dict):
        return value
    from litellm.litellm_core_utils.prompt_templates.common_utils import (
        convert_custom_tool_format_to_chat_shape,
        convert_custom_tool_format_to_responses_shape,
    )

    convert = convert_custom_tool_format_to_chat_shape if to_chat else convert_custom_tool_format_to_responses_shape
    return convert(value)


def _convert_tool_envelope(obj: object, *, to_chat: bool) -> object:
    if not isinstance(obj, dict):
        return obj
    tool_type: Final = obj.get("type")
    payload_keys: Final = _TOOL_PAYLOAD_KEYS.get(tool_type)
    if payload_keys is None:
        return obj
    nested: Final = obj.get(tool_type)
    nested_source: Final = nested if isinstance(nested, dict) else _EMPTY_TOOL_PAYLOAD
    payload: Final = {
        key: _convert_tool_payload_value(key, nested_source[key] if key in nested_source else obj[key], to_chat=to_chat)
        for key in payload_keys
        if key in nested_source or key in obj
    }
    if "name" not in payload:
        return obj
    return {"type": tool_type, tool_type: payload} if to_chat else {"type": tool_type, **payload}


def _normalize_tool_dialect(
    data: dict, *, to_chat: bool
) -> dict:  # mutable-ok: the parsed request body contract is a plain dict
    tools: Final = data.get("tools")
    tool_choice: Final = data.get("tool_choice")
    normalized_tools: Final = (
        [_convert_tool_envelope(tool, to_chat=to_chat) for tool in tools] if isinstance(tools, list) else tools
    )
    normalized_choice: Final = _convert_tool_envelope(tool_choice, to_chat=to_chat)
    if normalized_tools == tools and normalized_choice == tool_choice:
        return data
    replaceable: Final = (("tools", normalized_tools), ("tool_choice", normalized_choice))
    return {**data, **{key: value for key, value in replaceable if key in data}}


def _is_chat_completions_body(data: Mapping[str, object]) -> bool:
    messages: Final = data.get("messages")
    if isinstance(messages, list) and messages:
        return True
    return "messages" in data and "input" not in data


_CURSOR_THINKING_SEPARATOR: Final = "-thinking-"
_CURSOR_FAST_SUFFIX: Final = "-fast"
_CURSOR_THINKING_LEVELS: Final[frozenset[str]] = frozenset(get_args(REASONING_EFFORT))


class _CursorModelVariant(NamedTuple):
    base_model: str
    reasoning_effort: str | None


def _parse_cursor_model_variant(model: str) -> _CursorModelVariant:
    stripped: Final = model.removesuffix(_CURSOR_FAST_SUFFIX)
    base, separator, level = stripped.rpartition(_CURSOR_THINKING_SEPARATOR)
    if separator and base and level in _CURSOR_THINKING_LEVELS:
        return _CursorModelVariant(base, level)
    return _CursorModelVariant(stripped, None)


def _router_can_serve(model: str, llm_router: "Router | None") -> bool:
    if llm_router is None:
        return False
    if llm_router.is_recognized_model(model):
        return True
    if model in llm_router.team_public_model_names:
        return True
    return bool(llm_router.pattern_router.get_pattern(model))


def _resolve_cursor_model_variant(
    data: dict, llm_router: "Router | None"
) -> dict:  # mutable-ok: the parsed request body contract is a plain dict
    model: Final = data.get("model")
    if not isinstance(model, str) or _router_can_serve(model, llm_router):
        return data
    variant: Final = _parse_cursor_model_variant(model)
    if variant.base_model == model or not _router_can_serve(variant.base_model, llm_router):
        return data
    resolved: Final = {**data, "model": variant.base_model}
    if variant.reasoning_effort is None:
        return resolved
    if _is_chat_completions_body(data):
        if "reasoning_effort" in data:
            return resolved
        return {**resolved, "reasoning_effort": variant.reasoning_effort}
    reasoning: Final = data.get("reasoning")
    if isinstance(reasoning, dict):
        if reasoning.get("effort"):
            return resolved
        return {**resolved, "reasoning": {**reasoning, "effort": variant.reasoning_effort}}
    return {**resolved, "reasoning": {"effort": variant.reasoning_effort}}


async def _resolve_cursor_model_variant_before_auth(request: Request) -> None:
    from litellm.proxy.proxy_server import llm_router

    try:
        raw_body: Final = await read_request_body(request=request)
    except (json.JSONDecodeError, ProxyException):
        return
    resolved: Final = _resolve_cursor_model_variant(raw_body, llm_router)
    if resolved is not raw_body:
        safe_set_request_parsed_body(request=request, parsed_body=resolved)


@router.post(
    "/v1/responses",
    dependencies=[Depends(user_api_key_auth)],
    tags=["responses"],
    responses=RESPONSES_API_CREATE_RESPONSE_SCHEMAS,
)
@router.post(
    "/responses",
    dependencies=[Depends(user_api_key_auth)],
    tags=["responses"],
    responses=RESPONSES_API_CREATE_RESPONSE_SCHEMAS,
)
@router.post(
    "/openai/v1/responses",
    dependencies=[Depends(user_api_key_auth)],
    tags=["responses"],
    responses=RESPONSES_API_CREATE_RESPONSE_SCHEMAS,
)
async def responses_api(
    request: Request,
    fastapi_response: Response,
    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
    """
    Follows the OpenAI Responses API spec: https://platform.openai.com/docs/api-reference/responses

    Supports background mode with polling_via_cache for partial response retrieval.
    When background=true and polling_via_cache is enabled, returns a polling_id immediately
    and streams the response in the background, updating Redis cache.

    ```bash
    # Normal request
    curl -X POST http://localhost:4000/v1/responses \
    -H "Content-Type: application/json" \
    -H "Authorization: Bearer $LITELLM_MASTER_KEY" \
    -d '{
        "model": "gpt-4o",
        "input": "Tell me about AI"
    }'

    # Background request with polling
    curl -X POST http://localhost:4000/v1/responses \
    -H "Content-Type: application/json" \
    -H "Authorization: Bearer $LITELLM_MASTER_KEY" \
    -d '{
        "model": "gpt-4o",
        "input": "Tell me about AI",
        "background": true
    }'
    ```
    """
    from litellm.proxy.proxy_server import (
        general_settings,
        llm_router,
        native_background_mode,
        polling_cache_ttl,
        polling_via_cache_enabled,
        proxy_config,
        proxy_logging_obj,
        read_request_body,
        redis_usage_cache,
        select_data_generator,
        user_api_base,
        user_max_tokens,
        user_model,
        user_request_timeout,
        user_temperature,
        version,
    )

    native_data_generator: Final = partial(select_data_generator, responses_stream_errors=True)
    data = await read_request_body(request=request)  # rebind-ok: pre-existing rebinding on a rename-only line

    # Check if polling via cache should be used for this request
    from litellm.proxy.response_polling.polling_handler import (
        should_use_polling_for_request,
    )

    should_use_polling: Final = should_use_polling_for_request(
        background_mode=data.get("background", False),
        polling_via_cache_enabled=polling_via_cache_enabled,
        redis_cache=redis_usage_cache,
        model=data.get("model", ""),
        llm_router=llm_router,
        native_background_mode=native_background_mode,
    )

    # If polling is enabled, use polling mode
    if should_use_polling:
        from litellm.proxy.response_polling.background_streaming import (
            background_streaming_task,
        )
        from litellm.proxy.response_polling.polling_handler import (
            ResponsePollingHandler,
        )

        verbose_proxy_logger.info("Starting background response with polling for model=%s", data.get("model"))

        # Run pre-call checks (rate limits, guardrails, budget) BEFORE creating
        # polling ID. This ensures rate-limited requests get a synchronous 429
        # instead of a polling ID that immediately fails in the background task.
        processor = ProxyBaseLLMRequestProcessing(data=data)
        try:
            data, _logging_obj = await processor.common_processing_pre_call_logic(
                request=request,
                general_settings=general_settings,
                proxy_logging_obj=proxy_logging_obj,
                user_api_key_dict=user_api_key_dict,
                version=version,
                proxy_config=proxy_config,
                user_model=user_model,
                user_temperature=user_temperature,
                user_request_timeout=user_request_timeout,
                user_max_tokens=user_max_tokens,
                user_api_base=user_api_base,
                model=None,
                route_type="aresponses",
                llm_router=llm_router,
            )
            raise_if_required_body_param_missing(route_type="aresponses", data=data, llm_router=llm_router)
        except Exception as e:
            raise await processor.handle_llm_api_exception(
                e=e,
                user_api_key_dict=user_api_key_dict,
                proxy_logging_obj=proxy_logging_obj,
                version=version,
            )

        # Initialize polling handler with configured TTL (from global config)
        polling_handler: Final = ResponsePollingHandler(
            redis_cache=redis_usage_cache,
            ttl=polling_cache_ttl,  # Global var set at startup
        )

        # Generate polling ID
        polling_id: Final = ResponsePollingHandler.generate_polling_id()

        # Create initial state in Redis
        initial_state: Final = await polling_handler.create_initial_state(
            polling_id=polling_id,
            request_data=data,
        )

        # Start background task to stream and update cache.
        # Pass pre-processed data so the background task skips pre-call logic
        # (rate limits, guardrails already checked above).
        asyncio.create_task(
            background_streaming_task(
                polling_id=polling_id,
                data=data.copy(),
                polling_handler=polling_handler,
                request=request,
                fastapi_response=fastapi_response,
                user_api_key_dict=user_api_key_dict,
                general_settings=general_settings,
                llm_router=llm_router,
                proxy_config=proxy_config,
                proxy_logging_obj=proxy_logging_obj,
                select_data_generator=native_data_generator,
                user_model=user_model,
                user_temperature=user_temperature,
                user_request_timeout=user_request_timeout,
                user_max_tokens=user_max_tokens,
                user_api_base=user_api_base,
                version=version,
            )
        )

        # Return OpenAI Response object format (initial state)
        # https://platform.openai.com/docs/api-reference/responses/object
        return initial_state

    # Normal response flow
    processor = ProxyBaseLLMRequestProcessing(data=data)
    try:
        response: Final = await processor.base_process_llm_request(
            request=request,
            fastapi_response=fastapi_response,
            user_api_key_dict=user_api_key_dict,
            route_type="aresponses",
            proxy_logging_obj=proxy_logging_obj,
            llm_router=llm_router,
            general_settings=general_settings,
            proxy_config=proxy_config,
            select_data_generator=native_data_generator,
            model=None,
            user_model=user_model,
            user_temperature=user_temperature,
            user_request_timeout=user_request_timeout,
            user_max_tokens=user_max_tokens,
            user_api_base=user_api_base,
            version=version,
        )

        # Store in managed objects table if background mode is enabled
        if data.get("background") and isinstance(response, ResponsesAPIResponse):
            if response.status in ["queued", "in_progress"]:
                from litellm_enterprise.proxy.hooks.managed_files import (
                    PROXY_LiteLLMManagedFiles,
                )

                managed_files_obj: Final = cast(
                    PROXY_LiteLLMManagedFiles | None,
                    proxy_logging_obj.get_proxy_hook("managed_files"),
                )

                if managed_files_obj and llm_router:
                    try:
                        # Get the actual deployment model_id from hidden params
                        hidden_params: Final = getattr(response, "_hidden_params", {}) or {}
                        model_id: Final = hidden_params.get("model_id", None)

                        if not model_id:
                            verbose_proxy_logger.warning(
                                "No model_id found in response hidden params for response %s, skipping managed object storage",
                                response.id,
                            )
                            raise Exception("No model_id found in response hidden params")
                        # Store in managed objects table
                        await managed_files_obj.store_unified_object_id(
                            unified_object_id=response.id,
                            file_object=response,
                            litellm_parent_otel_span=None,
                            model_object_id=response.id,
                            file_purpose="response",
                            user_api_key_dict=user_api_key_dict,
                        )

                        verbose_proxy_logger.info(
                            "Stored background response %s in managed objects table with unified_id=%s",
                            response.id,
                            response.id,
                        )
                    except Exception as e:
                        verbose_proxy_logger.error(
                            "Failed to store background response in managed objects table: %s", e
                        )

        return response
    except ModifyResponseException as e:
        # Guardrail passthrough: return violation message in Responses API format (200)
        _data: Final = e.request_data
        await proxy_logging_obj.post_call_failure_hook(
            user_api_key_dict=user_api_key_dict,
            original_exception=e,
            request_data=_data,
        )

        if data.get("stream") is True:
            block_chunks: Final = OpenAIResponsesHandler().build_block_sse_chunks(e)

            async def _blocked_stream() -> AsyncGenerator[str, None]:
                for chunk in block_chunks:
                    yield chunk.decode()
                yield "data: [DONE]\n\n"

            return await create_response(generator=_blocked_stream(), media_type="text/event-stream", headers={})
        return build_blocked_response(e)
    except Exception as e:
        raise await processor.handle_llm_api_exception(
            e=e,
            user_api_key_dict=user_api_key_dict,
            proxy_logging_obj=proxy_logging_obj,
            version=version,
        )


@router.get(
    "/cursor/models",
    dependencies=(_user_api_key_auth_dep,),
    tags=_RESPONSES_TAGS,
)
@router.get(
    "/cursor/v1/models",
    dependencies=(_user_api_key_auth_dep,),
    tags=_RESPONSES_TAGS,
)
async def cursor_model_list(
    user_api_key_dict: UserAPIKeyAuth = _user_api_key_auth_dep,
):
    """
    OpenAI-compatible model listing for the Cursor BYOK base URL.

    Clients pointed at `<proxy>/cursor` as an OpenAI-compatible base URL resolve and
    verify models via `GET {base}/models` (the OpenAI SDK contract). Without this
    route those requests fall through to the Cursor Cloud Agents passthrough, which
    demands a Cursor API key and 401s, so key verification silently fails before any
    chat request is ever sent. Delegates to the standard `/v1/models` handler.
    """
    from litellm.proxy.proxy_server import model_list

    return await model_list(user_api_key_dict=user_api_key_dict)


@router.post(
    "/cursor/chat/completions",
    dependencies=[
        Depends(_resolve_cursor_model_variant_before_auth),
        Depends(user_api_key_auth),
    ],
    tags=["responses"],
)
async def cursor_chat_completions(
    request: Request,
    fastapi_response: Response,
    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
    """
    Cursor BYOK endpoint. Accepts both request shapes Cursor sends to its OpenAI-compatible
    base URL and always answers in chat completions format.

    Cursor agent mode sends Responses API format bodies (`input`, flat tool defs, `reasoning`,
    custom tools) to the chat/completions path while expecting chat completions responses;
    those are routed through the Responses API pipeline and converted back. Genuine chat
    completions bodies (`messages` present) are routed through the standard chat completions
    pipeline, after normalizing each level of the `tools` array and `tool_choice` to the chat
    completions shapes OpenAI requires. Cursor mixes Responses API shapes into chat bodies
    per level, independently: a flat tool def (`{"type": "custom", "name": "ApplyPatch", ...}`)
    gets nested under `custom`, and a flat grammar format
    (`{"type": "grammar", "definition", "syntax"}`) gets wrapped as
    `{"type": "grammar", "grammar": {...}}` wherever it appears, including inside tool defs
    Cursor already sent pre-nested.

    ```bash
    curl -X POST http://localhost:4000/cursor/chat/completions \
    -H "Content-Type: application/json" \
    -H "Authorization: Bearer $LITELLM_MASTER_KEY" \
    -d '{
        "model": "gpt-4o",
        "input": [{"role": "user", "content": "Hello"}]
    }'
    Responds back in chat completions format.
    ```
    """
    from litellm.completion_extras.litellm_responses_transformation.handler import (
        responses_api_bridge,
    )
    from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper
    from litellm.proxy.proxy_server import (
        async_data_generator,
        chat_completion,
        general_settings,
        llm_router,
        proxy_config,
        proxy_logging_obj,
        user_api_base,
        user_max_tokens,
        user_model,
        user_request_timeout,
        user_temperature,
        version,
    )
    from litellm.types.llms.openai import ResponsesAPIResponse
    from litellm.types.utils import ModelResponse

    raw_body: Final = await read_request_body(request=request)

    if _is_chat_completions_body(raw_body):
        # Genuine chat completions body (Cursor sends these for models whose BYOK it
        # already fixed); delegate so behavior matches /chat/completions exactly.
        # Keyed on messages CONTENT, not key presence: Cursor can send a null or
        # empty messages stub alongside a real agent-mode input array
        normalized: Final = _normalize_tool_dialect(raw_body, to_chat=True)
        if normalized is not raw_body:
            safe_set_request_parsed_body(request=request, parsed_body=normalized)
        return await chat_completion(
            request=request,
            fastapi_response=fastapi_response,
            model=None,
            user_api_key_dict=user_api_key_dict,
        )

    # OpenAI's Responses API rejects chat-completions-only stream_options
    # (Cursor sends include_usage); usage arrives via response.completed anyway.
    # Rebuild rather than pop: _read_request_body can return the request-scope
    # cached parsed-body dict itself, and removing keys from it corrupts the
    # cache's key snapshot so later readers get an empty body
    body_without_stream_options: Final = {key: value for key, value in raw_body.items() if key != "stream_options"}

    data: Final = _normalize_tool_dialect(body_without_stream_options, to_chat=False)

    processor: Final = ProxyBaseLLMRequestProcessing(data=data)

    def cursor_data_generator(response, user_api_key_dict, request_data, request=None):
        """
        Custom generator that transforms Responses API streaming chunks to chat completion chunks.

        This generator is used for the cursor endpoint to convert Responses API format responses
        to chat completion format that Cursor IDE expects.

        Args:
            response: The streaming Responses API event iterator (router-wrapped or not)
            user_api_key_dict: User API key authentication dict
            request_data: Request data containing model, logging_obj, etc.
            request: The originating FastAPI request, forwarded for disconnect handling

        Returns:
            Async generator that yields SSE-formatted chat completion chunks
        """
        # Any async-iterable here is a Responses API event stream needing conversion.
        # Class-identity checks miss router-wrapped streams (e.g.
        # HiddenParamsAsyncIteratorWrapper around LiteLLMCompletionStreamingIterator),
        # which previously leaked raw Responses events to the client.
        if hasattr(response, "__anext__"):
            # Transform Responses API iterator to chat completion iterator
            # Cast to AsyncIterator[str] since the stream implements __aiter__/__anext__
            completion_stream: Final = responses_api_bridge.transformation_handler.get_model_response_iterator(
                streaming_response=cast(AsyncIterator[str], response),
                sync_stream=False,
                json_mode=False,
            )
            # Wrap in CustomStreamWrapper to get the async generator
            logging_obj: Final = request_data.get("litellm_logging_obj")
            streamwrapper: Final = CustomStreamWrapper(
                completion_stream=completion_stream,
                model=request_data.get("model", ""),
                custom_llm_provider=None,
                logging_obj=logging_obj,
            )
            # Use async_data_generator to format as SSE
            return async_data_generator(
                response=streamwrapper,
                user_api_key_dict=user_api_key_dict,
                request_data=request_data,
                request=request,
            )
        # Otherwise, use the default generator
        return async_data_generator(
            response=response,
            user_api_key_dict=user_api_key_dict,
            request_data=request_data,
            request=request,
        )

    try:
        response: Final = await processor.base_process_llm_request(
            request=request,
            fastapi_response=fastapi_response,
            user_api_key_dict=user_api_key_dict,
            route_type="aresponses",
            proxy_logging_obj=proxy_logging_obj,
            llm_router=llm_router,
            general_settings=general_settings,
            proxy_config=proxy_config,
            select_data_generator=cursor_data_generator,
            model=None,
            user_model=user_model,
            user_temperature=user_temperature,
            user_request_timeout=user_request_timeout,
            user_max_tokens=user_max_tokens,
            user_api_base=user_api_base,
            version=version,
        )

        # Transform non-streaming Responses API response to chat completions format
        if isinstance(response, ResponsesAPIResponse):
            logging_obj: Final = processor.data.get("litellm_logging_obj")
            transformed_response: Final = responses_api_bridge.transformation_handler.transform_response(
                model=processor.data.get("model", ""),
                raw_response=response,
                model_response=ModelResponse(),
                logging_obj=cast(Any, logging_obj),
                request_data=processor.data,
                messages=processor.data.get("input", []),
                optional_params={},
                litellm_params={},
                encoding=None,
                api_key=None,
                json_mode=None,
            )
            return transformed_response

        # Streaming responses are already transformed by cursor_select_data_generator
        return response
    except Exception as e:
        raise await processor.handle_llm_api_exception(
            e=e,
            user_api_key_dict=user_api_key_dict,
            proxy_logging_obj=proxy_logging_obj,
            version=version,
        )


@router.get(
    "/v1/responses/{response_id}",
    dependencies=[Depends(user_api_key_auth)],
    tags=["responses"],
    responses=RESPONSES_API_RESPONSE_SCHEMAS,
)
@router.get(
    "/responses/{response_id}",
    dependencies=[Depends(user_api_key_auth)],
    tags=["responses"],
    responses=RESPONSES_API_RESPONSE_SCHEMAS,
)
@router.get(
    "/openai/v1/responses/{response_id}",
    dependencies=[Depends(user_api_key_auth)],
    tags=["responses"],
    responses=RESPONSES_API_RESPONSE_SCHEMAS,
)
async def get_response(
    response_id: str,
    request: Request,
    fastapi_response: Response,
    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
    """
    Get a response by ID.
    
    Supports both:
    - Polling IDs (litellm_poll_*): Returns cumulative cached content from background responses
    - Provider response IDs: Passes through to provider API
    
    Follows the OpenAI Responses API spec: https://platform.openai.com/docs/api-reference/responses/get
    
    ```bash
    # Get polling response
    curl -X GET http://localhost:4000/v1/responses/litellm_poll_abc123 \
    -H "Authorization: Bearer $LITELLM_MASTER_KEY"
    
    # Get provider response
    curl -X GET http://localhost:4000/v1/responses/resp_abc123 \
    -H "Authorization: Bearer $LITELLM_MASTER_KEY"
    ```
    """
    from litellm.proxy.proxy_server import (
        general_settings,
        llm_router,
        proxy_config,
        proxy_logging_obj,
        read_request_body,
        redis_usage_cache,
        select_data_generator,
        user_api_base,
        user_max_tokens,
        user_model,
        user_request_timeout,
        user_temperature,
        version,
    )
    from litellm.proxy.response_polling.polling_handler import ResponsePollingHandler

    # Check if this is a polling ID
    if ResponsePollingHandler.is_polling_id(response_id):
        # Handle polling response
        if not redis_usage_cache:
            raise HTTPException(
                status_code=500,
                detail="Redis cache not configured. Polling requires Redis.",
            )

        polling_handler: Final = ResponsePollingHandler(redis_cache=redis_usage_cache)

        # Get current state from cache
        state: Final = await polling_handler.get_state(response_id)

        if not state:
            raise HTTPException(
                status_code=404,
                detail=f"Polling response {response_id} not found or expired",
            )

        # Return the whole state directly (OpenAI Response object format)
        # https://platform.openai.com/docs/api-reference/responses/object
        return state

    # Normal provider response flow
    data: Final = await read_request_body(request=request)
    data["response_id"] = response_id
    processor: Final = ProxyBaseLLMRequestProcessing(data=data)
    try:
        return await processor.base_process_llm_request(
            request=request,
            fastapi_response=fastapi_response,
            user_api_key_dict=user_api_key_dict,
            route_type="aget_responses",
            proxy_logging_obj=proxy_logging_obj,
            llm_router=llm_router,
            general_settings=general_settings,
            proxy_config=proxy_config,
            select_data_generator=select_data_generator,
            model=None,
            user_model=user_model,
            user_temperature=user_temperature,
            user_request_timeout=user_request_timeout,
            user_max_tokens=user_max_tokens,
            user_api_base=user_api_base,
            version=version,
        )
    except Exception as e:
        raise await processor.handle_llm_api_exception(
            e=e,
            user_api_key_dict=user_api_key_dict,
            proxy_logging_obj=proxy_logging_obj,
            version=version,
        )


@router.delete(
    "/v1/responses/{response_id}",
    dependencies=[Depends(user_api_key_auth)],
    tags=["responses"],
    responses=DELETE_RESPONSE_SCHEMAS,
)
@router.delete(
    "/responses/{response_id}",
    dependencies=[Depends(user_api_key_auth)],
    tags=["responses"],
    responses=DELETE_RESPONSE_SCHEMAS,
)
@router.delete(
    "/openai/v1/responses/{response_id}",
    dependencies=[Depends(user_api_key_auth)],
    tags=["responses"],
    responses=DELETE_RESPONSE_SCHEMAS,
)
async def delete_response(
    response_id: str,
    request: Request,
    fastapi_response: Response,
    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
    """
    Delete a response by ID.
    
    Supports both:
    - Polling IDs (litellm_poll_*): Deletes from Redis cache
    - Provider response IDs: Passes through to provider API
    
    Follows the OpenAI Responses API spec: https://platform.openai.com/docs/api-reference/responses/delete
    
    ```bash
    curl -X DELETE http://localhost:4000/v1/responses/resp_abc123 \
    -H "Authorization: Bearer $LITELLM_MASTER_KEY"
    ```
    """
    from litellm.proxy.proxy_server import (
        general_settings,
        llm_router,
        proxy_config,
        proxy_logging_obj,
        read_request_body,
        redis_usage_cache,
        select_data_generator,
        user_api_base,
        user_max_tokens,
        user_model,
        user_request_timeout,
        user_temperature,
        version,
    )
    from litellm.proxy.response_polling.polling_handler import ResponsePollingHandler

    # Check if this is a polling ID
    if ResponsePollingHandler.is_polling_id(response_id):
        # Handle polling response deletion
        if not redis_usage_cache:
            raise HTTPException(status_code=500, detail="Redis cache not configured.")

        polling_handler: Final = ResponsePollingHandler(redis_cache=redis_usage_cache)

        # Get state to verify access
        state: Final = await polling_handler.get_state(response_id)

        if not state:
            raise HTTPException(status_code=404, detail=f"Polling response {response_id} not found")

        # Delete from cache
        success: Final = await polling_handler.delete_polling(response_id)

        if success:
            return DeleteResponseResult(id=response_id, object="response", deleted=True)
        else:
            raise HTTPException(status_code=500, detail="Failed to delete polling response")

    # Normal provider response flow
    data: Final = await read_request_body(request=request)
    data["response_id"] = response_id
    processor: Final = ProxyBaseLLMRequestProcessing(data=data)
    try:
        return await processor.base_process_llm_request(
            request=request,
            fastapi_response=fastapi_response,
            user_api_key_dict=user_api_key_dict,
            route_type="adelete_responses",
            proxy_logging_obj=proxy_logging_obj,
            llm_router=llm_router,
            general_settings=general_settings,
            proxy_config=proxy_config,
            select_data_generator=select_data_generator,
            model=None,
            user_model=user_model,
            user_temperature=user_temperature,
            user_request_timeout=user_request_timeout,
            user_max_tokens=user_max_tokens,
            user_api_base=user_api_base,
            version=version,
        )
    except Exception as e:
        raise await processor.handle_llm_api_exception(
            e=e,
            user_api_key_dict=user_api_key_dict,
            proxy_logging_obj=proxy_logging_obj,
            version=version,
        )


@router.get(
    "/v1/responses/{response_id}/input_items",
    dependencies=[Depends(user_api_key_auth)],
    tags=["responses"],
    responses=RESPONSE_ITEM_LIST_SCHEMAS,
)
@router.get(
    "/responses/{response_id}/input_items",
    dependencies=[Depends(user_api_key_auth)],
    tags=["responses"],
    responses=RESPONSE_ITEM_LIST_SCHEMAS,
)
@router.get(
    "/openai/v1/responses/{response_id}/input_items",
    dependencies=[Depends(user_api_key_auth)],
    tags=["responses"],
    responses=RESPONSE_ITEM_LIST_SCHEMAS,
)
async def get_response_input_items(
    response_id: str,
    request: Request,
    fastapi_response: Response,
    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
    """List input items for a response."""
    from litellm.proxy.proxy_server import (
        general_settings,
        llm_router,
        proxy_config,
        proxy_logging_obj,
        read_request_body,
        select_data_generator,
        user_api_base,
        user_max_tokens,
        user_model,
        user_request_timeout,
        user_temperature,
        version,
    )

    data: Final = await read_request_body(request=request)
    data["response_id"] = response_id
    processor: Final = ProxyBaseLLMRequestProcessing(data=data)
    try:
        return await processor.base_process_llm_request(
            request=request,
            fastapi_response=fastapi_response,
            user_api_key_dict=user_api_key_dict,
            route_type="alist_input_items",
            proxy_logging_obj=proxy_logging_obj,
            llm_router=llm_router,
            general_settings=general_settings,
            proxy_config=proxy_config,
            select_data_generator=select_data_generator,
            model=None,
            user_model=user_model,
            user_temperature=user_temperature,
            user_request_timeout=user_request_timeout,
            user_max_tokens=user_max_tokens,
            user_api_base=user_api_base,
            version=version,
        )
    except Exception as e:
        raise await processor.handle_llm_api_exception(
            e=e,
            user_api_key_dict=user_api_key_dict,
            proxy_logging_obj=proxy_logging_obj,
            version=version,
        )


@router.post(
    "/v1/responses/compact",
    dependencies=[Depends(user_api_key_auth)],
    tags=["responses"],
)
@router.post(
    "/responses/compact",
    dependencies=[Depends(user_api_key_auth)],
    tags=["responses"],
)
@router.post(
    "/openai/v1/responses/compact",
    dependencies=[Depends(user_api_key_auth)],
    tags=["responses"],
)
async def compact_response(
    request: Request,
    fastapi_response: Response,
    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
    """
    Compact a response by running a compaction pass over a conversation.
    
    Returns encrypted, opaque items that can be used to reduce context size.
    
    Follows the OpenAI Responses API spec: https://platform.openai.com/docs/api-reference/responses/compact
    
    ```bash
    curl -X POST http://localhost:4000/v1/responses/compact \
    -H "Content-Type: application/json" \
    -H "Authorization: Bearer $LITELLM_MASTER_KEY" \
    -d '{
        "model": "gpt-4o",
        "input": [{"role": "user", "content": "Hello"}]
    }'
    ```
    """
    from litellm.proxy.proxy_server import (
        general_settings,
        llm_router,
        proxy_config,
        proxy_logging_obj,
        read_request_body,
        select_data_generator,
        user_api_base,
        user_max_tokens,
        user_model,
        user_request_timeout,
        user_temperature,
        version,
    )

    data: Final = await read_request_body(request=request)
    processor: Final = ProxyBaseLLMRequestProcessing(data=data)
    try:
        return await processor.base_process_llm_request(
            request=request,
            fastapi_response=fastapi_response,
            user_api_key_dict=user_api_key_dict,
            route_type="acompact_responses",
            proxy_logging_obj=proxy_logging_obj,
            llm_router=llm_router,
            general_settings=general_settings,
            proxy_config=proxy_config,
            select_data_generator=select_data_generator,
            model=None,
            user_model=user_model,
            user_temperature=user_temperature,
            user_request_timeout=user_request_timeout,
            user_max_tokens=user_max_tokens,
            user_api_base=user_api_base,
            version=version,
        )
    except Exception as e:
        raise await processor.handle_llm_api_exception(
            e=e,
            user_api_key_dict=user_api_key_dict,
            proxy_logging_obj=proxy_logging_obj,
            version=version,
        )


class _ResponsesApiErrorDetail(TypedDict):
    message: ReadOnly[str]
    type: ReadOnly[str]
    param: ReadOnly[str | None]
    code: ReadOnly[str | None]


class _ResponsesApiErrorBody(TypedDict):
    error: ReadOnly[_ResponsesApiErrorDetail]


class _ResponsesInputTokensResult(TokenCountSourceFields):
    object: ReadOnly[str]
    input_tokens: ReadOnly[int]


class _TokenCountPayload(TypedDict):
    model: ReadOnly[str]
    messages: ReadOnly[tuple[Mapping[str, object], ...]]
    tools: ReadOnly[object]


class _TokenCounter(Protocol):
    def __call__(self, request: TokenCountRequest, call_endpoint: bool) -> Awaitable[TokenCountResponse]: ...


def _proxy_token_counter() -> _TokenCounter:
    from litellm.proxy.proxy_server import token_counter

    return token_counter


_token_counter_dep: Final = Depends(_proxy_token_counter)


def _responses_invalid_request_response(message: str, param: str | None, code: str | None) -> JSONResponse:
    body: Final[_ResponsesApiErrorBody] = {
        "error": {
            "message": message,
            "type": "invalid_request_error",
            "param": param,
            "code": code,
        }
    }
    return JSONResponse(status_code=400, content=body)


def _missing_responses_param_response(param: str) -> JSONResponse:
    return _responses_invalid_request_response(
        message=f"Missing required parameter: '{param}'.",
        param=param,
        code="missing_required_parameter",
    )


def _responses_input_as_token_count_messages(
    input_value: str | ResponseInputParam,
    instructions: str | None,
) -> tuple[Mapping[str, object], ...]:
    from litellm.responses.litellm_completion_transformation.transformation import (
        LiteLLMCompletionResponsesConfig,
    )

    request_params: Final[ResponsesAPIOptionalRequestParams] = {"instructions": instructions}
    transformed: Final = LiteLLMCompletionResponsesConfig.transform_responses_api_input_to_messages(
        input=input_value,
        responses_api_request=request_params,
    )
    return tuple(
        message if isinstance(message, dict) else message.model_dump(exclude_none=True) for message in transformed
    )


@router.post(
    "/v1/responses/input_tokens",
    dependencies=(_user_api_key_auth_dep,),
    tags=_RESPONSES_TAGS,
)
@router.post(
    "/responses/input_tokens",
    dependencies=(_user_api_key_auth_dep,),
    tags=_RESPONSES_TAGS,
)
@router.post(
    "/openai/v1/responses/input_tokens",
    dependencies=(_user_api_key_auth_dep,),
    tags=_RESPONSES_TAGS,
)
async def responses_input_tokens(
    request: Request,
    token_counter: _TokenCounter = _token_counter_dep,
):
    """
    Count the input tokens of a Responses API request without calling the model.

    Follows the OpenAI Responses API spec: https://platform.openai.com/docs/api-reference/responses/input-tokens

    ```bash
    curl -X POST http://localhost:4000/v1/responses/input_tokens \
    -H "Content-Type: application/json" \
    -H "Authorization: Bearer $LITELLM_MASTER_KEY" \
    -d '{
        "model": "gpt-4o",
        "input": "Hello, how are you?"
    }'
    ```

    Returns: `{"object": "response.input_tokens", "input_tokens": <count>}`
    """
    data: Final = await read_request_body(request=request)
    model_name: Final = data.get("model")
    input_value: Final = data.get("input")
    if not isinstance(model_name, str) or not model_name:
        return _missing_responses_param_response("model")
    if input_value is None:
        return _missing_responses_param_response("input")
    if isinstance(input_value, (str, list)) and not input_value:
        return _responses_invalid_request_response(
            message="""One of "input" or "previous_response_id" or 'prompt' or 'conversation' must be provided.""",
            param=None,
            code="missing_required_parameter",
        )

    try:
        payload: Final[_TokenCountPayload] = {
            "model": model_name,
            "messages": _responses_input_as_token_count_messages(
                input_value=input_value,
                instructions=data.get("instructions"),
            ),
            "tools": data.get("tools"),
        }
        token_request: Final = TokenCountRequest.model_validate(payload)
    except Exception as e:
        return _responses_invalid_request_response(
            message=f"Invalid request for token counting: {e}", param=None, code=None
        )

    token_response: Final = await token_counter(request=token_request, call_endpoint=True)
    result: Final[_ResponsesInputTokensResult] = {
        "object": "response.input_tokens",
        "input_tokens": token_response.total_tokens,
        **count_source_fields(token_response),
    }
    return result


@router.post(
    "/v1/responses/{response_id}/cancel",
    dependencies=[Depends(user_api_key_auth)],
    tags=["responses"],
)
@router.post(
    "/responses/{response_id}/cancel",
    dependencies=[Depends(user_api_key_auth)],
    tags=["responses"],
)
@router.post(
    "/openai/v1/responses/{response_id}/cancel",
    dependencies=[Depends(user_api_key_auth)],
    tags=["responses"],
)
async def cancel_response(
    response_id: str,
    request: Request,
    fastapi_response: Response,
    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
    """
    Cancel a response by ID.
    
    Supports both:
    - Polling IDs (litellm_poll_*): Cancels background response and updates status in Redis
    - Provider response IDs: Passes through to provider API
    
    Follows the OpenAI Responses API spec: https://platform.openai.com/docs/api-reference/responses/cancel
    
    ```bash
    # Cancel polling response
    curl -X POST http://localhost:4000/v1/responses/litellm_poll_abc123/cancel \
    -H "Authorization: Bearer $LITELLM_MASTER_KEY"
    
    # Cancel provider response
    curl -X POST http://localhost:4000/v1/responses/resp_abc123/cancel \
    -H "Authorization: Bearer $LITELLM_MASTER_KEY"
    ```
    """
    from litellm.proxy.proxy_server import (
        general_settings,
        llm_router,
        proxy_config,
        proxy_logging_obj,
        read_request_body,
        redis_usage_cache,
        select_data_generator,
        user_api_base,
        user_max_tokens,
        user_model,
        user_request_timeout,
        user_temperature,
        version,
    )
    from litellm.proxy.response_polling.polling_handler import ResponsePollingHandler

    # Check if this is a polling ID
    if ResponsePollingHandler.is_polling_id(response_id):
        # Handle polling response cancellation
        if not redis_usage_cache:
            raise HTTPException(status_code=500, detail="Redis cache not configured.")

        polling_handler: Final = ResponsePollingHandler(redis_cache=redis_usage_cache)

        # Get current state to verify it exists
        state: Final = await polling_handler.get_state(response_id)

        if not state:
            raise HTTPException(status_code=404, detail=f"Polling response {response_id} not found")

        # Cancel the polling response (sets status to "cancelled")
        success: Final = await polling_handler.cancel_polling(response_id)

        if success:
            # Fetch the updated state with cancelled status
            updated_state: Final = await polling_handler.get_state(response_id)

            # Return the whole state directly (now with status="cancelled")
            return updated_state
        else:
            raise HTTPException(status_code=500, detail="Failed to cancel polling response")

    # Normal provider response flow
    data: Final = await read_request_body(request=request)
    data["response_id"] = response_id
    processor: Final = ProxyBaseLLMRequestProcessing(data=data)
    try:
        return await processor.base_process_llm_request(
            request=request,
            fastapi_response=fastapi_response,
            user_api_key_dict=user_api_key_dict,
            route_type="acancel_responses",
            proxy_logging_obj=proxy_logging_obj,
            llm_router=llm_router,
            general_settings=general_settings,
            proxy_config=proxy_config,
            select_data_generator=select_data_generator,
            model=None,
            user_model=user_model,
            user_temperature=user_temperature,
            user_request_timeout=user_request_timeout,
            user_max_tokens=user_max_tokens,
            user_api_base=user_api_base,
            version=version,
        )
    except Exception as e:
        raise await processor.handle_llm_api_exception(
            e=e,
            user_api_key_dict=user_api_key_dict,
            proxy_logging_obj=proxy_logging_obj,
            version=version,
        )


def _resolve_responses_ws_session_limit_seconds() -> float:
    from litellm.proxy.proxy_server import general_settings

    field: Final = "responses_websocket_session_limit_seconds"
    raw: Final = _RESPONSES_WS_CONFIG_VALUE_ADAPTER.validate_python(general_settings.get(field))
    try:
        return ConfigGeneralSettings.model_validate(
            {} if raw is None else {field: raw}
        ).responses_websocket_session_limit_seconds
    except ValidationError as e:
        default: Final = DEFAULT_RESPONSES_WEBSOCKET_SESSION_LIMIT_SECONDS
        verbose_proxy_logger.warning("invalid general_settings.%s=%r (%s); using default %ss", field, raw, e, default)
        return default


async def _read_ws_model_from_first_frame(
    websocket: WebSocket,
    query_model: str | None = None,
) -> tuple[str, str] | None:
    """Read the first WS frame and return (model, raw_message), or None on error.

    Sends an appropriate error frame and closes the socket before returning None.
    The session-duration deadline is enforced by the caller, not here.
    """
    try:
        first_message: Final = await websocket.receive_text()
    except WebSocketDisconnect:
        return None
    except Exception:
        verbose_proxy_logger.exception("Responses WebSocket error reading first message")
        await websocket.close(code=1011, reason="Internal server error")
        return None

    try:
        first_event: Final = json.loads(first_message)
    except json.JSONDecodeError:
        await websocket.send_text(
            json.dumps(
                {
                    "type": "error",
                    "error": {
                        "type": "invalid_request_error",
                        "message": "First message is not valid JSON.",
                    },
                }
            )
        )
        await websocket.close(code=1008, reason="Invalid JSON in first message")
        return None

    if not isinstance(first_event, dict) or first_event.get("type") != "response.create":
        await websocket.send_text(
            json.dumps(
                {
                    "type": "error",
                    "error": {
                        "type": "invalid_request_error",
                        "message": "First message must be a response.create JSON object.",
                    },
                }
            )
        )
        await websocket.close(code=1008, reason="Invalid first message")
        return None

    model: Final = query_model or _extract_model_from_first_ws_event(first_event)
    if not model:
        await websocket.send_text(
            json.dumps(
                {
                    "type": "error",
                    "error": {
                        "type": "invalid_request_error",
                        "message": "No model provided. Supply ?model=<model> in the URL or include 'model' in the first response.create event.",
                    },
                }
            )
        )
        await websocket.close(code=1008, reason="No model provided")
        return None

    return model, first_message


def _extract_model_from_first_ws_event(first_event: object) -> str | None:
    """Extract model from a response.create WS event, handling flat and nested formats.

    Flat:   {"type": "response.create", "model": "gpt-4o", ...}
    Nested: {"type": "response.create", "response": {"model": "gpt-4o", ...}}
    """
    if not isinstance(first_event, dict):
        return None
    nested: Final = first_event.get("response")
    return (nested.get("model") if isinstance(nested, dict) else None) or first_event.get("model")


class _ResponseCreateRoutingHints(LiteLLMBaseModel):
    model_config = ConfigDict(extra="ignore", frozen=True)

    input: str | Sequence[object] | None = None
    previous_response_id: str | None = None
    response: "_ResponseCreateRoutingHints | None" = None


def _routing_hints_from_first_ws_frame(first_message: str) -> Mapping[str, object]:
    try:
        frame: Final = _ResponseCreateRoutingHints.model_validate_json(first_message)
    except ValidationError:
        return EMPTY_MAPPING
    nested: Final = frame.response or frame
    hints: Final = {
        "input": frame.input if nested.input is None else nested.input,
        "previous_response_id": (
            frame.previous_response_id if nested.previous_response_id is None else nested.previous_response_id
        ),
    }
    return MappingProxyType({key: value for key, value in hints.items() if value is not None})


def _responses_ws_failure_frame(failure: Exception) -> str:
    raw_status: Final = getattr(failure, "status_code", None)
    status: Final = raw_status if isinstance(raw_status, int) and not isinstance(raw_status, bool) else 500
    error_type: Final = (
        "rate_limit_exceeded" if status == 429 else "invalid_request_error" if 400 <= status < 500 else "server_error"
    )
    return json.dumps({"type": "error", "status": status, "error": {"type": error_type, "message": str(failure)}})


async def _enforce_responses_ws_first_frame_model_auth(
    request: Request,
    model: str,
    user_api_key_dict: UserAPIKeyAuth,
    llm_router: "Router | None",
) -> None:
    from litellm.proxy.auth.user_api_key_auth import (
        enforce_key_and_fallback_model_access,
        run_centralized_common_checks,
    )
    from litellm.proxy.proxy_server import (
        general_settings,
        llm_model_list,
        master_key,
        user_custom_auth,
    )

    request_data: Final = {"model": model}
    route: Final = request.scope.get("path") or "/v1/responses"
    if master_key is None and not (
        general_settings.get("enable_jwt_auth", False)
        or general_settings.get("enable_oauth2_auth", False)
        or general_settings.get("enable_oauth2_proxy_auth", False)
    ):
        return
    if user_custom_auth is not None and not general_settings.get("custom_auth_run_common_checks", False):
        return
    await enforce_key_and_fallback_model_access(
        valid_token=user_api_key_dict,
        request_data=request_data,
        route=route,
        request=request,
        llm_model_list=llm_model_list,
        llm_router=llm_router,
    )
    await run_centralized_common_checks(
        user_api_key_auth_obj=user_api_key_dict,
        request=request,
        request_data=request_data,
        route=route,
    )


async def _responses_websocket_session(
    websocket: WebSocket,
    model: str | None,
    user_api_key_dict: UserAPIKeyAuth,
) -> None:
    from litellm.proxy.proxy_server import (
        general_settings,
        llm_router,
        proxy_config,
        proxy_logging_obj,
        user_api_base,
        user_max_tokens,
        user_model,
        user_request_timeout,
        user_temperature,
        version,
    )
    from litellm.proxy.route_llm_request import route_request

    result: Final = await _read_ws_model_from_first_frame(websocket, query_model=model)
    if result is None:
        return
    resolved_model, first_message = result

    data: dict[str, object] = {
        "model": resolved_model,
        "websocket": websocket,
        "first_message": first_message,
    }

    # Construct a synthetic Request for pre-call processing
    headers_list: Final = list(websocket.scope.get("headers") or [])
    scope: Final[dict[str, object]] = {
        "type": "http",
        "method": "POST",
        "path": "/v1/responses",
        "headers": headers_list,
    }
    request: Final = Request(scope=scope)
    request._url = websocket.url  # pyright: ignore[reportPrivateUsage]  # Starlette WebSocket URL storage

    _body_bytes: Final = json.dumps({"model": resolved_model}).encode()

    async def return_body():
        return _body_bytes

    request.body = return_body

    # Phase 1: pre-call processing (auth, guardrails, rate limits)
    base_llm_response_processor: Final = ProxyBaseLLMRequestProcessing(data=data)
    try:
        if not model:
            await _enforce_responses_ws_first_frame_model_auth(
                request=request,
                model=resolved_model,
                user_api_key_dict=user_api_key_dict,
                llm_router=llm_router,
            )
        (
            data,
            litellm_logging_obj,
        ) = await base_llm_response_processor.common_processing_pre_call_logic(
            request=request,
            general_settings=general_settings,
            user_api_key_dict=user_api_key_dict,
            version=version,
            proxy_logging_obj=proxy_logging_obj,
            proxy_config=proxy_config,
            user_model=user_model,
            user_temperature=user_temperature,
            user_request_timeout=user_request_timeout,
            user_max_tokens=user_max_tokens,
            user_api_base=user_api_base,
            model=resolved_model,
            route_type="_aresponses_websocket",
        )
    except Exception as e:
        verbose_proxy_logger.exception("Responses WebSocket pre-call error")
        try:
            await websocket.send_text(
                json.dumps(
                    {
                        "type": "error",
                        "error": {
                            "type": "invalid_request_error",
                            "message": str(e),
                        },
                    }
                )
            )
        except Exception:
            pass
        await websocket.close(code=1008, reason="Pre-call error")
        return

    routed_data: Final = dict(
        data, user_api_key_dict=user_api_key_dict, **_routing_hints_from_first_ws_frame(first_message)
    )
    # Phase 2: route to upstream provider
    try:
        llm_call: Final = await route_request(
            data=routed_data,
            route_type="_aresponses_websocket",
            llm_router=llm_router,
            user_model=user_model,
        )
        failure: Final = await llm_call
        if isinstance(failure, Exception):
            await proxy_logging_obj.post_call_failure_hook(
                user_api_key_dict=user_api_key_dict,
                original_exception=failure,
                request_data=routed_data,
            )
    except Exception as e:
        verbose_proxy_logger.exception("Responses WebSocket error")
        with contextlib.suppress(Exception):
            await websocket.send_text(_responses_ws_failure_frame(e))
        await proxy_logging_obj.post_call_failure_hook(
            user_api_key_dict=user_api_key_dict,
            original_exception=e,
            request_data=routed_data,
        )
        await websocket.close(code=1011, reason="Internal server error")


@router.websocket("/v1/responses")
@router.websocket("/responses")
async def responses_websocket_endpoint(
    websocket: WebSocket,
    model: str | None = fastapi.Query(None, description="The model to use for the responses WebSocket session."),
    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth_websocket),
):
    """
    Responses API WebSocket mode endpoint.

    Keeps a persistent WebSocket connection for response.create events,
    enabling lower-latency agentic workflows with many tool-call round trips.

    Follows the OpenAI split: the bearer token is validated at connection time
    (before accept); the model is resolved either from the ?model= query param
    or from the first response.create frame, whichever is present.

    The session is bounded by a lifetime measured from accept, configured via
    general_settings.responses_websocket_session_limit_seconds (60-7200,
    default 3600). There is no separate first-frame deadline, so
    pre-established connections may sit idle until their first response.create.

    See: https://developers.openai.com/api/docs/guides/websocket-mode/
    """
    # Accept the WebSocket handshake. Key was already validated by the Depends
    # above; we can safely accept regardless of whether ?model= was supplied.
    requested_protocols: Final = [
        p.strip() for p in (websocket.headers.get("sec-websocket-protocol") or "").split(",") if p.strip()
    ]
    accept_kwargs: Final[dict] = {}
    if requested_protocols:
        accept_kwargs["subprotocol"] = requested_protocols[0]
    await websocket.accept(**accept_kwargs)

    limit_seconds: Final = _resolve_responses_ws_session_limit_seconds()
    session_task: Final = asyncio.ensure_future(
        _responses_websocket_session(websocket=websocket, model=model, user_api_key_dict=user_api_key_dict)
    )
    try:
        await asyncio.wait_for(asyncio.shield(session_task), timeout=limit_seconds)
    except asyncio.TimeoutError:
        verbose_proxy_logger.info("Responses WebSocket closed: session duration limit reached")
        session_task.cancel()
        with contextlib.suppress(Exception):
            await websocket.close(code=1000, reason="Session duration limit reached")
    finally:
        if not session_task.done():
            session_task.cancel()
            with contextlib.suppress(asyncio.CancelledError, Exception):
                await session_task