diff --git a/litellm/litellm_core_utils/litellm_logging.py b/litellm/litellm_core_utils/litellm_logging.py
index 78ff073fdcb..047990e782f 100644
--- a/litellm/litellm_core_utils/litellm_logging.py
+++ b/litellm/litellm_core_utils/litellm_logging.py
@@ -77,6 +77,7 @@ from litellm.litellm_core_utils.classifier_logging import (
 from litellm.litellm_core_utils.core_helpers import (
     get_provider_response_headers_from_hidden_params,
     is_expected_client_error,
+    process_response_headers,
     proxy_stamped_kubernetes_pod_routing,
     proxy_stamped_used_client_oauth_token,
     reconstruct_model_name,
@@ -4618,6 +4619,29 @@ class Logging(LiteLLMLoggingBaseClass):
         row is built from this response rather than re-parsed from the served body."""
         self.bridged_model_response = response
 
+    def price_bridged_provider_response(self, response: ModelResponse | ResponsesAPIResponse) -> None:
+        """Stamps the cost and headers of the provider response a bridge translates into its client's shape.
+
+        The client shape has no field for some of the usage the price depends on (built-in tool calls,
+        the 1h cache-write split), and the translation carries the stamped cost over, so the request is
+        billed for what the provider returned. The headers get the round of processing the call that made
+        the response gave them, which also names the provider request id `llm_provider-request-id`."""
+        set_hidden_param(response, "response_cost", self.response_cost_calculator(result=response))
+        provider_headers: Final = (get_hidden_params(response) or {}).get("additional_headers")
+        if isinstance(provider_headers, dict):
+            set_hidden_param(
+                response,
+                "additional_headers",
+                process_response_headers(provider_headers, preserve_litellm_internal_headers=True),
+            )
+
+    def price_bridged_provider_stream(self, stream: object) -> None:
+        """Prices a provider stream a bridge wraps for its client, as the call that opened it did.
+
+        Nothing has streamed yet, so this only records the zero cost breakdown the proxy builds the
+        response headers from; the request is billed when the stream ends."""
+        self.response_cost_calculator(result=stream)
+
     def record_bridged_responses_api_response(self, response: ResponsesAPIResponse) -> None:
         self.record_bridged_model_response(self._translate_responses_api_response_to_model_response(response))
 
diff --git a/litellm/llms/anthropic/pass_through/responses_adapters/handler.py b/litellm/llms/anthropic/pass_through/responses_adapters/handler.py
index b1347b0c79e..68219fc0c65 100644
--- a/litellm/llms/anthropic/pass_through/responses_adapters/handler.py
+++ b/litellm/llms/anthropic/pass_through/responses_adapters/handler.py
@@ -5,18 +5,35 @@ Used when the target model is an OpenAI or Azure model.
 """
 
 from collections.abc import AsyncIterator, Callable, Coroutine, Mapping, Sequence
-from typing import Final, TypeAlias
+from typing import Final, NoReturn, TypeAlias
 
+import httpx
+from aiohttp import ClientSession
 from pydantic import TypeAdapter, ValidationError
 
 import litellm
+from litellm.constants import request_timeout
 from litellm.integrations.anthropic_cache_control_hook import (
     AnthropicCacheControlHook,
     supports_openai_prompt_cache_breakpoint,
 )
 from litellm.litellm_core_utils.core_helpers import drops_unsupported_params
+from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
 from litellm.litellm_core_utils.prompt_templates.common_utils import cache_control_as_prompt_cache_breakpoint
-from litellm.responses.utils import auto_reasoning_summary
+from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler
+from litellm.native.responses import execute_responses
+from litellm.responses.main import (
+    apply_managed_file_id_mapping,
+    base_llm_http_handler,
+    reasoning_for_deployment,
+    requested_reasoning,
+    resolve_model_provider_for_responses,
+    resolve_responses_api_provider_config,
+    with_provider_affinity_header,
+)
+from litellm.responses.streaming_iterator import BaseResponsesAPIStreamingIterator
+from litellm.responses.utils import ResponsesAPIRequestUtils, auto_reasoning_summary
+from litellm.router_utils.reasoning_effort_capability import deployment_reasoning_effort_levels
 from litellm.types.integrations.custom_logger import CONVERTED_STREAM_KEYS
 from litellm.types.llms.anthropic import (
     AllAnthropicMessageValues,
@@ -29,6 +46,8 @@ from litellm.types.llms.anthropic_messages.anthropic_response import (
     AnthropicMessagesResponse,
 )
 from litellm.types.llms.openai import ResponsesAPIResponse
+from litellm.types.router import GenericLiteLLMParams
+from litellm.types.utils import CallTypes
 from litellm.utils import ProviderConfigManager
 
 from ..adapters.handler import request_safeguards_evaluator
@@ -43,6 +62,12 @@ BlockRewrite: TypeAlias = Callable[[Mapping[str, object]], Mapping[str, object]]
 _BLOCK_LIST: Final = TypeAdapter(list[dict[str, object]])
 
 _ADAPTER: Final = LiteLLMAnthropicToResponsesAPIAdapter()
+_STRING_KEYED_DICT: Final = TypeAdapter(dict[str, object])
+_CLIENT_HEADERS: Final = TypeAdapter(dict[str, str])
+_STRINGS: Final = TypeAdapter(tuple[str, ...])
+
+
+_NEVER_FORWARDED: Final = CONVERTED_STREAM_KEYS | {"anthropic_messages"}
 
 
 def _record_bridged_response(responses_kwargs: Mapping[str, object], result: ResponsesAPIResponse) -> None:
@@ -52,8 +77,10 @@ def _record_bridged_response(responses_kwargs: Mapping[str, object], result: Res
 
 
 def _forwarded_kwargs(extra_kwargs: Mapping[str, object] | None) -> Mapping[str, object]:
-    """The litellm-specific kwargs forwarded verbatim onto the Responses API request."""
-    return extra_kwargs or {}
+    """The litellm-specific kwargs (credentials, routing, metadata) that carry over to the Responses API call."""
+    return {
+        key: value for key, value in (extra_kwargs or {}).items() if key not in _NEVER_FORWARDED and value is not None
+    }
 
 
 def _provider_returns_encrypted_reasoning(model: str, custom_llm_provider: object) -> bool:
@@ -119,7 +146,7 @@ def _rewritten_blocks(content: object, rewrite: BlockRewrite) -> Sequence[Mappin
         return ()
 
 
-def _build_responses_kwargs(
+def _build_responses_request(
     *,
     max_tokens: int,
     messages: AnthropicRequestMessages,
@@ -139,9 +166,7 @@ def _build_responses_kwargs(
     output_format: AnthropicOutputSchema | None = None,
     extra_kwargs: Mapping[str, object] | None = None,
 ) -> dict[str, object]:
-    """
-    Build the kwargs dict to pass directly to litellm.responses() / litellm.aresponses().
-    """
+    """The Responses API request the Messages request translates to, without the forwarded litellm kwargs."""
     forwarded_kwargs: Final = _forwarded_kwargs(extra_kwargs)
     reject_malformed_tool_result_text(messages, model, forwarded_kwargs)
     block_rewrite: Final = _prompt_cache_block_rewrite(model, forwarded_kwargs)
@@ -186,7 +211,7 @@ def _build_responses_kwargs(
         request_data["output_format"] = output_format
 
     anthropic_request: Final = AnthropicMessagesRequest(**request_data)
-    responses_kwargs: Final = _ADAPTER.translate_request(
+    responses_request: Final = _ADAPTER.translate_request(
         anthropic_request,
         include_encrypted_reasoning=_provider_returns_encrypted_reasoning(
             model, forwarded_kwargs.get("custom_llm_provider")
@@ -197,7 +222,7 @@ def _build_responses_kwargs(
 
     # Normalize reasoning effort based on model capabilities
     # (e.g. "max" → "xhigh"/"high", "minimal" → "low" if unsupported)
-    reasoning: Final = responses_kwargs.get("reasoning")
+    reasoning: Final = responses_request.get("reasoning")
     if isinstance(reasoning, dict):
         effort: Final[object] = reasoning.get("effort")
         if isinstance(effort, str):
@@ -213,49 +238,238 @@ def _build_responses_kwargs(
                 deployment_model_info=forwarded_kwargs.get("model_info"),
             )
             if normalized != effort:
-                responses_kwargs["reasoning"] = {**reasoning, "effort": normalized}
+                responses_request["reasoning"] = {**reasoning, "effort": normalized}
 
     if stream:
-        responses_kwargs["stream"] = True
-
-    # Forward litellm-specific kwargs (api_key, api_base, logging obj, etc.)
-    excluded: Final = CONVERTED_STREAM_KEYS | {"anthropic_messages"}
-    for key, value in forwarded_kwargs.items():
-        if key == "litellm_logging_obj" and value is not None:
-            from litellm.litellm_core_utils.litellm_logging import (
-                Logging as LiteLLMLoggingObject,
-            )
-            from litellm.types.utils import CallTypes
-
-            if isinstance(value, LiteLLMLoggingObject):
-                # Keep call_type as anthropic_messages so spend_logs are billed
-                # against /v1/messages; the success handler translates the
-                # Responses API result back to a ModelResponse for the row.
-                setattr(value, "call_type", CallTypes.anthropic_messages.value)
-            responses_kwargs[key] = value
-        elif key not in excluded and key not in responses_kwargs and value is not None:
-            responses_kwargs[key] = value
+        responses_request["stream"] = True
 
     explicit_prompt_cache_key: Final = forwarded_kwargs.get("prompt_cache_key")
     if explicit_prompt_cache_key is not None:
-        responses_kwargs["prompt_cache_key"] = explicit_prompt_cache_key
+        responses_request["prompt_cache_key"] = explicit_prompt_cache_key
 
     deployment_include: Final = forwarded_kwargs.get("include")
-    bridge_include: Final = responses_kwargs.get("include")
+    bridge_include: Final = responses_request.get("include")
     if isinstance(deployment_include, list) and isinstance(bridge_include, list):
-        responses_kwargs["include"] = [
+        responses_request["include"] = [
             *bridge_include,
             *(item for item in deployment_include if item not in bridge_include),
         ]
 
-    return responses_kwargs
+    return responses_request
+
+
+def _optional_str(value: object) -> str | None:
+    return value if isinstance(value, str) else None
+
+
+def _string_keyed_dict(value: object) -> dict[str, object] | None:
+    try:
+        return _STRING_KEYED_DICT.validate_python(value) if value is not None else None
+    except ValidationError:
+        return None
+
+
+def _client_headers(value: object) -> Mapping[str, str] | None:
+    try:
+        return _CLIENT_HEADERS.validate_python(value) if isinstance(value, dict) else None
+    except ValidationError:
+        return None
+
+
+def _shared_session(value: object) -> ClientSession | None:
+    return value if isinstance(value, ClientSession) else None
+
+
+def _http_client(value: object) -> HTTPHandler | AsyncHTTPHandler | None:
+    return value if isinstance(value, (HTTPHandler, AsyncHTTPHandler)) else None
+
+
+def _timeout(value: object) -> float | httpx.Timeout:
+    return value if isinstance(value, (float, int, httpx.Timeout)) and value else request_timeout
+
+
+def _allowed_openai_params(value: object) -> Sequence[str] | None:
+    try:
+        return _STRINGS.validate_python(value) if value is not None else None
+    except ValidationError:
+        return None
+
+
+def _native_responses_call(
+    request: Mapping[str, object],
+    model: str,
+    forwarded_kwargs: Mapping[str, object],
+    logging_obj: LiteLLMLoggingObj,
+    is_async: bool,
+) -> (
+    ResponsesAPIResponse
+    | BaseResponsesAPIStreamingIterator
+    | Coroutine[object, object, ResponsesAPIResponse | BaseResponsesAPIStreamingIterator]
+):
+    """Resolve the deployment and hand the translated request to the native Responses executor."""
+    litellm_params: Final = GenericLiteLLMParams(**forwarded_kwargs)
+    provider_model, custom_llm_provider = resolve_model_provider_for_responses(
+        model=model,
+        custom_llm_provider=_optional_str(forwarded_kwargs.get("custom_llm_provider")),
+        litellm_params=litellm_params,
+        local_vars={},
+    )
+    if custom_llm_provider is None:
+        raise ValueError(f"No provider resolved for {model}")
+    provider_config: Final = resolve_responses_api_provider_config(
+        provider_model, custom_llm_provider, forwarded_kwargs.get("model_info"), litellm_params.api_base
+    )
+    if provider_config is None:
+        raise ValueError(f"{custom_llm_provider} has no Responses API config for {provider_model}")
+    request_input, request_tools = apply_managed_file_id_mapping(
+        input=request["input"],
+        tools=request.get("tools"),
+        kwargs=dict(forwarded_kwargs),
+        local_vars={},
+    )
+    params: Final = {
+        **forwarded_kwargs,
+        **request,
+        "input": request_input,
+        "tools": request_tools,
+        "custom_llm_provider": custom_llm_provider,
+    }
+    deployment_model_info: Final = forwarded_kwargs.get("model_info")
+    reasoning: Final = reasoning_for_deployment(
+        requested_reasoning(params, provider_model, custom_llm_provider),
+        provider_model,
+        custom_llm_provider,
+        deployment_model_info,
+        forwarded_kwargs,
+    )
+    return execute_responses(
+        http_handler=base_llm_http_handler,
+        model=provider_model,
+        input=request_input,
+        provider_config=provider_config,
+        requested_params=ResponsesAPIRequestUtils.get_requested_response_api_optional_param(
+            {key: value for key, value in {**params, "reasoning": reasoning}.items() if key != "reasoning_effort"}
+        ),
+        allowed_openai_params=_allowed_openai_params(forwarded_kwargs.get("allowed_openai_params")),
+        declared_reasoning_efforts=deployment_reasoning_effort_levels(deployment_model_info) or (),
+        custom_llm_provider=custom_llm_provider,
+        litellm_params=litellm_params,
+        logging_obj=logging_obj,
+        user=_optional_str(request.get("user")),
+        extra_headers=with_provider_affinity_header(
+            ResponsesAPIRequestUtils.merge_client_forwarded_headers(
+                extra_headers=_string_keyed_dict(forwarded_kwargs.get("extra_headers")),
+                client_headers=_client_headers(forwarded_kwargs.get("headers")),
+            ),
+            litellm_params,
+            _string_keyed_dict(request.get("metadata")),
+            forwarded_kwargs,
+            provider_model,
+            custom_llm_provider,
+        ),
+        extra_body=_string_keyed_dict(forwarded_kwargs.get("extra_body")),
+        timeout=_timeout(forwarded_kwargs.get("timeout")),
+        is_async=is_async,
+        stream=request.get("stream") is True,
+        client=_http_client(forwarded_kwargs.get("client")),
+        shared_session=_shared_session(forwarded_kwargs.get("shared_session")),
+        litellm_metadata=_string_keyed_dict(forwarded_kwargs.get("litellm_metadata")) or {},
+        request_kwargs=dict(forwarded_kwargs),
+    )
+
+
+_OPENAI_CHAT_COMPLETIONS_MODEL_PREFIX: Final = "openai/chat_completions/"
+
+
+def _requests_chat_completions_api(params: Mapping[str, object]) -> bool:
+    model: Final = params.get("model")
+    return bool(params.get("use_chat_completions_api")) or (
+        isinstance(model, str) and model.startswith(_OPENAI_CHAT_COMPLETIONS_MODEL_PREFIX)
+    )
+
+
+def _native_logging_obj(
+    logging_obj: LiteLLMLoggingObj | None, forwarded_kwargs: Mapping[str, object], request: Mapping[str, object]
+) -> LiteLLMLoggingObj | None:
+    """The logging object when the native executor can serve the call, None when only `aresponses` can.
+
+    The executor skips the prompt management hooks `aresponses` runs, so a call that needs them re-enters it.
+    So does a deployment that asks for the chat completions API, which `aresponses` bridges to chat.
+    """
+    if logging_obj is None:
+        return None
+    params: Final = {**forwarded_kwargs, **request}
+    if _requests_chat_completions_api(params):
+        return None
+    if logging_obj.should_run_prompt_management_hooks(
+        non_default_params=params, prompt_id=_optional_str(params.get("prompt_id"))
+    ):
+        return None
+    return logging_obj
+
+
+def _reentrant_kwargs(
+    forwarded_kwargs: Mapping[str, object], request: Mapping[str, object], logging_obj: LiteLLMLoggingObj | None
+) -> dict[str, object]:
+    if logging_obj is not None:
+        setattr(logging_obj, "call_type", CallTypes.anthropic_messages.value)
+    return {**forwarded_kwargs, **request}
+
+
+def _raise_mapped(error: Exception, model: str, forwarded_kwargs: Mapping[str, object]) -> NoReturn:
+    raise litellm.exception_type(
+        model=model,
+        custom_llm_provider=_optional_str(forwarded_kwargs.get("custom_llm_provider")),
+        original_exception=error,
+        extra_kwargs=dict(forwarded_kwargs),
+    ) from error
+
+
+def _priced(
+    response: ResponsesAPIResponse | BaseResponsesAPIStreamingIterator, logging_obj: LiteLLMLoggingObj
+) -> ResponsesAPIResponse | BaseResponsesAPIStreamingIterator:
+    """The response, priced as `aresponses` would have priced it before the bridge translates it."""
+    if isinstance(response, ResponsesAPIResponse):
+        logging_obj.price_bridged_provider_response(response)
+    else:
+        logging_obj.price_bridged_provider_stream(response)
+    return response
+
+
+def _native_responses(
+    request: Mapping[str, object],
+    model: str,
+    forwarded_kwargs: Mapping[str, object],
+    logging_obj: LiteLLMLoggingObj,
+) -> ResponsesAPIResponse | BaseResponsesAPIStreamingIterator:
+    try:
+        response: Final = _native_responses_call(request, model, forwarded_kwargs, logging_obj, is_async=False)
+    except Exception as e:  # noqa: BLE001  # every provider failure maps to the litellm exception aresponses would raise
+        _raise_mapped(e, model, forwarded_kwargs)
+    if isinstance(response, Coroutine):
+        raise TypeError("The sync Responses executor returned a coroutine")
+    return _priced(response, logging_obj)
+
+
+async def _native_aresponses(
+    request: Mapping[str, object],
+    model: str,
+    forwarded_kwargs: Mapping[str, object],
+    logging_obj: LiteLLMLoggingObj,
+) -> ResponsesAPIResponse | BaseResponsesAPIStreamingIterator:
+    try:
+        response: Final = _native_responses_call(request, model, forwarded_kwargs, logging_obj, is_async=True)
+        settled: Final = await response if isinstance(response, Coroutine) else response
+    except Exception as e:  # noqa: BLE001  # every provider failure maps to the litellm exception aresponses would raise
+        _raise_mapped(e, model, forwarded_kwargs)
+    return _priced(settled, logging_obj)
 
 
 class LiteLLMMessagesToResponsesAPIHandler:
     """
     Handles Anthropic /v1/messages requests for OpenAI / Azure models by
-    calling litellm.responses() / litellm.aresponses() directly and translating
-    the response back to Anthropic format.
+    calling the native Responses executor and translating the response back
+    to Anthropic format.
     """
 
     @staticmethod
@@ -278,7 +492,7 @@ class LiteLLMMessagesToResponsesAPIHandler:
         output_format: AnthropicOutputSchema | None = None,
         **kwargs: object,
     ) -> AnthropicMessagesResponse | AsyncIterator[bytes]:
-        responses_kwargs: Final = _build_responses_kwargs(
+        request: Final = _build_responses_request(
             max_tokens=max_tokens,
             messages=messages,
             model=model,
@@ -297,15 +511,22 @@ class LiteLLMMessagesToResponsesAPIHandler:
             output_format=output_format,
             extra_kwargs=kwargs,
         )
+        forwarded_kwargs: Final = _forwarded_kwargs(kwargs)
+        logging_obj: Final = litellm_logging_obj_from_kwargs(kwargs)
 
         safeguards_evaluator: Final = request_safeguards_evaluator(kwargs, messages)
-        result: Final = await litellm.aresponses(**responses_kwargs)
+        native_logging_obj: Final = _native_logging_obj(logging_obj, forwarded_kwargs, request)
+        result: Final = (
+            await litellm.aresponses(**_reentrant_kwargs(forwarded_kwargs, request, logging_obj))
+            if native_logging_obj is None
+            else await _native_aresponses(request, model, forwarded_kwargs, native_logging_obj)
+        )
 
         if stream:
             wrapper: Final = AnthropicResponsesStreamWrapper(
                 responses_stream=result,
                 model=local_model_name(model, kwargs.get("custom_llm_provider")),
-                litellm_logging_obj=litellm_logging_obj_from_kwargs(responses_kwargs),
+                litellm_logging_obj=logging_obj,
                 safeguards_evaluator=safeguards_evaluator,
             )
             return wrapper.async_anthropic_sse_wrapper()
@@ -314,7 +535,7 @@ class LiteLLMMessagesToResponsesAPIHandler:
             raise ValueError(f"Expected ResponsesAPIResponse, got {type(result)}")
 
         served: Final = await with_safeguard_results(_ADAPTER.translate_response(result), safeguards_evaluator)
-        _record_bridged_response(responses_kwargs, result)
+        _record_bridged_response(kwargs, result)
         return served
 
     @staticmethod
@@ -363,8 +584,7 @@ class LiteLLMMessagesToResponsesAPIHandler:
                 **kwargs,
             )
 
-        # Sync path
-        responses_kwargs: Final = _build_responses_kwargs(
+        request: Final = _build_responses_request(
             max_tokens=max_tokens,
             messages=messages,
             model=model,
@@ -383,14 +603,20 @@ class LiteLLMMessagesToResponsesAPIHandler:
             output_format=output_format,
             extra_kwargs=kwargs,
         )
-
-        result: Final = litellm.responses(**responses_kwargs)
+        forwarded_kwargs: Final = _forwarded_kwargs(kwargs)
+        logging_obj: Final = litellm_logging_obj_from_kwargs(kwargs)
+        native_logging_obj: Final = _native_logging_obj(logging_obj, forwarded_kwargs, request)
+        result: Final = (
+            litellm.responses(**_reentrant_kwargs(forwarded_kwargs, request, logging_obj))
+            if native_logging_obj is None
+            else _native_responses(request, model, forwarded_kwargs, native_logging_obj)
+        )
 
         if stream:
             wrapper: Final = AnthropicResponsesStreamWrapper(
                 responses_stream=result,
                 model=local_model_name(model, kwargs.get("custom_llm_provider")),
-                litellm_logging_obj=litellm_logging_obj_from_kwargs(responses_kwargs),
+                litellm_logging_obj=logging_obj,
             )
             return wrapper.async_anthropic_sse_wrapper()
 
diff --git a/litellm/native/responses.py b/litellm/native/responses.py
index 14364df1b30..0feb452240c 100644
--- a/litellm/native/responses.py
+++ b/litellm/native/responses.py
@@ -106,6 +106,16 @@ def execute_responses(
         litellm_metadata=litellm_metadata,
         shared_session=shared_session,
     )
+    if isinstance(response, Coroutine):
+        return _stamped_when_done(response, litellm_metadata, custom_llm_provider)
+    return _stamped(response, litellm_metadata, custom_llm_provider)
+
+
+def _stamped(
+    response: ResponsesAPIResponse | BaseResponsesAPIStreamingIterator,
+    litellm_metadata: Mapping[str, object] | None,
+    custom_llm_provider: str,
+) -> ResponsesAPIResponse | BaseResponsesAPIStreamingIterator:
     if not isinstance(response, ResponsesAPIResponse):
         return response
     stamped: Final = ResponsesAPIRequestUtils.update_responses_api_response_id_with_model_id(
@@ -115,3 +125,11 @@ def execute_responses(
     )
     stamped.hidden_params["custom_llm_provider"] = custom_llm_provider
     return stamped
+
+
+async def _stamped_when_done(
+    response: Coroutine[object, object, ResponsesAPIResponse | BaseResponsesAPIStreamingIterator],
+    litellm_metadata: Mapping[str, object] | None,
+    custom_llm_provider: str,
+) -> ResponsesAPIResponse | BaseResponsesAPIStreamingIterator:
+    return _stamped(await response, litellm_metadata, custom_llm_provider)
diff --git a/litellm/responses/main.py b/litellm/responses/main.py
index 8b4fe4835a5..26b78297556 100644
--- a/litellm/responses/main.py
+++ b/litellm/responses/main.py
@@ -627,7 +627,7 @@ class _AsyncPromptManagementOutcome:
     deployment_model_info: object
 
 
-def _resolve_responses_api_provider_config(
+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(
@@ -672,7 +672,7 @@ def _will_bridge_to_chat_completions(
         return True
     return isinstance(
         plan_responses_route(
-            _resolve_responses_api_provider_config(
+            resolve_responses_api_provider_config(
                 _dispatched_model_name(normalized_model[0], custom_llm_provider, api_base),
                 custom_llm_provider,
                 model_info,
@@ -1132,7 +1132,69 @@ def _strip_responses_routing_prefix(model: str) -> str:
     return model[len(_RESPONSES_ROUTING_PREFIX) :]
 
 
-def _resolve_model_provider_for_responses(
+def with_provider_affinity_header(
+    extra_headers: dict[str, object] | None,
+    litellm_params: GenericLiteLLMParams,
+    metadata: Mapping[str, object] | None,
+    kwargs: Mapping[str, object],
+    model: str,
+    custom_llm_provider: str | None,
+) -> dict[str, object] | None:
+    if litellm_params.provider_affinity_header is None:
+        return extra_headers
+    try:
+        return 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"),
+                }
+            ),
+        )
+    except ValueError as affinity_error:
+        raise litellm.BadRequestError(
+            message=str(affinity_error), model=model, llm_provider=custom_llm_provider
+        ) from affinity_error
+
+
+def requested_reasoning(params: Mapping[str, object], model: str, custom_llm_provider: str | None) -> Reasoning | None:
+    current_reasoning: Final = cast(  # cast-ok: prompt-managed reasoning arrives as a plain dict
+        Reasoning | None, params.get("reasoning")
+    )
+    reasoning_effort: Final = params.get("reasoning_effort")
+    if current_reasoning is None and reasoning_effort is not None:
+        return LiteLLMResponsesTransformationHandler().map_reasoning_effort(
+            reasoning_effort, model, custom_llm_provider
+        )
+    return current_reasoning
+
+
+def reasoning_for_deployment(
+    reasoning: Reasoning | None,
+    model: str,
+    custom_llm_provider: str | None,
+    deployment_model_info: object,
+    kwargs: Mapping[str, object],
+) -> Reasoning | None:
+    """The requested reasoning with its effort resolved onto the deployment's declared reasoning_effort_levels."""
+    return cast(  # cast-ok: the resolver hands a Reasoning back in the shape it was given
+        Reasoning | None,
+        resolve_deployment_reasoning_effort(
+            reasoning,
+            model=model,
+            custom_llm_provider=custom_llm_provider,
+            deployment_model_info=deployment_model_info,
+            model_group=model_group_from_metadata(kwargs.get("litellm_metadata")),
+            drop_params=litellm.drop_params is True or normalize_drop_params(kwargs.get("drop_params")) is True,
+        ).value,
+    )
+
+
+def resolve_model_provider_for_responses(
     model: str,
     custom_llm_provider: str | None,
     litellm_params: GenericLiteLLMParams,
@@ -1157,7 +1219,7 @@ def _resolve_model_provider_for_responses(
     return _strip_responses_routing_prefix(provider_model), resolved_provider
 
 
-def _apply_managed_file_id_mapping(
+def apply_managed_file_id_mapping(
     input: str | ResponseInputParam,
     tools: Iterable[ToolParam] | None,
     kwargs: dict[str, Any],
@@ -1456,27 +1518,9 @@ def responses(
 
         # 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
+        effective_extra_headers: Final = with_provider_affinity_header(
+            extra_headers, litellm_params, metadata, kwargs, model, custom_llm_provider
+        )
 
         #########################################################
         # MOCK RESPONSE LOGIC
@@ -1484,7 +1528,7 @@ def responses(
         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, custom_llm_provider = resolve_model_provider_for_responses(
             model=model,
             custom_llm_provider=custom_llm_provider,
             litellm_params=litellm_params,
@@ -1494,7 +1538,7 @@ def responses(
         #########################################################
         # 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)
+        input, tools = apply_managed_file_id_mapping(input=input, tools=tools, kwargs=kwargs, local_vars=local_vars)
 
         #########################################################
         # Native MCP Responses API
@@ -1537,7 +1581,7 @@ def responses(
         if custom_llm_provider is None:
             responses_api_provider_config = None
         else:
-            responses_api_provider_config = _resolve_responses_api_provider_config(
+            responses_api_provider_config = resolve_responses_api_provider_config(
                 model, custom_llm_provider, deployment_model_info, litellm_params.api_base
             )
 
@@ -1553,27 +1597,9 @@ def responses(
             _raise_responses_compatibility_failure(compatibility_failure, model, custom_llm_provider)
 
         local_vars.update(kwargs)
-        current_reasoning: Final = cast(  # cast-ok: prompt-managed reasoning arrives as a plain dict
-            Reasoning | None, local_vars.get("reasoning")
-        )
-        reasoning_effort: Final[object] = local_vars.get("reasoning_effort")
-        request_reasoning: Final = (
-            LiteLLMResponsesTransformationHandler().map_reasoning_effort(reasoning_effort, model, custom_llm_provider)
-            if current_reasoning is None and reasoning_effort is not None
-            else current_reasoning
-        )
-        routing_kwargs: Final = cast("Mapping[str, object]", kwargs)  # cast-ok: kwargs is the untyped request dict
-        deployment_reasoning: Final = cast(  # cast-ok: the resolver hands a Reasoning back in the shape it was given
-            Reasoning | None,
-            resolve_deployment_reasoning_effort(
-                request_reasoning,
-                model=model,
-                custom_llm_provider=custom_llm_provider,
-                deployment_model_info=deployment_model_info,
-                model_group=model_group_from_metadata(routing_kwargs.get("litellm_metadata")),
-                drop_params=litellm.drop_params is True
-                or normalize_drop_params(routing_kwargs.get("drop_params")) is True,
-            ).value,
+        request_reasoning: Final = requested_reasoning(local_vars, model, custom_llm_provider)
+        deployment_reasoning: Final = reasoning_for_deployment(
+            request_reasoning, model, custom_llm_provider, deployment_model_info, kwargs
         )
         response_api_optional_params: Final[ResponsesAPIOptionalRequestParams] = (
             ResponsesAPIRequestUtils.get_requested_response_api_optional_param(
@@ -2407,7 +2433,7 @@ def compact_responses(
         # get llm provider logic
         litellm_params: Final = GenericLiteLLMParams(**kwargs)
 
-        model, custom_llm_provider = _resolve_model_provider_for_responses(
+        model, custom_llm_provider = resolve_model_provider_for_responses(
             model=model,
             custom_llm_provider=custom_llm_provider,
             litellm_params=litellm_params,
diff --git a/litellm/responses/utils.py b/litellm/responses/utils.py
index c026b2466be..3acdc198557 100644
--- a/litellm/responses/utils.py
+++ b/litellm/responses/utils.py
@@ -340,7 +340,7 @@ class ResponsesAPIRequestUtils:
     def update_responses_api_response_id_with_model_id(
         responses_api_response: ResponsesAPIResponse,
         custom_llm_provider: str | None,
-        litellm_metadata: dict[str, object] | None = None,
+        litellm_metadata: Mapping[str, object] | None = None,
     ) -> ResponsesAPIResponse: 
         ...
 
@@ -349,7 +349,7 @@ class ResponsesAPIRequestUtils:
     def update_responses_api_response_id_with_model_id(
         responses_api_response: dict[str, object],
         custom_llm_provider: str | None,
-        litellm_metadata: dict[str, object] | None = None,
+        litellm_metadata: Mapping[str, object] | None = None,
     ) -> dict[str, object]:
         ...
 
@@ -359,7 +359,7 @@ class ResponsesAPIRequestUtils:
     def update_responses_api_response_id_with_model_id(
         responses_api_response: ResponsesAPIResponse | dict[str, Any],
         custom_llm_provider: str | None,
-        litellm_metadata: dict[str, Any] | None = None,
+        litellm_metadata: Mapping[str, Any] | None = None,
     ) -> ResponsesAPIResponse | dict[str, object]:
         """Update the responses_api_response_id with model_id and custom_llm_provider.
 
@@ -1051,7 +1051,7 @@ class ResponsesAPIRequestUtils:
     def _update_container_ids_in_response(
         responses_api_response: ResponsesAPIResponse | dict[str, Any],
         custom_llm_provider: str | None,
-        litellm_metadata: dict[str, Any] | None = None,
+        litellm_metadata: Mapping[str, Any] | None = None,
     ) -> ResponsesAPIResponse | dict[str, object]:
         """Encode container IDs in the response output with provider/model info.
 
@@ -1241,7 +1241,7 @@ class ResponseAPILoggingUtils:
                 output_tokens: Final = usage_input.get("output_tokens")
                 if isinstance(input_tokens, int) and isinstance(output_tokens, int):
                     usage_input["total_tokens"] = input_tokens + output_tokens
-            response_api_usage = ResponseAPIUsage(**usage_input)
+            response_api_usage = ResponseAPIUsage.model_validate(usage_input)
         else:
             response_api_usage = usage_input
         prompt_tokens: Final[int] = response_api_usage.input_tokens or 0
diff --git a/tests/unit/llms/anthropic/pass_through/responses_adapters/test_responses_adapters_handler.py b/tests/unit/llms/anthropic/pass_through/responses_adapters/test_responses_adapters_handler.py
index 8fe6bd31082..d6a8a3e781e 100644
--- a/tests/unit/llms/anthropic/pass_through/responses_adapters/test_responses_adapters_handler.py
+++ b/tests/unit/llms/anthropic/pass_through/responses_adapters/test_responses_adapters_handler.py
@@ -2,28 +2,34 @@ import datetime
 import json
 import os
 import sys
-from collections.abc import Mapping
+from collections.abc import Callable, Coroutine, Mapping
 from typing import Final, cast
 from unittest.mock import AsyncMock, patch
 
+import httpx
 import pytest
 import respx
 
 sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../../../..")))
 
 import litellm
+from litellm.integrations.custom_prompt_management import CustomPromptManagement
+from litellm.litellm_core_utils.litellm_logging import Logging
 from litellm.litellm_core_utils.prompt_templates.common_utils import (
     TOOL_RESULT_IMAGE_BOUNDARY,
     TOOL_RESULT_IMAGE_PLACEHOLDER,
 )
 from litellm.llms.anthropic.pass_through.responses_adapters.handler import (
     LiteLLMMessagesToResponsesAPIHandler,
-    _build_responses_kwargs,
+    _build_responses_request,
 )
 from litellm.router import Router
-from litellm.types.integrations.custom_logger import CONVERTED_STREAM_KEYS, converted_stream_requested
+from litellm.types.integrations.custom_logger import CONVERTED_STREAM_KEYS
+from litellm.types.llms.openai import AllMessageValues
+from litellm.types.prompts.init_prompts import PromptSpec
+from litellm.types.utils import StandardCallbackDynamicParams
 
-MESSAGES = [{"role": "user", "content": "hello"}]
+MESSAGES: Final[list[dict[str, object]]] = [{"role": "user", "content": "hello"}]
 CLAUDE_CODE_USER_ID = json.dumps({"device_id": "d" * 64, "account_uuid": "", "session_id": "session-abc"})
 
 RESPONSES_SSE_BODY = (
@@ -39,20 +45,8 @@ RESPONSES_SSE_BODY = (
 )
 
 
-def test_build_responses_kwargs_never_asks_the_nested_call_to_replay_a_stream():
-    responses_kwargs: Final = _build_responses_kwargs(
-        max_tokens=1024,
-        messages=MESSAGES,
-        model="azure/gpt-5.6-luna",
-        extra_kwargs={"custom_llm_provider": "azure", **{key: True for key in CONVERTED_STREAM_KEYS}},
-    )
-
-    assert not converted_stream_requested(responses_kwargs)
-    assert responses_kwargs["custom_llm_provider"] == "azure"
-
-
-def test_build_responses_kwargs_derives_prompt_cache_key_from_claude_code_session_id():
-    responses_kwargs = _build_responses_kwargs(
+def test_build_responses_request_derives_prompt_cache_key_from_claude_code_session_id():
+    responses_kwargs = _build_responses_request(
         max_tokens=1024,
         messages=MESSAGES,
         model="openai/gpt-5.6-luna",
@@ -63,8 +57,8 @@ def test_build_responses_kwargs_derives_prompt_cache_key_from_claude_code_sessio
     assert responses_kwargs["prompt_cache_key"] == "session-abc"
 
 
-def test_build_responses_kwargs_sets_no_prompt_cache_key_for_plain_user_id():
-    responses_kwargs = _build_responses_kwargs(
+def test_build_responses_request_sets_no_prompt_cache_key_for_plain_user_id():
+    responses_kwargs = _build_responses_request(
         max_tokens=1024,
         messages=MESSAGES,
         model="openai/gpt-5.6-luna",
@@ -75,8 +69,8 @@ def test_build_responses_kwargs_sets_no_prompt_cache_key_for_plain_user_id():
     assert "prompt_cache_key" not in responses_kwargs
 
 
-def test_build_responses_kwargs_prefers_explicit_prompt_cache_key_over_derived():
-    responses_kwargs = _build_responses_kwargs(
+def test_build_responses_request_prefers_explicit_prompt_cache_key_over_derived():
+    responses_kwargs = _build_responses_request(
         max_tokens=1024,
         messages=MESSAGES,
         model="openai/gpt-5.6-luna",
@@ -87,8 +81,8 @@ def test_build_responses_kwargs_prefers_explicit_prompt_cache_key_over_derived()
     assert responses_kwargs["prompt_cache_key"] == "explicit-key"
 
 
-def test_build_responses_kwargs_asks_openai_for_encrypted_reasoning_without_thinking():
-    responses_kwargs = _build_responses_kwargs(
+def test_build_responses_request_asks_openai_for_encrypted_reasoning_without_thinking():
+    responses_kwargs = _build_responses_request(
         max_tokens=1024,
         messages=MESSAGES,
         model="openai/gpt-5.6-luna",
@@ -98,8 +92,8 @@ def test_build_responses_kwargs_asks_openai_for_encrypted_reasoning_without_thin
     assert "reasoning" not in responses_kwargs
 
 
-def test_build_responses_kwargs_skips_include_for_a_responses_provider_that_rejects_it():
-    responses_kwargs = _build_responses_kwargs(
+def test_build_responses_request_skips_include_for_a_responses_provider_that_rejects_it():
+    responses_kwargs = _build_responses_request(
         max_tokens=1024,
         messages=MESSAGES,
         model="perplexity/sonar",
@@ -117,11 +111,11 @@ def test_build_responses_kwargs_skips_include_for_a_responses_provider_that_reje
         ("openai/gpt-5.6-luna", "openai", "detailed"),
     ],
 )
-def test_build_responses_kwargs_auto_summary_asks_each_target_for_a_level_it_accepts(
+def test_build_responses_request_auto_summary_asks_each_target_for_a_level_it_accepts(
     monkeypatch: pytest.MonkeyPatch, model: str, provider: str, summary: str
 ) -> None:
     monkeypatch.setenv("LITELLM_REASONING_AUTO_SUMMARY", "true")
-    responses_kwargs: Final = _build_responses_kwargs(
+    responses_kwargs: Final = _build_responses_request(
         max_tokens=1024,
         messages=MESSAGES,
         model=model,
@@ -131,8 +125,8 @@ def test_build_responses_kwargs_auto_summary_asks_each_target_for_a_level_it_acc
     assert cast(Mapping[str, object], responses_kwargs["reasoning"])["summary"] == summary
 
 
-def test_build_responses_kwargs_keeps_the_deployment_include_next_to_encrypted_reasoning():
-    responses_kwargs = _build_responses_kwargs(
+def test_build_responses_request_keeps_the_deployment_include_next_to_encrypted_reasoning():
+    responses_kwargs = _build_responses_request(
         max_tokens=1024,
         messages=MESSAGES,
         model="openai/gpt-5.6-luna",
@@ -141,8 +135,8 @@ def test_build_responses_kwargs_keeps_the_deployment_include_next_to_encrypted_r
     assert responses_kwargs["include"] == ["reasoning.encrypted_content", "file_search_call.results"]
 
 
-def test_build_responses_kwargs_without_metadata_sets_no_prompt_cache_key():
-    responses_kwargs = _build_responses_kwargs(
+def test_build_responses_request_without_metadata_sets_no_prompt_cache_key():
+    responses_kwargs = _build_responses_request(
         max_tokens=1024,
         messages=MESSAGES,
         model="openai/gpt-5.6-luna",
@@ -193,8 +187,6 @@ async def test_streaming_hands_the_logging_object_the_message_id_the_caller_is_s
     The bridge mints the ``msg_`` id itself, and it is the only request id a streaming
     /v1/messages caller ever sees, so the spend row has to be keyed on that same value.
     """
-    from litellm.litellm_core_utils.litellm_logging import Logging
-
     monkeypatch.setenv("OPENAI_API_KEY", "sk-lit6825-test")
     monkeypatch.setenv("DISABLE_AIOHTTP_TRANSPORT", "True")
     litellm.in_memory_llm_clients_cache.flush_cache()
@@ -424,7 +416,7 @@ def _request_with_client_cache_marks(cache_control: dict[str, str] | None) -> di
 def test_client_cache_control_reaches_gpt_5_6_as_a_prompt_cache_breakpoint_on_the_same_block(
     extra_kwargs: dict[str, object],
 ) -> None:
-    responses_kwargs: Final = _build_responses_kwargs(
+    responses_kwargs: Final = _build_responses_request(
         model="openai/gpt-5.6",
         extra_kwargs=extra_kwargs,
         **_request_with_client_cache_marks({**EPHEMERAL, "ttl": "1h"}),
@@ -452,7 +444,7 @@ def test_client_cache_control_reaches_gpt_5_6_as_a_prompt_cache_breakpoint_on_th
 
 
 def test_client_cache_control_on_a_screenshot_inside_a_tool_result_reaches_gpt_5_6_on_that_screenshot() -> None:
-    responses_kwargs: Final = _build_responses_kwargs(
+    responses_kwargs: Final = _build_responses_request(
         model="openai/gpt-5.6",
         extra_kwargs=OPENAI_API,
         max_tokens=1024,
@@ -531,7 +523,7 @@ RAW_QUESTION: Final = {"type": "input_text", "text": "q", "prompt_cache_breakpoi
 def test_raw_prompt_cache_breakpoints_reach_a_deployment_unless_drop_params_drops_them_for_a_model_that_takes_none(
     model: str, extra_kwargs: dict[str, object], log_output: object, question: dict[str, object]
 ) -> None:
-    responses_kwargs: Final = _build_responses_kwargs(
+    responses_kwargs: Final = _build_responses_request(
         model=model,
         extra_kwargs=extra_kwargs,
         max_tokens=1024,
@@ -566,10 +558,10 @@ def test_raw_prompt_cache_breakpoints_reach_a_deployment_unless_drop_params_drop
 def test_client_cache_control_changes_nothing_for_a_deployment_that_does_not_read_prompt_cache_breakpoint(
     model: str, extra_kwargs: dict[str, object]
 ) -> None:
-    marked: Final = _build_responses_kwargs(
+    marked: Final = _build_responses_request(
         model=model, extra_kwargs=extra_kwargs, **_request_with_client_cache_marks(EPHEMERAL)
     )
-    unmarked: Final = _build_responses_kwargs(
+    unmarked: Final = _build_responses_request(
         model=model, extra_kwargs=extra_kwargs, **_request_with_client_cache_marks(None)
     )
 
@@ -600,7 +592,7 @@ def _tool_result_turn_with(text_block: Mapping[str, object]) -> list[dict[str, o
 @pytest.mark.parametrize("text_block", MALFORMED_TEXT_BLOCKS)
 def test_malformed_tool_result_text_answers_400_naming_the_block(text_block: Mapping[str, object]) -> None:
     with pytest.raises(litellm.BadRequestError) as rejected:
-        _build_responses_kwargs(
+        _build_responses_request(
             model="openai/gpt-5.6",
             extra_kwargs=OPENAI_API,
             max_tokens=1024,
@@ -627,7 +619,7 @@ def test_many_malformed_tool_result_texts_answer_one_bounded_error() -> None:
     ]
 
     with pytest.raises(litellm.BadRequestError) as rejected:
-        _build_responses_kwargs(model="openai/gpt-5.6", extra_kwargs=OPENAI_API, max_tokens=1024, messages=messages)
+        _build_responses_request(model="openai/gpt-5.6", extra_kwargs=OPENAI_API, max_tokens=1024, messages=messages)
 
     assert rejected.value.status_code == 400
     assert f"tool_result {tool_use_id} content.0.text: Input should be a valid string" in rejected.value.message
@@ -636,7 +628,7 @@ def test_many_malformed_tool_result_texts_answer_one_bounded_error() -> None:
 
 @pytest.mark.parametrize("text_block", MALFORMED_TEXT_BLOCKS)
 def test_drop_params_drops_a_malformed_tool_result_text(text_block: Mapping[str, object]) -> None:
-    responses_kwargs: Final = _build_responses_kwargs(
+    responses_kwargs: Final = _build_responses_request(
         model="openai/gpt-5.6",
         extra_kwargs={**OPENAI_API, "drop_params": True},
         max_tokens=1024,
@@ -649,7 +641,7 @@ def test_drop_params_drops_a_malformed_tool_result_text(text_block: Mapping[str,
 
 
 def test_tool_result_text_block_without_text_still_reaches_the_model_as_empty_text() -> None:
-    responses_kwargs: Final = _build_responses_kwargs(
+    responses_kwargs: Final = _build_responses_request(
         model="openai/gpt-5.6",
         extra_kwargs=OPENAI_API,
         max_tokens=1024,
@@ -739,3 +731,224 @@ async def test_recomputed_cost_of_a_served_bridge_response_prices_the_reasoning_
     assert logging_obj.cost_breakdown is not None
     assert logging_obj.cost_breakdown["output_cost"] == pytest.approx(expected_output)
     assert logging_obj.cost_breakdown["reasoning_cost"] == pytest.approx(expected_reasoning)
+
+
+SEARCH_MODEL: Final = "gpt-bridge-search-cost"
+SEARCH_RATES: Final = {"input_cost_per_token": 2e-06, "output_cost_per_token": 1e-05}
+SEARCH_COST_PER_QUERY: Final = 0.03
+SEARCH_RESPONSE: Final = {
+    **REASONING_RESPONSE,
+    "id": "resp_bridge_search",
+    "model": SEARCH_MODEL,
+    "output": [
+        *(
+            {
+                "type": "web_search_call",
+                "id": f"ws_{i}",
+                "status": "completed",
+                "action": {"type": "search", "query": "q"},
+            }
+            for i in range(3)
+        ),
+        REASONING_RESPONSE["output"][1],
+    ],
+    "usage": {"input_tokens": 100, "output_tokens": 20, "total_tokens": 120},
+}
+
+
+@pytest.mark.asyncio
+async def test_served_bridge_response_is_billed_for_each_web_search_call_the_provider_made(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
+):
+    from litellm.proxy.common_request_processing import ProxyBaseLLMRequestProcessing
+
+    monkeypatch.setenv("OPENAI_API_KEY", "sk-bridge-search-test")
+    monkeypatch.setenv("DISABLE_AIOHTTP_TRANSPORT", "True")
+    monkeypatch.setitem(
+        litellm.model_cost,
+        SEARCH_MODEL,
+        {
+            "litellm_provider": "openai",
+            "mode": "responses",
+            "search_context_cost_per_query": {
+                f"search_context_size_{size}": SEARCH_COST_PER_QUERY for size in ("low", "medium", "high")
+            },
+            **SEARCH_RATES,
+        },
+    )
+    litellm.in_memory_llm_clients_cache.flush_cache()
+    respx_mock.post("https://api.openai.com/v1/responses").respond(status_code=200, json=SEARCH_RESPONSE)
+    logging_obj: Final = Logging(
+        model=SEARCH_MODEL,
+        messages=MESSAGES,
+        stream=False,
+        call_type="anthropic_messages",
+        start_time=datetime.datetime.now(datetime.timezone.utc),
+        litellm_call_id="bridge-search-0000-4000-8000-000000000001",
+        function_id="1234",
+    )
+
+    served: Final = await LiteLLMMessagesToResponsesAPIHandler.async_anthropic_messages_handler(
+        max_tokens=412,
+        messages=MESSAGES,
+        model=f"openai/{SEARCH_MODEL}",
+        custom_llm_provider="openai",
+        tools=[{"type": "web_search_20250305", "name": "web_search"}],
+        litellm_logging_obj=logging_obj,
+    )
+    assert isinstance(served, dict)
+
+    cost: Final = ProxyBaseLLMRequestProcessing._response_cost_from_logging_obj(
+        response=dict(served), logging_obj=logging_obj
+    )
+
+    expected_tokens: Final = 100 * SEARCH_RATES["input_cost_per_token"] + 20 * SEARCH_RATES["output_cost_per_token"]
+    assert cost == pytest.approx(expected_tokens + 3 * SEARCH_COST_PER_QUERY)
+    assert logging_obj.cost_breakdown is not None
+    assert logging_obj.cost_breakdown["tool_usage_cost"] == pytest.approx(3 * SEARCH_COST_PER_QUERY)
+
+
+def _messages_logging_obj(stream: bool) -> Logging:
+    return Logging(
+        model="gpt-5.6-luna",
+        messages=MESSAGES,
+        stream=stream,
+        call_type="anthropic_messages",
+        start_time=datetime.datetime.now(datetime.timezone.utc),
+        litellm_call_id="6825beef-0000-4000-8000-000000000004",
+        function_id="1234",
+    )
+
+
+def _untyped_bridge() -> Callable[..., Coroutine[object, object, object]]:
+    return LiteLLMMessagesToResponsesAPIHandler.async_anthropic_messages_handler
+
+
+async def _logged_bridge_call(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, upstream_response: httpx.Response, **extra: object
+) -> tuple[object, respx.Route]:
+    monkeypatch.setenv("OPENAI_API_KEY", "sk-native-test")
+    monkeypatch.setenv("DISABLE_AIOHTTP_TRANSPORT", "True")
+    litellm.in_memory_llm_clients_cache.flush_cache()
+    upstream: Final = respx_mock.post("https://api.openai.com/v1/responses").mock(return_value=upstream_response)
+    response: Final = await _untyped_bridge()(
+        max_tokens=1024,
+        messages=MESSAGES,
+        model="openai/gpt-5.6-luna",
+        custom_llm_provider="openai",
+        litellm_logging_obj=_messages_logging_obj(stream=False),
+        **extra,
+    )
+    return response, upstream
+
+
+@pytest.mark.asyncio
+async def test_bridge_answers_a_non_stream_request_with_a_message_even_when_the_caller_asked_for_stream_replay(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
+):
+    response, upstream = await _logged_bridge_call(
+        respx_mock,
+        monkeypatch,
+        httpx.Response(200, json=TOOL_CALL_RESPONSE),
+        **{key: True for key in CONVERTED_STREAM_KEYS},
+    )
+
+    assert isinstance(response, dict) and response["type"] == "message"
+    assert json.loads(upstream.calls.last.request.content).get("stream") is not True
+
+
+@pytest.mark.asyncio
+async def test_bridge_sends_the_deployment_prompt_cache_options_upstream(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
+):
+    _, upstream = await _logged_bridge_call(
+        respx_mock,
+        monkeypatch,
+        httpx.Response(200, json=TOOL_CALL_RESPONSE),
+        prompt_cache_options={"mode": "explicit"},
+    )
+
+    assert json.loads(upstream.calls.last.request.content)["prompt_cache_options"] == {"mode": "explicit"}
+
+
+@pytest.mark.asyncio
+async def test_bridge_raises_the_litellm_rate_limit_error_for_an_upstream_429(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
+):
+    with pytest.raises(litellm.RateLimitError):
+        await _logged_bridge_call(
+            respx_mock,
+            monkeypatch,
+            httpx.Response(429, json={"error": {"message": "slow down", "type": "rate_limit_error"}}),
+        )
+
+
+class _TemplatePromptManager(CustomPromptManagement):
+    async def async_get_chat_completion_prompt(
+        self,
+        model: str,
+        messages: list[AllMessageValues],
+        non_default_params: dict[str, object],
+        prompt_id: str | None,
+        prompt_variables: dict[str, object] | None,
+        dynamic_callback_params: StandardCallbackDynamicParams,
+        litellm_logging_obj: Logging,
+        prompt_spec: PromptSpec | None = None,
+        tools: list[dict[str, object]] | None = None,
+        prompt_label: str | None = None,
+        prompt_version: int | None = None,
+        ignore_prompt_manager_model: bool | None = False,
+        ignore_prompt_manager_optional_params: bool | None = False,
+    ) -> tuple[str, list[AllMessageValues], dict[str, object]]:
+        return model, [{"role": "user", "content": "from the template"}], non_default_params
+
+
+@pytest.mark.asyncio
+async def test_bridge_applies_the_prompt_template_named_by_prompt_id(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
+):
+    monkeypatch.setattr(litellm, "callbacks", [_TemplatePromptManager()])
+
+    _, upstream = await _logged_bridge_call(
+        respx_mock, monkeypatch, httpx.Response(200, json=TOOL_CALL_RESPONSE), prompt_id="bridge_template"
+    )
+
+    assert "from the template" in upstream.calls.last.request.content.decode()
+
+
+def _logged_sync_bridge_call(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, upstream_response: httpx.Response
+) -> tuple[object, respx.Route]:
+    monkeypatch.setenv("OPENAI_API_KEY", "sk-native-test")
+    monkeypatch.setenv("DISABLE_AIOHTTP_TRANSPORT", "True")
+    litellm.in_memory_llm_clients_cache.flush_cache()
+    upstream: Final = respx_mock.post("https://api.openai.com/v1/responses").mock(return_value=upstream_response)
+    response: Final = LiteLLMMessagesToResponsesAPIHandler.anthropic_messages_handler(
+        max_tokens=1024,
+        messages=MESSAGES,
+        model="openai/gpt-5.6-luna",
+        custom_llm_provider="openai",
+        litellm_logging_obj=_messages_logging_obj(stream=False),
+    )
+    return response, upstream
+
+
+def test_sync_bridge_answers_with_a_message_from_one_upstream_call(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
+):
+    response, upstream = _logged_sync_bridge_call(respx_mock, monkeypatch, httpx.Response(200, json=TOOL_CALL_RESPONSE))
+
+    assert isinstance(response, dict) and response["type"] == "message"
+    assert [block["id"] for block in response["content"] if block["type"] == "tool_use"] == ["call_curl"]
+    assert upstream.call_count == 1
+
+
+def test_sync_bridge_raises_the_litellm_rate_limit_error_for_an_upstream_429(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
+):
+    with pytest.raises(litellm.RateLimitError):
+        _logged_sync_bridge_call(
+            respx_mock,
+            monkeypatch,
+            httpx.Response(429, json={"error": {"message": "slow down", "type": "rate_limit_error"}}),
+        )
diff --git a/tests/unit/llms/anthropic/pass_through/responses_adapters/test_responses_adapters_transformation.py b/tests/unit/llms/anthropic/pass_through/responses_adapters/test_responses_adapters_transformation.py
index 829000c0c5f..914141ca4fe 100644
--- a/tests/unit/llms/anthropic/pass_through/responses_adapters/test_responses_adapters_transformation.py
+++ b/tests/unit/llms/anthropic/pass_through/responses_adapters/test_responses_adapters_transformation.py
@@ -2411,16 +2411,3 @@ class TestPromptCacheBreakpointToResponses:
             self.MARKED_SCREENSHOT_PART,
         ]
         assert log_output == self.LOG_OUTPUT
-
-    def test_prompt_cache_options_forwarded_to_responses_kwargs(self):
-        from litellm.llms.anthropic.pass_through.responses_adapters.handler import (
-            _build_responses_kwargs,
-        )
-
-        kwargs = _build_responses_kwargs(
-            max_tokens=16,
-            messages=[{"role": "user", "content": "hi"}],
-            model="openai/gpt-5.6",
-            extra_kwargs={"prompt_cache_options": {"mode": "explicit"}},
-        )
-        assert kwargs["prompt_cache_options"] == {"mode": "explicit"}
diff --git a/tests/unit/responses/test_bridge_executor.py b/tests/unit/responses/test_bridge_executor.py
new file mode 100644
index 00000000000..74e69fb6d24
--- /dev/null
+++ b/tests/unit/responses/test_bridge_executor.py
@@ -0,0 +1,64 @@
+import json
+from typing import Final
+
+import httpx
+import pytest
+import respx
+
+import litellm
+
+API_BASE: Final = "https://bridge-executor.invalid/v1"
+_CHAT_REPLY: Final = {
+    "id": "chatcmpl-1",
+    "object": "chat.completion",
+    "created": 1,
+    "model": "m",
+    "choices": [{"index": 0, "message": {"role": "assistant", "content": "hi"}, "finish_reason": "stop"}],
+    "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15},
+}
+
+
+@pytest.fixture(autouse=True)
+def _offline(monkeypatch: pytest.MonkeyPatch):
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    litellm.in_memory_llm_clients_cache.flush_cache()
+
+
+@pytest.mark.parametrize(
+    ("surface", "litellm_params", "model_info", "upstream_model"),
+    [
+        ("messages", {"model": "openai/gpt-4.1", "use_chat_completions_api": True}, None, "gpt-4.1"),
+        ("messages", {"model": "openai/gpt-4.1", "use_chat_completions_api": True}, {"mode": "responses"}, "gpt-4.1"),
+        ("messages", {"model": "openai/chat_completions/o3-pro"}, None, "o3-pro"),
+        ("responses", {"model": "openai/chat_completions/o3-pro"}, None, "o3-pro"),
+    ],
+)
+@pytest.mark.asyncio
+async def test_a_deployment_asking_for_chat_completions_is_sent_to_chat_without_the_prefix(
+    surface: str,
+    litellm_params: dict[str, object],
+    model_info: dict[str, object] | None,
+    upstream_model: str,
+    respx_mock: respx.MockRouter,
+):
+    chat: Final = respx_mock.post(f"{API_BASE}/chat/completions").mock(
+        return_value=httpx.Response(200, json=_CHAT_REPLY)
+    )
+    router: Final = litellm.Router(
+        model_list=[
+            {
+                "model_name": "m",
+                "litellm_params": {"api_base": API_BASE, "api_key": "k", **litellm_params},
+                **({"model_info": model_info} if model_info is not None else {}),
+            }
+        ],
+        num_retries=0,
+    )
+
+    if surface == "messages":
+        await router.aanthropic_messages(model="m", max_tokens=32, messages=[{"role": "user", "content": "hi"}])
+    else:
+        await router.aresponses(model="m", input="hi")
+
+    assert chat.call_count == 1
+    assert json.loads(chat.calls.last.request.content)["model"] == upstream_model
diff --git a/tests/unit/responses/test_responses_supported_endpoints_passthrough.py b/tests/unit/responses/test_responses_supported_endpoints_passthrough.py
index 4ff4963423e..a0bb774a1f6 100644
--- a/tests/unit/responses/test_responses_supported_endpoints_passthrough.py
+++ b/tests/unit/responses/test_responses_supported_endpoints_passthrough.py
@@ -15,7 +15,7 @@ import litellm
 from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
 from litellm.llms.openai.responses.transformation import OpenAIResponsesAPIConfig
 from litellm.llms.openai_like.responses.transformation import OpenAILikeResponsesConfig
-from litellm.responses.main import _resolve_responses_api_provider_config
+from litellm.responses.main import resolve_responses_api_provider_config
 from litellm.types.llms.openai import ResponsesAPIResponse
 from litellm.types.utils import ModelResponse
 
@@ -115,13 +115,13 @@ def _respx_interceptable_httpx_client(monkeypatch):
     ],
 )
 def test_resolver_opt_in_gates_openai_like_config(model_info, expected_type):
-    config = _resolve_responses_api_provider_config("my-model", "custom_openai", model_info, None)
+    config = resolve_responses_api_provider_config("my-model", "custom_openai", model_info, None)
     assert type(config) is expected_type
 
 
 def test_resolver_keeps_native_provider_config():
     """`openai/` already routes /v1/responses natively; the opt-in must not swap its config."""
-    config = _resolve_responses_api_provider_config("gpt-4.1", "openai", OPT_IN, None)
+    config = resolve_responses_api_provider_config("gpt-4.1", "openai", OPT_IN, None)
     assert type(config) is OpenAIResponsesAPIConfig