import ast
import asyncio
import copy
import json
import posixpath
import traceback
from base64 import b64encode
from collections.abc import AsyncGenerator, Callable, Iterable, Mapping, Sequence
from datetime import datetime
from itertools import groupby
from typing import Any, Final, TypedDict, cast
from urllib.parse import urlencode, urlparse

import httpx
from fastapi import (
    APIRouter,
    Depends,
    FastAPI,
    HTTPException,
    Request,
    Response,
    UploadFile,
    WebSocket,
    status,
)
from fastapi.responses import StreamingResponse
from starlette.datastructures import UploadFile as StarletteUploadFile
from starlette.websockets import WebSocketState
from websockets.asyncio.client import connect
from websockets.exceptions import (
    ConnectionClosedError,
    ConnectionClosedOK,
    InvalidStatus,
)
from websockets.frames import Close, CloseCode

import litellm
from litellm._logging import verbose_proxy_logger
from litellm._uuid import uuid
from litellm.constants import (
    MAXIMUM_TRACEBACK_LINES_TO_LOG,
    WEBSOCKET_CLOSE_REASON_MAX_BYTES,
)
from litellm.integrations.custom_guardrail import CustomGuardrail
from litellm.integrations.custom_logger import CustomLogger
from litellm.litellm_core_utils.core_helpers import (
    get_metadata_variable_name_from_kwargs,
    get_or_create_metadata_bucket,
)
from litellm.litellm_core_utils.internal_call_metadata import MODEL_ACCESS_GROUP_METADATA_KEY
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
from litellm.llms.base_llm.managed_resources.utils import (
    resolve_passthrough_managed_id_provider,
)
from litellm.llms.custom_httpx.http_handler import get_async_httpx_client
from litellm.passthrough import BasePassthroughUtils
from litellm.proxy._types import (
    ConfigFieldInfo,
    ConfigFieldUpdate,
    LiteLLMRoutes,
    PassThroughEndpointResponse,
    PassThroughGenericEndpoint,
    ProxyException,
    UserAPIKeyAuth,
)
from litellm.proxy.auth.auth_utils import request_dispatched_to_pass_through_endpoint
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
from litellm.proxy.common_request_processing import (
    ProxyBaseLLMRequestProcessing,
    open_sse_before_first_byte,
)
from litellm.proxy.common_utils.http_parsing_utils import (
    _read_request_body,
    _safe_get_request_headers,
)
from litellm.proxy.common_utils.sse_keepalive import (
    wrap_passthrough_sse_bytes_with_keepalive_pings,
)
from litellm.proxy.litellm_pre_call_utils import LiteLLMProxyRequestSetup
from litellm.proxy.utils import normalize_route_for_root_path
from litellm.repositories.team_repository import TeamRepository
from litellm.secret_managers.main import get_secret_str
from litellm.types.llms.custom_http import httpxSpecialProvider
from litellm.types.passthrough_endpoints.pass_through_endpoints import (
    LITELLM_PASS_THROUGH_CUSTOM_BODY_STATE_KEY,
    LITELLM_PASS_THROUGH_ENDPOINT_MARKER,
    LITELLM_PASS_THROUGH_RAW_BODY_STATE_KEY,
    EndpointType,
    PassthroughStandardLoggingPayload,
)
from litellm.types.utils import Usage

from .streaming_handler import PassThroughStreamingHandler
from .success_handler import PassThroughEndpointLogging
from .upstream_usage_headers import (
    UpstreamReportedUsage,
    apply_upstream_reported_usage,
)

router: Final = APIRouter()

pass_through_endpoint_logging: Final = PassThroughEndpointLogging()

# Global registry to track registered pass-through routes and prevent memory leaks
_registered_pass_through_routes: Final[dict[str, dict[str, str | bool | list[str] | Mapping[str, object]]]] = {}


def get_response_body(response: httpx.Response) -> dict | None:
    try:
        return response.json()
    except Exception:
        return None


async def set_env_variables_in_header(custom_headers: dict | None) -> dict | None:
    """
    checks if any headers on config.yaml are defined as os.environ/COHERE_API_KEY etc

    only runs for headers defined on config.yaml

    example header can be

    {"Authorization": "Bearer os.environ/COHERE_API_KEY"}
    """
    if custom_headers is None:
        return None
    headers: Final = {}
    for key, value in custom_headers.items():
        # langfuse Api requires base64 encoded headers - it's simpleer to just ask litellm users to set their langfuse public and secret keys
        # we can then get the b64 encoded keys here
        if key == "LANGFUSE_PUBLIC_KEY" or key == "LANGFUSE_SECRET_KEY":
            # langfuse requires b64 encoded headers - we construct that here
            _langfuse_public_key = custom_headers["LANGFUSE_PUBLIC_KEY"]
            _langfuse_secret_key = custom_headers["LANGFUSE_SECRET_KEY"]
            if isinstance(_langfuse_public_key, str) and _langfuse_public_key.startswith("os.environ/"):
                _langfuse_public_key = get_secret_str(_langfuse_public_key)
            if isinstance(_langfuse_secret_key, str) and _langfuse_secret_key.startswith("os.environ/"):
                _langfuse_secret_key = get_secret_str(_langfuse_secret_key)
            headers["Authorization"] = "Basic " + b64encode(
                f"{_langfuse_public_key}:{_langfuse_secret_key}".encode()
            ).decode("ascii")
        else:
            # for all other headers
            headers[key] = value
            if isinstance(value, str) and "os.environ/" in value:
                verbose_proxy_logger.debug("pass through endpoint - looking up 'os.environ/' variable")
                # get string section that is os.environ/
                start_index = value.find("os.environ/")
                _variable_name = value[start_index:]

                verbose_proxy_logger.debug(
                    "pass through endpoint - getting secret for variable name: %s",
                    _variable_name,
                )
                _secret_value = get_secret_str(_variable_name)
                if _secret_value is not None:
                    new_value = value.replace(_variable_name, _secret_value)
                    headers[key] = new_value
    return headers


async def chat_completion_pass_through_endpoint(
    fastapi_response: Response,
    request: Request,
    adapter_id: str,
    user_api_key_dict: UserAPIKeyAuth,
):
    from litellm.proxy.proxy_server import (
        add_litellm_data_to_request,
        general_settings,
        llm_router,
        proxy_config,
        proxy_logging_obj,
        user_api_base,
        user_max_tokens,
        user_model,
        user_request_timeout,
        user_temperature,
        version,
    )

    data = {}
    try:
        body: Final = await request.body()
        body_str: Final = body.decode()
        try:
            data = ast.literal_eval(body_str)
        except Exception:
            data = json.loads(body_str)

        data["adapter_id"] = adapter_id

        verbose_proxy_logger.debug("Request received by LiteLLM:\n%s", data)
        data["model"] = (
            general_settings.get("completion_model", None)  # server default
            or user_model  # model name passed via cli args
            or data.get("model", None)  # default passed in http request
        )
        if user_model:
            data["model"] = user_model

        data = await add_litellm_data_to_request(
            data=data,
            request=request,
            general_settings=general_settings,
            user_api_key_dict=user_api_key_dict,
            version=version,
            proxy_config=proxy_config,
        )

        # override with user settings, these are params passed via cli
        if user_temperature:
            data["temperature"] = user_temperature
        if user_request_timeout:
            data["request_timeout"] = user_request_timeout
        if user_max_tokens:
            data["max_tokens"] = user_max_tokens
        if user_api_base:
            data["api_base"] = user_api_base

        ### MODEL ALIAS MAPPING ###
        # check if model name in model alias map
        # get the actual model name
        if data["model"] in litellm.model_alias_map:
            data["model"] = litellm.model_alias_map[data["model"]]

        # Check key-specific aliases
        if (
            isinstance(data["model"], str)
            and user_api_key_dict.aliases
            and isinstance(user_api_key_dict.aliases, dict)
            and data["model"] in user_api_key_dict.aliases
        ):
            data["model"] = user_api_key_dict.aliases[data["model"]]

        ### CALL HOOKS ### - modify incoming data before calling the model
        data = await proxy_logging_obj.pre_call_hook(
            user_api_key_dict=user_api_key_dict, data=data, call_type="text_completion"
        )

        ### ROUTE THE REQUESTs ###
        router_model_names: Final = llm_router.model_names if llm_router is not None else []
        # skip router if user passed their key
        if "api_key" in data:
            llm_response = asyncio.create_task(litellm.aadapter_completion(**data))
        elif llm_router is not None and llm_router.is_recognized_model(data["model"]):
            llm_response = asyncio.create_task(llm_router.aadapter_completion(**data))
        elif (
            llm_router is not None
            and data["model"] not in router_model_names
            and (llm_router.default_deployment is not None or len(llm_router.pattern_router.patterns) > 0)
        ):  # check for wildcard routes or default deployment before checking deployment_names
            llm_response = asyncio.create_task(llm_router.aadapter_completion(**data))
        elif (
            llm_router is not None and data["model"] in llm_router.deployment_names
        ):  # model in router deployments, calling a specific deployment on the router (lowest priority)
            llm_response = asyncio.create_task(llm_router.aadapter_completion(**data, specific_deployment=True))
        elif user_model is not None:  # `litellm --model <your-model-name>`
            llm_response = asyncio.create_task(litellm.aadapter_completion(**data))
        else:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail={"error": "completion: Invalid model name passed in model=" + data.get("model", "")},
            )

        # Await the llm_response task
        response: Final = await llm_response

        hidden_params: Final = getattr(response, "_hidden_params", {}) or {}
        model_id: Final = hidden_params.get("model_id", None) or ""
        cache_key: Final = hidden_params.get("cache_key", None) or ""
        api_base: Final = hidden_params.get("api_base", None) or ""
        response_cost: Final = hidden_params.get("response_cost", None) or ""

        ### ALERTING ###
        asyncio.create_task(
            proxy_logging_obj.update_request_status(litellm_call_id=data.get("litellm_call_id", ""), status="success")
        )

        verbose_proxy_logger.debug("final response: %s", response)

        fastapi_response.headers.update(
            ProxyBaseLLMRequestProcessing.get_custom_headers(
                user_api_key_dict=user_api_key_dict,
                model_id=model_id,
                cache_key=cache_key,
                api_base=api_base,
                version=version,
                response_cost=response_cost,
            )
        )

        verbose_proxy_logger.debug("\nResponse from Litellm:\n%s", response)
        return response
    except Exception as e:
        await proxy_logging_obj.post_call_failure_hook(
            user_api_key_dict=user_api_key_dict, original_exception=e, request_data=data
        )
        verbose_proxy_logger.exception("litellm.proxy.proxy_server.completion(): Exception occured - %s", e)
        error_msg: Final = f"{e}"
        raise ProxyException(
            message=getattr(e, "message", error_msg),
            type=getattr(e, "type", "None"),
            param=getattr(e, "param", "None"),
            code=getattr(e, "status_code", 500),
        )


