diff --git a/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py b/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py
index 0f676a1feb2..e2f7646c0aa 100644
--- a/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py
+++ b/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py
@@ -40,6 +40,7 @@ from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
 from litellm.llms.custom_httpx.http_handler import get_async_httpx_client
 from litellm.passthrough import BasePassthroughUtils
 from litellm.proxy._types import (
+    CommonProxyErrors,
     ConfigFieldInfo,
     ConfigFieldUpdate,
     LiteLLMRoutes,
@@ -2155,6 +2156,102 @@ def _get_combined_pass_through_endpoints(
     return pass_through_endpoints + config_pass_through_endpoints
 
 
+async def _register_pass_through_endpoint(
+    endpoint: Union[Dict[str, Any], PassThroughGenericEndpoint],
+    app: FastAPI,
+    premium_user: bool,
+    visited_endpoints: set[str],
+) -> None:
+    endpoint_data: Dict[str, Any]
+    if isinstance(endpoint, PassThroughGenericEndpoint):
+        endpoint_data = endpoint.model_dump()
+    else:
+        endpoint_data = endpoint
+
+    if endpoint_data.get("id") is None:
+        endpoint_data["id"] = str(uuid.uuid4())
+    endpoint_id = cast(str, endpoint_data["id"])
+
+    target = endpoint_data.get("target")
+    path = endpoint_data.get("path")
+    if path is None:
+        raise ValueError("Path is required for pass-through endpoint")
+
+    custom_headers = await set_env_variables_in_header(
+        custom_headers=endpoint_data.get("headers")
+    )
+    forward_headers = endpoint_data.get("forward_headers")
+    merge_query_params = endpoint_data.get("merge_query_params")
+    default_query_params = endpoint_data.get("default_query_params")
+    auth = endpoint_data.get("auth")
+    dependencies = None
+
+    if auth is not None and str(auth).lower() == "true":
+        if premium_user is not True:
+            raise ValueError(
+                "Error Setting Authentication on Pass Through Endpoint: {}".format(
+                    CommonProxyErrors.not_premium_user.value
+                )
+            )
+        dependencies = [Depends(user_api_key_auth)]
+        if path not in LiteLLMRoutes.openai_routes.value:
+            LiteLLMRoutes.openai_routes.value.append(path)
+
+    if target is None:
+        return
+
+    guardrails = endpoint_data.get("guardrails")
+    methods = endpoint_data.get("methods")
+    cost_per_request = endpoint_data.get("cost_per_request")
+
+    verbose_proxy_logger.debug(
+        "Initializing pass through endpoint: %s (ID: %s)", path, endpoint_id
+    )
+    InitPassThroughEndpointHelpers.add_exact_path_route(
+        app=app,
+        path=path,
+        target=target,
+        custom_headers=custom_headers,
+        forward_headers=forward_headers,
+        merge_query_params=merge_query_params,
+        dependencies=dependencies,
+        cost_per_request=cost_per_request,
+        endpoint_id=endpoint_id,
+        guardrails=guardrails,
+        methods=methods,
+        default_query_params=default_query_params,
+    )
+
+    methods_for_key = methods if methods else ["GET", "POST", "PUT", "DELETE", "PATCH"]
+    methods_str = ",".join(sorted(methods_for_key))
+    visited_endpoints.add(f"{endpoint_id}:exact:{path}:{methods_str}")
+
+    if endpoint_data.get("include_subpath", False) is True:
+        if auth is not None and str(auth).lower() == "true":
+            wildcard_path = path.rstrip("/") + "/*"
+            if wildcard_path not in LiteLLMRoutes.openai_routes.value:
+                LiteLLMRoutes.openai_routes.value.append(wildcard_path)
+        InitPassThroughEndpointHelpers.add_subpath_route(
+            app=app,
+            path=path,
+            target=target,
+            custom_headers=custom_headers,
+            forward_headers=forward_headers,
+            merge_query_params=merge_query_params,
+            dependencies=dependencies,
+            cost_per_request=cost_per_request,
+            endpoint_id=endpoint_id,
+            guardrails=guardrails,
+            methods=methods,
+            default_query_params=default_query_params,
+        )
+        visited_endpoints.add(f"{endpoint_id}:subpath:{path}:{methods_str}")
+
+    verbose_proxy_logger.debug(
+        "Added new pass through endpoint: %s (ID: %s)", path, endpoint_id
+    )
+
+
 async def initialize_pass_through_endpoints(
     pass_through_endpoints: Union[List[Dict], List[PassThroughGenericEndpoint]],
 ):
@@ -2171,10 +2268,7 @@ async def initialize_pass_through_endpoints(
     Returns:
         None
     """
-    from litellm._uuid import uuid
-
     verbose_proxy_logger.debug("initializing pass through endpoints")
-    from litellm.proxy._types import CommonProxyErrors, LiteLLMRoutes
     from litellm.proxy.proxy_server import (
         app,
         config_passthrough_endpoints,
@@ -2201,105 +2295,14 @@ async def initialize_pass_through_endpoints(
         InitPassThroughEndpointHelpers.get_all_registered_pass_through_routes()
     )
 
-    visited_endpoints = set()
+    visited_endpoints: set[str] = set()
 
     for endpoint in combined_pass_through_endpoints:
-        if isinstance(endpoint, PassThroughGenericEndpoint):
-            endpoint = endpoint.model_dump()
-
-        # Auto-generate ID for backwards compatibility if not present
-        if endpoint.get("id") is None:
-            endpoint["id"] = str(uuid.uuid4())
-
-        # Get the endpoint_id as a string (guaranteed to be set at this point)
-        endpoint_id: str = endpoint["id"]
-
-        _target = endpoint.get("target", None)
-        _path: Optional[str] = endpoint.get("path", None)
-        if _path is None:
-            raise ValueError("Path is required for pass-through endpoint")
-        _custom_headers = endpoint.get("headers", None)
-        _custom_headers = await set_env_variables_in_header(
-            custom_headers=_custom_headers
-        )
-        _forward_headers = endpoint.get("forward_headers", None)
-        _merge_query_params = endpoint.get("merge_query_params", None)
-        _default_query_params = endpoint.get("default_query_params", None)
-        _auth = endpoint.get("auth", None)
-        _dependencies = None
-        if _auth is not None and str(_auth).lower() == "true":
-            if premium_user is not True:
-                raise ValueError(
-                    "Error Setting Authentication on Pass Through Endpoint: {}".format(
-                        CommonProxyErrors.not_premium_user.value
-                    )
-                )
-            _dependencies = [Depends(user_api_key_auth)]
-            if _path not in LiteLLMRoutes.openai_routes.value:
-                LiteLLMRoutes.openai_routes.value.append(_path)
-
-        if _target is None:
-            continue
-
-        # Get guardrails config if present
-        _guardrails = endpoint.get("guardrails", None)
-
-        # Get methods list if present (None means all methods for backward compatibility)
-        _methods = endpoint.get("methods", None)
-
-        # Add exact path route
-        verbose_proxy_logger.debug(
-            "Initializing pass through endpoint: %s (ID: %s)", _path, endpoint_id
-        )
-        InitPassThroughEndpointHelpers.add_exact_path_route(
+        await _register_pass_through_endpoint(
+            endpoint=endpoint,
             app=app,
-            path=_path,
-            target=_target,
-            custom_headers=_custom_headers,
-            forward_headers=_forward_headers,
-            merge_query_params=_merge_query_params,
-            dependencies=_dependencies,
-            cost_per_request=endpoint.get("cost_per_request", None),
-            endpoint_id=endpoint_id,
-            guardrails=_guardrails,
-            methods=_methods,
-            default_query_params=_default_query_params,
-        )
-
-        # Generate route key with methods for tracking
-        methods_for_key = (
-            _methods if _methods else ["GET", "POST", "PUT", "DELETE", "PATCH"]
-        )
-        methods_str = ",".join(sorted(methods_for_key))
-        visited_endpoints.add(f"{endpoint_id}:exact:{_path}:{methods_str}")
-
-        # Add wildcard route for sub-paths
-        if endpoint.get("include_subpath", False) is True:
-            # Register wildcard path in openai_routes so non-admin users
-            # can access subpath routes when auth is enabled
-            if _auth is not None and str(_auth).lower() == "true":
-                _wildcard_path = _path.rstrip("/") + "/*"
-                if _wildcard_path not in LiteLLMRoutes.openai_routes.value:
-                    LiteLLMRoutes.openai_routes.value.append(_wildcard_path)
-            InitPassThroughEndpointHelpers.add_subpath_route(
-                app=app,
-                path=_path,
-                target=_target,
-                custom_headers=_custom_headers,
-                forward_headers=_forward_headers,
-                merge_query_params=_merge_query_params,
-                dependencies=_dependencies,
-                cost_per_request=endpoint.get("cost_per_request", None),
-                endpoint_id=endpoint_id,
-                guardrails=_guardrails,
-                methods=_methods,
-                default_query_params=_default_query_params,
-            )
-
-            visited_endpoints.add(f"{endpoint_id}:subpath:{_path}:{methods_str}")
-
-        verbose_proxy_logger.debug(
-            "Added new pass through endpoint: %s (ID: %s)", _path, endpoint_id
+            premium_user=premium_user,
+            visited_endpoints=visited_endpoints,
         )
 
     # remove the ones that are not visited from the list
diff --git a/litellm/responses/file_search/emulated_handler.py b/litellm/responses/file_search/emulated_handler.py
index f770648e639..4700d673998 100644
--- a/litellm/responses/file_search/emulated_handler.py
+++ b/litellm/responses/file_search/emulated_handler.py
@@ -14,10 +14,10 @@ Flow:
 import json
 import time
 import uuid
-from typing import Any, Dict, Iterable, List, Optional, Tuple, cast
+from typing import Any, Dict, Iterable, List, Optional, Tuple, Union, cast
 
 from litellm._logging import verbose_logger
-from litellm.types.llms.openai import ResponsesAPIResponse
+from litellm.types.llms.openai import ResponseOutputItem, ResponsesAPIResponse
 from litellm.types.vector_stores import VectorStoreSearchResult
 
 # Keep ToolParam broad so we stay compatible with both dict and Pydantic forms
@@ -30,6 +30,7 @@ FILE_SEARCH_FUNCTION_NAME = "litellm_file_search"
 # Detection
 # ---------------------------------------------------------------------------
 
+
 def should_use_emulated_file_search(
     tools: Optional[Iterable[ToolParam]],
     provider_config: Any,  # BaseResponsesAPIConfig
@@ -37,9 +38,7 @@ def should_use_emulated_file_search(
     """Return True when there is a file_search tool and the provider can't handle it natively."""
     if not tools:
         return False
-    has_fs = any(
-        isinstance(t, dict) and t.get("type") == "file_search" for t in tools
-    )
+    has_fs = any(isinstance(t, dict) and t.get("type") == "file_search" for t in tools)
     if not has_fs:
         return False
     return provider_config is None or not provider_config.supports_native_file_search()
@@ -49,6 +48,7 @@ def should_use_emulated_file_search(
 # Tool conversion
 # ---------------------------------------------------------------------------
 
+
 def _build_function_tool(vector_store_ids: List[str]) -> Dict[str, Any]:
     """
     Create a Responses API function-tool definition that describes file search.
@@ -104,7 +104,7 @@ def _replace_file_search_tools(
     non_file_search: List[Dict[str, Any]] = []
     vector_store_ids: List[str] = []
 
-    for tool in (tools or []):
+    for tool in tools or []:
         if isinstance(tool, dict) and tool.get("type") == "file_search":
             ids = tool.get("vector_store_ids") or []
             vector_store_ids.extend(ids)
@@ -123,6 +123,7 @@ def _replace_file_search_tools(
 # Search execution
 # ---------------------------------------------------------------------------
 
+
 async def _run_vector_searches(
     queries: List[str],
     vector_store_ids: List[str],
@@ -150,7 +151,11 @@ async def _run_vector_searches(
                     vector_store_id=vs_id,
                     query=query,
                 )
-                results_data = response.get("data") if isinstance(response, dict) else getattr(response, "data", None)
+                results_data = (
+                    response.get("data")
+                    if isinstance(response, dict)
+                    else getattr(response, "data", None)
+                )
                 if results_data:
                     all_results.extend(results_data)
             except Exception as exc:
@@ -168,6 +173,7 @@ async def _run_vector_searches(
 # Result formatting
 # ---------------------------------------------------------------------------
 
+
 def _get_field(result: Any, key: str, default: Any = None) -> Any:
     """Read a field from either a dict/TypedDict or an attribute-based object."""
     if isinstance(result, dict):
@@ -319,13 +325,27 @@ def _build_message_output(
 def _extract_text_from_responses_output(response: ResponsesAPIResponse) -> str:
     """Pull the assistant's text from the provider's response."""
     for item in response.output:
-        item_type = item.get("type") if isinstance(item, dict) else getattr(item, "type", None)
+        item_type = (
+            item.get("type") if isinstance(item, dict) else getattr(item, "type", None)
+        )
         if item_type == "message":
-            content = item.get("content") if isinstance(item, dict) else getattr(item, "content", [])
-            for block in (content or []):
-                block_type = block.get("type") if isinstance(block, dict) else getattr(block, "type", None)
+            content = (
+                item.get("content")
+                if isinstance(item, dict)
+                else getattr(item, "content", [])
+            )
+            for block in content or []:
+                block_type = (
+                    block.get("type")
+                    if isinstance(block, dict)
+                    else getattr(block, "type", None)
+                )
                 if block_type == "output_text":
-                    raw = block.get("text") if isinstance(block, dict) else getattr(block, "text", "")
+                    raw = (
+                        block.get("text")
+                        if isinstance(block, dict)
+                        else getattr(block, "text", "")
+                    )
                     return str(raw) if raw is not None else ""
     return ""
 
@@ -345,13 +365,16 @@ def _synthesize_responses_api_response(
     synthesized _hidden_params so that billing callbacks see the total cost of
     both provider calls that the emulated flow makes.
     """
+    synthesized_output: List[Dict[str, Any]] = [file_search_call_output, message_output]
     synthesized = ResponsesAPIResponse(
         id=getattr(original_response, "id", f"resp_{uuid.uuid4().hex}"),
         object="response",
         created_at=getattr(original_response, "created_at", int(time.time())),
         status="completed",
         model=getattr(original_response, "model", ""),
-        output=[file_search_call_output, message_output],
+        output=cast(
+            List[Union[ResponseOutputItem, Dict[str, Any]]], synthesized_output
+        ),
         usage=getattr(original_response, "usage", None),
         error=None,
     )
@@ -359,9 +382,15 @@ def _synthesize_responses_api_response(
         hidden = dict(getattr(original_response, "_hidden_params") or {})
         if first_response is not None and hasattr(first_response, "_hidden_params"):
             first_hidden = getattr(first_response, "_hidden_params") or {}
-            first_cost = first_hidden.get("response_cost") if isinstance(first_hidden, dict) else getattr(first_hidden, "response_cost", None)
+            first_cost = (
+                first_hidden.get("response_cost")
+                if isinstance(first_hidden, dict)
+                else getattr(first_hidden, "response_cost", None)
+            )
             if first_cost is not None:
-                current_cost = hidden.get("response_cost") if isinstance(hidden, dict) else 0
+                current_cost = (
+                    hidden.get("response_cost") if isinstance(hidden, dict) else 0
+                )
                 hidden["response_cost"] = (current_cost or 0) + first_cost
         synthesized._hidden_params = hidden
     return synthesized
@@ -371,8 +400,12 @@ def _synthesize_responses_api_response(
 # Main entry point
 # ---------------------------------------------------------------------------
 
-async def _call_aresponses(input, model, tools, **kwargs):  # pragma: no cover – thin wrapper for patching in tests
+
+async def _call_aresponses(
+    input, model, tools, **kwargs
+):  # pragma: no cover – thin wrapper for patching in tests
     from litellm.responses.main import aresponses
+
     return await aresponses(input=input, model=model, tools=tools, **kwargs)
 
 
@@ -458,10 +491,17 @@ async def aresponses_with_emulated_file_search(
 
     for tool_call in file_search_calls:
         if isinstance(tool_call, dict):
-            call_id = tool_call.get("call_id") or tool_call.get("id") or file_search_call_id
+            call_id = str(
+                tool_call.get("call_id") or tool_call.get("id") or file_search_call_id
+            )
             raw_args = tool_call.get("arguments") or "{}"
         else:
-            call_id = getattr(tool_call, "call_id", None) or getattr(tool_call, "id", file_search_call_id)
+            raw_call_id = (
+                getattr(tool_call, "call_id", None)
+                or getattr(tool_call, "id", None)
+                or file_search_call_id
+            )
+            call_id = str(raw_call_id)
             raw_args = getattr(tool_call, "arguments", "{}") or "{}"
 
         try:
@@ -500,7 +540,11 @@ async def aresponses_with_emulated_file_search(
     # Including all output items (text blocks, reasoning, non-file-search calls) ensures providers
     # like Anthropic that emit text before the tool call have complete conversation context.
     # Serialize Pydantic model instances to plain dicts so the transformation layer can call .get().
-    original_input_items = list(input) if isinstance(input, (list, tuple)) else [{"role": "user", "content": str(input)}]
+    original_input_items = (
+        list(input)
+        if isinstance(input, (list, tuple))
+        else [{"role": "user", "content": str(input)}]
+    )
     first_response_output_items: List[Any] = []
     for _item in first_response.output:
         if isinstance(_item, dict):
@@ -510,11 +554,7 @@ async def aresponses_with_emulated_file_search(
         else:
             first_response_output_items.append(_item)
 
-    follow_up_input = (
-        original_input_items
-        + first_response_output_items
-        + tool_results
-    )
+    follow_up_input = original_input_items + first_response_output_items + tool_results
 
     # 6. Follow-up call — provider writes the final answer given search results.
     # Also an internal sub-call; billing is suppressed so the outer call fires once.
diff --git a/litellm/responses/main.py b/litellm/responses/main.py
index c429cfdbc39..16f97773fc6 100644
--- a/litellm/responses/main.py
+++ b/litellm/responses/main.py
@@ -77,9 +77,7 @@ def _has_file_search_tool(tools: Optional[Any]) -> 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
-    )
+    return any(isinstance(t, dict) and t.get("type") == "file_search" for t in tools)
 
 
 def mock_responses_api_response(
@@ -486,7 +484,9 @@ async def aresponses(
         prompt_variables = cast(Optional[dict], kwargs.get("prompt_variables", None))
         original_model = model
 
-        if isinstance(litellm_logging_obj, LiteLLMLoggingObj) and litellm_logging_obj.should_run_prompt_management_hooks(
+        if isinstance(
+            litellm_logging_obj, LiteLLMLoggingObj
+        ) and litellm_logging_obj.should_run_prompt_management_hooks(
             prompt_id=prompt_id, non_default_params=kwargs
         ):
             if isinstance(input, str):
@@ -514,9 +514,7 @@ async def aresponses(
             )
             input = cast(Union[str, ResponseInputParam], merged_input)
             if model != original_model:
-                _, custom_llm_provider, _, _ = litellm.get_llm_provider(
-                    model=model
-                )
+                _, custom_llm_provider, _, _ = litellm.get_llm_provider(model=model)
             kwargs.pop("prompt_id", None)
             kwargs["_async_prompt_merged_params"] = merged_optional_params
 
@@ -588,6 +586,88 @@ async def aresponses(
         )
 
 
+def _apply_prompt_management_to_responses_call(
+    input: Union[str, ResponseInputParam],
+    model: str,
+    custom_llm_provider: Optional[str],
+    litellm_logging_obj: Optional[LiteLLMLoggingObj],
+    kwargs: Dict[str, Any],
+    local_vars: Dict[str, Any],
+) -> tuple[Union[str, ResponseInputParam], str, Optional[str]]:
+    async_merged = kwargs.pop("_async_prompt_merged_params", None)
+    if async_merged is not None:
+        for key, value in async_merged.items():
+            local_vars[key] = value
+        return input, model, custom_llm_provider
+
+    prompt_id = cast(Optional[str], kwargs.get("prompt_id", None))
+    prompt_variables = cast(Optional[dict], kwargs.get("prompt_variables", None))
+    original_model = model
+
+    if isinstance(input, str):
+        client_input: List[AllMessageValues] = [{"role": "user", "content": input}]
+    else:
+        client_input = [
+            item  # type: ignore[misc]
+            for item in input
+            if isinstance(item, dict) and "role" in item
+        ]
+
+    if isinstance(
+        litellm_logging_obj, LiteLLMLoggingObj
+    ) and litellm_logging_obj.should_run_prompt_management_hooks(
+        prompt_id=prompt_id, non_default_params=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),
+        )
+        input = cast(Union[str, ResponseInputParam], merged_input)
+        local_vars["input"] = input
+        local_vars["model"] = model
+        if model != original_model:
+            _, custom_llm_provider, _, _ = litellm.get_llm_provider(model=model)
+            local_vars["custom_llm_provider"] = custom_llm_provider
+        for key, value in merged_optional_params.items():
+            local_vars[key] = value
+
+    return input, model, custom_llm_provider
+
+
+def _resolve_model_provider_for_responses(
+    model: str,
+    custom_llm_provider: Optional[str],
+    litellm_params: GenericLiteLLMParams,
+    local_vars: Dict[str, Any],
+) -> tuple[str, Optional[str]]:
+    (
+        model,
+        custom_llm_provider,
+        dynamic_api_key,
+        dynamic_api_base,
+    ) = litellm.get_llm_provider(
+        model=model,
+        custom_llm_provider=custom_llm_provider,
+        api_base=litellm_params.api_base,
+        api_key=litellm_params.api_key,
+    )
+    local_vars["custom_llm_provider"] = custom_llm_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 model, custom_llm_provider
+
+
 @client
 def responses(
     input: Union[str, ResponseInputParam],
@@ -659,80 +739,27 @@ def responses(
                 mock_response=litellm_params.mock_response
             )
 
-        (
-            model,
-            custom_llm_provider,
-            dynamic_api_key,
-            dynamic_api_base,
-        ) = litellm.get_llm_provider(
+        model, custom_llm_provider = _resolve_model_provider_for_responses(
             model=model,
             custom_llm_provider=custom_llm_provider,
-            api_base=litellm_params.api_base,
-            api_key=litellm_params.api_key,
+            litellm_params=litellm_params,
+            local_vars=local_vars,
         )
 
-        # Update local_vars with detected provider (fixes #19782)
-        local_vars["custom_llm_provider"] = custom_llm_provider
-
-        # Use dynamic credentials from get_llm_provider (e.g., when use_litellm_proxy=True)
-        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
-
         #########################################################
         # 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.
         #########################################################
-        _async_merged = kwargs.pop("_async_prompt_merged_params", None)
-        if _async_merged is not None:
-            for k, v in _async_merged.items():
-                local_vars[k] = v
-        else:
-            prompt_id = cast(Optional[str], kwargs.get("prompt_id", None))
-            prompt_variables = cast(
-                Optional[dict], kwargs.get("prompt_variables", None)
-            )
-            original_model = model
-
-            if isinstance(litellm_logging_obj, LiteLLMLoggingObj) and litellm_logging_obj.should_run_prompt_management_hooks(
-                prompt_id=prompt_id, non_default_params=kwargs
-            ):
-                if isinstance(input, str):
-                    client_input: List[AllMessageValues] = [
-                        {"role": "user", "content": input}
-                    ]
-                else:
-                    client_input = [
-                        item  # type: ignore[misc]
-                        for item in input
-                        if isinstance(item, dict) and "role" in item
-                    ]
-                (
-                    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),
-                )
-                input = cast(Union[str, ResponseInputParam], merged_input)
-                local_vars["input"] = input
-                local_vars["model"] = model
-                if model != original_model:
-                    _, custom_llm_provider, _, _ = litellm.get_llm_provider(
-                        model=model
-                    )
-                    local_vars["custom_llm_provider"] = custom_llm_provider
-                for k, v in merged_optional_params.items():
-                    local_vars[k] = v
+        input, model, custom_llm_provider = _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,
+        )
 
         #########################################################
         # Update input and tools with provider-specific file IDs if managed files are used
@@ -803,12 +830,16 @@ def responses(
             return run_async_function(aresponses_api_with_mcp, **mcp_call_kwargs)
 
         # get provider config
-        responses_api_provider_config: Optional[
-            BaseResponsesAPIConfig
-        ] = ProviderConfigManager.get_provider_responses_api_config(
-            model=model,
-            provider=custom_llm_provider,
-        )
+        responses_api_provider_config: Optional[BaseResponsesAPIConfig]
+        if custom_llm_provider is None:
+            responses_api_provider_config = None
+        else:
+            responses_api_provider_config = (
+                ProviderConfigManager.get_provider_responses_api_config(
+                    model=model,
+                    provider=custom_llm_provider,
+                )
+            )
 
         local_vars.update(kwargs)
         # Map reasoning_effort (from litellm_params/proxy config) to reasoning when not set
@@ -919,6 +950,9 @@ def responses(
         )
 
         # 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,