Add support for watsonx passthrough route

Signed-off-by: T K Chandra Hasan <t.k.chandra.hasan@ibm.com>

litellm/llms/watsonx/passthrough/transformation.py

@@ -0,0 +1,65 @@
+from typing import TYPE_CHECKING, List, Optional, Tuple
+
+from litellm.llms.base_llm.passthrough.transformation import BasePassthroughConfig
+from litellm.llms.watsonx.common_utils import IBMWatsonXMixin 
+from litellm.secret_managers.main import get_secret_str
+
+if TYPE_CHECKING:
+    from httpx import URL
+
+
+class WatsonxPassthroughConfig(IBMWatsonXMixin, BasePassthroughConfig):
+    """
+    Watsonx-specific passthrough configuration.
+    """
+
+    def is_streaming_request(self, endpoint: str, request_data: dict) -> bool:
+        """Check if request should be streamed"""
+        return request_data.get("stream", False)
+
+    def get_complete_url(
+        self,
+        api_base: Optional[str],
+        api_key: Optional[str],
+        model: str,
+        endpoint: str,
+        request_query_params: Optional[dict],
+        litellm_params: dict,
+    ) -> Tuple["URL", str]:
+        """
+        Construct complete Watsonx URL with version parameter.
+        
+        This ensures the version parameter is ALWAYS included in the URL,
+        solving the query parameter issue.
+        """
+        base_target_url = self.get_api_base(api_base) or self._get_base_url(api_base)
+
+        # Use the format_url helper to construct URL with query params
+        complete_url = self.format_url(
+            endpoint=endpoint,
+            base_target_url=base_target_url,
+            request_query_params=request_query_params,
+        )
+
+        return (complete_url, base_target_url)
+
+    @staticmethod
+    def get_api_base(
+        api_base: Optional[str] = None,
+    ) -> Optional[str]:
+        return api_base or get_secret_str("WATSONX_API_BASE")
+
+    @staticmethod
+    def get_api_key(
+        api_key: Optional[str] = None,
+    ) -> Optional[str]:
+        return api_key or get_secret_str("WATSON_API_KEY")
+
+    @staticmethod
+    def get_base_model(model: str) -> Optional[str]:
+        return model
+
+    def get_models(
+        self, api_key: Optional[str] = None, api_base: Optional[str] = None
+    ) -> List[str]:
+        return super().get_models(api_key, api_base)

litellm/proxy/_types.py

@@ -413,6 +413,7 @@ class LiteLLMRoutes(enum.Enum):
         "/vllm",
         "/mistral",
         "/milvus",
+        "/watsonx",
     ]
 
     #########################################################

litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py

@@ -2424,3 +2424,88 @@ def create_generic_websocket_passthrough_endpoint(
         _forward_headers=forward_headers,
         cost_per_request=cost_per_request,
     )
+
+@router.api_route(
+    "/watsonx/{endpoint:path}",
+    methods=["GET", "POST", "PUT", "DELETE", "PATCH"],
+    tags=["Watsonx Pass-through", "pass-through"],
+)
+async def watsonx_proxy_route(
+    endpoint: str,
+    request: Request,
+    fastapi_response: Response,
+    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
+):
+    """
+    Watsonx pass-through endpoint.
+    Allows using Watsonx APIs with automatic IAM token management and version parameter injection.
+
+    Example:
+        POST /watsonx/ml/v1/text/tokenization
+        POST /watsonx/ml/v1/text/generation
+    """
+    # Direct passthrough with WatsonxPassthroughConfig
+    from litellm.types.utils import LlmProviders
+    from litellm.utils import ProviderConfigManager
+
+    provider_config = ProviderConfigManager.get_provider_passthrough_config(
+        provider=LlmProviders.WATSONX,
+        model="",
+    )
+
+    if provider_config is None:
+        raise HTTPException(
+            status_code=404,
+            detail="Watsonx passthrough config not found"
+        )
+
+    # Get complete URL with version parameter
+    complete_url, _ = provider_config.get_complete_url(
+        api_base=None,
+        api_key=None,
+        model="",
+        endpoint=endpoint,
+        request_query_params=dict(request.query_params),
+        litellm_params={},
+    )
+
+    # Get auth headers with IAM token
+    auth_headers = provider_config.validate_environment(
+        headers={},
+        model="",
+        messages=[],
+        optional_params={},
+        litellm_params={},
+        api_key=None,
+        api_base=None,
+    )
+
+    # Check for streaming
+    is_streaming_request = False
+    if request.method == "POST":
+        if "multipart/form-data" not in request.headers.get("content-type", ""):
+            _request_body = await request.json()
+        else:
+            _request_body = await get_form_data(request)
+
+        if _request_body.get("stream"):
+            is_streaming_request = True
+
+    request_query_params = dict()
+    request_query_params["version"] = litellm.WATSONX_DEFAULT_API_VERSION
+
+    # Create pass-through endpoint
+    endpoint_func = create_pass_through_route(
+        endpoint=endpoint,
+        target=str(complete_url),
+        custom_headers=auth_headers,
+        is_streaming_request=is_streaming_request,
+        custom_llm_provider="watsonx",
+        query_params=request_query_params,
+    )
+
+    return await endpoint_func(
+        request,
+        fastapi_response,
+        user_api_key_dict,
+    )

litellm/utils.py

@@ -8798,6 +8798,12 @@ class ProviderConfigManager:
             )
 
             return AzurePassthroughConfig()
+        elif LlmProviders.WATSONX == provider:
+            from litellm.llms.watsonx.passthrough.transformation import (
+                WatsonxPassthroughConfig,
+            )
+
+            return WatsonxPassthroughConfig()
         return None
 
     @staticmethod