class HttpPassThroughEndpointHelpers(BasePassthroughUtils):
    @staticmethod
    def get_response_headers(
        headers: httpx.Headers,
        litellm_call_id: str | None = None,
        custom_headers: dict | None = None,
    ) -> dict:
        # Exclude headers that uvicorn writes itself (server, date) and
        # encoding/length headers that don't survive re-serialization.
        # If we forward the upstream's Server header, uvicorn adds its
        # own and strict HTTP parsers (e.g. aiohttp) reject the
        # response with "Duplicate 'Server' header found".
        excluded_headers: Final = {
            "transfer-encoding",
            "content-encoding",
            "content-length",
            "server",
            "date",
            "connection",
            "keep-alive",
        }

        return_headers: Final = {key: value for key, value in headers.items() if key.lower() not in excluded_headers}
        if litellm_call_id:
            return_headers["x-litellm-call-id"] = litellm_call_id
        if custom_headers:
            # Ensure custom headers don't override actual upstream response headers or let framework defaults (like content-length: 0) interfere.
            sanitized_custom_headers: Final = {
                key: value for key, value in custom_headers.items() if key.lower() not in excluded_headers
            }
            return_headers.update(sanitized_custom_headers)

        return return_headers

    @staticmethod
    def get_endpoint_type(url: str) -> EndpointType:
        parsed_url: Final = urlparse(url)
        if (
            ("generateContent") in url
            or ("streamGenerateContent") in url
            or ("rawPredict") in url
            or ("streamRawPredict") in url
        ):
            return EndpointType.VERTEX_AI
        elif parsed_url.hostname == "api.anthropic.com":
            return EndpointType.ANTHROPIC
        elif (
            parsed_url.hostname == "api.openai.com"
            or parsed_url.hostname == "openai.azure.com"
            or (parsed_url.hostname and "openai.com" in parsed_url.hostname)
        ):
            return EndpointType.OPENAI
        return EndpointType.GENERIC

    @staticmethod
    async def _make_non_streaming_http_request(
        request: Request,
        async_client: httpx.AsyncClient,
        url: str,
        headers: dict,
        requested_query_params: dict | None = None,
        custom_body: dict | None = None,
    ) -> httpx.Response:
        """
        Make a non-streaming HTTP request

        If request is GET, don't include a JSON body
        """
        if request.method == "GET":
            response = await async_client.request(
                method=request.method,
                url=url,
                headers=headers,
                params=requested_query_params,
            )
        else:
            response = await async_client.request(
                method=request.method,
                url=url,
                headers=headers,
                params=requested_query_params,
                json=custom_body,
            )
        return response

    @staticmethod
    async def non_streaming_http_request_handler(
        request: Request,
        async_client: httpx.AsyncClient,
        url: httpx.URL,
        headers: dict,
        requested_query_params: dict | None = None,
        _parsed_body: dict | None = None,
        forward_multipart: bool = False,
    ) -> httpx.Response:
        """
        Handle non-SSE HTTP requests

        Handles special cases when GET requests, multipart/form-data requests, and generic httpx requests.

        GET and generic requests are sent with httpx stream semantics so the caller can
        decide from the response headers whether to buffer the body (JSON, inspected for
        logging/guardrails) or relay it to the client without materializing it in memory
        (LIT-4009: large batch results files must not be buffered in proxy RSS).
        """
        if request.method == "GET":
            get_request: Final = async_client.build_request(
                request.method,
                url,
                headers=headers,
                params=requested_query_params,
            )
            return await async_client.send(get_request, stream=True)
        if HttpPassThroughEndpointHelpers.is_multipart(request) is True and forward_multipart:
            # Forward multipart via make_multipart_http_request even when _parsed_body is
            # non-empty (pass_through_request always injects litellm_logging_obj, etc.).
            # forward_multipart is False when custom_body was supplied (JSON body despite
            # multipart content-type) — those requests use the generic json= path.
            return await HttpPassThroughEndpointHelpers.make_multipart_http_request(
                request=request,
                async_client=async_client,
                url=url,
                headers=headers,
                requested_query_params=requested_query_params,
            )
        generic_request: Final = async_client.build_request(
            request.method,
            url,
            headers=headers,
            params=requested_query_params,
            json=_parsed_body,
        )
        return await async_client.send(generic_request, stream=True)

    @staticmethod
    def is_multipart(request: Request) -> bool:
        """Check if the request is a multipart/form-data request"""
        return "multipart/form-data" in request.headers.get("content-type", "")

    @staticmethod
    async def _build_request_files_from_upload_file(
        upload_file: UploadFile | StarletteUploadFile,
    ) -> tuple[str | None, bytes, str | None]:
        """Build a request files dict from an UploadFile object"""
        file_content: Final = await upload_file.read()
        return (upload_file.filename, file_content, upload_file.content_type)

    @staticmethod
    async def make_multipart_http_request(
        request: Request,
        async_client: httpx.AsyncClient,
        url: httpx.URL,
        headers: dict,
        requested_query_params: dict | None = None,
        stream: bool = False,
    ) -> httpx.Response:
        """Process multipart/form-data requests, handling both files and form fields.

        Iterates ``form.multi_items()`` rather than ``form.items()`` so repeated
        field names (e.g. several ``-F file=@...`` parts) are all forwarded;
        ``items()`` collapses duplicate keys to the last value. Files go out as a
        list of ``(field_name, (filename, content, content_type))`` tuples and
        repeated non-file fields are grouped into list values, both of which httpx
        encodes as separate multipart parts. A form with no file parts is sent
        entirely through ``files`` as ``(field_name, (None, value))`` tuples,
        because httpx downgrades a file-less ``data=`` payload to
        application/x-www-form-urlencoded.
        """
        form_items: Final = (await request.form()).multi_items()

        files: Final = [
            (
                field_name,
                await HttpPassThroughEndpointHelpers._build_request_files_from_upload_file(upload_file=field_value),
            )
            for field_name, field_value in form_items
            if isinstance(field_value, (StarletteUploadFile, UploadFile))
        ]

        non_file_items: Final = tuple(
            (field_name, field_value)
            for field_name, field_value in form_items
            if not isinstance(field_value, (StarletteUploadFile, UploadFile))
        )
        field_order: Final = {
            field_name: index
            for index, field_name in enumerate(dict.fromkeys(field_name for field_name, _ in non_file_items))
        }
        form_data_dict: Final = {
            field_name: [value for _, value in group]
            for field_name, group in groupby(
                sorted(non_file_items, key=lambda item: field_order[item[0]]),
                key=lambda item: item[0],
            )
        }

        multipart_files: Final = (
            files if files else tuple((field_name, (None, field_value)) for field_name, field_value in non_file_items)
        )
        multipart_data: Final = form_data_dict if files else None

        # Remove content-type header - httpx will set it correctly with the new boundary
        # when it creates the multipart body from files/data parameters
        headers_copy: Final = headers.copy()
        headers_copy.pop("content-type", None)

        # httpx.AsyncClient.request() does not accept stream=; use send() for streaming.
        if stream:
            req: Final = async_client.build_request(
                request.method,
                url,
                headers=headers_copy,
                params=requested_query_params,
                files=multipart_files,
                data=multipart_data,
            )
            return await async_client.send(req, stream=True)

        return await async_client.request(
            method=request.method,
            url=url,
            headers=headers_copy,
            params=requested_query_params,
            files=multipart_files,
            data=multipart_data,
        )

    @staticmethod
    def _init_kwargs_for_pass_through_endpoint(
        request: Request,
        user_api_key_dict: UserAPIKeyAuth,
        passthrough_logging_payload: PassthroughStandardLoggingPayload,
        logging_obj: LiteLLMLoggingObj,
        _parsed_body: dict | None = None,
        litellm_call_id: str | None = None,
    ) -> dict:
        """
        Filter out litellm params from the request body
        """
        from litellm.types.utils import all_litellm_params

        _parsed_body = _parsed_body or {}

        litellm_params_in_body: Final = {}
        for k in all_litellm_params:
            if k in _parsed_body:
                litellm_params_in_body[k] = _parsed_body.pop(k, None)

        _metadata = dict(
            LiteLLMProxyRequestSetup.get_sanitized_user_information_from_key(user_api_key_dict=user_api_key_dict)
        )

        litellm_metadata: Final = litellm_params_in_body.pop("litellm_metadata", None)
        metadata: Final = litellm_params_in_body.pop("metadata", None)
        if litellm_metadata:
            _metadata.update(litellm_metadata)
        if metadata:
            _metadata.update(metadata)

        _metadata = _update_metadata_with_tags_in_header(
            request=request,
            metadata=_metadata,
        )

        # Set internal keys after merging client-supplied metadata so a request
        # body that mirrors them cannot clobber the authenticated key or the
        # real parent span.
        _metadata["user_api_key"] = user_api_key_dict.api_key
        _metadata["litellm_parent_otel_span"] = user_api_key_dict.parent_otel_span
        _metadata["user_api_key_budget_reservation"] = user_api_key_dict.budget_reservation
        _metadata[MODEL_ACCESS_GROUP_METADATA_KEY] = user_api_key_dict.matched_model_access_groups
        # The per-model budget counters are keyed off these. get_sanitized_user_information_from_key
        # returns StandardLoggingUserAPIKeyMetadata, which carries no budget field, so without this
        # the post-call increment finds nothing and every passthrough request goes untracked and
        # unenforced. Set after the client merge so a request body cannot supply its own budget.
        #
        # Only for the built-in provider routes. `get_model_from_request` returns
        # None for a user-defined pass-through, deliberately: its body is forwarded
        # verbatim, so `model` there names an UPSTREAM model rather than a
        # LiteLLM-managed one. Enforcement is therefore skipped on those routes, and
        # charging a counter anyway would track spend that nothing can refuse, and
        # would attribute it to a budget the operator scoped to a LiteLLM model that
        # merely shares the name.
        if not request_dispatched_to_pass_through_endpoint(request):
            _metadata["user_api_key_model_max_budget"] = user_api_key_dict.model_max_budget
            _metadata["user_api_key_user_model_max_budget"] = user_api_key_dict.user_model_max_budget
            _metadata["user_api_key_end_user_model_max_budget"] = user_api_key_dict.end_user_model_max_budget
        _metadata.update(
            LiteLLMProxyRequestSetup.get_sanitized_user_information_from_key(user_api_key_dict=user_api_key_dict)
        )

        kwargs: Final = {
            "litellm_params": {
                **litellm_params_in_body,
                "metadata": _metadata,
                "proxy_server_request": {
                    "url": str(request.url),
                    "method": request.method,
                    "body": copy.copy(_parsed_body),  # use copy instead of deepcopy
                    "headers": request.headers,
                },
            },
            "call_type": "pass_through_endpoint",
            "litellm_call_id": litellm_call_id,
            "passthrough_logging_payload": passthrough_logging_payload,
        }

        logging_obj.model_call_details["passthrough_logging_payload"] = passthrough_logging_payload

        return kwargs

    @staticmethod
    def construct_target_url_with_subpath(base_target: str, subpath: str, include_subpath: bool | None) -> str:
        """
        Helper function to construct the full target URL with subpath handling.

        Args:
            base_target: The base target URL
            subpath: The captured subpath from the request
            include_subpath: Whether to include the subpath in the target URL

        Returns:
            The constructed full target URL
        """
        if not include_subpath:
            return base_target

        if not subpath:
            return base_target

        # Ensure base_target ends with / and subpath doesn't start with /
        if not base_target.endswith("/"):
            base_target = base_target + "/"
        subpath = subpath.removeprefix("/")

        # Resolve any '..' segments in the subpath so it cannot climb above
        # the base_target prefix that the operator configured. Preserve a
        # trailing slash on the original subpath since some upstreams treat
        # `/foo` and `/foo/` as different resources.
        trailing_slash: Final = subpath.endswith("/")
        safe_subpath = posixpath.normpath("/" + subpath).lstrip("/")
        if safe_subpath == ".":
            safe_subpath = ""
        if trailing_slash and safe_subpath and not safe_subpath.endswith("/"):
            safe_subpath += "/"

        return base_target + safe_subpath

    @staticmethod
    def join_base_and_endpoint_path(base_url: httpx.URL, endpoint_path: str) -> str:
        """
        Combine the path component of ``base_url`` with ``endpoint_path``.

        Preserves any path prefix configured on the base URL and resolves
        ``..`` segments in the endpoint so the result stays within the base
        path. A trailing slash on ``endpoint_path`` is preserved.
        """
        trailing_slash: Final = endpoint_path.endswith("/")
        base_path = base_url.path or ""
        if not base_path or base_path == "/":
            normalized_endpoint = posixpath.normpath("/" + endpoint_path.lstrip("/"))
            if trailing_slash and normalized_endpoint != "/":
                normalized_endpoint += "/"
            return normalized_endpoint

        base_path = base_path.rstrip("/")
        clean_endpoint: Final = endpoint_path.lstrip("/")
        combined = posixpath.normpath(base_path + "/" + clean_endpoint)
        # If normalization climbs out of the base path, fall back to base.
        if combined != base_path and not combined.startswith(base_path + "/"):
            return base_path + "/"
        if trailing_slash and not combined.endswith("/"):
            combined += "/"
        return combined

    @staticmethod
    def _update_stream_param_based_on_request_body(
        parsed_body: dict,
        stream: bool | None = None,
    ) -> bool | None:
        """
        If stream is provided in the request body, use it.
        Otherwise, use the stream parameter passed to the `pass_through_request` function
        """
        if "stream" in parsed_body:
            return parsed_body.get("stream", stream)
        return stream


def _carry_guardrail_logging_info(request_data: dict, guardrail_data: dict | None) -> None:
    """Copy guardrail logging entries from ``guardrail_data`` onto ``request_data``.

    Post-call guardrails run against a throwaway ``hook_data`` dict (its
    ``metadata`` is what ``_init_kwargs_for_pass_through_endpoint`` already
    stripped off ``_parsed_body``), so a block records the
    ``standard_logging_guardrail_information`` there and not on the dict the
    failure handler forwards to ``post_call_failure_hook``. Without this the
    otel guardrail span is emitted on allow but missing on block. Carry the
    entries over so the failure path matches the unified path.
    """
    if guardrail_data is None:
        return
    source_key: Final = get_metadata_variable_name_from_kwargs(guardrail_data)
    source_metadata: Final = guardrail_data.get(source_key) or {}
    entries: Final = source_metadata.get("standard_logging_guardrail_information")
    if not entries:
        return

    _, metadata = get_or_create_metadata_bucket(request_data)
    metadata.setdefault("standard_logging_guardrail_information", list(entries))


def _build_passthrough_failure_request_payload(
    parsed_body: dict | None,
    kwargs: dict | None,
    logging_obj: LiteLLMLoggingObj | None,
    custom_llm_provider: str | None,
    upstream_usage: UpstreamReportedUsage | None = None,
) -> dict:
    """Build the ``request_data`` dict passed to ``post_call_failure_hook``.

    Shared by the outer exception handler (LiteLLM-internal failures) and
    upstream HTTP error logging, so both failure paths report the same shape
    of request data (model, custom_llm_provider, litellm_logging_obj, ...).

    ``upstream_usage`` carries the cost and tokens an upstream reported on an
    error response. Spend tracking only attributes a recovered cost when it
    comes paired with a usage object, so both keys are written together.
    """
    request_payload: Final[dict] = dict(parsed_body or {})
    if kwargs:
        request_payload.update(kwargs)
    if logging_obj is not None:
        request_payload["litellm_logging_obj"] = logging_obj
    if "model" not in request_payload and parsed_body and isinstance(parsed_body, dict):
        request_payload["model"] = parsed_body.get("model", "")
    if "custom_llm_provider" not in request_payload and custom_llm_provider:
        request_payload["custom_llm_provider"] = custom_llm_provider
    if upstream_usage is not None:
        request_payload["response_cost"] = upstream_usage.response_cost or 0.0
        request_payload["combined_usage_object"] = Usage(total_tokens=upstream_usage.total_tokens or 0)
    return request_payload


async def _log_passthrough_upstream_failure(
    response: httpx.Response,
    user_api_key_dict: UserAPIKeyAuth,
    request_payload: dict,
) -> None:
    """Fire LiteLLM-side failure hooks (spend tracking, alerting callbacks) for
    an upstream 4xx/5xx passthrough response.

    Passthrough must return the upstream status/body/headers to the client
    unchanged, so this never raises or transforms the response - it only
    mirrors the monitoring side effect that ``post_call_failure_hook`` would
    have received had the error originated inside LiteLLM.
    """
    if response.status_code < 400:
        return
    from litellm.proxy.proxy_server import proxy_logging_obj

    try:
        response.raise_for_status()
    except httpx.HTTPStatusError:
        # Reported as an HTTPException, not the raw httpx error: ProxyLogging's
        # alerting path only excludes HTTPException/ProxyException from its
        # "High" severity llm_exceptions alert, treating everything else as an
        # operational LLM-API failure. An upstream 4xx/5xx returned unchanged
        # to the client is a user-facing error like any other, not something
        # ops needs paged for, so it must be excluded the same way auth and
        # rate-limit errors already are.
        synthetic_exception: Final = HTTPException(
            status_code=response.status_code,
            detail=f"Upstream passthrough request failed with status {response.status_code}",
        )
        try:
            await proxy_logging_obj.post_call_failure_hook(
                user_api_key_dict=user_api_key_dict,
                original_exception=synthetic_exception,
                request_data=request_payload,
                traceback_str=traceback.format_exc(limit=MAXIMUM_TRACEBACK_LINES_TO_LOG),
            )
        except Exception:  # noqa: BLE001 - a failing logging callback must never break the passthrough response
            verbose_proxy_logger.warning(
                "pass_through_endpoint: post_call_failure_hook raised for upstream error",
                exc_info=True,
            )


from litellm.passthrough.timeout_utils import (
    DEFAULT_PASS_THROUGH_REQUEST_TIMEOUT_SECONDS,  # noqa: F401 - re-exported for backward compat
    resolve_llm_passthrough_timeout,  # noqa: F401 - re-exported for backward compat
    resolve_pass_through_request_timeout,
)


