litellm/litellm/responses/main.py

import asyncio
import contextvars
import json
from collections.abc import Coroutine, Generator, Iterable, Mapping, Sequence
from contextlib import contextmanager
from dataclasses import dataclass
from functools import partial
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, Literal, NoReturn, Optional, TypeAlias, cast

import httpx
from pydantic import BaseModel, TypeAdapter, ValidationError
from typing_extensions import assert_never

import litellm
from litellm._logging import verbose_logger
from litellm.completion_extras.litellm_responses_transformation.transformation import (
    LiteLLMResponsesTransformationHandler,
)
from litellm.constants import DEFAULT_CHAT_COMPLETION_PARAM_VALUES, request_timeout
from litellm.integrations.anthropic_cache_control_hook import CARRY_UNMATCHED_MESSAGE_POINTS
from litellm.litellm_core_utils.asyncify import run_async_function
from litellm.litellm_core_utils.core_helpers import normalize_drop_params
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.litellm_core_utils.prompt_templates.common_utils import (
    update_responses_input_with_model_file_ids,
    update_responses_tools_with_model_file_ids,
)
from litellm.litellm_core_utils.provider_affinity import add_provider_affinity_header
from litellm.llms.base_llm.responses.transformation import BaseResponsesAPIConfig
from litellm.llms.custom_httpx.llm_http_handler import BaseLLMHTTPHandler
from litellm.llms.openai_like.responses.transformation import OpenAILikeResponsesConfig
from litellm.responses.litellm_completion_transformation.handler import (
    LiteLLMCompletionTransformationHandler,
)
from litellm.responses.mcp.request_context import MCPRequestContext
from litellm.responses.utils import ResponsesAPIRequestUtils
from litellm.types.llms.openai import (
    PromptObject,
    Reasoning,
    ResponseIncludable,
    ResponseInputParam,
    ResponsesAPIOptionalRequestParams,
    ResponsesAPIResponse,
    ToolChoice,
    ToolParam,
)

# Handle ResponseText import with fallback
if TYPE_CHECKING:
    from litellm.types.llms.openai import ResponseText
else:
    ResponseText = str  # Fallback for ResponseText import
from litellm.litellm_core_utils.get_litellm_params import get_litellm_params
from litellm.llms.openai.data_residency import infer_openai_data_residency
from litellm.secret_managers.main import get_secret_str
from litellm.types.responses.main import *
from litellm.types.responses.streaming_websocket import ResponsesWebSocketRequestDefaults
from litellm.types.router import GenericLiteLLMParams
from litellm.types.utils import all_litellm_params
from litellm.utils import (
    ProviderConfigManager,
    client,
)

if TYPE_CHECKING:
    from fastapi import WebSocket
    from mcp.types import Tool as MCPTool
else:
    MCPTool = Any

from .streaming_iterator import BaseResponsesAPIStreamingIterator

__all__ = (
    "acancel_responses",
    "acompact_responses",
    "adelete_responses",
    "aget_responses",
    "alist_input_items",
    "aresponses",
    "aresponses_api_with_mcp",
    "cancel_responses",
    "compact_responses",
    "delete_responses",
    "get_responses",
    "list_input_items",
    "mock_responses_api_response",
    "responses",
)

####### ENVIRONMENT VARIABLES ###################
# Initialize any necessary instances or variables here
base_llm_http_handler = BaseLLMHTTPHandler()
litellm_completion_transformation_handler: Final = LiteLLMCompletionTransformationHandler()
#################################################


def _has_file_search_tool(tools: Iterable[Mapping[str, object]] | None) -> bool:
    """Return True if any tool in the list has type 'file_search'."""
    if not tools:
        return False
    return any(isinstance(t, dict) and t.get("type") == "file_search" for t in tools)


def mock_responses_api_response(
    mock_response: str = "In a peaceful grove beneath a silver moon, a unicorn named Lumina discovered a hidden pool that reflected the stars. As she dipped her horn into the water, the pool began to shimmer, revealing a pathway to a magical realm of endless night skies. Filled with wonder, Lumina whispered a wish for all who dream to find their own hidden magic, and as she glanced back, her hoofprints sparkled like stardust.",
):
    return ResponsesAPIResponse(
        **{
            "id": "resp_67ccd2bed1ec8190b14f964abc0542670bb6a6b452d3795b",
            "object": "response",
            "created_at": 1741476542,
            "status": "completed",
            "error": None,
            "incomplete_details": None,
            "instructions": None,
            "max_output_tokens": None,
            "model": "gpt-4.1-2025-04-14",
            "output": [
                {
                    "type": "message",
                    "id": "msg_67ccd2bf17f0819081ff3bb2cf6508e60bb6a6b452d3795b",
                    "status": "completed",
                    "role": "assistant",
                    "content": [
                        {
                            "type": "output_text",
                            "text": mock_response,
                            "annotations": [],
                        }
                    ],
                }
            ],
            "parallel_tool_calls": True,
            "previous_response_id": None,
            "reasoning": {"effort": None, "summary": None},
            "store": True,
            "temperature": 1.0,
            "text": {"format": {"type": "text"}},
            "tool_choice": "auto",
            "tools": [],
            "top_p": 1.0,
            "truncation": "disabled",
            "usage": {
                "input_tokens": 36,
                "input_tokens_details": {"cached_tokens": 0},
                "output_tokens": 87,
                "output_tokens_details": {},
                "total_tokens": 123,
            },
            "user": None,
            "metadata": {},
        }
    )


