two version of R2R are hereHEADmaster

1 files changed, 901 insertions, 0 deletions

diff --git a/.venv/lib/python3.12/site-packages/litellm/proxy/litellm_pre_call_utils.py b/.venv/lib/python3.12/site-packages/litellm/proxy/litellm_pre_call_utils.py
new file mode 100644
index 00000000..ece5ecf4
--- /dev/null
+++ b/.venv/lib/python3.12/site-packages/litellm/proxy/litellm_pre_call_utils.py
@@ -0,0 +1,901 @@
+import asyncio
+import copy
+import time
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union
+
+from fastapi import Request
+from starlette.datastructures import Headers
+
+import litellm
+from litellm._logging import verbose_logger, verbose_proxy_logger
+from litellm._service_logger import ServiceLogging
+from litellm.proxy._types import (
+    AddTeamCallback,
+    CommonProxyErrors,
+    LitellmDataForBackendLLMCall,
+    SpecialHeaders,
+    TeamCallbackMetadata,
+    UserAPIKeyAuth,
+)
+from litellm.proxy.auth.route_checks import RouteChecks
+from litellm.router import Router
+from litellm.types.llms.anthropic import ANTHROPIC_API_HEADERS
+from litellm.types.services import ServiceTypes
+from litellm.types.utils import (
+    ProviderSpecificHeader,
+    StandardLoggingUserAPIKeyMetadata,
+    SupportedCacheControls,
+)
+
+service_logger_obj = ServiceLogging()  # used for tracking latency on OTEL
+
+
+if TYPE_CHECKING:
+    from litellm.proxy.proxy_server import ProxyConfig as _ProxyConfig
+
+    ProxyConfig = _ProxyConfig
+else:
+    ProxyConfig = Any
+
+
+def parse_cache_control(cache_control):
+    cache_dict = {}
+    directives = cache_control.split(", ")
+
+    for directive in directives:
+        if "=" in directive:
+            key, value = directive.split("=")
+            cache_dict[key] = value
+        else:
+            cache_dict[directive] = True
+
+    return cache_dict
+
+
+def _get_metadata_variable_name(request: Request) -> str:
+    """
+    Helper to return what the "metadata" field should be called in the request data
+
+    For all /thread or /assistant endpoints we need to call this "litellm_metadata"
+
+    For ALL other endpoints we call this "metadata
+    """
+    if RouteChecks._is_assistants_api_request(request):
+        return "litellm_metadata"
+
+    LITELLM_METADATA_ROUTES = [
+        "batches",
+        "/v1/messages",
+        "responses",
+    ]
+    if any(
+        [
+            litellm_metadata_route in request.url.path
+            for litellm_metadata_route in LITELLM_METADATA_ROUTES
+        ]
+    ):
+        return "litellm_metadata"
+    else:
+        return "metadata"
+
+
+def safe_add_api_version_from_query_params(data: dict, request: Request):
+    try:
+        if hasattr(request, "query_params"):
+            query_params = dict(request.query_params)
+            if "api-version" in query_params:
+                data["api_version"] = query_params["api-version"]
+    except KeyError:
+        pass
+    except Exception as e:
+        verbose_logger.exception(
+            "error checking api version in query params: %s", str(e)
+        )
+
+
+def convert_key_logging_metadata_to_callback(
+    data: AddTeamCallback, team_callback_settings_obj: Optional[TeamCallbackMetadata]
+) -> TeamCallbackMetadata:
+    if team_callback_settings_obj is None:
+        team_callback_settings_obj = TeamCallbackMetadata()
+    if data.callback_type == "success":
+        if team_callback_settings_obj.success_callback is None:
+            team_callback_settings_obj.success_callback = []
+
+        if data.callback_name not in team_callback_settings_obj.success_callback:
+            team_callback_settings_obj.success_callback.append(data.callback_name)
+    elif data.callback_type == "failure":
+        if team_callback_settings_obj.failure_callback is None:
+            team_callback_settings_obj.failure_callback = []
+
+        if data.callback_name not in team_callback_settings_obj.failure_callback:
+            team_callback_settings_obj.failure_callback.append(data.callback_name)
+    elif (
+        not data.callback_type or data.callback_type == "success_and_failure"
+    ):  # assume 'success_and_failure' = litellm.callbacks
+        if team_callback_settings_obj.success_callback is None:
+            team_callback_settings_obj.success_callback = []
+        if team_callback_settings_obj.failure_callback is None:
+            team_callback_settings_obj.failure_callback = []
+        if team_callback_settings_obj.callbacks is None:
+            team_callback_settings_obj.callbacks = []
+
+        if data.callback_name not in team_callback_settings_obj.success_callback:
+            team_callback_settings_obj.success_callback.append(data.callback_name)
+
+        if data.callback_name not in team_callback_settings_obj.failure_callback:
+            team_callback_settings_obj.failure_callback.append(data.callback_name)
+
+        if data.callback_name not in team_callback_settings_obj.callbacks:
+            team_callback_settings_obj.callbacks.append(data.callback_name)
+
+    for var, value in data.callback_vars.items():
+        if team_callback_settings_obj.callback_vars is None:
+            team_callback_settings_obj.callback_vars = {}
+        team_callback_settings_obj.callback_vars[var] = str(
+            litellm.utils.get_secret(value, default_value=value) or value
+        )
+
+    return team_callback_settings_obj
+
+
+def _get_dynamic_logging_metadata(
+    user_api_key_dict: UserAPIKeyAuth, proxy_config: ProxyConfig
+) -> Optional[TeamCallbackMetadata]:
+    callback_settings_obj: Optional[TeamCallbackMetadata] = None
+    if (
+        user_api_key_dict.metadata is not None
+        and "logging" in user_api_key_dict.metadata
+    ):
+        for item in user_api_key_dict.metadata["logging"]:
+            callback_settings_obj = convert_key_logging_metadata_to_callback(
+                data=AddTeamCallback(**item),
+                team_callback_settings_obj=callback_settings_obj,
+            )
+    elif (
+        user_api_key_dict.team_metadata is not None
+        and "callback_settings" in user_api_key_dict.team_metadata
+    ):
+        """
+        callback_settings = {
+            {
+            'callback_vars': {'langfuse_public_key': 'pk', 'langfuse_secret_key': 'sk_'},
+            'failure_callback': [],
+            'success_callback': ['langfuse', 'langfuse']
+        }
+        }
+        """
+        team_metadata = user_api_key_dict.team_metadata
+        callback_settings = team_metadata.get("callback_settings", None) or {}
+        callback_settings_obj = TeamCallbackMetadata(**callback_settings)
+        verbose_proxy_logger.debug(
+            "Team callback settings activated: %s", callback_settings_obj
+        )
+    elif user_api_key_dict.team_id is not None:
+        callback_settings_obj = (
+            LiteLLMProxyRequestSetup.add_team_based_callbacks_from_config(
+                team_id=user_api_key_dict.team_id, proxy_config=proxy_config
+            )
+        )
+    return callback_settings_obj
+
+
+def clean_headers(
+    headers: Headers, litellm_key_header_name: Optional[str] = None
+) -> dict:
+    """
+    Removes litellm api key from headers
+    """
+    special_headers = [v.value.lower() for v in SpecialHeaders._member_map_.values()]
+    special_headers = special_headers
+    if litellm_key_header_name is not None:
+        special_headers.append(litellm_key_header_name.lower())
+    clean_headers = {}
+    for header, value in headers.items():
+        if header.lower() not in special_headers:
+            clean_headers[header] = value
+    return clean_headers
+
+
+class LiteLLMProxyRequestSetup:
+    @staticmethod
+    def _get_timeout_from_request(headers: dict) -> Optional[float]:
+        """
+        Workaround for client request from Vercel's AI SDK.
+
+        Allow's user to set a timeout in the request headers.
+
+        Example:
+
+        ```js
+        const openaiProvider = createOpenAI({
+            baseURL: liteLLM.baseURL,
+            apiKey: liteLLM.apiKey,
+            compatibility: "compatible",
+            headers: {
+                "x-litellm-timeout": "90"
+            },
+        });
+        ```
+        """
+        timeout_header = headers.get("x-litellm-timeout", None)
+        if timeout_header is not None:
+            return float(timeout_header)
+        return None
+
+    @staticmethod
+    def _get_forwardable_headers(
+        headers: Union[Headers, dict],
+    ):
+        """
+        Get the headers that should be forwarded to the LLM Provider.
+
+        Looks for any `x-` headers and sends them to the LLM Provider.
+        """
+        forwarded_headers = {}
+        for header, value in headers.items():
+            if header.lower().startswith("x-") and not header.lower().startswith(
+                "x-stainless"
+            ):  # causes openai sdk to fail
+                forwarded_headers[header] = value
+
+        return forwarded_headers
+
+    @staticmethod
+    def get_openai_org_id_from_headers(
+        headers: dict, general_settings: Optional[Dict] = None
+    ) -> Optional[str]:
+        """
+        Get the OpenAI Org ID from the headers.
+        """
+        if (
+            general_settings is not None
+            and general_settings.get("forward_openai_org_id") is not True
+        ):
+            return None
+        for header, value in headers.items():
+            if header.lower() == "openai-organization":
+                verbose_logger.info(f"found openai org id: {value}, sending to llm")
+                return value
+        return None
+
+    @staticmethod
+    def add_headers_to_llm_call(
+        headers: dict, user_api_key_dict: UserAPIKeyAuth
+    ) -> dict:
+        """
+        Add headers to the LLM call
+
+        - Checks request headers for forwardable headers
+        - Checks if user information should be added to the headers
+        """
+
+        returned_headers = LiteLLMProxyRequestSetup._get_forwardable_headers(headers)
+
+        if litellm.add_user_information_to_llm_headers is True:
+            litellm_logging_metadata_headers = (
+                LiteLLMProxyRequestSetup.get_sanitized_user_information_from_key(
+                    user_api_key_dict=user_api_key_dict
+                )
+            )
+            for k, v in litellm_logging_metadata_headers.items():
+                if v is not None:
+                    returned_headers["x-litellm-{}".format(k)] = v
+
+        return returned_headers
+
+    @staticmethod
+    def add_litellm_data_for_backend_llm_call(
+        *,
+        headers: dict,
+        user_api_key_dict: UserAPIKeyAuth,
+        general_settings: Optional[Dict[str, Any]] = None,
+    ) -> LitellmDataForBackendLLMCall:
+        """
+        - Adds forwardable headers
+        - Adds org id
+        """
+        data = LitellmDataForBackendLLMCall()
+        if (
+            general_settings
+            and general_settings.get("forward_client_headers_to_llm_api") is True
+        ):
+            _headers = LiteLLMProxyRequestSetup.add_headers_to_llm_call(
+                headers, user_api_key_dict
+            )
+            if _headers != {}:
+                data["headers"] = _headers
+        _organization = LiteLLMProxyRequestSetup.get_openai_org_id_from_headers(
+            headers, general_settings
+        )
+        if _organization is not None:
+            data["organization"] = _organization
+
+        timeout = LiteLLMProxyRequestSetup._get_timeout_from_request(headers)
+        if timeout is not None:
+            data["timeout"] = timeout
+
+        return data
+
+    @staticmethod
+    def get_sanitized_user_information_from_key(
+        user_api_key_dict: UserAPIKeyAuth,
+    ) -> StandardLoggingUserAPIKeyMetadata:
+        user_api_key_logged_metadata = StandardLoggingUserAPIKeyMetadata(
+            user_api_key_hash=user_api_key_dict.api_key,  # just the hashed token
+            user_api_key_alias=user_api_key_dict.key_alias,
+            user_api_key_team_id=user_api_key_dict.team_id,
+            user_api_key_user_id=user_api_key_dict.user_id,
+            user_api_key_org_id=user_api_key_dict.org_id,
+            user_api_key_team_alias=user_api_key_dict.team_alias,
+            user_api_key_end_user_id=user_api_key_dict.end_user_id,
+            user_api_key_user_email=user_api_key_dict.user_email,
+        )
+        return user_api_key_logged_metadata
+
+    @staticmethod
+    def add_key_level_controls(
+        key_metadata: dict, data: dict, _metadata_variable_name: str
+    ):
+        if "cache" in key_metadata:
+            data["cache"] = {}
+            if isinstance(key_metadata["cache"], dict):
+                for k, v in key_metadata["cache"].items():
+                    if k in SupportedCacheControls:
+                        data["cache"][k] = v
+
+        ## KEY-LEVEL SPEND LOGS / TAGS
+        if "tags" in key_metadata and key_metadata["tags"] is not None:
+            data[_metadata_variable_name]["tags"] = (
+                LiteLLMProxyRequestSetup._merge_tags(
+                    request_tags=data[_metadata_variable_name].get("tags"),
+                    tags_to_add=key_metadata["tags"],
+                )
+            )
+        if "spend_logs_metadata" in key_metadata and isinstance(
+            key_metadata["spend_logs_metadata"], dict
+        ):
+            if "spend_logs_metadata" in data[_metadata_variable_name] and isinstance(
+                data[_metadata_variable_name]["spend_logs_metadata"], dict
+            ):
+                for key, value in key_metadata["spend_logs_metadata"].items():
+                    if (
+                        key not in data[_metadata_variable_name]["spend_logs_metadata"]
+                    ):  # don't override k-v pair sent by request (user request)
+                        data[_metadata_variable_name]["spend_logs_metadata"][
+                            key
+                        ] = value
+            else:
+                data[_metadata_variable_name]["spend_logs_metadata"] = key_metadata[
+                    "spend_logs_metadata"
+                ]
+
+        ## KEY-LEVEL DISABLE FALLBACKS
+        if "disable_fallbacks" in key_metadata and isinstance(
+            key_metadata["disable_fallbacks"], bool
+        ):
+            data["disable_fallbacks"] = key_metadata["disable_fallbacks"]
+        return data
+
+    @staticmethod
+    def _merge_tags(request_tags: Optional[list], tags_to_add: Optional[list]) -> list:
+        """
+        Helper function to merge two lists of tags, ensuring no duplicates.
+
+        Args:
+            request_tags (Optional[list]): List of tags from the original request
+            tags_to_add (Optional[list]): List of tags to add
+
+        Returns:
+            list: Combined list of unique tags
+        """
+        final_tags = []
+
+        if request_tags and isinstance(request_tags, list):
+            final_tags.extend(request_tags)
+
+        if tags_to_add and isinstance(tags_to_add, list):
+            for tag in tags_to_add:
+                if tag not in final_tags:
+                    final_tags.append(tag)
+
+        return final_tags
+
+    @staticmethod
+    def add_team_based_callbacks_from_config(
+        team_id: str,
+        proxy_config: ProxyConfig,
+    ) -> Optional[TeamCallbackMetadata]:
+        """
+        Add team-based callbacks from the config
+        """
+        team_config = proxy_config.load_team_config(team_id=team_id)
+        if len(team_config.keys()) == 0:
+            return None
+
+        callback_vars_dict = {**team_config.get("callback_vars", team_config)}
+        callback_vars_dict.pop("team_id", None)
+        callback_vars_dict.pop("success_callback", None)
+        callback_vars_dict.pop("failure_callback", None)
+
+        return TeamCallbackMetadata(
+            success_callback=team_config.get("success_callback", None),
+            failure_callback=team_config.get("failure_callback", None),
+            callback_vars=callback_vars_dict,
+        )
+
+    @staticmethod
+    def add_request_tag_to_metadata(
+        llm_router: Optional[Router],
+        headers: dict,
+        data: dict,
+    ) -> Optional[List[str]]:
+        tags = None
+
+        if llm_router and llm_router.enable_tag_filtering is True:
+            # Check request headers for tags
+            if "x-litellm-tags" in headers:
+                if isinstance(headers["x-litellm-tags"], str):
+                    _tags = headers["x-litellm-tags"].split(",")
+                    tags = [tag.strip() for tag in _tags]
+                elif isinstance(headers["x-litellm-tags"], list):
+                    tags = headers["x-litellm-tags"]
+        # Check request body for tags
+        if "tags" in data and isinstance(data["tags"], list):
+            tags = data["tags"]
+
+        return tags
+
+
+async def add_litellm_data_to_request(  # noqa: PLR0915
+    data: dict,
+    request: Request,
+    user_api_key_dict: UserAPIKeyAuth,
+    proxy_config: ProxyConfig,
+    general_settings: Optional[Dict[str, Any]] = None,
+    version: Optional[str] = None,
+):
+    """
+    Adds LiteLLM-specific data to the request.
+
+    Args:
+        data (dict): The data dictionary to be modified.
+        request (Request): The incoming request.
+        user_api_key_dict (UserAPIKeyAuth): The user API key dictionary.
+        general_settings (Optional[Dict[str, Any]], optional): General settings. Defaults to None.
+        version (Optional[str], optional): Version. Defaults to None.
+
+    Returns:
+        dict: The modified data dictionary.
+
+    """
+
+    from litellm.proxy.proxy_server import llm_router, premium_user
+
+    safe_add_api_version_from_query_params(data, request)
+
+    _headers = clean_headers(
+        request.headers,
+        litellm_key_header_name=(
+            general_settings.get("litellm_key_header_name")
+            if general_settings is not None
+            else None
+        ),
+    )
+
+    data.update(
+        LiteLLMProxyRequestSetup.add_litellm_data_for_backend_llm_call(
+            headers=_headers,
+            user_api_key_dict=user_api_key_dict,
+            general_settings=general_settings,
+        )
+    )
+
+    # Include original request and headers in the data
+    data["proxy_server_request"] = {
+        "url": str(request.url),
+        "method": request.method,
+        "headers": _headers,
+        "body": copy.copy(data),  # use copy instead of deepcopy
+    }
+
+    ## Dynamic api version (Azure OpenAI endpoints) ##
+    try:
+        query_params = request.query_params
+        # Convert query parameters to a dictionary (optional)
+        query_dict = dict(query_params)
+    except KeyError:
+        query_dict = {}
+
+    ## check for api version in query params
+    dynamic_api_version: Optional[str] = query_dict.get("api-version")
+
+    if dynamic_api_version is not None:  # only pass, if set
+        data["api_version"] = dynamic_api_version
+
+    ## Forward any LLM API Provider specific headers in extra_headers
+    add_provider_specific_headers_to_request(data=data, headers=_headers)
+
+    ## Cache Controls
+    headers = request.headers
+    verbose_proxy_logger.debug("Request Headers: %s", headers)
+    cache_control_header = headers.get("Cache-Control", None)
+    if cache_control_header:
+        cache_dict = parse_cache_control(cache_control_header)
+        data["ttl"] = cache_dict.get("s-maxage")
+
+    verbose_proxy_logger.debug("receiving data: %s", data)
+
+    _metadata_variable_name = _get_metadata_variable_name(request)
+
+    if _metadata_variable_name not in data:
+        data[_metadata_variable_name] = {}
+
+    # We want to log the "metadata" from the client side request. Avoid circular reference by not directly assigning metadata to itself.
+    if "metadata" in data and data["metadata"] is not None:
+        data[_metadata_variable_name]["requester_metadata"] = copy.deepcopy(
+            data["metadata"]
+        )
+
+    user_api_key_logged_metadata = (
+        LiteLLMProxyRequestSetup.get_sanitized_user_information_from_key(
+            user_api_key_dict=user_api_key_dict
+        )
+    )
+    data[_metadata_variable_name].update(user_api_key_logged_metadata)
+    data[_metadata_variable_name][
+        "user_api_key"
+    ] = (
+        user_api_key_dict.api_key
+    )  # this is just the hashed token. [TODO]: replace variable name in repo.
+
+    data[_metadata_variable_name]["user_api_end_user_max_budget"] = getattr(
+        user_api_key_dict, "end_user_max_budget", None
+    )
+
+    data[_metadata_variable_name]["litellm_api_version"] = version
+
+    if general_settings is not None:
+        data[_metadata_variable_name]["global_max_parallel_requests"] = (
+            general_settings.get("global_max_parallel_requests", None)
+        )
+
+    ### KEY-LEVEL Controls
+    key_metadata = user_api_key_dict.metadata
+    data = LiteLLMProxyRequestSetup.add_key_level_controls(
+        key_metadata=key_metadata,
+        data=data,
+        _metadata_variable_name=_metadata_variable_name,
+    )
+    ## TEAM-LEVEL SPEND LOGS/TAGS
+    team_metadata = user_api_key_dict.team_metadata or {}
+    if "tags" in team_metadata and team_metadata["tags"] is not None:
+        data[_metadata_variable_name]["tags"] = LiteLLMProxyRequestSetup._merge_tags(
+            request_tags=data[_metadata_variable_name].get("tags"),
+            tags_to_add=team_metadata["tags"],
+        )
+    if "spend_logs_metadata" in team_metadata and isinstance(
+        team_metadata["spend_logs_metadata"], dict
+    ):
+        if "spend_logs_metadata" in data[_metadata_variable_name] and isinstance(
+            data[_metadata_variable_name]["spend_logs_metadata"], dict
+        ):
+            for key, value in team_metadata["spend_logs_metadata"].items():
+                if (
+                    key not in data[_metadata_variable_name]["spend_logs_metadata"]
+                ):  # don't override k-v pair sent by request (user request)
+                    data[_metadata_variable_name]["spend_logs_metadata"][key] = value
+        else:
+            data[_metadata_variable_name]["spend_logs_metadata"] = team_metadata[
+                "spend_logs_metadata"
+            ]
+
+    # Team spend, budget - used by prometheus.py
+    data[_metadata_variable_name][
+        "user_api_key_team_max_budget"
+    ] = user_api_key_dict.team_max_budget
+    data[_metadata_variable_name][
+        "user_api_key_team_spend"
+    ] = user_api_key_dict.team_spend
+
+    # API Key spend, budget - used by prometheus.py
+    data[_metadata_variable_name]["user_api_key_spend"] = user_api_key_dict.spend
+    data[_metadata_variable_name][
+        "user_api_key_max_budget"
+    ] = user_api_key_dict.max_budget
+    data[_metadata_variable_name][
+        "user_api_key_model_max_budget"
+    ] = user_api_key_dict.model_max_budget
+
+    data[_metadata_variable_name]["user_api_key_metadata"] = user_api_key_dict.metadata
+    _headers = dict(request.headers)
+    _headers.pop(
+        "authorization", None
+    )  # do not store the original `sk-..` api key in the db
+    data[_metadata_variable_name]["headers"] = _headers
+    data[_metadata_variable_name]["endpoint"] = str(request.url)
+
+    # OTEL Controls / Tracing
+    # Add the OTEL Parent Trace before sending it LiteLLM
+    data[_metadata_variable_name][
+        "litellm_parent_otel_span"
+    ] = user_api_key_dict.parent_otel_span
+    _add_otel_traceparent_to_data(data, request=request)
+
+    ### END-USER SPECIFIC PARAMS ###
+    if user_api_key_dict.allowed_model_region is not None:
+        data["allowed_model_region"] = user_api_key_dict.allowed_model_region
+    start_time = time.time()
+    ## [Enterprise Only]
+    # Add User-IP Address
+    requester_ip_address = ""
+    if premium_user is True:
+        # Only set the IP Address for Enterprise Users
+
+        # logic for tracking IP Address
+        if (
+            general_settings is not None
+            and general_settings.get("use_x_forwarded_for") is True
+            and request is not None
+            and hasattr(request, "headers")
+            and "x-forwarded-for" in request.headers
+        ):
+            requester_ip_address = request.headers["x-forwarded-for"]
+        elif (
+            request is not None
+            and hasattr(request, "client")
+            and hasattr(request.client, "host")
+            and request.client is not None
+        ):
+            requester_ip_address = request.client.host
+    data[_metadata_variable_name]["requester_ip_address"] = requester_ip_address
+
+    # Check if using tag based routing
+    tags = LiteLLMProxyRequestSetup.add_request_tag_to_metadata(
+        llm_router=llm_router,
+        headers=dict(request.headers),
+        data=data,
+    )
+
+    if tags is not None:
+        data[_metadata_variable_name]["tags"] = tags
+
+    # Team Callbacks controls
+    callback_settings_obj = _get_dynamic_logging_metadata(
+        user_api_key_dict=user_api_key_dict, proxy_config=proxy_config
+    )
+    if callback_settings_obj is not None:
+        data["success_callback"] = callback_settings_obj.success_callback
+        data["failure_callback"] = callback_settings_obj.failure_callback
+
+        if callback_settings_obj.callback_vars is not None:
+            # unpack callback_vars in data
+            for k, v in callback_settings_obj.callback_vars.items():
+                data[k] = v
+
+    # Guardrails
+    move_guardrails_to_metadata(
+        data=data,
+        _metadata_variable_name=_metadata_variable_name,
+        user_api_key_dict=user_api_key_dict,
+    )
+
+    # Team Model Aliases
+    _update_model_if_team_alias_exists(
+        data=data,
+        user_api_key_dict=user_api_key_dict,
+    )
+
+    verbose_proxy_logger.debug(
+        "[PROXY] returned data from litellm_pre_call_utils: %s", data
+    )
+
+    ## ENFORCED PARAMS CHECK
+    # loop through each enforced param
+    # example enforced_params ['user', 'metadata', 'metadata.generation_name']
+    _enforced_params_check(
+        request_body=data,
+        general_settings=general_settings,
+        user_api_key_dict=user_api_key_dict,
+        premium_user=premium_user,
+    )
+
+    end_time = time.time()
+    asyncio.create_task(
+        service_logger_obj.async_service_success_hook(
+            service=ServiceTypes.PROXY_PRE_CALL,
+            duration=end_time - start_time,
+            call_type="add_litellm_data_to_request",
+            start_time=start_time,
+            end_time=end_time,
+            parent_otel_span=user_api_key_dict.parent_otel_span,
+        )
+    )
+
+    return data
+
+
+def _update_model_if_team_alias_exists(
+    data: dict,
+    user_api_key_dict: UserAPIKeyAuth,
+) -> None:
+    """
+    Update the model if the team alias exists
+
+    If a alias map has been set on a team, then we want to make the request with the model the team alias is pointing to
+
+    eg.
+        - user calls `gpt-4o`
+        - team.model_alias_map = {
+            "gpt-4o": "gpt-4o-team-1"
+        }
+        - requested_model = "gpt-4o-team-1"
+    """
+    _model = data.get("model")
+    if (
+        _model
+        and user_api_key_dict.team_model_aliases
+        and _model in user_api_key_dict.team_model_aliases
+    ):
+        data["model"] = user_api_key_dict.team_model_aliases[_model]
+    return
+
+
+def _get_enforced_params(
+    general_settings: Optional[dict], user_api_key_dict: UserAPIKeyAuth
+) -> Optional[list]:
+    enforced_params: Optional[list] = None
+    if general_settings is not None:
+        enforced_params = general_settings.get("enforced_params")
+        if "service_account_settings" in general_settings:
+            service_account_settings = general_settings["service_account_settings"]
+            if "enforced_params" in service_account_settings:
+                if enforced_params is None:
+                    enforced_params = []
+                enforced_params.extend(service_account_settings["enforced_params"])
+    if user_api_key_dict.metadata.get("enforced_params", None) is not None:
+        if enforced_params is None:
+            enforced_params = []
+        enforced_params.extend(user_api_key_dict.metadata["enforced_params"])
+    return enforced_params
+
+
+def _enforced_params_check(
+    request_body: dict,
+    general_settings: Optional[dict],
+    user_api_key_dict: UserAPIKeyAuth,
+    premium_user: bool,
+) -> bool:
+    """
+    If enforced params are set, check if the request body contains the enforced params.
+    """
+    enforced_params: Optional[list] = _get_enforced_params(
+        general_settings=general_settings, user_api_key_dict=user_api_key_dict
+    )
+    if enforced_params is None:
+        return True
+    if enforced_params is not None and premium_user is not True:
+        raise ValueError(
+            f"Enforced Params is an Enterprise feature. Enforced Params: {enforced_params}. {CommonProxyErrors.not_premium_user.value}"
+        )
+
+    for enforced_param in enforced_params:
+        _enforced_params = enforced_param.split(".")
+        if len(_enforced_params) == 1:
+            if _enforced_params[0] not in request_body:
+                raise ValueError(
+                    f"BadRequest please pass param={_enforced_params[0]} in request body. This is a required param"
+                )
+        elif len(_enforced_params) == 2:
+            # this is a scenario where user requires request['metadata']['generation_name'] to exist
+            if _enforced_params[0] not in request_body:
+                raise ValueError(
+                    f"BadRequest please pass param={_enforced_params[0]} in request body. This is a required param"
+                )
+            if _enforced_params[1] not in request_body[_enforced_params[0]]:
+                raise ValueError(
+                    f"BadRequest please pass param=[{_enforced_params[0]}][{_enforced_params[1]}] in request body. This is a required param"
+                )
+    return True
+
+
+def _add_guardrails_from_key_or_team_metadata(
+    key_metadata: Optional[dict],
+    team_metadata: Optional[dict],
+    data: dict,
+    metadata_variable_name: str,
+) -> None:
+    """
+    Helper add guardrails from key or team metadata to request data
+
+    Args:
+        key_metadata: The key metadata dictionary to check for guardrails
+        team_metadata: The team metadata dictionary to check for guardrails
+        data: The request data to update
+        metadata_variable_name: The name of the metadata field in data
+
+    """
+    from litellm.proxy.utils import _premium_user_check
+
+    for _management_object_metadata in [key_metadata, team_metadata]:
+        if _management_object_metadata and "guardrails" in _management_object_metadata:
+            if len(_management_object_metadata["guardrails"]) > 0:
+                _premium_user_check()
+
+            data[metadata_variable_name]["guardrails"] = _management_object_metadata[
+                "guardrails"
+            ]
+
+
+def move_guardrails_to_metadata(
+    data: dict,
+    _metadata_variable_name: str,
+    user_api_key_dict: UserAPIKeyAuth,
+):
+    """
+    Helper to add guardrails from request to metadata
+
+    - If guardrails set on API Key metadata then sets guardrails on request metadata
+    - If guardrails not set on API key, then checks request metadata
+    """
+    # Check key-level guardrails
+    _add_guardrails_from_key_or_team_metadata(
+        key_metadata=user_api_key_dict.metadata,
+        team_metadata=user_api_key_dict.team_metadata,
+        data=data,
+        metadata_variable_name=_metadata_variable_name,
+    )
+
+    # Check request-level guardrails
+    if "guardrails" in data:
+        data[_metadata_variable_name]["guardrails"] = data["guardrails"]
+        del data["guardrails"]
+
+    if "guardrail_config" in data:
+        data[_metadata_variable_name]["guardrail_config"] = data["guardrail_config"]
+        del data["guardrail_config"]
+
+
+def add_provider_specific_headers_to_request(
+    data: dict,
+    headers: dict,
+):
+    anthropic_headers = {}
+    # boolean to indicate if a header was added
+    added_header = False
+    for header in ANTHROPIC_API_HEADERS:
+        if header in headers:
+            header_value = headers[header]
+            anthropic_headers[header] = header_value
+            added_header = True
+
+    if added_header is True:
+        data["provider_specific_header"] = ProviderSpecificHeader(
+            custom_llm_provider="anthropic",
+            extra_headers=anthropic_headers,
+        )
+
+    return
+
+
+def _add_otel_traceparent_to_data(data: dict, request: Request):
+    from litellm.proxy.proxy_server import open_telemetry_logger
+
+    if data is None:
+        return
+    if open_telemetry_logger is None:
+        # if user is not use OTEL don't send extra_headers
+        # relevant issue: https://github.com/BerriAI/litellm/issues/4448
+        return
+
+    if litellm.forward_traceparent_to_llm_provider is True:
+        if request.headers:
+            if "traceparent" in request.headers:
+                # we want to forward this to the LLM Provider
+                # Relevant issue: https://github.com/BerriAI/litellm/issues/4419
+                # pass this in extra_headers
+                if "extra_headers" not in data:
+                    data["extra_headers"] = {}
+                _exra_headers = data["extra_headers"]
+                if "traceparent" not in _exra_headers:
+                    _exra_headers["traceparent"] = request.headers["traceparent"]