async def pass_through_request(
    request: Request,
    target: str,
    custom_headers: dict,
    user_api_key_dict: UserAPIKeyAuth,
    custom_body: dict | None = None,
    forward_headers: bool | None = False,
    merge_query_params: bool | None = False,
    query_params: dict | None = None,
    default_query_params: dict | None = None,
    stream: bool | None = None,
    cost_per_request: float | None = None,
    custom_llm_provider: str | None = None,
    guardrails_config: dict | None = None,
    timeout: float | None = None,
):
    """
    Pass through endpoint handler, makes the httpx request for pass-through endpoints and ensures logging hooks are called

    Args:
        request: The incoming request
        target: The target URL
        custom_headers: The custom headers
        user_api_key_dict: The user API key dictionary
        custom_body: The custom body
        forward_headers: Whether to forward headers
        merge_query_params: Whether to merge query params
        query_params: The query params
        default_query_params: The default query params to be applied if not overridden by client
        stream: Whether to stream the response
        cost_per_request: Optional field - cost per request to the target endpoint
        custom_llm_provider: Optional field - custom LLM provider for the endpoint
        guardrails_config: Optional field - guardrails configuration for passthrough endpoint
        timeout: Optional per-endpoint timeout in seconds. Falls back to
            general_settings.pass_through_request_timeout, then 600s.
    """
    from litellm.exceptions import ModifyResponseException
    from litellm.litellm_core_utils.litellm_logging import Logging
    from litellm.proxy.pass_through_endpoints.passthrough_guardrails import (
        PassthroughGuardrailHandler,
    )
    from litellm.proxy.proxy_server import proxy_logging_obj

    #########################################################
    # Initialize variables
    #########################################################
    litellm_call_id: Final = str(uuid.uuid4())
    url: httpx.URL | None = None

    # parsed request body
    _parsed_body: dict | None = None
    # kwargs for pass through endpoint, contains metadata, litellm_params, call_type, litellm_call_id, passthrough_logging_payload
    kwargs: dict | None = None
    logging_obj: Logging | None = None
    # the dict post-call guardrails wrote their logging info into; the failure
    # handler reuses it so a guardrail block still surfaces its span/logs
    post_call_guardrail_data: dict | None = None

    #########################################################
    try:
        url = httpx.URL(target)
        headers = custom_headers
        headers = HttpPassThroughEndpointHelpers.forward_headers_from_request(
            request_headers=_safe_get_request_headers(request).copy(),
            headers=headers,
            forward_headers=forward_headers,
        )

        requested_query_params: dict | None = query_params or dict(request.query_params)

        endpoint_type: Final[EndpointType] = HttpPassThroughEndpointHelpers.get_endpoint_type(str(url))

        # SigV4-signed callers (e.g. Bedrock) attach the exact bytes that were
        # signed via request.state; we must send those instead of re-encoding the
        # parsed dict (hooks mutate it, breaking the signature / Content-Length).
        # Tolerate request objects without `state` (test fixtures) and only honor
        # values httpx accepts for `content=`.
        _request_state: Final = getattr(request, "state", None)
        state_raw_body: str | bytes | None = (
            getattr(_request_state, LITELLM_PASS_THROUGH_RAW_BODY_STATE_KEY, None)
            if _request_state is not None
            else None
        )
        if state_raw_body is not None and not isinstance(state_raw_body, (str, bytes, bytearray)):
            state_raw_body = None

        # Skip body parsing for multipart requests - make_multipart_http_request will handle it
        # But if custom_body is provided (e.g., JSON parsed despite multipart content-type), use it
        is_multipart: Final = HttpPassThroughEndpointHelpers.is_multipart(request) and not custom_body

        if custom_body:
            _parsed_body = custom_body
        elif is_multipart:
            # Don't parse multipart body here - it will be handled by make_multipart_http_request
            _parsed_body = {}
        else:
            _parsed_body = await _read_request_body(request)
        verbose_proxy_logger.debug(
            "Pass through endpoint sending request to \nURL %s\nheaders: %s\nbody: %s\n",
            url,
            headers,
            _parsed_body,
        )

        ### COLLECT GUARDRAILS FOR PASSTHROUGH ENDPOINT ###
        # Passthrough endpoints are opt-in only for guardrails
        # When enabled, collect guardrails from org/team/key levels + passthrough-specific
        guardrails_to_run: Final = PassthroughGuardrailHandler.collect_guardrails(
            user_api_key_dict=user_api_key_dict,
            passthrough_guardrails_config=guardrails_config,
        )

        # Add guardrails to metadata if any should run
        if guardrails_to_run and len(guardrails_to_run) > 0:
            if _parsed_body is None:
                _parsed_body = {}
            if "metadata" not in _parsed_body:
                _parsed_body["metadata"] = {}
            _parsed_body["metadata"]["guardrails"] = guardrails_to_run
            verbose_proxy_logger.debug("Added guardrails to passthrough request metadata: %s", guardrails_to_run)

        ## LOGGING OBJECT ## - initialize before pre_call_hook so guardrails can access it
        # Surface the requested model (when the body carries one) so logging/spans
        # read e.g. ``chat gpt-4o`` instead of ``chat unknown``.
        passthrough_model: Final = (_parsed_body.get("model") if isinstance(_parsed_body, dict) else None) or "unknown"
        start_time: Final = datetime.now()
        logging_obj = Logging(
            model=passthrough_model,
            messages=[{"role": "user", "content": safe_dumps(_parsed_body)}],
            stream=False,
            call_type="pass_through_endpoint",
            start_time=start_time,
            litellm_call_id=litellm_call_id,
            function_id="1245",
        )

        # Store passthrough guardrails config on logging_obj for field targeting
        logging_obj.passthrough_guardrails_config = guardrails_config

        # Store logging_obj in data so guardrails can access it
        if _parsed_body is None:
            _parsed_body = {}
        _parsed_body["litellm_logging_obj"] = logging_obj

        ### CALL HOOKS ### - modify incoming data / reject request before calling the model
        _parsed_body = await proxy_logging_obj.pre_call_hook(
            user_api_key_dict=user_api_key_dict,
            data=_parsed_body,
            call_type="pass_through_endpoint",
        )
        resolved_timeout: Final = resolve_pass_through_request_timeout(timeout)
        async_client_obj: Final = get_async_httpx_client(
            llm_provider=httpxSpecialProvider.PassThroughEndpoint,
            params={"timeout": resolved_timeout},
        )
        async_client: Final = async_client_obj.client
        passthrough_logging_payload: Final = PassthroughStandardLoggingPayload(
            url=str(url),
            request_body=_parsed_body,
            request_method=getattr(request, "method", None),
            cost_per_request=cost_per_request,
        )
        kwargs = HttpPassThroughEndpointHelpers._init_kwargs_for_pass_through_endpoint(
            user_api_key_dict=user_api_key_dict,
            _parsed_body=_parsed_body,
            passthrough_logging_payload=passthrough_logging_payload,
            litellm_call_id=litellm_call_id,
            request=request,
            logging_obj=logging_obj,
        )

        # Store custom_llm_provider in kwargs and logging object if provided
        if custom_llm_provider:
            logging_obj.model_call_details["custom_llm_provider"] = custom_llm_provider
            logging_obj.model_call_details["litellm_params"] = kwargs.get("litellm_params", {})

        # done for supporting 'parallel_request_limiter.py' with pass-through endpoints
        logging_obj.update_environment_variables(
            model=passthrough_model,
            user="unknown",
            optional_params={},
            litellm_params=kwargs["litellm_params"],
            call_type="pass_through_endpoint",
        )
        logging_obj.model_call_details["litellm_call_id"] = litellm_call_id

        ## PASSTHROUGH MANAGED ID RESOLUTION (INPUT) ##
        # Resolve managed IDs in path, query params, and body back to raw
        # provider IDs before forwarding upstream.  Gated by feature flag and
        # enterprise managed-files hook.  Runs after pre_call_hook so
        # guardrails have already seen the managed IDs.
        from litellm.proxy.proxy_server import (
            general_settings as proxy_general_settings,
        )

        _managed_id_provider: Final = resolve_passthrough_managed_id_provider(custom_llm_provider)

        if proxy_general_settings.get("passthrough_managed_object_ids", False) and _managed_id_provider is not None:
            verbose_proxy_logger.debug(
                "pass_through_endpoint: managed-id input rewrite enabled for route=%s method=%s",
                request.url.path,
                request.method,
            )
            _passthrough_managed_hook = proxy_logging_obj.get_proxy_hook("managed_files")
            if _passthrough_managed_hook is not None:
                from litellm.proxy.pass_through_endpoints.managed_id_rewriter import (
                    rewrite_body_ids,
                    rewrite_path_ids,
                    rewrite_query_ids,
                )
                from litellm.proxy.proxy_server import (
                    prisma_client as _passthrough_prisma,
                )

                _original_path: Final = url.path
                _original_query_params: Final = requested_query_params
                _original_body: Final = _parsed_body
                _new_path: Final = await rewrite_path_ids(
                    url.path,
                    _managed_id_provider,
                    user_api_key_dict,
                    _passthrough_prisma,
                    _passthrough_managed_hook,
                )
                if _new_path != url.path:
                    url = url.copy_with(path=_new_path)
                requested_query_params = await rewrite_query_ids(
                    requested_query_params,
                    _managed_id_provider,
                    user_api_key_dict,
                    _passthrough_prisma,
                    _passthrough_managed_hook,
                )
                _parsed_body = await rewrite_body_ids(
                    _parsed_body,
                    _managed_id_provider,
                    user_api_key_dict,
                    _passthrough_prisma,
                    _passthrough_managed_hook,
                )
                verbose_proxy_logger.debug(
                    "pass_through_endpoint: managed-id input rewrite results path_changed=%s query_changed=%s body_changed=%s route=%s method=%s",
                    _new_path != _original_path,
                    requested_query_params is not _original_query_params,
                    _parsed_body is not _original_body,
                    request.url.path,
                    request.method,
                )
            else:
                verbose_proxy_logger.debug(
                    "pass_through_endpoint: managed-id input rewrite skipped (managed_files hook not available) route=%s method=%s",
                    request.url.path,
                    request.method,
                )

        # Apply default query parameters if provided, regardless of merge_query_params setting
        if default_query_params or merge_query_params:
            # Create a new URL with the merged query params
            url = url.copy_with(
                query=urlencode(
                    HttpPassThroughEndpointHelpers.get_merged_query_parameters(
                        existing_url=url,
                        request_query_params=requested_query_params,
                        default_query_params=default_query_params,
                    )
                ).encode("ascii")
            )
            requested_query_params = None

        ## PASSTHROUGH MANAGED LIST (DB-only response) ##
        # For GET /v1/files and GET /v1/batches passthrough routes, serve the
        # listing entirely from our DB so each caller only sees their own IDs.
        # Admins / master-key callers see all rows.  Gated on the same
        # conditions as INPUT/OUTPUT rewrite: feature flag, provider, AND
        # the managed_files hook must be present.  Without the hook no managed
        # IDs are ever minted or stored, so the DB is empty and intercepting
        # the list would silently hide the caller's real upstream files/batches.
        if (
            proxy_general_settings.get("passthrough_managed_object_ids", False)
            and _managed_id_provider is not None
            and request.method == "GET"
            and proxy_logging_obj.get_proxy_hook("managed_files") is not None
        ):
            from litellm.proxy.auth.auth_utils import get_request_route
            from litellm.proxy.pass_through_endpoints.managed_id_rewriter import (
                is_passthrough_list_route,
                list_passthrough_ids_from_db,
            )
            from litellm.proxy.proxy_server import prisma_client as _list_prisma

            if (
                is_passthrough_list_route(_managed_id_provider, request.method, get_request_route(request))
                and _list_prisma is not None
            ):
                _list_result: Final = await list_passthrough_ids_from_db(
                    provider=_managed_id_provider,
                    route=get_request_route(request),
                    user_api_key_dict=user_api_key_dict,
                    prisma_client=_list_prisma,
                    query_params=dict(request.query_params),
                )
                if _list_result is not None:
                    verbose_proxy_logger.debug(
                        "pass_through_endpoint: list served from DB route=%s count=%d",
                        request.url.path,
                        len(_list_result.get("data", [])),
                    )
                    return Response(
                        content=json.dumps(_list_result),
                        status_code=200,
                        media_type="application/json",
                    )

        requested_query_params_str = None
        if requested_query_params:
            requested_query_params_str = "&".join(f"{k}={v}" for k, v in requested_query_params.items())

        logging_url = str(url)
        if requested_query_params_str:
            if "?" in str(url):
                logging_url = str(url) + "&" + requested_query_params_str
            else:
                logging_url = str(url) + "?" + requested_query_params_str

        logging_obj.pre_call(
            input=[{"role": "user", "content": safe_dumps(_parsed_body)}],
            api_key="",
            additional_args={
                "complete_input_dict": _parsed_body,
                "api_base": str(logging_url),
                "headers": headers,
            },
        )
        stream = HttpPassThroughEndpointHelpers._update_stream_param_based_on_request_body(
            parsed_body=_parsed_body or {},
            stream=stream,
        )

        if stream:
            logging_obj.stream = True
            logging_obj.model_call_details["stream"] = True

            if is_multipart:
                response = await HttpPassThroughEndpointHelpers.make_multipart_http_request(
                    request=request,
                    async_client=async_client,
                    url=url,
                    headers=headers,
                    requested_query_params=requested_query_params,
                    stream=True,
                )
            else:
                # SigV4-signed callers (Bedrock) supply the exact pre-signed bytes;
                # otherwise httpx encodes the parsed JSON dict as before.
                req: Final = (
                    async_client.build_request(
                        request.method,
                        url,
                        params=requested_query_params,
                        headers=headers,
                        content=state_raw_body,
                    )
                    if state_raw_body is not None
                    else async_client.build_request(
                        request.method,
                        url,
                        params=requested_query_params,
                        headers=headers,
                        json=_parsed_body,
                    )
                )

                response = await async_client.send(req, stream=stream)

            upstream_usage = apply_upstream_reported_usage(
                logging_obj=logging_obj,
                headers=response.headers,
            )

            await _log_passthrough_upstream_failure(
                response=response,
                user_api_key_dict=user_api_key_dict,
                request_payload=_build_passthrough_failure_request_payload(
                    parsed_body=_parsed_body,
                    kwargs=kwargs,
                    logging_obj=logging_obj,
                    custom_llm_provider=custom_llm_provider,
                    upstream_usage=upstream_usage,
                ),
            )

            # Call response headers hook for streaming pass-through
            _response_headers = HttpPassThroughEndpointHelpers.get_response_headers(
                headers=response.headers,
                litellm_call_id=litellm_call_id,
            )
            callback_headers = await proxy_logging_obj.post_call_response_headers_hook(
                data=_parsed_body or {},
                user_api_key_dict=user_api_key_dict,
                response=response,
                request_headers=dict(request.headers),
            )
            if callback_headers:
                _response_headers.update(callback_headers)

            return StreamingResponse(
                wrap_passthrough_sse_bytes_with_keepalive_pings(
                    stream=_own_streamed_managed_ids(
                        stream=PassThroughStreamingHandler.chunk_processor(
                            response=response,
                            request_body=_parsed_body,
                            litellm_logging_obj=logging_obj,
                            endpoint_type=endpoint_type,
                            start_time=start_time,
                            passthrough_success_handler_obj=pass_through_endpoint_logging,
                            url_route=str(url),
                        ),
                        managed_id_provider=_managed_id_provider,
                        request=request,
                        user_api_key_dict=user_api_key_dict,
                    ),
                    ping_interval_seconds=litellm.sse_keepalive_ping_interval_seconds,
                    upstream_headers=response.headers,
                ),
                headers=_response_headers,
                status_code=response.status_code,
            )

        if state_raw_body is not None:
            # SigV4-signed callers (Bedrock) require the exact pre-signed bytes
            # to be forwarded so the signature/Content-Length stay valid.
            raw_body_request: Final = async_client.build_request(
                request.method,
                url,
                headers=headers,
                params=requested_query_params,
                content=state_raw_body,
            )
            response = await async_client.send(raw_body_request, stream=True)
        else:
            response = await HttpPassThroughEndpointHelpers.non_streaming_http_request_handler(
                request=request,
                async_client=async_client,
                url=url,
                headers=headers,
                requested_query_params=requested_query_params,
                _parsed_body=_parsed_body,
                forward_multipart=is_multipart,
            )
        verbose_proxy_logger.debug("response.headers= %s", response.headers)

        upstream_usage = apply_upstream_reported_usage(
            logging_obj=logging_obj,
            headers=response.headers,
        )

        if _is_streaming_response(response) is True:
            logging_obj.stream = True
            logging_obj.model_call_details["stream"] = True

            await _log_passthrough_upstream_failure(
                response=response,
                user_api_key_dict=user_api_key_dict,
                request_payload=_build_passthrough_failure_request_payload(
                    parsed_body=_parsed_body,
                    kwargs=kwargs,
                    logging_obj=logging_obj,
                    custom_llm_provider=custom_llm_provider,
                    upstream_usage=upstream_usage,
                ),
            )

            # Call response headers hook for detected streaming pass-through
            _response_headers = HttpPassThroughEndpointHelpers.get_response_headers(
                headers=response.headers,
                litellm_call_id=litellm_call_id,
            )
            callback_headers = await proxy_logging_obj.post_call_response_headers_hook(
                data=_parsed_body or {},
                user_api_key_dict=user_api_key_dict,
                response=response,
                request_headers=dict(request.headers),
            )
            if callback_headers:
                _response_headers.update(callback_headers)

            return StreamingResponse(
                wrap_passthrough_sse_bytes_with_keepalive_pings(
                    stream=_own_streamed_managed_ids(
                        stream=PassThroughStreamingHandler.chunk_processor(
                            response=response,
                            request_body=_parsed_body,
                            litellm_logging_obj=logging_obj,
                            endpoint_type=endpoint_type,
                            start_time=start_time,
                            passthrough_success_handler_obj=pass_through_endpoint_logging,
                            url_route=str(url),
                        ),
                        managed_id_provider=_managed_id_provider,
                        request=request,
                        user_api_key_dict=user_api_key_dict,
                    ),
                    ping_interval_seconds=litellm.sse_keepalive_ping_interval_seconds,
                    upstream_headers=response.headers,
                ),
                headers=_response_headers,
                status_code=response.status_code,
            )

        if not _should_buffer_passthrough_response(response):
            relay_custom_headers: Final = ProxyBaseLLMRequestProcessing.get_custom_headers(
                user_api_key_dict=user_api_key_dict,
                call_id=litellm_call_id,
                model_id=None,
                cache_key=None,
                api_base=str(url._uri_reference),
            )
            relay_callback_headers: Final = await proxy_logging_obj.post_call_response_headers_hook(
                data=_parsed_body or {},
                user_api_key_dict=user_api_key_dict,
                response=response,
                request_headers=dict(request.headers),
            )
            if relay_callback_headers:
                relay_custom_headers.update(relay_callback_headers)

            return StreamingResponse(
                _relay_passthrough_response_bytes(
                    response=response,
                    request_body=_parsed_body or {},
                    url_route=str(url),
                    start_time=start_time,
                    logging_obj=logging_obj,
                    custom_llm_provider=custom_llm_provider,
                    success_handler_kwargs=kwargs,
                ),
                status_code=response.status_code,
                headers=HttpPassThroughEndpointHelpers.get_response_headers(
                    headers=response.headers,
                    custom_headers=relay_custom_headers,
                ),
            )

        content = await response.aread()

        ## POST-CALL GUARDRAILS ##
        # Guardrails and managed-id rewriting only apply to successful upstream
        # responses; response_body itself is parsed unconditionally so the
        # failure-hook log payload below still reflects upstream error bodies.
        _content_modified = False
        response_body: dict | None = get_response_body(response)

        failure_request_payload: Final = _build_passthrough_failure_request_payload(
            parsed_body=_parsed_body,
            kwargs=kwargs,
            logging_obj=logging_obj,
            custom_llm_provider=custom_llm_provider,
            upstream_usage=upstream_usage,
        )
        failure_request_payload["response_body"] = response_body
        await _log_passthrough_upstream_failure(
            response=response,
            user_api_key_dict=user_api_key_dict,
            request_payload=failure_request_payload,
        )

        if response.status_code < 400 and response_body is not None and guardrails_to_run:
            # Build an enriched data dict: _parsed_body has been stripped of
            # `metadata` by both pre_call_hook and _init_kwargs_for_pass_through_endpoint,
            # so we re-attach the configured guardrails here so should_run_guardrail
            # sees them.
            hook_data: Final = dict(_parsed_body or {})
            existing_metadata = hook_data.get("metadata")
            if not isinstance(existing_metadata, dict):
                existing_metadata = {}
            hook_data["metadata"] = {
                **existing_metadata,
                "guardrails": guardrails_to_run,
            }
            post_call_guardrail_data = hook_data
            response_body = await proxy_logging_obj.post_call_success_hook(
                data=hook_data,
                user_api_key_dict=user_api_key_dict,
                response=response_body,
            )
            if isinstance(response_body, dict):
                content = json.dumps(response_body).encode("utf-8")
                _content_modified = True
            else:
                verbose_proxy_logger.debug(
                    "pass_through_endpoint: post_call_success_hook returned %s, expected dict — using original response",
                    type(response_body).__name__,
                )
        elif response_body is None:
            verbose_proxy_logger.debug(
                "pass_through_endpoint: response body not JSON-parseable, skipping post-call guardrails"
            )

        ## PASSTHROUGH MANAGED ID MINTING (OUTPUT) ##
        # Mint managed IDs for raw provider IDs in the response body and swap
        # them before the response reaches the client.  Runs after guardrails
        # so guardrails see the raw IDs (cleaner) and the client receives the
        # managed IDs.  Gated by feature flag and enterprise managed-files hook.
        if (
            proxy_general_settings.get("passthrough_managed_object_ids", False)
            and _managed_id_provider is not None
            and isinstance(response_body, dict)
            and response.status_code < 300
        ):
            verbose_proxy_logger.debug(
                "pass_through_endpoint: managed-id output rewrite enabled for route=%s method=%s status=%s",
                request.url.path,
                request.method,
                response.status_code,
            )
            _passthrough_managed_hook = proxy_logging_obj.get_proxy_hook("managed_files")
            if _passthrough_managed_hook is not None:
                from litellm.proxy.auth.auth_utils import get_request_route
                from litellm.proxy.pass_through_endpoints.managed_id_rewriter import (
                    rewrite_response_ids,
                )
                from litellm.proxy.proxy_server import (
                    prisma_client as _passthrough_prisma,
                )

                _new_body: Final = await rewrite_response_ids(
                    provider=_managed_id_provider,
                    method=request.method,
                    route=get_request_route(request),
                    body=response_body,
                    user_api_key_dict=user_api_key_dict,
                    prisma_client=_passthrough_prisma,
                    managed_files_hook=_passthrough_managed_hook,
                )
                if _new_body is not response_body:
                    response_body = _new_body
                    content = json.dumps(response_body).encode("utf-8")
                    _content_modified = True
                    verbose_proxy_logger.debug(
                        "pass_through_endpoint: managed-id output rewrite applied route=%s method=%s",
                        request.url.path,
                        request.method,
                    )
                else:
                    verbose_proxy_logger.debug(
                        "pass_through_endpoint: managed-id output rewrite no-op route=%s method=%s",
                        request.url.path,
                        request.method,
                    )
            else:
                verbose_proxy_logger.debug(
                    "pass_through_endpoint: managed-id output rewrite skipped (managed_files hook not available) route=%s method=%s",
                    request.url.path,
                    request.method,
                )

        ## LOG SUCCESS
        # Upstream errors are already logged via _log_passthrough_upstream_failure
        # above; the success handler has no status-code awareness of its own; so
        # calling it here for a 4xx/5xx would double-log the same request as both
        # a failure and a success (corrupting spend tracking).
        passthrough_logging_payload["response_body"] = response_body
        end_time: Final = datetime.now()
        if response.status_code < 400:
            GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue(
                async_coroutine=pass_through_endpoint_logging.pass_through_async_success_handler(
                    httpx_response=response,
                    response_body=response_body,
                    url_route=str(url),
                    result="",
                    start_time=start_time,
                    end_time=end_time,
                    logging_obj=logging_obj,
                    cache_hit=False,
                    request_body=_parsed_body or {},
                    custom_llm_provider=custom_llm_provider,
                    **kwargs,
                )
            )

        ## CUSTOM HEADERS - `x-litellm-*`
        custom_headers = ProxyBaseLLMRequestProcessing.get_custom_headers(
            user_api_key_dict=user_api_key_dict,
            call_id=litellm_call_id,
            model_id=None,
            cache_key=None,
            api_base=str(url._uri_reference),
        )

        # Call response headers hook
        callback_headers = await proxy_logging_obj.post_call_response_headers_hook(
            data=_parsed_body or {},
            user_api_key_dict=user_api_key_dict,
            response=response,
            request_headers=dict(request.headers),
        )
        if callback_headers:
            custom_headers.update(callback_headers)

        response_headers: Final = HttpPassThroughEndpointHelpers.get_response_headers(
            headers=response.headers,
            custom_headers=custom_headers,
        )
        if _content_modified:
            response_headers.pop("content-length", None)

        return Response(
            content=content,
            status_code=response.status_code,
            headers=response_headers,
        )
    except ModifyResponseException as e:
        verbose_proxy_logger.info(
            "pass_through_endpoint: Guardrail %s modified response: %s",
            e.guardrail_name,
            str(e.message or "")[:200],
        )
        try:
            await proxy_logging_obj.post_call_failure_hook(
                user_api_key_dict=user_api_key_dict,
                original_exception=e,
                request_data=e.request_data,
            )
        except Exception:
            verbose_proxy_logger.warning(
                "pass_through_endpoint: post_call_failure_hook raised during guardrail block",
                exc_info=True,
            )
        error_body: Final = {
            "error": {
                "message": e.message or "Response blocked by guardrail",
                "type": "content_filter",
                "guardrail_name": e.guardrail_name,
                "model": e.model,
            }
        }
        return Response(
            content=json.dumps(error_body),
            status_code=200,
            media_type="application/json",
        )
    except Exception as e:
        custom_headers = ProxyBaseLLMRequestProcessing.get_custom_headers(
            user_api_key_dict=user_api_key_dict,
            call_id=litellm_call_id,
            model_id=None,
            cache_key=None,
            api_base=str(url._uri_reference) if url else None,
        )
        if CustomGuardrail._is_guardrail_intervention(e):
            verbose_proxy_logger.warning(
                "pass_through_endpoint: request blocked by guardrail - %s",
                str(e),
            )
        else:
            verbose_proxy_logger.exception(
                "litellm.proxy.proxy_server.pass_through_endpoint(): Exception occured - %s", e
            )

        #########################################################
        # Monitoring: Trigger post_call_failure_hook
        # for pass through endpoint failure
        #########################################################
        request_payload: Final[dict] = _parsed_body or {}
        # add user_api_key_dict, litellm_call_id, passthrough_logging_payloa for logging
        if kwargs:
            for key, value in kwargs.items():
                request_payload[key] = value
        if logging_obj is not None:
            request_payload["litellm_logging_obj"] = logging_obj

        if "model" not in request_payload and _parsed_body and isinstance(_parsed_body, dict):
            request_payload["model"] = _parsed_body.get("model", "")
        if "custom_llm_provider" not in request_payload and custom_llm_provider:
            request_payload["custom_llm_provider"] = custom_llm_provider

        _carry_guardrail_logging_info(request_payload, post_call_guardrail_data)

        await proxy_logging_obj.post_call_failure_hook(
            user_api_key_dict=user_api_key_dict,
            original_exception=e,
            request_data=request_payload,
            traceback_str=traceback.format_exc(
                limit=MAXIMUM_TRACEBACK_LINES_TO_LOG,
            ),
        )

        #########################################################

        if isinstance(e, ProxyException):
            raise
        if isinstance(e, HTTPException):
            raise ProxyException(
                message=getattr(e, "message", str(getattr(e, "detail", str(e)))),
                type=getattr(e, "type", "None"),
                param=getattr(e, "param", "None"),
                code=getattr(e, "status_code", status.HTTP_400_BAD_REQUEST),
                headers=custom_headers,
            )
        else:
            error_msg: Final = f"{e}"
            raise ProxyException(
                message=getattr(e, "message", error_msg),
                type=getattr(e, "type", "None"),
                param=getattr(e, "param", "None"),
                code=getattr(e, "status_code", 500),
                headers=custom_headers,
            )