async def aresponses_api_with_mcp(
    input: str | ResponseInputParam,
    model: str,
    include: list[ResponseIncludable] | None = None,
    instructions: str | None = None,
    max_output_tokens: int | None = None,
    prompt: PromptObject | None = None,
    metadata: dict[str, object] | None = None,
    parallel_tool_calls: bool | None = None,
    previous_response_id: str | None = None,
    reasoning: Reasoning | None = None,
    store: bool | None = None,
    background: bool | None = None,
    stream: bool | None = None,
    temperature: float | None = None,
    text: Optional["ResponseText"] = None,
    tool_choice: ToolChoice | None = None,
    tools: Iterable[ToolParam] | None = None,
    top_p: float | None = None,
    truncation: Literal["auto", "disabled"] | None = None,
    user: str | None = None,
    # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
    # The extra values given here take precedence over values defined on the client or passed to this method.
    extra_headers: dict[str, object] | None = None,
    extra_query: dict[str, object] | None = None,
    extra_body: dict[str, object] | None = None,
    timeout: float | httpx.Timeout | None = None,
    # LiteLLM specific params,
    custom_llm_provider: str | None = None,
    **kwargs,
) -> ResponsesAPIResponse | BaseResponsesAPIStreamingIterator:
    """
    Async version of responses API with MCP integration.

    When MCP tools with server_url="litellm_proxy" are provided, this function will:
    1. Get available tools from the MCP server manager
    2. Insert the tools into the messages/input
    3. Call the standard responses API
    4. If require_approval="never" and tool calls are returned, automatically execute them
    """
    from litellm.responses.mcp.litellm_proxy_mcp_handler import (
        LiteLLM_Proxy_MCP_Handler,
    )

    # Parse MCP tools and separate from other tools
    (
        mcp_tools_with_litellm_proxy,
        other_tools,
    ) = await LiteLLM_Proxy_MCP_Handler._split_mcp_tools(tools)

    # Process MCP tools through the complete pipeline (fetch + filter + deduplicate + transform)
    # Extract user_api_key_auth from litellm_metadata (where it's added by add_user_api_key_auth_to_request_metadata)
    user_api_key_auth = kwargs.get("user_api_key_auth") or kwargs.get("litellm_metadata", {}).get("user_api_key_auth")

    # Extract MCP auth headers from request (for dynamic auth when fetching tools)
    mcp_auth_header: str | None = None
    mcp_server_auth_headers: dict[str, dict[str, str]] | None = None
    secret_fields = kwargs.get("secret_fields")
    if secret_fields and isinstance(secret_fields, dict):
        (
            mcp_auth_header,
            mcp_server_auth_headers,
            _,
            _,
        ) = ResponsesAPIRequestUtils.extract_mcp_headers_from_request(secret_fields=secret_fields, tools=tools)

    # Get original MCP tools (for events) and OpenAI tools (for LLM) by reusing existing methods
    (
        original_mcp_tools,
        tool_server_map,
    ) = await LiteLLM_Proxy_MCP_Handler._process_mcp_tools_without_openai_transform(
        user_api_key_auth=user_api_key_auth,
        mcp_tools_with_litellm_proxy=mcp_tools_with_litellm_proxy,
        litellm_trace_id=kwargs.get("litellm_trace_id"),
        mcp_auth_header=mcp_auth_header,
        mcp_server_auth_headers=mcp_server_auth_headers,
        request_tags=LiteLLM_Proxy_MCP_Handler._get_parent_request_tags(kwargs),
    )
    openai_tools: Final = LiteLLM_Proxy_MCP_Handler._transform_mcp_tools_to_openai(original_mcp_tools)

    # Combine with other tools
    all_tools: Final = openai_tools + other_tools if (openai_tools or other_tools) else None

    # Prepare call parameters for reuse
    call_params: Final = {
        "include": include,
        "instructions": instructions,
        "max_output_tokens": max_output_tokens,
        "prompt": prompt,
        "metadata": metadata,
        "parallel_tool_calls": parallel_tool_calls,
        "reasoning": reasoning,
        "store": store,
        "background": background,
        "stream": stream,
        "temperature": temperature,
        "text": text,
        "tool_choice": tool_choice,
        "top_p": top_p,
        "truncation": truncation,
        "user": user,
        "extra_headers": extra_headers,
        "extra_query": extra_query,
        "extra_body": extra_body,
        "timeout": timeout,
        "custom_llm_provider": custom_llm_provider,
        **kwargs,
        "_skip_mcp_handler": True,
    }

    # Handle MCP streaming if requested
    if stream and mcp_tools_with_litellm_proxy:
        # Generate MCP discovery events using the already processed tools
        from litellm._uuid import uuid
        from litellm.responses.mcp.mcp_streaming_iterator import (
            create_mcp_list_tools_events,
        )

        base_item_id: Final = f"mcp_{uuid.uuid4().hex[:8]}"
        mcp_discovery_events: Final = await create_mcp_list_tools_events(
            mcp_tools_with_litellm_proxy=mcp_tools_with_litellm_proxy,
            user_api_key_auth=user_api_key_auth,
            base_item_id=base_item_id,
            pre_processed_mcp_tools=original_mcp_tools,
        )

        mcp_streaming_response: Final = LiteLLM_Proxy_MCP_Handler._create_mcp_streaming_response(
            input=input,
            model=model,
            all_tools=all_tools,
            mcp_tools_with_litellm_proxy=mcp_tools_with_litellm_proxy,
            mcp_discovery_events=mcp_discovery_events,
            call_params=call_params,
            previous_response_id=previous_response_id,
            tool_server_map=tool_server_map,
            **kwargs,
        )
        await mcp_streaming_response._create_initial_response_iterator()
        if mcp_streaming_response._initial_creation_error is not None:
            raise mcp_streaming_response._initial_creation_error
        return mcp_streaming_response

    # Determine if we should auto-execute tools
    should_auto_execute = bool(mcp_tools_with_litellm_proxy) and LiteLLM_Proxy_MCP_Handler._should_auto_execute_tools(
        mcp_tools_with_litellm_proxy=mcp_tools_with_litellm_proxy
    )

    # Prepare parameters for the initial call
    initial_call_params: Final = LiteLLM_Proxy_MCP_Handler._prepare_initial_call_params(
        call_params=call_params, should_auto_execute=should_auto_execute
    )

    #########################################################
    # Make initial response API call
    #########################################################
    response: Final = await aresponses(
        input=input,
        model=model,
        tools=all_tools,
        previous_response_id=previous_response_id,
        **initial_call_params,
    )

    verbose_logger.debug("Initial response %s", response)

    #########################################################
    # Auto-Execute Tools Handling
    # If auto-execute tools is True, then we need to execute the tool calls
    #########################################################
    if should_auto_execute and isinstance(response, ResponsesAPIResponse):
        tool_calls: Final = LiteLLM_Proxy_MCP_Handler._extract_tool_calls_from_response(response=response)

        if tool_calls:
            user_api_key_auth = kwargs.get("litellm_metadata", {}).get("user_api_key_auth")

            # Extract MCP auth headers from the request to pass to MCP server
            secret_fields = kwargs.get("secret_fields")
            (
                mcp_auth_header,
                mcp_server_auth_headers,
                oauth2_headers,
                raw_headers_from_request,
            ) = ResponsesAPIRequestUtils.extract_mcp_headers_from_request(
                secret_fields=secret_fields,
                tools=tools,
            )

            tool_results: Final = await LiteLLM_Proxy_MCP_Handler._execute_tool_calls(
                tool_server_map=tool_server_map,
                tool_calls=tool_calls,
                user_api_key_auth=user_api_key_auth,
                mcp_auth_header=mcp_auth_header,
                mcp_server_auth_headers=mcp_server_auth_headers,
                oauth2_headers=oauth2_headers,
                raw_headers=raw_headers_from_request,
                litellm_call_id=kwargs.get("litellm_call_id"),
                litellm_trace_id=kwargs.get("litellm_trace_id"),
                request_tags=LiteLLM_Proxy_MCP_Handler._get_parent_request_tags(kwargs),
                guardrail_context=MCPRequestContext.resolve_guardrail_context(
                    MappingProxyType({**kwargs, "metadata": metadata, "model": model})
                ),
            )

            if tool_results:
                persistence_disabled: Final = LiteLLM_Proxy_MCP_Handler._is_persistence_disabled(call_params)

                follow_up_input: Final = LiteLLM_Proxy_MCP_Handler._create_follow_up_input(
                    response=response,
                    tool_results=tool_results,
                    original_input=input,
                    preserve_reasoning=persistence_disabled,
                )

                # Prepare parameters for follow-up call (restores original stream setting)
                follow_up_call_params: Final = LiteLLM_Proxy_MCP_Handler._prepare_follow_up_call_params(
                    call_params=call_params, original_stream_setting=stream or False
                )

                # Create tool execution events for streaming if needed
                tool_execution_events = []
                if stream:
                    tool_execution_events = LiteLLM_Proxy_MCP_Handler._create_tool_execution_events(
                        tool_calls=tool_calls, tool_results=tool_results
                    )

                final_response = await LiteLLM_Proxy_MCP_Handler._make_follow_up_call(
                    follow_up_input=follow_up_input,
                    model=model,
                    all_tools=all_tools,
                    response_id=previous_response_id if persistence_disabled else response.id,
                    **follow_up_call_params,
                )

                # If streaming and we have tool execution events, wrap the response
                if (
                    stream
                    and tool_execution_events
                    and (hasattr(final_response, "__aiter__") or hasattr(final_response, "__iter__"))
                ):
                    from litellm.responses.mcp.mcp_streaming_iterator import (
                        MCPEnhancedStreamingIterator,
                    )

                    final_response = MCPEnhancedStreamingIterator(
                        tool_server_map=tool_server_map,
                        base_iterator=final_response,
                        mcp_events=tool_execution_events,
                        user_api_key_auth=user_api_key_auth,
                    )

                # Add custom output elements to the final response (for non-streaming)
                elif isinstance(final_response, ResponsesAPIResponse):
                    # Fetch MCP tools again for output elements (without OpenAI transformation)
                    (
                        mcp_tools_for_output,
                        _,
                    ) = await LiteLLM_Proxy_MCP_Handler._process_mcp_tools_without_openai_transform(
                        user_api_key_auth=user_api_key_auth,
                        mcp_tools_with_litellm_proxy=mcp_tools_with_litellm_proxy,
                        mcp_auth_header=mcp_auth_header,
                        mcp_server_auth_headers=mcp_server_auth_headers,
                        request_tags=LiteLLM_Proxy_MCP_Handler._get_parent_request_tags(kwargs),
                    )
                    final_response = LiteLLM_Proxy_MCP_Handler._add_mcp_output_elements_to_response(
                        response=final_response,
                        mcp_tools_fetched=mcp_tools_for_output,
                        tool_results=tool_results,
                    )
                return final_response

    return response


def _bridges_to_chat_completions(
    responses_api_provider_config: BaseResponsesAPIConfig | None, use_chat_completions_api: bool
) -> bool:
    """Whether the request reaches its provider as a chat completion, not a Responses call."""
    return responses_api_provider_config is None or use_chat_completions_api is True


_RESPONSES_ONLY_REQUEST_FIELDS_NEVER_BRIDGED: Final = frozenset({"client_metadata"})


def _bridge_kwargs(
    kwargs: Mapping[str, object],
    responses_api_provider_config: BaseResponsesAPIConfig | None,
    allowed_openai_params: Sequence[str] | None,
) -> Mapping[str, object]:
    if responses_api_provider_config is None:
        return MappingProxyType(
            {key: value for key, value in kwargs.items() if key not in _RESPONSES_ONLY_REQUEST_FIELDS_NEVER_BRIDGED}
        )
    forwarded_keys: Final = frozenset(
        (
            *litellm.OPENAI_CHAT_COMPLETION_PARAMS,
            *DEFAULT_CHAT_COMPLETION_PARAM_VALUES,
            *all_litellm_params,
            *GenericLiteLLMParams.model_fields,
            *(allowed_openai_params or ()),
        )
    ).difference(_RESPONSES_ONLY_REQUEST_FIELDS_NEVER_BRIDGED)
    return MappingProxyType({key: value for key, value in kwargs.items() if key in forwarded_keys})


_ResponsesCompatibilityFailure: TypeAlias = Literal["encrypted_task_unsupported"]


def _encrypted_task_support_failure(
    responses_api_provider_config: BaseResponsesAPIConfig | None, use_chat_completions_api: bool
) -> _ResponsesCompatibilityFailure | None:
    if (
        responses_api_provider_config is None
        or _bridges_to_chat_completions(responses_api_provider_config, use_chat_completions_api)
        or not responses_api_provider_config.supports_encrypted_agent_messages()
    ):
        return "encrypted_task_unsupported"
    return None


def _raise_responses_compatibility_failure(
    failure: _ResponsesCompatibilityFailure, model: str, custom_llm_provider: str | None
) -> NoReturn:
    match failure:
        case "encrypted_task_unsupported":
            raise litellm.exception_type(
                model=model,
                custom_llm_provider=custom_llm_provider,
                original_exception=ValueError(
                    "Encrypted task classification requires a compatible native Responses deployment"
                ),
            )
        case _:
            assert_never(failure)


def _deployment_passes_through_responses(model_info: object) -> bool:
    """Whether ``model_info.supported_endpoints`` opts the deployment into native ``{api_base}/responses``."""
    if not isinstance(model_info, dict):
        return False
    supported_endpoints: Final = model_info.get("supported_endpoints")
    return isinstance(supported_endpoints, (list, tuple)) and "/v1/responses" in supported_endpoints


def _deployment_model_info_after_prompt_swap(
    requested_provider: str | None, resolved_provider: str | None, model_info: object
) -> object:
    """Deployment metadata only describes the upstream while the prompt manager keeps its provider."""
    return model_info if resolved_provider == requested_provider else None


@dataclass(frozen=True, slots=True)
class _AsyncPromptManagementOutcome:
    merged_optional_params: Mapping[str, object]
    deployment_model_info: object


def _resolve_responses_api_provider_config(
    model: str, custom_llm_provider: str, model_info: object, api_base: str | None
) -> BaseResponsesAPIConfig | None:
    provider_config: Final = ProviderConfigManager.get_provider_responses_api_config(
        model=model, provider=custom_llm_provider, api_base=api_base
    )
    if provider_config is not None or not _deployment_passes_through_responses(model_info):
        return provider_config
    return OpenAILikeResponsesConfig()


def _api_base_kwarg(kwargs: Mapping[str, object]) -> str | None:
    api_base: Final = kwargs.get("api_base")
    return api_base if isinstance(api_base, str) else None


def _will_bridge_to_chat_completions(
    model: str,
    custom_llm_provider: str | None,
    use_chat_completions_api: bool,
    model_info: object,
    api_base: str | None,
) -> bool:
    """``_bridges_to_chat_completions`` for callers running before the provider config is resolved.

    Resolving the config is a pure lookup, so this asks the same question the dispatch
    asks rather than restating its condition. Both callers resolve the provider before
    this runs, so the only way to be wrong is a prompt manager that moves the model
    across the bridge boundary, which would leave the deferred points to a pass that
    never comes.
    """
    normalized_model: Final = _normalize_openai_chat_completions_responses_model(model)
    if custom_llm_provider is None:
        return True
    return _bridges_to_chat_completions(
        _resolve_responses_api_provider_config(normalized_model[0], custom_llm_provider, model_info, api_base),
        use_chat_completions_api or normalized_model[1],
    )


@contextmanager
def _prompt_management_sees_a_provisional_message_list(
    kwargs: dict[str, Any],  # mutable-ok: the signal is read and popped out of the caller's own kwargs
    bridged: bool,
) -> Generator[None, None]:
    """Tell the cache-control hook that this layer's messages are not the ones sent upstream.

    A Responses request keeps its system prompt in ``instructions``, which only becomes a
    system message when the chat-completion bridge builds one, so a role-targeted point
    is placed by the bridge's pass rather than this one.

    Only raised for a request that will be bridged. A provider serving Responses natively
    gets no second pass, so this layer is the last one that can place anything and handing
    a point forward there drops it.
    """
    if not bridged:
        yield
        return
    kwargs[CARRY_UNMATCHED_MESSAGE_POINTS] = True
    try:
        yield
    finally:
        kwargs.pop(CARRY_UNMATCHED_MESSAGE_POINTS, None)


@client
async def aresponses(
    input: str | ResponseInputParam,
    model: str,
    include: list[ResponseIncludable] | None = None,
    instructions: str | None = None,
    max_output_tokens: int | None = None,
    prompt: PromptObject | None = None,
    metadata: dict[str, object] | None = None,
    parallel_tool_calls: bool | None = None,
    previous_response_id: str | None = None,
    reasoning: Reasoning | None = None,
    store: bool | None = None,
    background: bool | None = None,
    stream: bool | None = None,
    temperature: float | None = None,
    text: Optional["ResponseText"] = None,
    text_format: type["BaseModel"] | dict | None = None,
    tool_choice: ToolChoice | None = None,
    tools: Iterable[ToolParam] | None = None,
    top_p: float | None = None,
    truncation: Literal["auto", "disabled"] | None = None,
    user: str | None = None,
    service_tier: str | None = None,
    safety_identifier: str | None = None,
    # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
    # The extra values given here take precedence over values defined on the client or passed to this method.
    extra_headers: dict[str, object] | None = None,
    extra_query: dict[str, object] | None = None,
    extra_body: dict[str, object] | None = None,
    timeout: float | httpx.Timeout | None = None,
    # LiteLLM specific params,
    custom_llm_provider: str | None = None,
    **kwargs,
) -> ResponsesAPIResponse | BaseResponsesAPIStreamingIterator:
    """
    Async: Handles responses API requests by reusing the synchronous function
    """
    local_vars: Final = locals()
    try:
        loop: Final = asyncio.get_event_loop()
        kwargs["aresponses"] = True

        # Convert text_format to text parameter if provided
        text = ResponsesAPIRequestUtils.convert_text_format_to_text_param(text_format=text_format, text=text)
        if text is not None:
            # Update local_vars to include the converted text parameter
            local_vars["text"] = text

        # get custom llm provider so we can use this for mapping exceptions
        if custom_llm_provider is None:
            _, custom_llm_provider, _, _ = litellm.get_llm_provider(
                model=model, api_base=local_vars.get("base_url", None)
            )
            # Update local_vars with detected provider (fixes #19782)
            local_vars["custom_llm_provider"] = custom_llm_provider

        #########################################################
        # ASYNC PROMPT MANAGEMENT
        # Run the async hook here so async-only prompt loggers are honoured.
        # Then pop prompt_id from kwargs so the sync responses() path does NOT
        # re-run the hook (which would double-prepend template messages).
        # Pass merged_optional_params via an internal kwarg so responses()
        # can apply them to local_vars without re-invoking the hook.
        #########################################################
        litellm_logging_obj: Final = kwargs.get("litellm_logging_obj", None)
        prompt_id: Final = cast(str | None, kwargs.get("prompt_id", None))
        prompt_variables: Final = cast(dict | None, kwargs.get("prompt_variables", None))
        original_model: Final = model

        if isinstance(
            litellm_logging_obj, LiteLLMLoggingObj
        ) and litellm_logging_obj.should_run_prompt_management_hooks(prompt_id=prompt_id, non_default_params=kwargs):
            client_input: Final = ResponsesAPIRequestUtils.responses_input_to_chat_messages(input)
            with _prompt_management_sees_a_provisional_message_list(
                kwargs,
                bridged=_will_bridge_to_chat_completions(
                    model,
                    custom_llm_provider,
                    bool(kwargs.get("use_chat_completions_api")),
                    kwargs.get("model_info"),
                    _api_base_kwarg(kwargs),
                ),
            ):
                (
                    model,
                    merged_input,
                    merged_optional_params,
                ) = await litellm_logging_obj.async_get_chat_completion_prompt(
                    model=model,
                    messages=client_input,
                    non_default_params=kwargs,
                    prompt_id=prompt_id,
                    prompt_variables=prompt_variables,
                    prompt_label=kwargs.get("prompt_label", None),
                    prompt_version=kwargs.get("prompt_version", None),
                    request_kwargs=kwargs,
                )
            input = cast(
                str | ResponseInputParam,
                ResponsesAPIRequestUtils.merge_prompt_management_input(
                    original_input=input,
                    client_input=client_input,
                    merged_input=merged_input,
                ),
            )
            requested_provider: Final = custom_llm_provider
            if model != original_model:
                custom_llm_provider = _resolve_prompt_swapped_provider(
                    original_model=original_model,
                    swapped_model=model,
                    custom_llm_provider=custom_llm_provider,
                    kwargs=kwargs,
                    prompt_id=prompt_id,
                )
            kwargs.pop("prompt_id", None)
            kwargs["_async_prompt_merged_params"] = _AsyncPromptManagementOutcome(
                merged_optional_params=merged_optional_params,
                deployment_model_info=_deployment_model_info_after_prompt_swap(
                    requested_provider, custom_llm_provider, kwargs.get("model_info")
                ),
            )

        func: Final = partial(
            responses,
            input=input,
            model=model,
            include=include,
            instructions=instructions,
            max_output_tokens=max_output_tokens,
            prompt=prompt,
            metadata=metadata,
            parallel_tool_calls=parallel_tool_calls,
            previous_response_id=previous_response_id,
            reasoning=reasoning,
            store=store,
            background=background,
            stream=stream,
            temperature=temperature,
            text=text,
            tool_choice=tool_choice,
            tools=tools,
            top_p=top_p,
            truncation=truncation,
            user=user,
            extra_headers=extra_headers,
            extra_query=extra_query,
            extra_body=extra_body,
            timeout=timeout,
            custom_llm_provider=custom_llm_provider,
            service_tier=service_tier,
            safety_identifier=safety_identifier,
            **kwargs,
        )

        ctx: Final = contextvars.copy_context()
        func_with_context: Final = partial(ctx.run, func)
        init_response: Final = await loop.run_in_executor(None, func_with_context)

        if asyncio.iscoroutine(init_response):
            response = await init_response
        else:
            response = init_response

        # Update the responses_api_response_id with the model_id
        if isinstance(response, ResponsesAPIResponse):
            response = ResponsesAPIRequestUtils._update_responses_api_response_id_with_model_id(
                responses_api_response=response,
                litellm_metadata=kwargs.get("litellm_metadata", {}),
                custom_llm_provider=custom_llm_provider,
            )
            # Stamp custom_llm_provider so callbacks can identify the provider
            # (mirrors litellm/main.py:1371 for chat completions)
            response._hidden_params["custom_llm_provider"] = custom_llm_provider

        if response is None:
            raise ValueError(f"Got an unexpected None response from the Responses API: {response}")

        return response
    except Exception as e:
        raise litellm.exception_type(
            model=model,
            custom_llm_provider=custom_llm_provider,
            original_exception=e,
            completion_kwargs=local_vars,
            extra_kwargs=kwargs,
        )


def _resolve_prompt_swapped_provider(
    original_model: str,
    swapped_model: str,
    custom_llm_provider: str | None,
    kwargs: Mapping[str, object],
    prompt_id: str | None,
) -> str:
    swapped_provider: Final = litellm.get_llm_provider(model=swapped_model)[1]
    if kwargs.get("api_key") is None and kwargs.get("api_base") is None:
        return swapped_provider
    try:
        original_provider: Final = custom_llm_provider or litellm.get_llm_provider(model=original_model)[1]
    except litellm.BadRequestError:
        return swapped_provider
    if swapped_provider == original_provider:
        return swapped_provider
    raise litellm.BadRequestError(
        message=(
            f"prompt_id '{prompt_id}' swaps model '{original_model}' -> '{swapped_model}', which changes the "
            f"provider from '{original_provider}' to '{swapped_provider}' after credentials for "
            f"'{original_provider}' were already resolved. Refusing to send them to '{swapped_provider}'. "
            "Point the request at a model whose provider matches the prompt's metadata.model, or set "
            "ignore_prompt_manager_model on the prompt to keep the requested model."
        ),
        model=swapped_model,
        llm_provider=swapped_provider,
    )


def _apply_prompt_management_to_responses_call(
    input: str | ResponseInputParam,
    model: str,
    custom_llm_provider: str | None,
    litellm_logging_obj: LiteLLMLoggingObj | None,
    kwargs: dict[str, Any],
    local_vars: dict[str, object],
    use_chat_completions_api: bool,
) -> tuple[str | ResponseInputParam, str, str | None, object]:
    """Returns the prompt-managed input, model and provider, plus the deployment metadata that still
    describes the upstream (``None`` once the prompt manager moved the request to another provider)."""
    async_outcome: Final[_AsyncPromptManagementOutcome | None] = kwargs.pop("_async_prompt_merged_params", None)
    if async_outcome is not None:
        for key, value in async_outcome.merged_optional_params.items():
            local_vars[key] = value
        return input, model, custom_llm_provider, async_outcome.deployment_model_info

    prompt_id: Final = cast(str | None, kwargs.get("prompt_id", None))
    prompt_variables: Final = cast(dict | None, kwargs.get("prompt_variables", None))
    original_model: Final = model

    client_input: Final = ResponsesAPIRequestUtils.responses_input_to_chat_messages(input)

    if isinstance(litellm_logging_obj, LiteLLMLoggingObj) and litellm_logging_obj.should_run_prompt_management_hooks(
        prompt_id=prompt_id, non_default_params=kwargs
    ):
        with _prompt_management_sees_a_provisional_message_list(
            kwargs,
            bridged=_will_bridge_to_chat_completions(
                model,
                custom_llm_provider,
                use_chat_completions_api,
                kwargs.get("model_info"),
                _api_base_kwarg(kwargs),
            ),
        ):
            (
                model,
                merged_input,
                merged_optional_params,
            ) = litellm_logging_obj.get_chat_completion_prompt(
                model=model,
                messages=client_input,
                non_default_params=kwargs,
                prompt_id=prompt_id,
                prompt_variables=prompt_variables,
                prompt_label=kwargs.get("prompt_label", None),
                prompt_version=kwargs.get("prompt_version", None),
                request_kwargs=kwargs,
            )
        input = cast(
            str | ResponseInputParam,
            ResponsesAPIRequestUtils.merge_prompt_management_input(
                original_input=input,
                client_input=client_input,
                merged_input=merged_input,
            ),
        )
        local_vars["input"] = input
        local_vars["model"] = model
        resolved_provider: Final = (
            custom_llm_provider
            if model == original_model
            else _resolve_prompt_swapped_provider(
                original_model=original_model,
                swapped_model=model,
                custom_llm_provider=custom_llm_provider,
                kwargs=kwargs,
                prompt_id=prompt_id,
            )
        )
        local_vars["custom_llm_provider"] = resolved_provider
        for key, value in merged_optional_params.items():
            local_vars[key] = value
        return (
            input,
            model,
            resolved_provider,
            _deployment_model_info_after_prompt_swap(custom_llm_provider, resolved_provider, kwargs.get("model_info")),
        )

    return input, model, custom_llm_provider, kwargs.get("model_info")