def _update_metadata_with_tags_in_header(request: Request, metadata: dict) -> dict:
    """
    If tags are in the request headers, add them to the metadata

    Used for google and vertex JS SDKs, and Azure passthrough
    Checks both 'tags' and 'x-litellm-tags' headers
    """
    tags_to_add: Final = []

    # Check for 'tags' header first
    _tags = request.headers.get("tags")
    if _tags:
        tags_to_add.extend([tag.strip() for tag in _tags.split(",")])

    _tags = request.headers.get("x-litellm-tags")
    if _tags:
        tags_to_add.extend([tag.strip() for tag in _tags.split(",")])

    # Only add tags key if there are tags to add
    if tags_to_add:
        if "tags" not in metadata:
            metadata["tags"] = []
        metadata["tags"].extend(tags_to_add)

    return metadata


class _PassThroughRequestEnvelope(TypedDict, total=False):
    query_params: Mapping[str, object] | None
    custom_body: Mapping[str, object] | None
    stream: bool | None


async def _parse_request_data_by_content_type(
    request: Request,
) -> tuple[object, object, None, bool | None]:
    """
    Parse request data based on content type.

    Handles JSON, multipart/form-data, and URL-encoded form data.

    Returns:
        Tuple of (query_params_data, custom_body_data, file_data, stream)
    """
    content_type: Final = request.headers.get("content-type", "")

    query_params_data = None
    custom_body_data = None
    file_data: Final = None
    stream = None

    if "application/json" in content_type:
        # ✅ Handle JSON
        try:
            body: _PassThroughRequestEnvelope = await request.json()
            query_params_data = body.get("query_params")
            custom_body_data = body.get("custom_body")
            stream = body.get("stream")
        except json.JSONDecodeError:
            # Handle requests with no body (e.g., DELETE requests)
            pass
    elif "multipart/form-data" in content_type:
        # ✅ Try to parse as JSON first (handles misconfigured clients sending JSON with multipart content-type)
        # If that fails, skip parsing - pass_through_request will handle actual multipart
        try:
            body = await request.json()
            # Successfully parsed as JSON - treat as JSON body
            query_params_data = body.get("query_params")
            custom_body_data = body.get("custom_body")
            stream = body.get("stream")
            # If custom_body is not set, use the entire body
            if custom_body_data is None and body:
                custom_body_data = body
        except (json.JSONDecodeError, Exception):
            # Not JSON - this is actual multipart data
            # Skip parsing here to avoid consuming the request body stream
            # make_multipart_http_request will handle it
            pass

    elif "application/x-www-form-urlencoded" in content_type:
        # ✅ Handle URL-encoded form data
        form: Final = await request.form()
        query_params_data = form.get("query_params")
        custom_body_data = form.get("custom_body")

    else:
        # ✅ Fallback: maybe no body, just query params
        query_params_data = dict(request.query_params) or None

    return query_params_data, custom_body_data, file_data, stream