# Opt-in via model id (mirrors the `responses/` prefix pattern on chat completions).
_OPENAI_CHAT_COMPLETIONS_RESPONSES_MODEL_PREFIX: Final = "openai/chat_completions/"


def _normalize_openai_chat_completions_responses_model(model: str) -> tuple[str, bool]:
    """
    Strip `openai/chat_completions/<name>` → `openai/<name>` and return True when the
    prefix was applied (same effect as use_chat_completions_api=True).
    """
    if not model.startswith(_OPENAI_CHAT_COMPLETIONS_RESPONSES_MODEL_PREFIX):
        return model, False
    remainder: Final = model[len(_OPENAI_CHAT_COMPLETIONS_RESPONSES_MODEL_PREFIX) :]
    if not remainder:
        return model, False
    return f"openai/{remainder}", True


def _pop_use_chat_completions_api_kw(kwargs: dict[str, object]) -> bool:
    """Pop use_chat_completions_api; True when the chat-completions bridge is requested."""
    use_cc: Final = kwargs.pop("use_chat_completions_api", None)
    return bool(use_cc)


_RESPONSES_ROUTING_PREFIX: Final = "responses/"


def _strip_responses_routing_prefix(model: str) -> str:
    if not model.startswith(_RESPONSES_ROUTING_PREFIX):
        return model
    return model[len(_RESPONSES_ROUTING_PREFIX) :]


def _resolve_model_provider_for_responses(
    model: str,
    custom_llm_provider: str | None,
    litellm_params: GenericLiteLLMParams,
    local_vars: dict[str, object],
) -> tuple[str, str | None]:
    if custom_llm_provider is not None and not litellm_params.custom_llm_provider:
        litellm_params.custom_llm_provider = custom_llm_provider
    (
        provider_model,
        resolved_provider,
        dynamic_api_key,
        dynamic_api_base,
    ) = litellm.get_llm_provider(
        model=model,
        litellm_params=litellm_params,
    )
    local_vars["custom_llm_provider"] = resolved_provider
    if dynamic_api_key is not None:
        litellm_params.api_key = dynamic_api_key
    if dynamic_api_base is not None:
        litellm_params.api_base = dynamic_api_base
    return _strip_responses_routing_prefix(provider_model), resolved_provider


def _apply_managed_file_id_mapping(
    input: str | ResponseInputParam,
    tools: Iterable[ToolParam] | None,
    kwargs: dict[str, Any],
    local_vars: dict[str, object],
) -> tuple[str | ResponseInputParam, Iterable[ToolParam] | None]:
    model_file_id_mapping: Final = kwargs.get("model_file_id_mapping")
    model_info_id = kwargs.get("model_info", {}).get("id") if isinstance(kwargs.get("model_info"), dict) else None

    input = cast(
        str | ResponseInputParam,
        update_responses_input_with_model_file_ids(
            input=input,
            model_id=model_info_id,
            model_file_id_mapping=model_file_id_mapping,
        ),
    )
    local_vars["input"] = input

    if tools:
        tools = cast(
            Iterable[ToolParam] | None,
            update_responses_tools_with_model_file_ids(
                tools=cast(list[dict[str, object]] | None, tools),
                model_id=model_info_id,
                model_file_id_mapping=model_file_id_mapping,
            ),
        )
        local_vars["tools"] = tools

    return input, tools


def _responses_try_dispatch_mcp_gateway(
    *,
    tools: Iterable[ToolParam] | None,
    input: str | ResponseInputParam,
    model: str,
    include: list[ResponseIncludable] | None,
    instructions: str | None,
    max_output_tokens: int | None,
    prompt: PromptObject | None,
    metadata: dict[str, object] | None,
    parallel_tool_calls: bool | None,
    previous_response_id: str | None,
    reasoning: Reasoning | None,
    store: bool | None,
    background: bool | None,
    stream: bool | None,
    temperature: float | None,
    text: Any,
    tool_choice: ToolChoice | None,
    top_p: float | None,
    truncation: Literal["auto", "disabled"] | None,
    user: str | None,
    extra_headers: dict[str, object] | None,
    extra_query: dict[str, object] | None,
    extra_body: dict[str, object] | None,
    timeout: float | httpx.Timeout | None,
    custom_llm_provider: str | None,
    kwargs: dict[str, object],
    _is_async: bool,
    skip_mcp_handler: bool,
) -> Any | None:
    """Return a response when MCP gateway handles the call; otherwise None."""
    from litellm.responses.mcp.litellm_proxy_mcp_handler import (
        LiteLLM_Proxy_MCP_Handler,
    )

    if skip_mcp_handler or not LiteLLM_Proxy_MCP_Handler._should_use_litellm_mcp_gateway(tools=tools):
        return None
    mcp_call_kwargs: Final = {
        "input": input,
        "model": model,
        "include": include,
        "instructions": instructions,
        "max_output_tokens": max_output_tokens,
        "prompt": prompt,
        "metadata": metadata,
        "parallel_tool_calls": parallel_tool_calls,
        "previous_response_id": previous_response_id,
        "reasoning": reasoning,
        "store": store,
        "background": background,
        "stream": stream,
        "temperature": temperature,
        "text": text,
        "tool_choice": tool_choice,
        "tools": tools,
        "top_p": top_p,
        "truncation": truncation,
        "user": user,
        "extra_headers": extra_headers,
        "extra_query": extra_query,
        "extra_body": extra_body,
        "timeout": timeout,
        "custom_llm_provider": custom_llm_provider,
        **kwargs,
    }
    if _is_async:
        return aresponses_api_with_mcp(**mcp_call_kwargs)
    return run_async_function(aresponses_api_with_mcp, **mcp_call_kwargs)


def _responses_try_dispatch_emulated_file_search(
    *,
    tools: Iterable[ToolParam] | None,
    input: str | ResponseInputParam,
    model: str,
    responses_api_provider_config: BaseResponsesAPIConfig | None,
    use_chat_completions_api: bool,
    include: list[ResponseIncludable] | None,
    instructions: str | None,
    max_output_tokens: int | None,
    prompt: PromptObject | None,
    metadata: dict[str, object] | None,
    parallel_tool_calls: bool | None,
    previous_response_id: str | None,
    reasoning: Reasoning | None,
    store: bool | None,
    background: bool | None,
    stream: bool | None,
    temperature: float | None,
    text: Any,
    tool_choice: ToolChoice | None,
    top_p: float | None,
    truncation: Literal["auto", "disabled"] | None,
    user: str | None,
    service_tier: str | None,
    safety_identifier: str | None,
    text_format: type[BaseModel] | dict | None,
    allowed_openai_params: list[str] | None,
    extra_headers: dict[str, object] | None,
    extra_query: dict[str, object] | None,
    extra_body: dict[str, object] | None,
    timeout: float | httpx.Timeout | None,
    custom_llm_provider: str | None,
    kwargs: dict[str, object],
    _is_async: bool,
) -> ResponsesAPIResponse | Coroutine[object, object, ResponsesAPIResponse] | None:
    """Return a response when emulated file_search handles the call; otherwise None."""
    if not _has_file_search_tool(tools) or not (
        responses_api_provider_config is None
        or use_chat_completions_api is True
        or not responses_api_provider_config.supports_native_file_search()
    ):
        return None
    from litellm.responses.file_search.emulated_handler import (
        aresponses_with_emulated_file_search,
    )

    _internal_skip: Final = {"litellm_call_id", "aresponses"}
    emulated_kwargs: Final = {
        "include": include,
        "instructions": instructions,
        "max_output_tokens": max_output_tokens,
        "prompt": prompt,
        "metadata": metadata,
        "parallel_tool_calls": parallel_tool_calls,
        "previous_response_id": previous_response_id,
        "reasoning": reasoning,
        "store": store,
        "background": background,
        "stream": stream,
        "temperature": temperature,
        "text": text,
        "tool_choice": tool_choice,
        "top_p": top_p,
        "truncation": truncation,
        "user": user,
        "service_tier": service_tier,
        "safety_identifier": safety_identifier,
        "text_format": text_format,
        "allowed_openai_params": allowed_openai_params,
        "extra_headers": extra_headers,
        "extra_query": extra_query,
        "extra_body": extra_body,
        "timeout": timeout,
        "custom_llm_provider": custom_llm_provider,
        **({"use_chat_completions_api": True} if use_chat_completions_api else {}),
        **{k: v for k, v in kwargs.items() if k not in _internal_skip},
    }
    if _is_async:
        return aresponses_with_emulated_file_search(input=input, model=model, tools=tools, **emulated_kwargs)
    return run_async_function(
        aresponses_with_emulated_file_search,
        input=input,
        model=model,
        tools=tools,
        **emulated_kwargs,
    )


@client
def responses(
    input: str | ResponseInputParam,
    model: str,
    include: list[ResponseIncludable] | None = None,
    instructions: str | None = None,
    max_output_tokens: int | None = None,
    prompt: PromptObject | None = None,
    metadata: dict[str, object] | None = None,
    parallel_tool_calls: bool | None = None,
    previous_response_id: str | None = None,
    reasoning: Reasoning | None = None,
    store: bool | None = None,
    background: bool | None = None,
    stream: bool | None = None,
    temperature: float | None = None,
    text: Optional["ResponseText"] = None,
    text_format: type["BaseModel"] | dict | None = None,
    tool_choice: ToolChoice | None = None,
    tools: Iterable[ToolParam] | None = None,
    top_p: float | None = None,
    truncation: Literal["auto", "disabled"] | None = None,
    user: str | None = None,
    service_tier: str | None = None,
    safety_identifier: str | None = None,
    # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
    # The extra values given here take precedence over values defined on the client or passed to this method.
    extra_headers: dict[str, object] | None = None,
    extra_query: dict[str, object] | None = None,
    extra_body: dict[str, object] | None = None,
    timeout: float | httpx.Timeout | None = None,
    # LiteLLM specific params,
    allowed_openai_params: list[str] | None = None,
    custom_llm_provider: str | None = None,
    **kwargs,
):
    """
    Synchronous version of the Responses API.
    Uses the synchronous HTTP handler to make requests.
    """
    local_vars: Final = locals()

    try:
        litellm_logging_obj: Final[LiteLLMLoggingObj] = kwargs.get("litellm_logging_obj")
        litellm_call_id: Final[str | None] = kwargs.get("litellm_call_id", None)
        _is_async: Final = kwargs.pop("aresponses", False) is True
        skip_mcp_handler: Final = kwargs.pop("_skip_mcp_handler", False)
        require_encrypted_task_support: Final = kwargs.pop("_require_encrypted_task_support", False) is True
        use_chat_completions_api = _pop_use_chat_completions_api_kw(kwargs)

        client_headers: Final = kwargs.get("headers")
        extra_headers = ResponsesAPIRequestUtils.merge_client_forwarded_headers(
            extra_headers=extra_headers,
            client_headers=client_headers if isinstance(client_headers, dict) else None,
        )
        local_vars["extra_headers"] = extra_headers

        # Convert text_format to text parameter if provided
        text = ResponsesAPIRequestUtils.convert_text_format_to_text_param(text_format=text_format, text=text)
        if text is not None:
            # Update local_vars to include the converted text parameter
            local_vars["text"] = text

        #########################################################
        # PROMPT MANAGEMENT
        # If aresponses() already ran the async hook, it pops prompt_id and
        # passes the result via _async_prompt_merged_params — apply those
        # directly and skip the sync hook to avoid double-merging.
        #########################################################
        _stripped_model, _from_chat_completions_prefix = _normalize_openai_chat_completions_responses_model(model)
        model = _stripped_model
        local_vars["model"] = model
        use_chat_completions_api = use_chat_completions_api or _from_chat_completions_prefix

        if custom_llm_provider is None:
            _, custom_llm_provider, _, _ = litellm.get_llm_provider(
                model=model, api_base=local_vars.get("base_url", None)
            )
            local_vars["custom_llm_provider"] = custom_llm_provider

        input, model, custom_llm_provider, deployment_model_info = _apply_prompt_management_to_responses_call(
            input=input,
            model=model,
            custom_llm_provider=custom_llm_provider,
            litellm_logging_obj=litellm_logging_obj,
            kwargs=kwargs,
            local_vars=local_vars,
            use_chat_completions_api=use_chat_completions_api,
        )

        # get llm provider logic
        litellm_params: Final = GenericLiteLLMParams(**kwargs)
        try:
            effective_extra_headers: Final = (
                add_provider_affinity_header(
                    headers=extra_headers or MappingProxyType({}),
                    litellm_params=MappingProxyType(
                        {
                            "provider_affinity_header": litellm_params.provider_affinity_header,
                            "litellm_session_id": kwargs.get("litellm_session_id"),
                            "session_id": kwargs.get("session_id"),
                            "metadata": metadata,
                            "litellm_metadata": kwargs.get("litellm_metadata"),
                        }
                    ),
                )
                if litellm_params.provider_affinity_header is not None
                else extra_headers
            )
        except ValueError as affinity_error:
            raise litellm.BadRequestError(
                message=str(affinity_error), model=model, llm_provider=custom_llm_provider
            ) from affinity_error

        #########################################################
        # MOCK RESPONSE LOGIC
        #########################################################
        if litellm_params.mock_response and isinstance(litellm_params.mock_response, str):
            return mock_responses_api_response(mock_response=litellm_params.mock_response)

        model, custom_llm_provider = _resolve_model_provider_for_responses(
            model=model,
            custom_llm_provider=custom_llm_provider,
            litellm_params=litellm_params,
            local_vars=local_vars,
        )

        #########################################################
        # Update input and tools with provider-specific file IDs if managed files are used
        #########################################################
        input, tools = _apply_managed_file_id_mapping(input=input, tools=tools, kwargs=kwargs, local_vars=local_vars)

        #########################################################
        # Native MCP Responses API
        #########################################################
        _mcp_dispatch: Final = _responses_try_dispatch_mcp_gateway(
            tools=tools,
            input=input,
            model=model,
            include=include,
            instructions=instructions,
            max_output_tokens=max_output_tokens,
            prompt=prompt,
            metadata=metadata,
            parallel_tool_calls=parallel_tool_calls,
            previous_response_id=previous_response_id,
            reasoning=reasoning,
            store=store,
            background=background,
            stream=stream,
            temperature=temperature,
            text=text,
            tool_choice=tool_choice,
            top_p=top_p,
            truncation=truncation,
            user=user,
            extra_headers=effective_extra_headers,
            extra_query=extra_query,
            extra_body=extra_body,
            timeout=timeout,
            custom_llm_provider=custom_llm_provider,
            kwargs=kwargs,
            _is_async=_is_async,
            skip_mcp_handler=skip_mcp_handler,
        )
        if _mcp_dispatch is not None:
            return _mcp_dispatch

        # get provider config
        responses_api_provider_config: BaseResponsesAPIConfig | None
        if custom_llm_provider is None:
            responses_api_provider_config = None
        else:
            responses_api_provider_config = _resolve_responses_api_provider_config(
                model, custom_llm_provider, deployment_model_info, litellm_params.api_base
            )

        if (
            require_encrypted_task_support
            and (
                compatibility_failure := _encrypted_task_support_failure(
                    responses_api_provider_config, use_chat_completions_api
                )
            )
            is not None
        ):
            _raise_responses_compatibility_failure(compatibility_failure, model, custom_llm_provider)

        local_vars.update(kwargs)
        # Map reasoning_effort (from litellm_params/proxy config) to reasoning when not set
        if reasoning is None and "reasoning_effort" in local_vars:
            _mapped = LiteLLMResponsesTransformationHandler()._map_reasoning_effort(local_vars.pop("reasoning_effort"))
            if _mapped is not None:
                reasoning = _mapped
                local_vars["reasoning"] = _mapped
        # Get ResponsesAPIOptionalRequestParams with only valid parameters
        response_api_optional_params: Final[ResponsesAPIOptionalRequestParams] = (
            ResponsesAPIRequestUtils.get_requested_response_api_optional_param(local_vars)
        )

        _file_search_dispatch: Final = _responses_try_dispatch_emulated_file_search(
            tools=tools,
            input=input,
            model=model,
            responses_api_provider_config=responses_api_provider_config,
            use_chat_completions_api=use_chat_completions_api,
            include=include,
            instructions=instructions,
            max_output_tokens=max_output_tokens,
            prompt=prompt,
            metadata=metadata,
            parallel_tool_calls=parallel_tool_calls,
            previous_response_id=previous_response_id,
            reasoning=reasoning,
            store=store,
            background=background,
            stream=stream,
            temperature=temperature,
            text=text,
            tool_choice=tool_choice,
            top_p=top_p,
            truncation=truncation,
            user=user,
            service_tier=service_tier,
            safety_identifier=safety_identifier,
            text_format=text_format,
            allowed_openai_params=allowed_openai_params,
            extra_headers=effective_extra_headers,
            extra_query=extra_query,
            extra_body=extra_body,
            timeout=timeout,
            custom_llm_provider=custom_llm_provider,
            kwargs=kwargs,
            _is_async=_is_async,
        )
        if _file_search_dispatch is not None:
            return _file_search_dispatch

        if _bridges_to_chat_completions(responses_api_provider_config, use_chat_completions_api):
            bridge_kwargs: Final = _bridge_kwargs(kwargs, responses_api_provider_config, allowed_openai_params)
            return litellm_completion_transformation_handler.response_api_handler(
                model=model,
                input=input,
                responses_api_request=response_api_optional_params,
                custom_llm_provider=custom_llm_provider,
                _is_async=_is_async,
                stream=stream,
                extra_headers=effective_extra_headers,
                extra_body=extra_body,
                timeout=timeout if timeout is not None else request_timeout,
                allowed_openai_params=allowed_openai_params,
                **bridge_kwargs,
            )

        # Get optional parameters for the responses API
        request_drop_params: Final = kwargs.get("drop_params")
        responses_api_request_params: Final[dict] = ResponsesAPIRequestUtils.get_optional_params_responses_api(
            model=model,
            responses_api_provider_config=responses_api_provider_config,
            response_api_optional_params=response_api_optional_params,
            allowed_openai_params=allowed_openai_params,
            drop_params=normalize_drop_params(request_drop_params),
        )

        litellm_logging_obj.update_from_kwargs(
            kwargs=kwargs,
            model=model,
            user=user,
            optional_params=dict(responses_api_request_params),
            litellm_params={
                **responses_api_request_params,
                "aresponses": _is_async,
                "litellm_call_id": litellm_call_id,
                "model_info": kwargs.get("model_info"),
                "data_residency": infer_openai_data_residency(custom_llm_provider, litellm_params.api_base),
                "metadata": (kwargs["litellm_metadata"] if "litellm_metadata" in kwargs else kwargs.get("metadata")),
                "provider_affinity_header": litellm_params.provider_affinity_header,
            },
            custom_llm_provider=custom_llm_provider,
        )

        # Decode any litellm-encoded encrypted-content item IDs back to their original IDs
        input = ResponsesAPIRequestUtils._restore_encrypted_content_item_ids_in_input(input)

        # Call the handler with _is_async flag instead of directly calling the async handler
        if custom_llm_provider is None:
            raise ValueError("custom_llm_provider is required but passed as None")

        response = base_llm_http_handler.response_api_handler(
            model=model,
            input=input,
            responses_api_provider_config=responses_api_provider_config,
            response_api_optional_request_params=responses_api_request_params,
            custom_llm_provider=custom_llm_provider,
            litellm_params=litellm_params,
            logging_obj=litellm_logging_obj,
            extra_headers=effective_extra_headers,
            extra_body=extra_body,
            timeout=timeout or request_timeout,
            _is_async=_is_async,
            client=kwargs.get("client"),
            fake_stream=responses_api_provider_config.should_fake_stream(
                model=model, stream=stream, custom_llm_provider=custom_llm_provider
            ),
            litellm_metadata=kwargs.get("litellm_metadata", {}),
            shared_session=kwargs.get("shared_session"),
        )

        # Update the responses_api_response_id with the model_id
        if isinstance(response, ResponsesAPIResponse):
            response = ResponsesAPIRequestUtils._update_responses_api_response_id_with_model_id(
                responses_api_response=response,
                litellm_metadata=kwargs.get("litellm_metadata", {}),
                custom_llm_provider=custom_llm_provider,
            )
            # Stamp custom_llm_provider so callbacks can identify the provider
            # (mirrors litellm/main.py:1371 for chat completions)
            response._hidden_params["custom_llm_provider"] = custom_llm_provider

        return response
    except Exception as e:
        raise litellm.exception_type(
            model=model,
            custom_llm_provider=custom_llm_provider,
            original_exception=e,
            completion_kwargs=local_vars,
            extra_kwargs=kwargs,
        )