def create_pass_through_route(
    endpoint,
    target: str,
    custom_headers: Mapping[str, object] | None = None,
    _forward_headers: bool | None = False,
    _merge_query_params: bool | None = False,
    dependencies: list | None = None,
    include_subpath: bool | None = False,
    cost_per_request: float | None = None,
    custom_llm_provider: str | None = None,
    is_streaming_request: bool | None = False,
    query_params: dict | None = None,
    default_query_params: dict | None = None,
    guardrails: dict[str, object] | None = None,
    config_file_path: str | None = None,
    timeout: float | None = None,
):
    # check if target is an adapter.py or a url
    from litellm._uuid import uuid
    from litellm.proxy.types_utils.utils import get_instance_fn

    try:
        if isinstance(target, CustomLogger):
            adapter = target
        else:
            adapter = get_instance_fn(value=target, config_file_path=config_file_path)
        adapter_id: Final = str(uuid.uuid4())
        litellm.adapters = [{"id": adapter_id, "adapter": adapter}]

        async def endpoint_func(
            request: Request,
            fastapi_response: Response,
            user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
            subpath: str = "",  # captures sub-paths when include_subpath=True
        ):
            return await chat_completion_pass_through_endpoint(
                fastapi_response=fastapi_response,
                request=request,
                adapter_id=adapter_id,
                user_api_key_dict=user_api_key_dict,
            )

    except Exception:
        verbose_proxy_logger.debug("Defaulting to target being a url.")

        async def endpoint_func(
            request: Request,
            fastapi_response: Response,
            user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
            subpath: str = "",  # captures sub-paths when include_subpath=True
        ):
            from litellm.proxy.auth.auth_utils import (  # noqa: PLC0415
                get_request_route,
            )
            from litellm.proxy.pass_through_endpoints.pass_through_endpoints import (
                InitPassThroughEndpointHelpers,
            )

            path: Final = get_request_route(request)

            # Parse request data based on content type
            (
                query_params_data,
                custom_body_data,
                file_data,
                stream,
            ) = await _parse_request_data_by_content_type(request)

            if not InitPassThroughEndpointHelpers.is_registered_pass_through_route(route=path):
                raise HTTPException(
                    status_code=404,
                    detail=f"Pass-through endpoint {endpoint} not found. This could have been deleted or not yet added to the proxy.",
                )

            passthrough_params: Final = InitPassThroughEndpointHelpers.get_registered_pass_through_route(
                route=path, method=request.method
            )
            if (
                passthrough_params is None
                and InitPassThroughEndpointHelpers.get_registered_pass_through_route(route=path) is not None
            ):
                raise HTTPException(
                    status_code=status.HTTP_405_METHOD_NOT_ALLOWED,
                    detail=f"Method {request.method} is not allowed for pass-through endpoint {path}.",
                )
            target_params: Final = {
                "target": target,
                "custom_headers": custom_headers,
                "forward_headers": _forward_headers,
                "merge_query_params": _merge_query_params,
                "cost_per_request": cost_per_request,
                "guardrails": None,
                "timeout": timeout,
            }

            if passthrough_params is not None:
                target_params.update(passthrough_params.get("passthrough_params", {}))

            # Extract and cast parameters with proper types
            param_target: Final = target_params.get("target") or target
            param_custom_headers: Final = target_params.get("custom_headers", custom_headers)
            param_forward_headers: Final = target_params.get("forward_headers", _forward_headers)
            param_merge_query_params: Final = target_params.get("merge_query_params", _merge_query_params)
            param_cost_per_request: Final = target_params.get("cost_per_request", cost_per_request)
            param_guardrails: Final = target_params.get("guardrails", None)
            param_default_query_params: Final = target_params.get("default_query_params", None)
            param_timeout: Final = target_params.get("timeout", timeout)

            # Construct the full target URL with subpath if needed
            full_target: Final = HttpPassThroughEndpointHelpers.construct_target_url_with_subpath(
                base_target=cast(str, param_target),
                subpath=subpath,
                include_subpath=include_subpath,
            )

            # Ensure custom_headers is a dict. Botocore returns a HeadersDict
            # for SigV4-prepared requests, which is a Mapping but not a dict.
            headers_dict: Final = dict(param_custom_headers) if isinstance(param_custom_headers, Mapping) else {}

            # Ensure query_params and custom_body are dicts or None
            final_query_params: Final = query_params_data if isinstance(query_params_data, dict) else {}
            if query_params:
                final_query_params.update(query_params)
            # Programmatic callers set LITELLM_PASS_THROUGH_CUSTOM_BODY_STATE_KEY on
            # request.state (see Bedrock proxy). Parsed JSON envelope otherwise.
            state_custom_body: Final[dict | None] = getattr(
                request.state,
                LITELLM_PASS_THROUGH_CUSTOM_BODY_STATE_KEY,
                None,
            )
            final_custom_body: dict | None = None
            if isinstance(state_custom_body, dict):
                final_custom_body = state_custom_body
            elif isinstance(custom_body_data, dict):
                final_custom_body = custom_body_data

            is_stream: Final = bool(is_streaming_request or stream)

            async def _relay() -> Response:
                try:
                    return await pass_through_request(
                        request=request,
                        target=full_target,
                        custom_headers=headers_dict,
                        user_api_key_dict=user_api_key_dict,
                        forward_headers=cast(bool | None, param_forward_headers),
                        merge_query_params=cast(bool | None, param_merge_query_params),
                        query_params=final_query_params,
                        default_query_params=cast(dict | None, param_default_query_params),
                        stream=is_stream,
                        custom_body=final_custom_body,
                        cost_per_request=cast(float | None, param_cost_per_request),
                        custom_llm_provider=custom_llm_provider,
                        guardrails_config=cast(dict | None, param_guardrails),
                        timeout=cast(float | None, param_timeout),
                    )
                finally:
                    if hasattr(request.state, LITELLM_PASS_THROUGH_CUSTOM_BODY_STATE_KEY):
                        delattr(request.state, LITELLM_PASS_THROUGH_CUSTOM_BODY_STATE_KEY)
                    if hasattr(request.state, LITELLM_PASS_THROUGH_RAW_BODY_STATE_KEY):
                        delattr(request.state, LITELLM_PASS_THROUGH_RAW_BODY_STATE_KEY)

            # The upstream withholds its response headers until its first token, so
            # the whole time-to-first-token is spent inside _relay with nothing on
            # the wire. Off unless an operator sets an interval.
            return await open_sse_before_first_byte(
                _relay(),
                ping_interval_seconds=(litellm.sse_keepalive_ping_interval_seconds if is_stream else None),
            )

    setattr(endpoint_func, LITELLM_PASS_THROUGH_ENDPOINT_MARKER, True)
    return endpoint_func


def create_websocket_passthrough_route(
    endpoint: str,
    target: str,
    custom_headers: dict | None = None,
    _forward_headers: bool | None = False,
    dependencies: list | None = None,
    cost_per_request: float | None = None,
):
    """
    Create a WebSocket passthrough route function.

    Args:
        endpoint: The endpoint path (for logging purposes)
        target: The target WebSocket URL (e.g., "wss://api.example.com/ws")
        custom_headers: Custom headers to include in the WebSocket connection
        _forward_headers: Whether to forward incoming headers
        dependencies: FastAPI dependencies to inject

    Returns:
        A WebSocket passthrough function that can be registered with app.websocket()
    """
    from litellm.proxy.auth.user_api_key_auth import user_api_key_auth_websocket

    async def websocket_endpoint_func(
        websocket: WebSocket,
        user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth_websocket),
        **kwargs,  # For additional query parameters
    ):
        """
        WebSocket passthrough endpoint function.

        This function handles the WebSocket connection by:
        1. Accepting the incoming WebSocket connection
        2. Establishing a connection to the target WebSocket
        3. Forwarding messages bidirectionally
        4. Handling connection cleanup
        """
        return await websocket_passthrough_request(
            websocket=websocket,
            target=target,
            custom_headers=custom_headers or {},
            user_api_key_dict=user_api_key_dict,
            forward_headers=_forward_headers,
            endpoint=endpoint,
            cost_per_request=cost_per_request,
            accept_websocket=True,  # Generic usage should accept the WebSocket
        )

    return websocket_endpoint_func


def _rewrite_vertex_live_setup_model(text_data: str, setup_model_rewriter: Callable[[str], str] | None) -> str:
    """
    Rewrite the model of a Vertex AI Live ``setup`` frame, leaving every other frame byte-identical
    """
    if setup_model_rewriter is None:
        return text_data
    try:
        message: Final = json.loads(text_data)
    except json.JSONDecodeError:
        return text_data
    if not isinstance(message, dict):
        return text_data
    setup: Final = message.get("setup")
    if not isinstance(setup, dict):
        return text_data
    setup_model: Final = setup.get("model")
    if not isinstance(setup_model, str):
        return text_data
    rewritten_model: Final = setup_model_rewriter(setup_model)
    if rewritten_model == setup_model:
        return text_data
    return json.dumps({**message, "setup": {**setup, "model": rewritten_model}})  # mutable-ok: one-shot json payload


def _truncated_close_reason(reason: str) -> str:
    """
    Fit a close reason inside the byte budget a WebSocket close frame allows, without splitting a character
    """
    encoded: Final = reason.encode("utf-8")
    if len(encoded) <= WEBSOCKET_CLOSE_REASON_MAX_BYTES:
        return reason
    return encoded[:WEBSOCKET_CLOSE_REASON_MAX_BYTES].decode("utf-8", errors="ignore")


SENDABLE_CLOSE_CODES: Final = frozenset(CloseCode) - frozenset(
    {CloseCode.NO_STATUS_RCVD, CloseCode.ABNORMAL_CLOSURE, CloseCode.TLS_HANDSHAKE}
)


def _client_socket_is_open(websocket: WebSocket) -> bool:
    """
    Starlette tracks the two halves separately and raises on a second close, so both have to still be live
    """
    return (
        websocket.client_state != WebSocketState.DISCONNECTED
        and websocket.application_state != WebSocketState.DISCONNECTED
    )


def _upstream_close_to_relay(task_results: Iterable[object]) -> Close | None:
    """
    The upstream close worth telling the client about: anything other than a plain, reasonless normal close.

    Codes outside ``SENDABLE_CLOSE_CODES`` and the private range never travel on the wire (1006 for a socket that
    died without a close frame, 1005 for one that sent no code), so relaying them would build an invalid frame
    """
    upstream_close: Final = next((result for result in task_results if isinstance(result, Close)), None)
    if upstream_close is None:
        return None
    if upstream_close.code == 1000 and upstream_close.reason == "":
        return None
    if upstream_close.code not in SENDABLE_CLOSE_CODES and not 3000 <= upstream_close.code < 5000:
        return None
    return upstream_close