@client
async def adelete_responses(
    response_id: str,
    # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
    # The extra values given here take precedence over values defined on the client or passed to this method.
    extra_headers: dict[str, object] | None = None,
    extra_query: dict[str, object] | None = None,
    extra_body: dict[str, object] | None = None,
    timeout: float | httpx.Timeout | None = None,
    # LiteLLM specific params,
    custom_llm_provider: str | None = None,
    **kwargs,
) -> DeleteResponseResult:
    """
    Async version of the DELETE Responses API

    DELETE /v1/responses/{response_id} endpoint in the responses API

    """
    local_vars: Final = locals()
    try:
        loop: Final = asyncio.get_event_loop()
        kwargs["adelete_responses"] = True

        # get custom llm provider from response_id
        decoded_response_id: Final[DecodedResponseId] = ResponsesAPIRequestUtils._decode_responses_api_response_id(
            response_id=response_id,
        )
        response_id = decoded_response_id.get("response_id") or response_id
        custom_llm_provider = decoded_response_id.get("custom_llm_provider") or custom_llm_provider

        func: Final = partial(
            delete_responses,
            response_id=response_id,
            custom_llm_provider=custom_llm_provider,
            extra_headers=extra_headers,
            extra_query=extra_query,
            extra_body=extra_body,
            timeout=timeout,
            **kwargs,
        )

        ctx: Final = contextvars.copy_context()
        func_with_context: Final = partial(ctx.run, func)
        init_response: Final = await loop.run_in_executor(None, func_with_context)

        if asyncio.iscoroutine(init_response):
            response = await init_response
        else:
            response = init_response
        return response
    except Exception as e:
        raise litellm.exception_type(
            model=None,
            custom_llm_provider=custom_llm_provider,
            original_exception=e,
            completion_kwargs=local_vars,
            extra_kwargs=kwargs,
        )


@client
def delete_responses(
    response_id: str,
    # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
    # The extra values given here take precedence over values defined on the client or passed to this method.
    extra_headers: dict[str, object] | None = None,
    extra_query: dict[str, object] | None = None,
    extra_body: dict[str, object] | None = None,
    timeout: float | httpx.Timeout | None = None,
    # LiteLLM specific params,
    custom_llm_provider: str | None = None,
    **kwargs,
) -> DeleteResponseResult | Coroutine[object, object, DeleteResponseResult]:
    """
    Synchronous version of the DELETE Responses API

    DELETE /v1/responses/{response_id} endpoint in the responses API

    """
    local_vars: Final = locals()
    try:
        litellm_logging_obj: Final[LiteLLMLoggingObj] = kwargs.get("litellm_logging_obj")
        litellm_call_id: Final[str | None] = kwargs.get("litellm_call_id", None)
        _is_async: Final = kwargs.pop("adelete_responses", False) is True

        # get llm provider logic
        litellm_params: Final = GenericLiteLLMParams(**kwargs)

        # get custom llm provider from response_id
        decoded_response_id: Final[DecodedResponseId] = ResponsesAPIRequestUtils._decode_responses_api_response_id(
            response_id=response_id,
        )
        response_id = decoded_response_id.get("response_id") or response_id
        custom_llm_provider = decoded_response_id.get("custom_llm_provider") or custom_llm_provider

        if custom_llm_provider is None:
            raise ValueError("custom_llm_provider is required but passed as None")

        # get provider config
        responses_api_provider_config: Final[BaseResponsesAPIConfig | None] = (
            ProviderConfigManager.get_provider_responses_api_config(
                model=None,
                provider=custom_llm_provider,
                api_base=litellm_params.api_base,
            )
        )

        if responses_api_provider_config is None:
            raise ValueError(f"DELETE responses is not supported for {custom_llm_provider}")

        local_vars.update(kwargs)

        # Pre Call logging
        litellm_logging_obj.update_from_kwargs(
            kwargs=local_vars,
            model=None,
            optional_params={
                "response_id": response_id,
            },
            litellm_params={
                "litellm_call_id": litellm_call_id,
            },
            custom_llm_provider=custom_llm_provider,
        )

        # Call the handler with _is_async flag instead of directly calling the async handler
        response: Final = base_llm_http_handler.delete_response_api_handler(
            response_id=response_id,
            custom_llm_provider=custom_llm_provider,
            responses_api_provider_config=responses_api_provider_config,
            litellm_params=litellm_params,
            logging_obj=litellm_logging_obj,
            extra_headers=extra_headers,
            extra_body=extra_body,
            timeout=timeout or request_timeout,
            _is_async=_is_async,
            client=kwargs.get("client"),
            shared_session=kwargs.get("shared_session"),
        )

        return response
    except Exception as e:
        raise litellm.exception_type(
            model=None,
            custom_llm_provider=custom_llm_provider,
            original_exception=e,
            completion_kwargs=local_vars,
            extra_kwargs=kwargs,
        )


@client
async def aget_responses(
    response_id: str,
    # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
    # The extra values given here take precedence over values defined on the client or passed to this method.
    extra_headers: dict[str, object] | None = None,
    extra_query: dict[str, object] | None = None,
    extra_body: dict[str, object] | None = None,
    timeout: float | httpx.Timeout | None = None,
    # LiteLLM specific params,
    custom_llm_provider: str | None = None,
    **kwargs,
) -> ResponsesAPIResponse:
    """
    Async: Fetch a response by its ID.

    GET /v1/responses/{response_id} endpoint in the responses API

    Args:
        response_id: The ID of the response to fetch.
        custom_llm_provider: Optional provider name. If not specified, will be decoded from response_id.

    Returns:
        The response object with complete information about the stored response.
    """
    local_vars: Final = locals()
    try:
        loop: Final = asyncio.get_event_loop()
        kwargs["aget_responses"] = True

        # get custom llm provider from response_id
        decoded_response_id: Final[DecodedResponseId] = ResponsesAPIRequestUtils._decode_responses_api_response_id(
            response_id=response_id,
        )
        response_id = decoded_response_id.get("response_id") or response_id
        custom_llm_provider = decoded_response_id.get("custom_llm_provider") or custom_llm_provider

        func: Final = partial(
            get_responses,
            response_id=response_id,
            custom_llm_provider=custom_llm_provider,
            extra_headers=extra_headers,
            extra_query=extra_query,
            extra_body=extra_body,
            timeout=timeout,
            **kwargs,
        )

        ctx: Final = contextvars.copy_context()
        func_with_context: Final = partial(ctx.run, func)
        init_response: Final = await loop.run_in_executor(None, func_with_context)

        if asyncio.iscoroutine(init_response):
            response = await init_response
        else:
            response = init_response

        # Update the responses_api_response_id with the model_id
        if isinstance(response, ResponsesAPIResponse):
            response = ResponsesAPIRequestUtils._update_responses_api_response_id_with_model_id(
                responses_api_response=response,
                litellm_metadata=kwargs.get("litellm_metadata", {}),
                custom_llm_provider=custom_llm_provider,
            )
        return response
    except Exception as e:
        raise litellm.exception_type(
            model=None,
            custom_llm_provider=custom_llm_provider,
            original_exception=e,
            completion_kwargs=local_vars,
            extra_kwargs=kwargs,
        )


@client
def get_responses(
    response_id: str,
    # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
    # The extra values given here take precedence over values defined on the client or passed to this method.
    extra_headers: dict[str, object] | None = None,
    extra_query: dict[str, object] | None = None,
    extra_body: dict[str, object] | None = None,
    timeout: float | httpx.Timeout | None = None,
    # LiteLLM specific params,
    custom_llm_provider: str | None = None,
    **kwargs,
) -> ResponsesAPIResponse | Coroutine[object, object, ResponsesAPIResponse]:
    """
    Fetch a response by its ID.

    GET /v1/responses/{response_id} endpoint in the responses API

    Args:
        response_id: The ID of the response to fetch.
        custom_llm_provider: Optional provider name. If not specified, will be decoded from response_id.

    Returns:
        The response object with complete information about the stored response.
    """
    local_vars: Final = locals()
    try:
        litellm_logging_obj: Final[LiteLLMLoggingObj] = kwargs.get("litellm_logging_obj")
        litellm_call_id: Final[str | None] = kwargs.get("litellm_call_id", None)
        _is_async: Final = kwargs.pop("aget_responses", False) is True

        # get llm provider logic
        litellm_params: Final = GenericLiteLLMParams(**kwargs)

        # get custom llm provider from response_id
        decoded_response_id: Final[DecodedResponseId] = ResponsesAPIRequestUtils._decode_responses_api_response_id(
            response_id=response_id,
        )
        response_id = decoded_response_id.get("response_id") or response_id
        custom_llm_provider = decoded_response_id.get("custom_llm_provider") or custom_llm_provider

        if custom_llm_provider is None:
            raise ValueError("custom_llm_provider is required but passed as None")

        # get provider config
        responses_api_provider_config: Final[BaseResponsesAPIConfig | None] = (
            ProviderConfigManager.get_provider_responses_api_config(
                model=None,
                provider=custom_llm_provider,
                api_base=litellm_params.api_base,
            )
        )

        if responses_api_provider_config is None:
            raise ValueError(f"GET responses is not supported for {custom_llm_provider}")

        local_vars.update(kwargs)

        # Pre Call logging
        litellm_logging_obj.update_from_kwargs(
            kwargs=local_vars,
            model=None,
            optional_params={
                "response_id": response_id,
            },
            litellm_params={
                "litellm_call_id": litellm_call_id,
            },
            custom_llm_provider=custom_llm_provider,
        )

        # Call the handler with _is_async flag instead of directly calling the async handler
        response = base_llm_http_handler.get_responses(
            response_id=response_id,
            custom_llm_provider=custom_llm_provider,
            responses_api_provider_config=responses_api_provider_config,
            litellm_params=litellm_params,
            logging_obj=litellm_logging_obj,
            extra_headers=extra_headers,
            extra_body=extra_body,
            timeout=timeout or request_timeout,
            _is_async=_is_async,
            client=kwargs.get("client"),
            shared_session=kwargs.get("shared_session"),
        )

        # Update the responses_api_response_id with the model_id
        if isinstance(response, ResponsesAPIResponse):
            response = ResponsesAPIRequestUtils._update_responses_api_response_id_with_model_id(
                responses_api_response=response,
                litellm_metadata=kwargs.get("litellm_metadata", {}),
                custom_llm_provider=custom_llm_provider,
            )

        return response
    except Exception as e:
        raise litellm.exception_type(
            model=None,
            custom_llm_provider=custom_llm_provider,
            original_exception=e,
            completion_kwargs=local_vars,
            extra_kwargs=kwargs,
        )