async def websocket_passthrough_request(
    websocket: WebSocket,
    target: str,
    custom_headers: dict,
    user_api_key_dict: UserAPIKeyAuth,
    forward_headers: bool | None = False,
    endpoint: str | None = None,
    cost_per_request: float | None = None,
    accept_websocket: bool = True,
    setup_model_rewriter: Callable[[str], str] | None = None,
):
    """
    WebSocket passthrough request handler.

    Args:
        websocket: The incoming WebSocket connection
        target: The target WebSocket URL
        custom_headers: Custom headers to include in the connection
        user_api_key_dict: The user API key dictionary
        forward_headers: Whether to forward incoming headers
        endpoint: The endpoint path (for logging purposes)
        cost_per_request: Optional field - cost per request to the target endpoint
        setup_model_rewriter: Optional rewrite of the setup frame's model before it reaches the upstream
    """
    from litellm.litellm_core_utils.litellm_logging import Logging
    from litellm.proxy.proxy_server import proxy_logging_obj
    from litellm.types.passthrough_endpoints.pass_through_endpoints import (
        PassthroughStandardLoggingPayload,
    )

    # Initialize tracking variables
    start_time: Final = datetime.now()
    websocket_messages: Final[list[dict[str, object]]] = []
    litellm_call_id: Final = str(uuid.uuid4())

    verbose_proxy_logger.info("WebSocket passthrough (%s): Starting WebSocket connection to %s", endpoint, target)

    # Only accept the WebSocket if requested (for generic usage)
    if accept_websocket:
        await websocket.accept()
        verbose_proxy_logger.debug("WebSocket passthrough (%s): WebSocket connection accepted", endpoint)

    # Prepare headers for the upstream connection
    upstream_headers: Final = custom_headers.copy()

    if forward_headers:
        # Forward relevant headers from the incoming request
        incoming_headers: Final = dict(websocket.headers)
        for header_name, header_value in incoming_headers.items():
            # Only forward certain headers to avoid conflicts
            if header_name.lower() in [
                "authorization",
                "x-api-key",
                "x-goog-user-project",
            ]:
                upstream_headers[header_name] = header_value

    # Initialize logging object similar to HTTP passthrough
    logging_obj: Final = Logging(
        model="unknown",
        messages=[{"role": "user", "content": "WebSocket connection"}],
        stream=True,  # WebSockets are inherently streaming
        call_type="pass_through_endpoint",
        start_time=start_time,
        litellm_call_id=litellm_call_id,
        function_id="websocket_passthrough",
    )

    # Create passthrough logging payload
    passthrough_logging_payload: Final = PassthroughStandardLoggingPayload(
        url=target,
        request_body={},  # WebSocket doesn't have a traditional request body
        request_method="WEBSOCKET",
        cost_per_request=cost_per_request,
    )

    # Create a dummy request object for WebSocket connections to maintain compatibility
    # with the existing _init_kwargs_for_pass_through_endpoint function
    class DummyRequest:
        def __init__(self, url: str, method: str = "WEBSOCKET", headers: dict | None = None):
            self.url = url
            self.method = method
            self.headers = headers or {}

        def __str__(self):
            return f"DummyRequest(url={self.url}, method={self.method})"

    dummy_request: Final = DummyRequest(
        url=target,
        method="WEBSOCKET",
        headers=dict(websocket.headers) if hasattr(websocket, "headers") else {},
    )

    # Initialize kwargs for logging using the same pattern as HTTP passthrough
    kwargs: Final = HttpPassThroughEndpointHelpers._init_kwargs_for_pass_through_endpoint(
        user_api_key_dict=user_api_key_dict,
        _parsed_body={},  # WebSocket doesn't have a traditional request body
        passthrough_logging_payload=passthrough_logging_payload,
        litellm_call_id=litellm_call_id,
        request=dummy_request,
        logging_obj=logging_obj,
    )

    # Update logging environment variables
    logging_obj.update_environment_variables(
        model="unknown",
        user="unknown",
        optional_params={},
        litellm_params=dict(kwargs.get("litellm_params", {})),
        call_type="pass_through_endpoint",
    )
    logging_obj.model_call_details["litellm_call_id"] = litellm_call_id

    # Pre-call logging
    logging_obj.pre_call(
        input=[{"role": "user", "content": "WebSocket connection"}],
        api_key="",
        additional_args={
            "complete_input_dict": {},
            "api_base": target,
            "headers": upstream_headers,
        },
    )

    ### CALL HOOKS ### - modify incoming data / reject request before calling the model
    websocket_data: dict[str, object] = {}
    websocket_data = await proxy_logging_obj.pre_call_hook(
        user_api_key_dict=user_api_key_dict,
        data=websocket_data,
        call_type="pass_through_endpoint",
    )

    try:
        verbose_proxy_logger.debug(
            "WebSocket passthrough (%s): Establishing upstream connection to %s", endpoint, target
        )
        async with connect(
            target,
            additional_headers=upstream_headers,
        ) as upstream_ws:
            verbose_proxy_logger.info(
                "WebSocket passthrough (%s): Upstream connection established successfully", endpoint
            )

            async def forward_client_to_upstream() -> None:
                """Forward messages from client to upstream WebSocket"""
                try:
                    while True:
                        message = await websocket.receive()
                        message_type = message.get("type")
                        if message_type == "websocket.disconnect":
                            await upstream_ws.close()
                            break

                        text_data: str | None = message.get("text")
                        bytes_data: bytes | None = message.get("bytes")

                        if text_data is not None:
                            # Try to extract model from client setup message for Vertex AI Live
                            if endpoint and "/vertex_ai/live" in endpoint:
                                verbose_proxy_logger.debug(
                                    "WebSocket passthrough (%s): Processing client message for model extraction",
                                    endpoint,
                                )
                                try:
                                    client_message = json.loads(text_data)
                                    if isinstance(client_message, dict) and "setup" in client_message:
                                        setup_data = client_message["setup"]
                                        verbose_proxy_logger.debug(
                                            "WebSocket passthrough (%s): Found setup data in client message: %s",
                                            endpoint,
                                            setup_data,
                                        )
                                        if isinstance(setup_data, dict) and "model" in setup_data:
                                            extracted_model = _extract_model_from_vertex_ai_setup(setup_data)
                                            if extracted_model:
                                                kwargs["model"] = extracted_model
                                                kwargs["custom_llm_provider"] = "vertex_ai-language-models"
                                                # Update logging object with correct model
                                                logging_obj.model = extracted_model
                                                logging_obj.model_call_details["model"] = extracted_model
                                                logging_obj.model_call_details["custom_llm_provider"] = "vertex_ai"
                                                verbose_proxy_logger.info(
                                                    "WebSocket passthrough (%s): Successfully extracted model '%s' and set provider to 'vertex_ai' from client setup message",
                                                    endpoint,
                                                    extracted_model,
                                                )
                                            else:
                                                verbose_proxy_logger.warning(
                                                    "WebSocket passthrough (%s): Failed to extract model from client setup data: %s",
                                                    endpoint,
                                                    setup_data,
                                                )
                                        else:
                                            verbose_proxy_logger.debug(
                                                "WebSocket passthrough (%s): Setup data does not contain model field: %s",
                                                endpoint,
                                                setup_data,
                                            )
                                    else:
                                        verbose_proxy_logger.debug(
                                            "WebSocket passthrough (%s): Client message does not contain setup data",
                                            endpoint,
                                        )
                                except (json.JSONDecodeError, KeyError, TypeError) as e:
                                    verbose_proxy_logger.debug(
                                        "WebSocket passthrough (%s): Client message is not a valid setup message: %s",
                                        endpoint,
                                        e,
                                    )
                                    # Not a JSON message or doesn't contain setup data

                            await upstream_ws.send(_rewrite_vertex_live_setup_model(text_data, setup_model_rewriter))
                        elif bytes_data is not None:
                            await upstream_ws.send(bytes_data)
                except asyncio.CancelledError:
                    raise
                except Exception:
                    verbose_proxy_logger.exception(
                        "WebSocket passthrough (%s): error forwarding client message", endpoint
                    )
                    await upstream_ws.close()

            async def forward_upstream_to_client() -> Close | None:
                """Forward messages from upstream to client WebSocket, returning the upstream's close frame"""
                try:
                    # Wait for the first response from upstream
                    raw_response = await upstream_ws.recv(decode=False)
                    # Ensure raw_response is bytes before decoding
                    if isinstance(raw_response, str):
                        raw_response = raw_response.encode("utf-8")
                    setup_response: Final[Mapping[str, object]] = json.loads(raw_response.decode("utf-8"))
                    verbose_proxy_logger.debug("Setup response: %s", setup_response)

                    # Extract model and provider from setup response for Vertex AI Live
                    if endpoint and "/vertex_ai/live" in endpoint:
                        verbose_proxy_logger.debug(
                            "WebSocket passthrough (%s): Processing server setup response for model extraction",
                            endpoint,
                        )
                        extracted_model: Final = _extract_model_from_vertex_ai_setup(setup_response)
                        if extracted_model:
                            kwargs["model"] = extracted_model
                            kwargs["custom_llm_provider"] = "vertex_ai_language_models"
                            # Update logging object with correct model
                            logging_obj.model = extracted_model
                            logging_obj.model_call_details["model"] = extracted_model
                            logging_obj.model_call_details["custom_llm_provider"] = "vertex_ai_language_models"
                            verbose_proxy_logger.debug(
                                "WebSocket passthrough (%s): Successfully extracted model '%s' and set provider to 'vertex_ai' from server setup response",
                                endpoint,
                                extracted_model,
                            )
                        else:
                            verbose_proxy_logger.warning(
                                "WebSocket passthrough (%s): Failed to extract model from server setup response: %s",
                                endpoint,
                                setup_response,
                            )
                    else:
                        verbose_proxy_logger.debug(
                            "WebSocket passthrough (%s): Not a Vertex AI Live endpoint, skipping model extraction",
                            endpoint,
                        )

                    # Send the setup response to the client
                    await websocket.send_text(json.dumps(setup_response))

                    # Now continuously forward messages from upstream to client
                    async for upstream_message in upstream_ws:
                        if isinstance(upstream_message, bytes):
                            await websocket.send_bytes(upstream_message)
                            # Parse and collect for cost tracking
                            try:
                                message_data: dict[str, object] = json.loads(upstream_message.decode())
                                websocket_messages.append(message_data)
                            except (json.JSONDecodeError, UnicodeDecodeError):
                                pass
                        else:
                            await websocket.send_text(upstream_message)
                            # Parse and collect for cost tracking
                            try:
                                message_data = json.loads(upstream_message)
                                websocket_messages.append(message_data)
                            except json.JSONDecodeError:
                                pass

                except (ConnectionClosedOK, ConnectionClosedError) as e:
                    verbose_proxy_logger.debug("Upstream WebSocket connection closed: %s", e)
                    return e.rcvd
                except asyncio.CancelledError:
                    verbose_proxy_logger.debug("asyncio.CancelledError in forward_upstream_to_client")
                    raise
                except Exception as e:
                    verbose_proxy_logger.debug("Exception in forward_upstream_to_client: %s", e)
                    verbose_proxy_logger.exception(
                        "WebSocket passthrough (%s): error forwarding upstream message", endpoint
                    )
                    raise

            # Create tasks for bidirectional message forwarding
            tasks: Final = [
                asyncio.create_task(forward_client_to_upstream()),
                asyncio.create_task(forward_upstream_to_client()),
            ]

            done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)

            # Cancel remaining tasks
            for task in pending:
                task.cancel()
                try:
                    await task
                except asyncio.CancelledError:
                    pass

            # Check for exceptions in completed tasks
            for task in done:
                exception = task.exception()
                if exception is not None:
                    raise exception

            upstream_close: Final = _upstream_close_to_relay(task.result() for task in done)
            if upstream_close is not None and _client_socket_is_open(websocket):
                await websocket.close(
                    code=upstream_close.code,
                    reason=_truncated_close_reason(upstream_close.reason),
                )

            end_time: Final = datetime.now()

            # Update passthrough logging payload with response data
            passthrough_logging_payload["response_body"] = websocket_messages
            passthrough_logging_payload["end_time"] = end_time

            # Remove logging_obj from kwargs to avoid duplicate keyword argument
            success_kwargs: Final = kwargs.copy()
            success_kwargs.pop("logging_obj", None)

            # # Add user authentication context for database logging
            # if user_api_key_dict:
            #     success_kwargs.setdefault('litellm_params', {})
            #     success_kwargs['litellm_params'].update({
            #         'proxy_server_request': {
            #             'body': {
            #                 'user': user_api_key_dict.user_id,
            #                 'team_id': user_api_key_dict.team_id,
            #                 'end_user_id': user_api_key_dict.end_user_id,
            #             }
            #         }
            #     })
            #     # Also add the user_api_key for direct access
            #     success_kwargs['user_api_key'] = user_api_key_dict.api_key

            # Create a dummy httpx.Response for WebSocket connections
            class MockWebSocketResponse:
                def __init__(self, target_url: str):
                    self.status_code = 200
                    self.text = "WebSocket connection successful"
                    self.headers: dict[str, str] = {}
                    self.request = MockWebSocketRequest(target_url)

            class MockWebSocketRequest:
                def __init__(self, target_url: str):
                    self.method = "WEBSOCKET"
                    self.url = target_url

            mock_response: Final = MockWebSocketResponse(target)

            # Use the same success handler as HTTP passthrough endpoints
            GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue(
                async_coroutine=pass_through_endpoint_logging.pass_through_async_success_handler(
                    httpx_response=mock_response,
                    response_body=websocket_messages,
                    url_route=endpoint or "",
                    result="websocket_connection_successful",
                    start_time=start_time,
                    end_time=end_time,
                    logging_obj=logging_obj,
                    cache_hit=False,
                    request_body={},
                    **success_kwargs,
                )
            )

            # Call the proxy logging success hook
            if proxy_logging_obj:
                await proxy_logging_obj.post_call_success_hook(
                    data={},
                    user_api_key_dict=user_api_key_dict,
                    response={"status": "websocket_connection_successful"},
                )

    except InvalidStatus as exc:
        verbose_proxy_logger.exception("WebSocket passthrough (%s): upstream rejected WebSocket connection", endpoint)

        # Prepare request payload for logging
        request_payload = {}
        if kwargs:
            for key, value in kwargs.items():
                request_payload[key] = value
        if logging_obj is not None:
            request_payload["litellm_logging_obj"] = logging_obj

        # Log the connection failure using the same pattern as HTTP
        await proxy_logging_obj.post_call_failure_hook(
            user_api_key_dict=user_api_key_dict,
            original_exception=exc,
            request_data=request_payload,
            traceback_str=traceback.format_exc(
                limit=MAXIMUM_TRACEBACK_LINES_TO_LOG,
            ),
        )

        if _client_socket_is_open(websocket):
            await websocket.close(
                code=getattr(exc, "status_code", 1011),
                reason="Upstream connection rejected",
            )
    except Exception as e:
        verbose_proxy_logger.exception(
            "WebSocket passthrough (%s): unexpected error while proxying WebSocket", endpoint
        )

        # Prepare request payload for logging
        request_payload = {}
        if kwargs:
            for key, value in kwargs.items():
                request_payload[key] = value
        if logging_obj is not None:
            request_payload["litellm_logging_obj"] = logging_obj

        # Log the unexpected error using the same pattern as HTTP
        await proxy_logging_obj.post_call_failure_hook(
            user_api_key_dict=user_api_key_dict,
            original_exception=e,
            request_data=request_payload,
            traceback_str=traceback.format_exc(
                limit=MAXIMUM_TRACEBACK_LINES_TO_LOG,
            ),
        )

        if _client_socket_is_open(websocket):
            await websocket.close(code=1011, reason="WebSocket passthrough error")
    finally:
        if _client_socket_is_open(websocket):
            await websocket.close()


def _is_streaming_response(response: httpx.Response) -> bool:
    _content_type: Final = response.headers.get("content-type")
    if _content_type is not None and "text/event-stream" in _content_type:
        return True
    return False


def _own_streamed_managed_ids(
    stream: AsyncGenerator[bytes, None],
    managed_id_provider: str | None,
    request: Request,
    user_api_key_dict: UserAPIKeyAuth,
) -> AsyncGenerator[bytes, None]:
    from litellm.proxy.proxy_server import general_settings, prisma_client, proxy_logging_obj

    if (
        managed_id_provider is None
        or not general_settings.get("passthrough_managed_object_ids", False)
        or prisma_client is None
        or proxy_logging_obj.get_proxy_hook("managed_files") is None
    ):
        return stream
    from litellm.proxy.auth.auth_utils import get_request_route
    from litellm.proxy.pass_through_endpoints.managed_id_rewriter import (
        rewrite_streamed_response_ids,
    )

    return rewrite_streamed_response_ids(
        stream=stream,
        provider=managed_id_provider,
        method=request.method,
        route=get_request_route(request),
        user_api_key_dict=user_api_key_dict,
        prisma_client=prisma_client,
    )


def _should_buffer_passthrough_response(response: httpx.Response) -> bool:
    """
    Decide from the response headers whether the body must be read into memory.

    JSON bodies (and upstream errors) stay buffered: spend logging, guardrails and
    managed-id rewriting inspect them, and they are small in practice. Everything
    else (jsonl batch results, octet-stream files, ...) is relayed to the client
    chunk by chunk so a large body is never resident in full (LIT-4009). A missing
    content-type is buffered because the body cannot be classified.
    """
    if response.status_code >= 400:
        return True
    content_type_header: Final[str] = response.headers.get("content-type", "")
    media_type: Final = content_type_header.split(";")[0].strip().lower()
    return media_type in ("", "application/json") or media_type.endswith("+json")


async def _relay_passthrough_response_bytes(
    response: httpx.Response,
    request_body: dict,
    url_route: str,
    start_time: datetime,
    logging_obj: LiteLLMLoggingObj,
    custom_llm_provider: str | None,
    success_handler_kwargs: dict,
) -> AsyncGenerator[bytes, None]:
    """
    Yield upstream bytes to the client without accumulating them, then fire the
    passthrough success handler with response_body=None (uninspected body). The
    finally block also runs on client disconnect (GeneratorExit) so partial
    downloads still produce a spend-log row, mirroring chunk_processor; a
    disconnect additionally logs a warning with the number of bytes relayed so
    partial deliveries are distinguishable from complete ones in proxy logs.
    """
    bytes_relayed = 0
    upstream_fully_relayed = False
    try:
        async for chunk in response.aiter_bytes():
            bytes_relayed += len(chunk)
            yield chunk
        upstream_fully_relayed = True
    finally:
        if not upstream_fully_relayed:
            verbose_proxy_logger.warning(
                "Passthrough stream for %s ended before upstream body was fully relayed; %s bytes were sent to the client",
                url_route,
                bytes_relayed,
            )
        await response.aclose()
        GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue(
            async_coroutine=pass_through_endpoint_logging.pass_through_async_success_handler(
                httpx_response=response,
                response_body=None,
                url_route=url_route,
                result="",
                start_time=start_time,
                end_time=datetime.now(),
                logging_obj=logging_obj,
                cache_hit=False,
                request_body=request_body,
                custom_llm_provider=custom_llm_provider,
                **success_handler_kwargs,
            )
        )


def _extract_model_from_vertex_ai_setup(setup_response: Mapping[str, object]) -> str | None:
    """
    Extract the model name from Vertex AI Live setup response.

    The setup response can contain a model field in two formats:
    1. Direct: {"model": "projects/.../models/gemini-2.0-flash-live-preview-04-09"}
    2. Nested: {"setup": {"model": "projects/.../models/gemini-2.0-flash-live-preview-04-09"}}

    We extract just the model name: "gemini-2.0-flash-live-preview-04-09"
    """
    try:
        # Handle both direct model field and nested setup.model field
        model_path = None
        if isinstance(setup_response, dict):
            if "model" in setup_response:
                model_path = setup_response["model"]
            elif (
                "setup" in setup_response
                and isinstance(setup_response["setup"], dict)
                and "model" in setup_response["setup"]
            ):
                model_path = setup_response["setup"]["model"]

        if isinstance(model_path, str) and "/models/" in model_path:
            # Extract the model name after the last "/models/"
            model_name: Final = model_path.split("/models/")[-1]
            return model_name
    except Exception as e:
        verbose_proxy_logger.debug("Error extracting model from setup response: %s", e)
    return None


class SafeRouteAdder:
    """
    Wrapper class for adding routes to FastAPI app.
    Only adds routes if they don't already exist on the app.
    """

    @staticmethod
    def _is_path_registered(app: FastAPI, path: str, methods: list[str]) -> bool:
        """
        Check if a path with any of the specified methods is already registered on the app.

        Args:
            app: The FastAPI application instance
            path: The path to check (e.g., "/v1/chat/completions")
            methods: List of HTTP methods to check (e.g., ["GET", "POST"])

        Returns:
            True if the path is already registered with any of the methods, False otherwise
        """
        for route in app.routes:
            # Use getattr to safely access route attributes
            route_path = getattr(route, "path", None)
            route_methods = getattr(route, "methods", None)

            if route_path == path and route_methods is not None:
                # Check if any of the methods overlap
                if any(method in route_methods for method in methods):
                    return True
        return False

    @staticmethod
    def add_api_route_if_not_exists(
        app: FastAPI,
        path: str,
        endpoint: Callable[..., object],
        methods: list[str],
        dependencies: list | None = None,
    ) -> bool:
        """
        Add an API route to the app only if it doesn't already exist.

        Args:
            app: The FastAPI application instance
            path: The path for the route
            endpoint: The endpoint function/callable
            methods: List of HTTP methods
            dependencies: Optional list of dependencies

        Returns:
            True if route was added, False if it already existed
        """
        if SafeRouteAdder._is_path_registered(app=app, path=path, methods=methods):
            verbose_proxy_logger.debug(
                "Skipping route registration - path %s with methods %s already registered on app",
                path,
                methods,
            )
            return False

        app.add_api_route(
            path=path,
            endpoint=endpoint,
            methods=methods,
            dependencies=dependencies,
        )
        verbose_proxy_logger.debug(
            "Successfully added route: %s with methods %s",
            path,
            methods,
        )
        return True