@client
async def alist_input_items(
    response_id: str,
    after: str | None = None,
    before: str | None = None,
    include: list[str] | None = None,
    limit: int = 20,
    order: Literal["asc", "desc"] = "desc",
    extra_headers: dict[str, object] | None = None,
    timeout: float | httpx.Timeout | None = None,
    custom_llm_provider: str | None = None,
    **kwargs,
) -> dict:
    """Async: List input items for a response"""
    local_vars: Final = locals()
    try:
        loop: Final = asyncio.get_event_loop()
        kwargs["alist_input_items"] = True

        decoded_response_id: Final = ResponsesAPIRequestUtils._decode_responses_api_response_id(response_id=response_id)
        response_id = decoded_response_id.get("response_id") or response_id
        custom_llm_provider = decoded_response_id.get("custom_llm_provider") or custom_llm_provider

        func: Final = partial(
            list_input_items,
            response_id=response_id,
            after=after,
            before=before,
            include=include,
            limit=limit,
            order=order,
            extra_headers=extra_headers,
            timeout=timeout,
            custom_llm_provider=custom_llm_provider,
            **kwargs,
        )

        ctx: Final = contextvars.copy_context()
        func_with_context: Final = partial(ctx.run, func)
        init_response: Final = await loop.run_in_executor(None, func_with_context)

        if asyncio.iscoroutine(init_response):
            response = await init_response
        else:
            response = init_response
        return response
    except Exception as e:
        raise litellm.exception_type(
            model=None,
            custom_llm_provider=custom_llm_provider,
            original_exception=e,
            completion_kwargs=local_vars,
            extra_kwargs=kwargs,
        )


@client
def list_input_items(
    response_id: str,
    after: str | None = None,
    before: str | None = None,
    include: list[str] | None = None,
    limit: int = 20,
    order: Literal["asc", "desc"] = "desc",
    extra_headers: dict[str, object] | None = None,
    timeout: float | httpx.Timeout | None = None,
    custom_llm_provider: str | None = None,
    **kwargs,
) -> dict | Coroutine[object, object, dict]:
    """List input items for a response"""
    local_vars: Final = locals()
    try:
        litellm_logging_obj: Final[LiteLLMLoggingObj] = kwargs.get("litellm_logging_obj")
        litellm_call_id: Final[str | None] = kwargs.get("litellm_call_id", None)
        _is_async: Final = kwargs.pop("alist_input_items", False) is True

        litellm_params: Final = GenericLiteLLMParams(**kwargs)

        decoded_response_id: Final = ResponsesAPIRequestUtils._decode_responses_api_response_id(response_id=response_id)
        response_id = decoded_response_id.get("response_id") or response_id
        custom_llm_provider = decoded_response_id.get("custom_llm_provider") or custom_llm_provider

        if custom_llm_provider is None:
            raise ValueError("custom_llm_provider is required but passed as None")

        responses_api_provider_config: Final[BaseResponsesAPIConfig | None] = (
            ProviderConfigManager.get_provider_responses_api_config(
                model=None,
                provider=custom_llm_provider,
                api_base=litellm_params.api_base,
            )
        )

        if responses_api_provider_config is None:
            raise ValueError(f"list_input_items is not supported for {custom_llm_provider}")

        local_vars.update(kwargs)

        litellm_logging_obj.update_from_kwargs(
            kwargs=local_vars,
            model=None,
            optional_params={"response_id": response_id},
            litellm_params={"litellm_call_id": litellm_call_id},
            custom_llm_provider=custom_llm_provider,
        )

        response: Final = base_llm_http_handler.list_responses_input_items(
            response_id=response_id,
            custom_llm_provider=custom_llm_provider,
            responses_api_provider_config=responses_api_provider_config,
            litellm_params=litellm_params,
            logging_obj=litellm_logging_obj,
            after=after,
            before=before,
            include=include,
            limit=limit,
            order=order,
            extra_headers=extra_headers,
            timeout=timeout or request_timeout,
            _is_async=_is_async,
            client=kwargs.get("client"),
            shared_session=kwargs.get("shared_session"),
        )

        return response
    except Exception as e:
        raise litellm.exception_type(
            model=None,
            custom_llm_provider=custom_llm_provider,
            original_exception=e,
            completion_kwargs=local_vars,
            extra_kwargs=kwargs,
        )


@client
async def acancel_responses(
    response_id: str,
    # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
    # The extra values given here take precedence over values defined on the client or passed to this method.
    extra_headers: dict[str, object] | None = None,
    extra_query: dict[str, object] | None = None,
    extra_body: dict[str, object] | None = None,
    timeout: float | httpx.Timeout | None = None,
    # LiteLLM specific params,
    custom_llm_provider: str | None = None,
    **kwargs,
) -> ResponsesAPIResponse:
    """
    Async version of the POST Cancel Responses API

    POST /v1/responses/{response_id}/cancel endpoint in the responses API

    """
    local_vars: Final = locals()
    try:
        loop: Final = asyncio.get_event_loop()
        kwargs["acancel_responses"] = True

        # get custom llm provider from response_id
        decoded_response_id: Final[DecodedResponseId] = ResponsesAPIRequestUtils._decode_responses_api_response_id(
            response_id=response_id,
        )
        response_id = decoded_response_id.get("response_id") or response_id
        custom_llm_provider = decoded_response_id.get("custom_llm_provider") or custom_llm_provider

        func: Final = partial(
            cancel_responses,
            response_id=response_id,
            custom_llm_provider=custom_llm_provider,
            extra_headers=extra_headers,
            extra_query=extra_query,
            extra_body=extra_body,
            timeout=timeout,
            **kwargs,
        )

        ctx: Final = contextvars.copy_context()
        func_with_context: Final = partial(ctx.run, func)
        init_response: Final = await loop.run_in_executor(None, func_with_context)

        if asyncio.iscoroutine(init_response):
            response = await init_response
        else:
            response = init_response
        return response
    except Exception as e:
        raise litellm.exception_type(
            model=None,
            custom_llm_provider=custom_llm_provider,
            original_exception=e,
            completion_kwargs=local_vars,
            extra_kwargs=kwargs,
        )


@client
def cancel_responses(
    response_id: str,
    # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
    # The extra values given here take precedence over values defined on the client or passed to this method.
    extra_headers: dict[str, object] | None = None,
    extra_query: dict[str, object] | None = None,
    extra_body: dict[str, object] | None = None,
    timeout: float | httpx.Timeout | None = None,
    # LiteLLM specific params,
    custom_llm_provider: str | None = None,
    **kwargs,
) -> ResponsesAPIResponse | Coroutine[object, object, ResponsesAPIResponse]:
    """
    Synchronous version of the POST Responses API

    POST /v1/responses/{response_id}/cancel endpoint in the responses API

    """
    local_vars: Final = locals()
    try:
        litellm_logging_obj: Final[LiteLLMLoggingObj] = kwargs.get("litellm_logging_obj")
        litellm_call_id: Final[str | None] = kwargs.get("litellm_call_id", None)
        _is_async: Final = kwargs.pop("acancel_responses", False) is True

        # get llm provider logic
        litellm_params: Final = GenericLiteLLMParams(**kwargs)

        # get custom llm provider from response_id
        decoded_response_id: Final[DecodedResponseId] = ResponsesAPIRequestUtils._decode_responses_api_response_id(
            response_id=response_id,
        )
        response_id = decoded_response_id.get("response_id") or response_id
        custom_llm_provider = decoded_response_id.get("custom_llm_provider") or custom_llm_provider

        if custom_llm_provider is None:
            raise ValueError("custom_llm_provider is required but passed as None")

        # get provider config
        responses_api_provider_config: Final[BaseResponsesAPIConfig | None] = (
            ProviderConfigManager.get_provider_responses_api_config(
                model=None,
                provider=custom_llm_provider,
                api_base=litellm_params.api_base,
            )
        )

        if responses_api_provider_config is None:
            raise ValueError(f"CANCEL responses is not supported for {custom_llm_provider}")

        local_vars.update(kwargs)

        # Pre Call logging
        litellm_logging_obj.update_from_kwargs(
            kwargs=local_vars,
            model=None,
            optional_params={
                "response_id": response_id,
            },
            litellm_params={
                "litellm_call_id": litellm_call_id,
            },
            custom_llm_provider=custom_llm_provider,
        )

        # Call the handler with _is_async flag instead of directly calling the async handler
        response: Final = base_llm_http_handler.cancel_response_api_handler(
            response_id=response_id,
            custom_llm_provider=custom_llm_provider,
            responses_api_provider_config=responses_api_provider_config,
            litellm_params=litellm_params,
            logging_obj=litellm_logging_obj,
            extra_headers=extra_headers,
            extra_body=extra_body,
            timeout=timeout or request_timeout,
            _is_async=_is_async,
            client=kwargs.get("client"),
            shared_session=kwargs.get("shared_session"),
        )

        return response
    except Exception as e:
        raise litellm.exception_type(
            model=None,
            custom_llm_provider=custom_llm_provider,
            original_exception=e,
            completion_kwargs=local_vars,
            extra_kwargs=kwargs,
        )


@client
async def acompact_responses(
    input: str | ResponseInputParam,
    model: str,
    instructions: str | None = None,
    previous_response_id: str | None = None,
    # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
    # The extra values given here take precedence over values defined on the client or passed to this method.
    extra_headers: dict[str, object] | None = None,
    extra_query: dict[str, object] | None = None,
    extra_body: dict[str, object] | None = None,
    timeout: float | httpx.Timeout | None = None,
    # LiteLLM specific params,
    custom_llm_provider: str | None = None,
    **kwargs,
) -> ResponsesAPIResponse:
    """
    Async version of the POST Compact Responses API

    POST /v1/responses/compact endpoint in the responses API

    Runs a compaction pass over a conversation, returning encrypted, opaque items.
    """
    local_vars: Final = locals()
    try:
        loop: Final = asyncio.get_event_loop()
        kwargs["acompact_responses"] = True

        # get custom llm provider so we can use this for mapping exceptions
        if custom_llm_provider is None:
            _, custom_llm_provider, _, _ = litellm.get_llm_provider(
                model=model, api_base=local_vars.get("base_url", None)
            )
            # Update local_vars with detected provider (fixes #19782)
            local_vars["custom_llm_provider"] = custom_llm_provider

        func: Final = partial(
            compact_responses,
            input=input,
            model=model,
            instructions=instructions,
            previous_response_id=previous_response_id,
            extra_headers=extra_headers,
            extra_query=extra_query,
            extra_body=extra_body,
            timeout=timeout,
            custom_llm_provider=custom_llm_provider,
            **kwargs,
        )

        ctx: Final = contextvars.copy_context()
        func_with_context: Final = partial(ctx.run, func)
        init_response: Final = await loop.run_in_executor(None, func_with_context)

        if asyncio.iscoroutine(init_response):
            response = await init_response
        else:
            response = init_response

        # Update the responses_api_response_id with the model_id
        if isinstance(response, ResponsesAPIResponse):
            response = ResponsesAPIRequestUtils._update_responses_api_response_id_with_model_id(
                responses_api_response=response,
                litellm_metadata=kwargs.get("litellm_metadata", {}),
                custom_llm_provider=custom_llm_provider,
            )

        return response
    except Exception as e:
        raise litellm.exception_type(
            model=model,
            custom_llm_provider=custom_llm_provider,
            original_exception=e,
            completion_kwargs=local_vars,
            extra_kwargs=kwargs,
        )