class InitPassThroughEndpointHelpers:
    @staticmethod
    def add_exact_path_route(
        app: FastAPI,
        path: str,
        target: str,
        custom_headers: dict | None,
        forward_headers: bool | None,
        merge_query_params: bool | None,
        dependencies: list | None,
        cost_per_request: float | None,
        endpoint_id: str,
        guardrails: dict | None = None,
        methods: list[str] | None = None,
        default_query_params: dict | None = None,
        config_file_path: str | None = None,
        auth: bool = False,
        timeout: float | None = None,
    ):
        """Add exact path route for pass-through endpoint"""
        # Default to all methods if none specified (backward compatibility)
        if methods is None or len(methods) == 0:
            methods = ["GET", "POST", "PUT", "DELETE", "PATCH"]

        # Create route key that includes methods for uniqueness
        methods_str: Final = ",".join(sorted(methods))
        route_key: Final = f"{endpoint_id}:exact:{path}:{methods_str}"

        # Check if this exact route is already registered
        if route_key in _registered_pass_through_routes:
            verbose_proxy_logger.debug(
                "Updating duplicate exact pass through endpoint: %s with methods %s (already registered)",
                path,
                methods,
            )

        verbose_proxy_logger.debug(
            "adding exact pass through endpoint: %s, methods: %s, dependencies: %s",
            path,
            methods,
            dependencies,
        )

        # Use SafeRouteAdder to only add route if it doesn't exist on the app
        SafeRouteAdder.add_api_route_if_not_exists(
            app=app,
            path=path,
            endpoint=create_pass_through_route(
                path,
                target,
                custom_headers,
                forward_headers,
                merge_query_params,
                dependencies,
                cost_per_request=cost_per_request,
                default_query_params=default_query_params,
                guardrails=guardrails,
                config_file_path=config_file_path,
                timeout=timeout,
            ),
            methods=methods,
            dependencies=dependencies,
        )

        # Always register/update the route metadata (headers, target) even if FastAPI route exists
        _registered_pass_through_routes[route_key] = {
            "endpoint_id": endpoint_id,
            "path": path,
            "type": "exact",
            "methods": methods,
            "auth": auth,
            "passthrough_params": {
                "target": target,
                "custom_headers": custom_headers,
                "forward_headers": forward_headers,
                "merge_query_params": merge_query_params,
                "default_query_params": default_query_params,
                "dependencies": dependencies,
                "cost_per_request": cost_per_request,
                "guardrails": guardrails,
                "timeout": timeout,
            },
        }

    @staticmethod
    def add_subpath_route(
        app: FastAPI,
        path: str,
        target: str,
        custom_headers: dict | None,
        forward_headers: bool | None,
        merge_query_params: bool | None,
        dependencies: list | None,
        cost_per_request: float | None,
        endpoint_id: str,
        guardrails: dict | None = None,
        methods: list[str] | None = None,
        default_query_params: dict | None = None,
        config_file_path: str | None = None,
        auth: bool = False,
        timeout: float | None = None,
    ):
        """Add wildcard route for sub-paths"""
        # Default to all methods if none specified (backward compatibility)
        if methods is None or len(methods) == 0:
            methods = ["GET", "POST", "PUT", "DELETE", "PATCH"]

        wildcard_path: Final = f"{path}/{{subpath:path}}"
        methods_str: Final = ",".join(sorted(methods))
        route_key: Final = f"{endpoint_id}:subpath:{path}:{methods_str}"

        # Check if this subpath route is already registered
        if route_key in _registered_pass_through_routes:
            verbose_proxy_logger.debug(
                "Updating duplicate wildcard pass through endpoint: %s with methods %s (already registered)",
                wildcard_path,
                methods,
            )

        verbose_proxy_logger.debug(
            "adding wildcard pass through endpoint: %s, methods: %s, dependencies: %s",
            wildcard_path,
            methods,
            dependencies,
        )

        # Use SafeRouteAdder to only add route if it doesn't exist on the app
        SafeRouteAdder.add_api_route_if_not_exists(
            app=app,
            path=wildcard_path,
            endpoint=create_pass_through_route(
                path,
                target,
                custom_headers,
                forward_headers,
                merge_query_params,
                dependencies,
                include_subpath=True,
                cost_per_request=cost_per_request,
                default_query_params=default_query_params,
                guardrails=guardrails,
                config_file_path=config_file_path,
                timeout=timeout,
            ),
            methods=methods,
            dependencies=dependencies,
        )

        # Register the route to prevent duplicates only if it was added
        _registered_pass_through_routes[route_key] = {
            "endpoint_id": endpoint_id,
            "path": path,
            "type": "subpath",
            "methods": methods,
            "auth": auth,
            "passthrough_params": {
                "target": target,
                "custom_headers": custom_headers,
                "forward_headers": forward_headers,
                "merge_query_params": merge_query_params,
                "default_query_params": default_query_params,
                "dependencies": dependencies,
                "cost_per_request": cost_per_request,
                "guardrails": guardrails,
                "timeout": timeout,
            },
        }

    @staticmethod
    def remove_endpoint_routes(endpoint_id: str):
        """Remove all routes for a specific endpoint ID from the registry
        and clean up corresponding entries from LiteLLMRoutes.openai_routes."""
        keys_to_remove: Final = [
            key for key, value in _registered_pass_through_routes.items() if value["endpoint_id"] == endpoint_id
        ]
        for key in keys_to_remove:
            route_info = _registered_pass_through_routes[key]
            path = route_info.get("path")
            if isinstance(path, str):
                openai_routes = LiteLLMRoutes.openai_routes.value
                if path in openai_routes:
                    openai_routes.remove(path)
                if route_info.get("type") == "subpath":
                    wildcard_path = path.rstrip("/") + "/*"
                    if wildcard_path in openai_routes:
                        openai_routes.remove(wildcard_path)
            del _registered_pass_through_routes[key]
            verbose_proxy_logger.debug("Removed pass-through route from registry: %s", key)

    @staticmethod
    def clear_all_pass_through_routes():
        """Clear all pass-through routes from the registry"""
        _registered_pass_through_routes.clear()

    @staticmethod
    def get_all_registered_pass_through_routes() -> list[str]:
        """Get all registered pass-through endpoints from the registry"""
        return list(_registered_pass_through_routes.keys())

    @staticmethod
    def _route_for_registry_lookup(route: str) -> str:
        """
        Normalize an incoming route to the bare path stored in the registry.

        Registry keys store root-stripped paths. Callers should pass routes from
        ``get_request_route()`` (already stripped); prefixed ``request.url.path``
        values are stripped via ``normalize_route_for_root_path``.
        """
        normalized_route: Final = normalize_route_for_root_path(route)
        return normalized_route if normalized_route is not None else route

    @staticmethod
    def is_registered_pass_through_route(route: str) -> bool:
        """
        Check if route is a registered pass-through endpoint from DB

        Uses the in-memory registry to avoid additional DB queries
        Optimized for minimal latency

        Args:
            route: The route to check

        Returns:
            bool: True if route is a registered pass-through endpoint, False otherwise
        """
        ## CHECK IF MAPPED PASS THROUGH ENDPOINT
        normalized_route: Final = normalize_route_for_root_path(route)
        if normalized_route is not None:
            for mapped_route in LiteLLMRoutes.mapped_pass_through_routes.value:
                if normalized_route.startswith(mapped_route):
                    return True

        comparison_route: Final = InitPassThroughEndpointHelpers._route_for_registry_lookup(route)

        # Fast path: check if any registered route key contains this path
        # Keys are in format: "{endpoint_id}:exact:{path}:{methods}" or "{endpoint_id}:subpath:{path}:{methods}"
        # For backward compatibility, also support old format: "{endpoint_id}:exact:{path}" or "{endpoint_id}:subpath:{path}"
        # Extract unique paths from keys for quick checking
        for key in _registered_pass_through_routes:
            parts = key.split(":", 3)  # Split into [endpoint_id, type, path, methods?]
            if len(parts) >= 3:
                route_type = parts[1]
                registered_path = parts[2]
                if route_type == "exact" and comparison_route == registered_path:
                    return True
                elif route_type == "subpath":
                    if comparison_route == registered_path or comparison_route.startswith(registered_path + "/"):
                        return True

        return False

    @staticmethod
    def get_registered_pass_through_route(route: str, method: str | None = None) -> dict[str, Any] | None:
        """Get passthrough params for a given route and optionally filter by HTTP method"""
        comparison_route: Final = InitPassThroughEndpointHelpers._route_for_registry_lookup(route)
        for key in _registered_pass_through_routes:
            parts = key.split(":", 3)  # Split into [endpoint_id, type, path, methods?]
            if len(parts) >= 3:
                route_type = parts[1]
                registered_path = parts[2]

                # Get the methods for this route. Prefer the registered metadata,
                # but keep supporting test fixtures / older registry entries that
                # only encoded methods in the route key.
                methods_entry = _registered_pass_through_routes[key].get("methods", [])
                route_methods: list[str] = methods_entry if isinstance(methods_entry, list) else []
                if not route_methods and len(parts) == 4:
                    route_methods = parts[3].split(",")

                # Check if path matches
                path_matches = False
                if route_type == "exact" and comparison_route == registered_path:
                    path_matches = True
                elif route_type == "subpath":
                    if comparison_route == registered_path or comparison_route.startswith(registered_path + "/"):
                        path_matches = True

                # If path matches and method filter is provided, check if method is allowed
                if path_matches:
                    if method is None or not route_methods or method in route_methods:
                        return _registered_pass_through_routes[key]

        return None


def _get_combined_pass_through_endpoints(
    pass_through_endpoints: list[dict] | list[PassThroughGenericEndpoint],
    config_pass_through_endpoints: list[dict],
):
    """Get combined pass-through endpoints from db + config"""
    return pass_through_endpoints + config_pass_through_endpoints


async def _register_pass_through_endpoint(
    endpoint: dict[str, object] | PassThroughGenericEndpoint,
    app: FastAPI,
    premium_user: bool,
    visited_endpoints: set[str],
    config_file_path: str | None = None,
) -> 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: Final = cast(str, endpoint_data["id"])

    target: Final[str | None] = endpoint_data.get("target")
    path: Final[str | None] = endpoint_data.get("path")
    if path is None:
        raise ValueError("Path is required for pass-through endpoint")

    custom_headers: Final = await set_env_variables_in_header(custom_headers=endpoint_data.get("headers"))
    forward_headers: Final = endpoint_data.get("forward_headers")
    merge_query_params: Final = endpoint_data.get("merge_query_params")
    default_query_params: Final = endpoint_data.get("default_query_params")
    auth: Final[bool | str | None] = endpoint_data.get("auth")
    dependencies = None
    auth_enforced: Final = auth is not None and str(auth).lower() == "true"

    if auth_enforced:
        # Authentication on a pass-through endpoint used to be enterprise-only.
        # That left OSS with no safe configuration: auth=True raised at startup
        # unless the operator had a license. The safe option must always be free,
        # and unauthenticated forwarding should require explicit opt-in.
        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: Final = endpoint_data.get("guardrails")
    methods: Final = endpoint_data.get("methods")
    cost_per_request: Final = endpoint_data.get("cost_per_request")
    timeout: Final = endpoint_data.get("timeout")

    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,
        config_file_path=config_file_path,
        auth=auth_enforced,
        timeout=timeout,
    )

    methods_for_key: Final = methods if methods else ["GET", "POST", "PUT", "DELETE", "PATCH"]
    methods_str: Final = ",".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: Final = 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,
            config_file_path=config_file_path,
            auth=auth_enforced,
            timeout=timeout,
        )
        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: list[dict] | list[PassThroughGenericEndpoint],
    config_file_path: str | None = None,
):
    """
    1. Create a global list of pass-through endpoints (db + config)
    2. Clear all existing pass-through endpoints from the FastAPI app routes
    3. Add new endpoints to the in-memory registry

    Initialize a list of pass-through endpoints by adding them to the FastAPI app routes

    Args:
        pass_through_endpoints: List of pass-through endpoints to initialize
        config_file_path: Path to the operator's config.yaml when this call
            originates from a YAML-load. Threaded through to
            ``create_pass_through_route`` so an operator using
            ``s3://``/``gcs://`` ``custom_handler`` in their config still
            loads. Callers from the DB-overlay / runtime API path must leave
            this ``None`` so the runtime gate in ``get_instance_fn`` fires.

    Returns:
        None
    """
    verbose_proxy_logger.debug("initializing pass through endpoints")
    from litellm.proxy.proxy_server import (
        app,
        config_passthrough_endpoints,
        premium_user,
    )

    ## get combined pass-through endpoints from db + config
    combined_pass_through_endpoints: list[dict | PassThroughGenericEndpoint]

    if config_passthrough_endpoints is not None:
        combined_pass_through_endpoints = _get_combined_pass_through_endpoints(
            pass_through_endpoints, config_passthrough_endpoints
        )
    else:
        combined_pass_through_endpoints = pass_through_endpoints

    ## clear all existing pass-through endpoints from the FastAPI app routes
    # InitPassThroughEndpointHelpers.clear_all_pass_through_routes()

    # get a list of all registered pass-through endpoints
    # mark the ones that are visited in the list
    # remove the ones that are not visited from the list
    registered_pass_through_endpoints: Final = InitPassThroughEndpointHelpers.get_all_registered_pass_through_routes()

    visited_endpoints: Final[set[str]] = set()

    for endpoint in combined_pass_through_endpoints:
        await _register_pass_through_endpoint(
            endpoint=endpoint,
            app=app,
            premium_user=premium_user,
            visited_endpoints=visited_endpoints,
            config_file_path=config_file_path,
        )

    # Drop stale registry entries by their exact route key. registered_pass_through_endpoints
    # holds route keys ("{id}:{type}:{path}:{methods}"), not endpoint ids, so remove_endpoint_routes
    # (which matches on endpoint_id) never matched and left the registry growing every reload cycle.
    # We pop the key directly and leave openai_routes alone: its append is path-deduped, and the path
    # is still owned by the live endpoint that was just re-registered under a new id this same cycle.
    for endpoint_key in registered_pass_through_endpoints:
        if endpoint_key not in visited_endpoints:
            _registered_pass_through_routes.pop(endpoint_key, None)


def _get_pass_through_endpoints_from_config() -> list[PassThroughGenericEndpoint]:
    """
    Get pass-through endpoints defined in the config file.
    These are read-only and cannot be edited via the UI.
    Malformed endpoints are logged and skipped; they do not crash the function.
    """
    from pydantic import ValidationError

    from litellm.proxy.proxy_server import config_passthrough_endpoints

    if config_passthrough_endpoints is None or len(config_passthrough_endpoints) == 0:
        return []

    returned_endpoints: Final[list[PassThroughGenericEndpoint]] = []
    for endpoint in config_passthrough_endpoints:
        try:
            if isinstance(endpoint, dict):
                endpoint_dict = dict(endpoint)
                endpoint_dict["is_from_config"] = True
                returned_endpoints.append(PassThroughGenericEndpoint.model_validate(endpoint_dict))
            elif isinstance(endpoint, PassThroughGenericEndpoint):
                # Create a copy with is_from_config=True
                endpoint_dict = endpoint.model_dump()
                endpoint_dict["is_from_config"] = True
                returned_endpoints.append(PassThroughGenericEndpoint.model_validate(endpoint_dict))
        except ValidationError as e:
            verbose_proxy_logger.warning(
                "Skipping malformed pass-through endpoint from config: %s",
                e,
                exc_info=False,
            )

    return returned_endpoints


async def _get_pass_through_endpoints_from_db(
    endpoint_id: str | None = None,
    user_api_key_dict: UserAPIKeyAuth | None = None,
) -> list[PassThroughGenericEndpoint]:
    from litellm.proxy._types import LitellmUserRoles
    from litellm.proxy.proxy_server import get_config_general_settings

    try:
        if user_api_key_dict is None:
            user_api_key_dict = UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN)
        response: Final[ConfigFieldInfo] = await get_config_general_settings(
            field_name="pass_through_endpoints", user_api_key_dict=user_api_key_dict
        )
    except Exception:
        return []

    pass_through_endpoint_data: Final[list | None] = response.field_value
    if pass_through_endpoint_data is None:
        return []

    returned_endpoints: Final[list[PassThroughGenericEndpoint]] = []
    if endpoint_id is None:
        # Return all endpoints from DB, mark as not from config
        for endpoint in pass_through_endpoint_data:
            if isinstance(endpoint, dict):
                endpoint_dict = dict(endpoint)
                endpoint_dict["is_from_config"] = False
                returned_endpoints.append(PassThroughGenericEndpoint.model_validate(endpoint_dict))
            elif isinstance(endpoint, PassThroughGenericEndpoint):
                endpoint_dict = endpoint.model_dump()
                endpoint_dict["is_from_config"] = False
                returned_endpoints.append(PassThroughGenericEndpoint.model_validate(endpoint_dict))
    else:
        # Find specific endpoint by ID
        found_endpoint: Final = _find_endpoint_by_id(pass_through_endpoint_data, endpoint_id)
        if found_endpoint is not None:
            endpoint_dict = (
                found_endpoint.model_dump()
                if isinstance(found_endpoint, PassThroughGenericEndpoint)
                else dict(found_endpoint)
            )
            endpoint_dict["is_from_config"] = False
            returned_endpoints.append(PassThroughGenericEndpoint.model_validate(endpoint_dict))

    return returned_endpoints