@client
def compact_responses(
    input: str | ResponseInputParam,
    model: str,
    instructions: str | None = None,
    previous_response_id: str | None = None,
    # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
    # The extra values given here take precedence over values defined on the client or passed to this method.
    extra_headers: dict[str, object] | None = None,
    extra_query: dict[str, object] | None = None,
    extra_body: dict[str, object] | None = None,
    timeout: float | httpx.Timeout | None = None,
    # LiteLLM specific params,
    custom_llm_provider: str | None = None,
    **kwargs,
) -> ResponsesAPIResponse | Coroutine[object, object, ResponsesAPIResponse]:
    """
    Synchronous version of the POST Compact Responses API

    POST /v1/responses/compact endpoint in the responses API

    Runs a compaction pass over a conversation, returning encrypted, opaque items.
    """
    local_vars: Final = locals()
    try:
        litellm_logging_obj: Final[LiteLLMLoggingObj] = kwargs.get("litellm_logging_obj")
        litellm_call_id: Final[str | None] = kwargs.get("litellm_call_id", None)
        _is_async: Final = kwargs.pop("acompact_responses", False) is True

        # get llm provider logic
        litellm_params: Final = GenericLiteLLMParams(**kwargs)

        model, custom_llm_provider = _resolve_model_provider_for_responses(
            model=model,
            custom_llm_provider=custom_llm_provider,
            litellm_params=litellm_params,
            local_vars=local_vars,
        )

        if custom_llm_provider is None:
            raise ValueError("custom_llm_provider is required but passed as None")

        # get provider config
        responses_api_provider_config: Final[BaseResponsesAPIConfig | None] = (
            ProviderConfigManager.get_provider_responses_api_config(
                model=model,
                provider=custom_llm_provider,
                api_base=litellm_params.api_base,
            )
        )

        if responses_api_provider_config is None:
            raise ValueError(f"COMPACT responses is not supported for {custom_llm_provider}")

        local_vars.update(kwargs)

        # Build optional params for compact endpoint
        response_api_optional_params: Final[ResponsesAPIOptionalRequestParams] = (
            ResponsesAPIRequestUtils.get_requested_response_api_optional_param(local_vars)
        )

        # Get optional parameters for the responses API
        request_drop_params: Final = kwargs.get("drop_params")
        responses_api_request_params: Final[dict] = ResponsesAPIRequestUtils.get_optional_params_responses_api(
            model=model,
            responses_api_provider_config=responses_api_provider_config,
            response_api_optional_params=response_api_optional_params,
            allowed_openai_params=None,
            drop_params=normalize_drop_params(request_drop_params),
        )

        # Pre Call logging
        litellm_logging_obj.update_from_kwargs(
            kwargs=local_vars,
            model=model,
            optional_params=dict(responses_api_request_params),
            litellm_params={
                **responses_api_request_params,
                "litellm_call_id": litellm_call_id,
                "data_residency": infer_openai_data_residency(custom_llm_provider, litellm_params.api_base),
            },
            custom_llm_provider=custom_llm_provider,
        )

        # Decode any litellm-encoded encrypted-content item IDs back to their original IDs
        # before forwarding to the upstream provider.
        input = ResponsesAPIRequestUtils._restore_encrypted_content_item_ids_in_input(input)

        # Call the handler with _is_async flag instead of directly calling the async handler
        response = base_llm_http_handler.compact_response_api_handler(
            model=model,
            input=input,
            responses_api_provider_config=responses_api_provider_config,
            response_api_optional_request_params=responses_api_request_params,
            litellm_params=litellm_params,
            logging_obj=litellm_logging_obj,
            custom_llm_provider=custom_llm_provider,
            extra_headers=extra_headers,
            extra_body=extra_body,
            timeout=timeout or request_timeout,
            _is_async=_is_async,
            client=kwargs.get("client"),
            shared_session=kwargs.get("shared_session"),
        )

        # Update the responses_api_response_id with the model_id
        if isinstance(response, ResponsesAPIResponse):
            response = ResponsesAPIRequestUtils._update_responses_api_response_id_with_model_id(
                responses_api_response=response,
                litellm_metadata=kwargs.get("litellm_metadata", {}),
                custom_llm_provider=custom_llm_provider,
            )

        return response
    except Exception as e:
        raise litellm.exception_type(
            model=model,
            custom_llm_provider=custom_llm_provider,
            original_exception=e,
            completion_kwargs=local_vars,
            extra_kwargs=kwargs,
        )


# ---------------------------------------------------------------------------
# Responses API WebSocket mode
# ---------------------------------------------------------------------------


def _build_litellm_metadata_for_ws(kwargs: dict) -> dict:
    metadata: Final[dict] = {**(kwargs.get("litellm_metadata") or {})}
    guardrails: Final = (kwargs.get("metadata") or {}).get("guardrails") or kwargs.get("guardrails") or []
    if guardrails:
        metadata["guardrails"] = guardrails
    return metadata


_JSON_OBJECT_ADAPTER: Final = TypeAdapter(dict[str, object] | None)


def _deployment_reasoning_default(kwargs: Mapping[str, object]) -> Reasoning | dict[str, object] | None:
    if kwargs.get("reasoning") is not None:
        return None
    reasoning_effort: Final = kwargs.get("reasoning_effort")
    if isinstance(reasoning_effort, str):
        return LiteLLMResponsesTransformationHandler()._map_reasoning_effort(reasoning_effort)
    return _JSON_OBJECT_ADAPTER.validate_python(reasoning_effort) if isinstance(reasoning_effort, Mapping) else None


_RESPONSES_WS_ROUTING_HINT_KEYS: Final = frozenset({"input", "previous_response_id"})


def _first_ws_frame_with_routed_input(first_message: str, routed_input: object) -> str:
    try:
        frame: Final = _JSON_OBJECT_ADAPTER.validate_json(first_message)
    except ValidationError:
        return first_message
    if frame is None or routed_input is None:
        return first_message
    raw_nested: Final = frame.get("response")
    nested: Final = _JSON_OBJECT_ADAPTER.validate_python(raw_nested) if isinstance(raw_nested, Mapping) else None
    if nested is not None and nested.get("input") is not None:
        if nested["input"] == routed_input:
            return first_message
        return json.dumps({**frame, "response": {**nested, "input": routed_input}})
    if frame.get("input") == routed_input:
        return first_message
    return json.dumps({**frame, "input": routed_input})


def _build_responses_websocket_request_defaults(kwargs: Mapping[str, object]) -> ResponsesWebSocketRequestDefaults:
    default_reasoning: Final = _deployment_reasoning_default(kwargs)
    candidate_params: Final[dict[str, object]] = {
        **kwargs,
        **({"reasoning": default_reasoning} if default_reasoning is not None else {}),
    }
    fill_missing: Final = ResponsesAPIRequestUtils.get_requested_response_api_optional_param(candidate_params)
    return ResponsesWebSocketRequestDefaults(
        fill_missing=MappingProxyType(dict(fill_missing)),
        overrides=MappingProxyType(_JSON_OBJECT_ADAPTER.validate_python(kwargs.get("extra_body")) or {}),
    )


@client
async def _aresponses_websocket(
    model: str,
    websocket: "WebSocket",
    api_base: str | None = None,
    api_key: str | None = None,
    timeout: float | None = None,
    **kwargs,
) -> Exception | None:
    """
    Private function to handle the Responses API WebSocket mode.

    For PROXY use only. Returns the provider failure that ended the connection, if any.

    Resolves the LLM provider from ``model``, looks up the matching
    ``BaseResponsesAPIConfig``, and hands off to
    ``BaseLLMHTTPHandler.async_responses_websocket``.
    """
    litellm_logging_obj: Final[LiteLLMLoggingObj] = kwargs.get("litellm_logging_obj")
    user: Final = kwargs.get("user", None)
    litellm_params: Final = GenericLiteLLMParams(**kwargs)
    litellm_params_dict: Final = get_litellm_params(**kwargs)

    (
        provider_model,
        _custom_llm_provider,
        dynamic_api_key,
        dynamic_api_base,
    ) = litellm.get_llm_provider(
        model=model,
        api_base=api_base,
        api_key=api_key,
    )
    resolved_model: Final = _strip_responses_routing_prefix(provider_model)

    litellm_params_dict["data_residency"] = infer_openai_data_residency(
        _custom_llm_provider,
        dynamic_api_base or litellm_params.api_base or litellm.api_base,
    )

    litellm_logging_obj.update_from_kwargs(
        kwargs=kwargs,
        model=resolved_model,
        user=user,
        optional_params={},
        litellm_params=litellm_params_dict,
        custom_llm_provider=_custom_llm_provider,
    )

    resolved_api_base: Final = dynamic_api_base or litellm_params.api_base or litellm.api_base or None
    responses_api_provider_config: BaseResponsesAPIConfig | None = None
    if _custom_llm_provider is not None:
        responses_api_provider_config = ProviderConfigManager.get_provider_responses_api_config(
            model=resolved_model,
            provider=litellm.LlmProviders(_custom_llm_provider),
            api_base=resolved_api_base,
        )

    resolved_api_key: Final = (
        dynamic_api_key
        or litellm_params.api_key
        or litellm.api_key
        or litellm.openai_key
        or get_secret_str("OPENAI_API_KEY")
    )

    # Extract params that we're passing explicitly to avoid duplicates in **kwargs
    _explicit_keys: Final = {
        "user_api_key_dict",
        "litellm_metadata",
        "custom_llm_provider",
        "model",
        "websocket",
        "litellm_logging_obj",
        "api_base",
        "api_key",
        "timeout",
        "first_message",
        *_RESPONSES_WS_ROUTING_HINT_KEYS,
    }
    remaining_kwargs: Final = {k: v for k, v in kwargs.items() if k not in _explicit_keys}
    deployment_kwargs: Final = {k: v for k, v in kwargs.items() if k not in _RESPONSES_WS_ROUTING_HINT_KEYS}
    first_message: Final = kwargs.get("first_message")

    return await base_llm_http_handler.async_responses_websocket(
        model=resolved_model,
        websocket=websocket,
        logging_obj=litellm_logging_obj,
        responses_api_provider_config=responses_api_provider_config,
        api_base=resolved_api_base,
        api_key=resolved_api_key,
        timeout=timeout,
        first_message=(
            _first_ws_frame_with_routed_input(first_message, kwargs.get("input"))
            if isinstance(first_message, str)
            else None
        ),
        user_api_key_dict=kwargs.get("user_api_key_dict"),
        litellm_metadata=_build_litellm_metadata_for_ws(kwargs),
        custom_llm_provider=_custom_llm_provider,
        request_defaults=_build_responses_websocket_request_defaults(deployment_kwargs),
        **remaining_kwargs,
    )