async def _filter_endpoints_by_team_allowed_routes(
    team_id: str,
    pass_through_endpoints: list[PassThroughGenericEndpoint],
    prisma_client,
) -> list[PassThroughGenericEndpoint]:
    """
    Filter pass-through endpoints based on team's allowed_passthrough_routes metadata.

    Args:
        team_id: The team ID to check permissions for
        pass_through_endpoints: List of endpoints to filter
        prisma_client: Database client

    Returns:
        Filtered list of endpoints based on team permissions

    Raises:
        HTTPException: If team is not found
    """
    # retrieve team from db
    team: Final = await TeamRepository(prisma_client).table.find_unique(
        where={"team_id": team_id},
    )
    if team is None:
        raise HTTPException(
            status_code=404,
            detail={"error": "Team not found"},
        )

    # retrieve team metadata
    team_metadata: Final = cast(  # cast-ok: prisma types the Json column as str; reads hand back the decoded value
        "Mapping[str, object] | None", team.metadata
    )
    if team_metadata is not None and team_metadata.get("allowed_passthrough_routes") is not None:
        ## FILTER pass_through_endpoints by allowed_passthrough_routes
        pass_through_endpoints = [
            endpoint
            for endpoint in pass_through_endpoints
            if endpoint.path
            in cast(  # cast-ok: guarded above; team metadata stores this key as a list of route paths
                "Sequence[str]", team_metadata.get("allowed_passthrough_routes")
            )
        ]

    return pass_through_endpoints


@router.get(
    "/config/pass_through_endpoint",
    dependencies=[Depends(user_api_key_auth)],
    response_model=PassThroughEndpointResponse,
)
@router.get(
    "/config/pass_through_endpoint/team/{team_id}",
    dependencies=[Depends(user_api_key_auth)],
    response_model=PassThroughEndpointResponse,
)
async def get_pass_through_endpoints(
    endpoint_id: str | None = None,
    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
    team_id: str | None = None,
):
    """
    GET configured pass through endpoint.

    If no endpoint_id given, return all configured endpoints.
    """  ## Get existing pass-through endpoint field value
    from litellm.proxy._types import CommonProxyErrors
    from litellm.proxy.proxy_server import prisma_client

    if prisma_client is None:
        raise HTTPException(
            status_code=500,
            detail={"error": CommonProxyErrors.db_not_connected_error.value},
        )

    # Get endpoints from DB (editable via UI)
    db_endpoints: Final = await _get_pass_through_endpoints_from_db(
        endpoint_id=endpoint_id, user_api_key_dict=user_api_key_dict
    )

    # Get endpoints from config file (read-only, not editable via UI)
    config_endpoints: Final = _get_pass_through_endpoints_from_config()

    # Merge: config endpoints not in DB + all DB endpoints (DB overrides config for same path)
    db_paths: Final = {ep.path for ep in db_endpoints}
    config_only_endpoints: Final = [ep for ep in config_endpoints if ep.path not in db_paths]
    if endpoint_id is not None:
        # When filtering by endpoint_id, only return if found in DB (config endpoints use generated IDs)
        pass_through_endpoints = db_endpoints
    else:
        pass_through_endpoints = config_only_endpoints + db_endpoints

    if team_id is not None:
        pass_through_endpoints = await _filter_endpoints_by_team_allowed_routes(
            team_id=team_id,
            pass_through_endpoints=pass_through_endpoints,
            prisma_client=prisma_client,
        )

    return PassThroughEndpointResponse(endpoints=pass_through_endpoints)


@router.post(
    "/config/pass_through_endpoint/{endpoint_id}",
    dependencies=[Depends(user_api_key_auth)],
)
async def update_pass_through_endpoints(
    endpoint_id: str,
    data: PassThroughGenericEndpoint,
    request: Request,
    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
    """
    Update a pass-through endpoint by ID.
    """
    from litellm.proxy.proxy_server import (
        get_config_general_settings,
        update_config_general_settings,
    )

    ## Get existing pass-through endpoint field value
    try:
        response: Final[ConfigFieldInfo] = await get_config_general_settings(
            field_name="pass_through_endpoints", user_api_key_dict=user_api_key_dict
        )
    except Exception:
        raise HTTPException(
            status_code=404,
            detail={"error": "No pass-through endpoints found"},
        )

    pass_through_endpoint_data: Final[list | None] = response.field_value
    if pass_through_endpoint_data is None:
        raise HTTPException(
            status_code=404,
            detail={"error": "No pass-through endpoints found"},
        )

    # Find the endpoint to update
    found_endpoint: Final = _find_endpoint_by_id(pass_through_endpoint_data, endpoint_id)

    if found_endpoint is None:
        raise HTTPException(
            status_code=404,
            detail={"error": f"Endpoint with ID '{endpoint_id}' not found"},
        )

    # Find the index for updating the list
    endpoint_index = None
    for idx, endpoint in enumerate(pass_through_endpoint_data):
        _endpoint = PassThroughGenericEndpoint(**endpoint) if isinstance(endpoint, dict) else endpoint
        if _endpoint.id == endpoint_id:
            endpoint_index = idx
            break

    if endpoint_index is None:
        raise HTTPException(
            status_code=404,
            detail={"error": f"Could not find index for endpoint with ID '{endpoint_id}'"},
        )

    # Only merge fields the caller explicitly sent so omitted fields keep their
    # stored value. Without exclude_unset, defaults like auth=True would overwrite
    # an existing auth=false entry on any unrelated edit.
    # Exclude is_from_config as it's a response-only field (computed at read time)
    update_data: Final = data.model_dump(exclude_unset=True, exclude_none=True, exclude={"is_from_config"})

    # Start with existing endpoint data
    endpoint_dict: Final = found_endpoint.model_dump()

    # Update with new data (only explicitly provided values)
    endpoint_dict.update(update_data)

    # Preserve existing ID if not provided in update and endpoint has ID
    if "id" not in update_data and found_endpoint.id is not None:
        endpoint_dict["id"] = found_endpoint.id

    # Remove is_from_config before saving - it's a response-only field (computed at read time)
    endpoint_dict.pop("is_from_config", None)

    # Create updated endpoint object
    updated_endpoint: Final = PassThroughGenericEndpoint.model_validate(endpoint_dict)

    # Update the list
    pass_through_endpoint_data[endpoint_index] = endpoint_dict

    # Remove old routes from registry before they get re-registered
    InitPassThroughEndpointHelpers.remove_endpoint_routes(endpoint_id)

    ## Update db
    updated_data: Final = ConfigFieldUpdate(
        field_name="pass_through_endpoints",
        field_value=pass_through_endpoint_data,
        config_type="general_settings",
    )

    await update_config_general_settings(data=updated_data, user_api_key_dict=user_api_key_dict)

    # Re-register the route with updated headers
    _custom_headers: dict | None = updated_endpoint.headers or {}
    _custom_headers = await set_env_variables_in_header(custom_headers=_custom_headers)

    route_app: Final[FastAPI] = request.app
    if updated_endpoint.include_subpath:
        InitPassThroughEndpointHelpers.add_subpath_route(
            app=route_app,
            path=updated_endpoint.path,
            target=updated_endpoint.target,
            custom_headers=_custom_headers,
            forward_headers=None,  # Defaults not available in model? assuming None logic handles it
            merge_query_params=None,
            dependencies=None,
            cost_per_request=updated_endpoint.cost_per_request,
            endpoint_id=updated_endpoint.id or endpoint_id or "",
            guardrails=getattr(updated_endpoint, "guardrails", None),
            methods=updated_endpoint.methods,
            default_query_params=updated_endpoint.default_query_params,
            auth=updated_endpoint.auth,
            timeout=updated_endpoint.timeout,
        )
    else:
        InitPassThroughEndpointHelpers.add_exact_path_route(
            app=route_app,
            path=updated_endpoint.path,
            target=updated_endpoint.target,
            custom_headers=_custom_headers,
            forward_headers=None,
            merge_query_params=None,
            dependencies=None,
            cost_per_request=updated_endpoint.cost_per_request,
            endpoint_id=updated_endpoint.id or endpoint_id or "",
            guardrails=getattr(updated_endpoint, "guardrails", None),
            methods=updated_endpoint.methods,
            default_query_params=updated_endpoint.default_query_params,
            auth=updated_endpoint.auth,
            timeout=updated_endpoint.timeout,
        )

    return PassThroughEndpointResponse(endpoints=[updated_endpoint] if updated_endpoint else [])


@router.post(
    "/config/pass_through_endpoint",
    dependencies=[Depends(user_api_key_auth)],
)
async def create_pass_through_endpoints(
    data: PassThroughGenericEndpoint,
    request: Request,
    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
    """
    Create new pass-through endpoint
    """
    from litellm._uuid import uuid
    from litellm.proxy.proxy_server import (
        get_config_general_settings,
        update_config_general_settings,
    )

    ## Get existing pass-through endpoint field value

    try:
        response: ConfigFieldInfo = await get_config_general_settings(
            field_name="pass_through_endpoints", user_api_key_dict=user_api_key_dict
        )
    except Exception:
        response = ConfigFieldInfo(field_name="pass_through_endpoints", field_value=None)

    ## Auto-generate ID if not provided
    # Exclude is_from_config as it's a response-only field (computed at read time)
    data_dict: Final = data.model_dump(exclude={"is_from_config"})
    if data_dict.get("id") is None:
        data_dict["id"] = str(uuid.uuid4())

    if response.field_value is None:
        response.field_value = [data_dict]
    elif isinstance(response.field_value, list):
        response.field_value.append(data_dict)

    ## Update db
    updated_data: Final = ConfigFieldUpdate(
        field_name="pass_through_endpoints",
        field_value=response.field_value,
        config_type="general_settings",
    )
    await update_config_general_settings(data=updated_data, user_api_key_dict=user_api_key_dict)

    # Return the created endpoint with the generated ID
    created_endpoint: Final = PassThroughGenericEndpoint.model_validate(data_dict)

    # Register the new route
    _custom_headers: dict | None = created_endpoint.headers or {}
    _custom_headers = await set_env_variables_in_header(custom_headers=_custom_headers)

    route_app: Final[FastAPI] = request.app
    if created_endpoint.include_subpath:
        InitPassThroughEndpointHelpers.add_subpath_route(
            app=route_app,
            path=created_endpoint.path,
            target=created_endpoint.target,
            custom_headers=_custom_headers,
            forward_headers=None,
            merge_query_params=None,
            dependencies=None,
            cost_per_request=created_endpoint.cost_per_request,
            endpoint_id=created_endpoint.id or "",
            guardrails=getattr(created_endpoint, "guardrails", None),
            methods=created_endpoint.methods,
            default_query_params=created_endpoint.default_query_params,
            auth=created_endpoint.auth,
            timeout=created_endpoint.timeout,
        )
    else:
        InitPassThroughEndpointHelpers.add_exact_path_route(
            app=route_app,
            path=created_endpoint.path,
            target=created_endpoint.target,
            custom_headers=_custom_headers,
            forward_headers=None,
            merge_query_params=None,
            dependencies=None,
            cost_per_request=created_endpoint.cost_per_request,
            endpoint_id=created_endpoint.id or "",
            guardrails=getattr(created_endpoint, "guardrails", None),
            methods=created_endpoint.methods,
            default_query_params=created_endpoint.default_query_params,
            auth=created_endpoint.auth,
            timeout=created_endpoint.timeout,
        )

    return PassThroughEndpointResponse(endpoints=[created_endpoint])


@router.delete(
    "/config/pass_through_endpoint",
    dependencies=[Depends(user_api_key_auth)],
    response_model=PassThroughEndpointResponse,
)
async def delete_pass_through_endpoints(
    endpoint_id: str,
    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
    """
    Delete a pass-through endpoint by ID.

    Returns - the deleted endpoint
    """
    from litellm.proxy.proxy_server import (
        get_config_general_settings,
        update_config_general_settings,
    )

    ## Get existing pass-through endpoint field value

    try:
        response: ConfigFieldInfo = await get_config_general_settings(
            field_name="pass_through_endpoints", user_api_key_dict=user_api_key_dict
        )
    except Exception:
        response = ConfigFieldInfo(field_name="pass_through_endpoints", field_value=None)

    ## Update field by removing endpoint
    pass_through_endpoint_data: Final[list | None] = response.field_value
    if response.field_value is None or pass_through_endpoint_data is None:
        raise HTTPException(
            status_code=400,
            detail={"error": "There are no pass-through endpoints setup."},
        )

    # Find the endpoint to delete
    found_endpoint: Final = _find_endpoint_by_id(pass_through_endpoint_data, endpoint_id)

    if found_endpoint is None:
        raise HTTPException(
            status_code=400,
            detail={"error": f"Endpoint with ID '{endpoint_id}' was not found in pass-through endpoint list."},
        )

    # Find the index for deleting from the list
    endpoint_index = None
    for idx, endpoint in enumerate(pass_through_endpoint_data):
        _endpoint = PassThroughGenericEndpoint(**endpoint) if isinstance(endpoint, dict) else endpoint
        if _endpoint.id == endpoint_id:
            endpoint_index = idx
            break

    if endpoint_index is None:
        raise HTTPException(
            status_code=400,
            detail={"error": f"Could not find index for endpoint with ID '{endpoint_id}'"},
        )

    # Remove the endpoint
    pass_through_endpoint_data.pop(endpoint_index)
    response_obj: Final = found_endpoint

    # Remove routes from registry
    InitPassThroughEndpointHelpers.remove_endpoint_routes(endpoint_id)

    ## Update db
    updated_data: Final = ConfigFieldUpdate(
        field_name="pass_through_endpoints",
        field_value=pass_through_endpoint_data,
        config_type="general_settings",
    )
    await update_config_general_settings(data=updated_data, user_api_key_dict=user_api_key_dict)

    return PassThroughEndpointResponse(endpoints=[response_obj])


def _find_endpoint_by_id(
    endpoints_data: list,
    endpoint_id: str,
) -> PassThroughGenericEndpoint | None:
    """
    Find an endpoint by ID.

    Args:
        endpoints_data: List of endpoint data (dicts or PassThroughGenericEndpoint objects)
        endpoint_id: ID to search for

    Returns:
        Found endpoint or None if not found
    """
    for endpoint in endpoints_data:
        _endpoint: PassThroughGenericEndpoint | None = None
        if isinstance(endpoint, dict):
            _endpoint = PassThroughGenericEndpoint(**endpoint)
        elif isinstance(endpoint, PassThroughGenericEndpoint):
            _endpoint = endpoint

        # Only compare IDs to IDs
        if _endpoint is not None and _endpoint.id == endpoint_id:
            return _endpoint

    return None


async def initialize_pass_through_endpoints_in_db():
    """
    Gets all pass-through endpoints from db and initializes them in the proxy server.
    """
    pass_through_endpoints: Final = await _get_pass_through_endpoints_from_db()
    await initialize_pass_through_endpoints(pass_through_endpoints=pass_through_endpoints)