fix(llms): sanitize trailing slashes in provider get_complete_url (#42887)

* fix(llms): sanitize trailing slashes in provider get_complete_url

* test(llms): move trailing-slash url tests into each provider's mapped unit file

* fix(llms): strip trailing slashes in Perplexity search and Ollama embedding URLs and run the Topaz tests in CI

* fix(llms): strip trailing slashes in search, DashScope, VolcEngine, GigaChat and Gradient AI URL builders

* fix(fastcrw): strip a trailing slash from api_base and expect one slash in the GigaChat tests

* fix(gigachat): strip a trailing slash from the pass-through api_base

* test(integration): cover trailing-slash api_base across provider URL builders

Wire cells for chat, streaming, messages, responses, embeddings, search and pass-through with and without a trailing slash, an owned-proxy module for the flag-gated providers, env bases and a worker kill, two chaos bursts, and a Topaz SDK cell

---------

Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>

.github/workflows/test-unit.yml

@@ -346,6 +346,7 @@ jobs:
               tests/unit/llms/tinyfish
               tests/unit/llms/together_ai
               tests/unit/llms/tool_loop
+              tests/unit/llms/topaz
               tests/unit/llms/triton
               tests/unit/llms/v0
               tests/unit/llms/valkey

litellm/llms/dashscope/chat/transformation.py

@@ -82,7 +82,7 @@ class DashScopeChatConfig(OpenAIGPTConfig):
         """
         If api_base is not provided, use the default DashScope /chat/completions endpoint.
         """
-        resolved_api_base: Final = self._resolve_chat_api_base(api_base)
+        resolved_api_base: Final = self._resolve_chat_api_base(api_base).rstrip("/")
         if resolved_api_base.endswith("/chat/completions"):
             return resolved_api_base
         return f"{resolved_api_base}/chat/completions"

litellm/llms/deepseek/chat/transformation.py

@@ -363,10 +363,9 @@ class DeepSeekChatConfig(OpenAIGPTConfig):
         """
         If api_base is not provided, use the default DeepSeek /chat/completions endpoint.
         """
-        if not api_base:
-            api_base = "https://api.deepseek.com/beta"
+        clean_api_base: Final = (api_base or "https://api.deepseek.com/beta").rstrip("/")
 
-        if not api_base.endswith("/chat/completions"):
-            api_base = f"{api_base}/chat/completions"
+        if not clean_api_base.endswith("/chat/completions"):
+            return f"{clean_api_base}/chat/completions"
 
-        return api_base
+        return clean_api_base

litellm/llms/exa_ai/search/transformation.py

@@ -92,13 +92,11 @@ class ExaAISearchConfig(BaseSearchConfig):
         """
         Get complete URL for Search endpoint.
         """
-        api_base = api_base or get_secret_str("EXA_API_BASE") or self.EXA_AI_API_BASE
-
-        # Append "/search" to the api base if it's not already there
-        if not api_base.endswith("/search"):
-            api_base = f"{api_base}/search"
-
-        return api_base
+        resolved_api_base: Final = api_base or get_secret_str("EXA_API_BASE") or self.EXA_AI_API_BASE
+        clean_api_base: Final = resolved_api_base.rstrip("/")
+        if clean_api_base.endswith("/search"):
+            return clean_api_base
+        return f"{clean_api_base}/search"
 
     def transform_search_request(
         self,

litellm/llms/fastcrw/search/transformation.py

@@ -83,13 +83,11 @@ class FastCRWSearchConfig(BaseSearchConfig):
         """
         Get complete URL for Search endpoint.
         """
-        api_base = api_base or get_secret_str("CRW_API_BASE") or self.FASTCRW_API_BASE
-
-        # Append "/search" to the api base if it's not already there
-        if not api_base.endswith("/search"):
-            api_base = f"{api_base}/search"
-
-        return api_base
+        resolved_api_base: Final = api_base or get_secret_str("CRW_API_BASE") or self.FASTCRW_API_BASE
+        clean_api_base: Final = resolved_api_base.rstrip("/")
+        if clean_api_base.endswith("/search"):
+            return clean_api_base
+        return f"{clean_api_base}/search"
 
     def transform_search_request(
         self,

litellm/llms/firecrawl/search/transformation.py

@@ -85,13 +85,11 @@ class FirecrawlSearchConfig(BaseSearchConfig):
         """
         Get complete URL for Search endpoint.
         """
-        api_base = api_base or get_secret_str("FIRECRAWL_API_BASE") or self.FIRECRAWL_API_BASE
-
-        # Append "/search" to the api base if it's not already there
-        if not api_base.endswith("/search"):
-            api_base = f"{api_base}/search"
-
-        return api_base
+        resolved_api_base: Final = api_base or get_secret_str("FIRECRAWL_API_BASE") or self.FIRECRAWL_API_BASE
+        clean_api_base: Final = resolved_api_base.rstrip("/")
+        if clean_api_base.endswith("/search"):
+            return clean_api_base
+        return f"{clean_api_base}/search"
 
     def transform_search_request(
         self,

litellm/llms/gigachat/passthrough/transformation.py

@@ -193,7 +193,8 @@ class GigaChatPassthroughConfig(BasePassthroughConfig):
 
     @staticmethod
     def get_api_base(api_base: str | None = None) -> str | None:
-        return api_base or get_secret_str("GIGACHAT_API_BASE") or GIGACHAT_BASE_URL
+        resolved_api_base: Final = api_base or get_secret_str("GIGACHAT_API_BASE") or GIGACHAT_BASE_URL
+        return resolved_api_base.rstrip("/") if resolved_api_base else None
 
     @staticmethod
     def get_api_key(

litellm/llms/gigachat/utils.py

@@ -22,4 +22,4 @@ def convert_usage(usage_data: Mapping[str, int]) -> Usage:
 
 
 def get_api_base(api_base: str | None = None) -> str | None:
-    return api_base or get_secret_str("GIGACHAT_API_BASE") or GIGACHAT_BASE_URL
+    return (api_base or get_secret_str("GIGACHAT_API_BASE") or GIGACHAT_BASE_URL).rstrip("/")

litellm/llms/gradient_ai/chat/transformation.py

@@ -103,15 +103,13 @@ class GradientAIConfig(OpenAILikeChatConfig):
         litellm_params: dict,
         stream: bool | None = None,
     ) -> str:
-        gradient_ai_endpoint: Final = get_secret_str("GRADIENT_AI_AGENT_ENDPOINT")
-        complete_url = f"{GRADIENT_AI_SERVERLESS_ENDPOINT}/v1/chat/completions"
-
-        if api_base and api_base != GRADIENT_AI_SERVERLESS_ENDPOINT:
-            complete_url = f"{api_base}/api/v1/chat/completions"
-        elif gradient_ai_endpoint and gradient_ai_endpoint != GRADIENT_AI_SERVERLESS_ENDPOINT:
-            complete_url = f"{gradient_ai_endpoint}/api/v1/chat/completions"
-
-        return complete_url
+        clean_api_base: Final = (api_base or "").rstrip("/")
+        if clean_api_base and clean_api_base != GRADIENT_AI_SERVERLESS_ENDPOINT:
+            return f"{clean_api_base}/api/v1/chat/completions"
+        agent_endpoint: Final = (get_secret_str("GRADIENT_AI_AGENT_ENDPOINT") or "").rstrip("/")
+        if agent_endpoint and agent_endpoint != GRADIENT_AI_SERVERLESS_ENDPOINT:
+            return f"{agent_endpoint}/api/v1/chat/completions"
+        return f"{GRADIENT_AI_SERVERLESS_ENDPOINT}/v1/chat/completions"
 
     def _get_openai_compatible_provider_info(
         self, api_base: str | None, api_key: str | None

litellm/llms/heroku/chat/transformation.py

@@ -6,7 +6,7 @@ this is OpenAI compatible - no translation needed / occurs
 
 import os
 from collections.abc import Coroutine
-from typing import Literal, overload
+from typing import Final, Literal, overload
 
 from litellm.litellm_core_utils.prompt_templates.common_utils import (
     handle_messages_with_content_list_to_str_conversion,
@@ -71,14 +71,16 @@ class HerokuChatConfig(OpenAIGPTConfig):
         litellm_params: dict,
         stream: bool | None = None,
     ) -> str:
-        api_base, _ = self._get_openai_compatible_provider_info(api_base, api_key)
+        resolved_base, _ = self._get_openai_compatible_provider_info(api_base, api_key)
 
-        if not api_base:
+        if not resolved_base:
             raise HerokuError(
                 "No api base was set. Please provide an api_base, or set the HEROKU_API_BASE environment variable."
             )
 
-        if not api_base.endswith("/v1/chat/completions"):
-            api_base = f"{api_base}/v1/chat/completions"
+        clean_api_base: Final = resolved_base.rstrip("/")
 
-        return api_base
+        if not clean_api_base.endswith("/v1/chat/completions"):
+            return f"{clean_api_base}/v1/chat/completions"
+
+        return clean_api_base

litellm/llms/jina_ai/embedding/transformation.py

@@ -106,7 +106,12 @@ class JinaAIEmbeddingConfig(BaseEmbeddingConfig):
         litellm_params: dict,
         stream: bool | None = None,
     ) -> str:
-        return f"{api_base}/embeddings" if api_base else "https://api.jina.ai/v1/embeddings"
+        if api_base:
+            clean_api_base: Final = api_base.rstrip("/")
+            if not clean_api_base.endswith("/embeddings"):
+                return f"{clean_api_base}/embeddings"
+            return clean_api_base
+        return "https://api.jina.ai/v1/embeddings"
 
     def transform_embedding_request(
         self,

litellm/llms/linkup/search/transformation.py

@@ -87,13 +87,11 @@ class LinkupSearchConfig(BaseSearchConfig):
         """
         Get complete URL for Search endpoint.
         """
-        api_base = api_base or get_secret_str("LINKUP_API_BASE") or self.LINKUP_API_BASE
-
-        # Append "/search" to the api base if it's not already there
-        if not api_base.endswith("/search"):
-            api_base = f"{api_base}/search"
-
-        return api_base
+        resolved_api_base: Final = api_base or get_secret_str("LINKUP_API_BASE") or self.LINKUP_API_BASE
+        clean_api_base: Final = resolved_api_base.rstrip("/")
+        if clean_api_base.endswith("/search"):
+            return clean_api_base
+        return f"{clean_api_base}/search"
 
     def transform_search_request(
         self,

litellm/llms/modelscope/chat/transformation.py

@@ -86,10 +86,9 @@ class ModelScopeChatConfig(OpenAIGPTConfig):
         """
         If api_base is not provided, use the default ModelScope /chat/completions endpoint.
         """
-        if not api_base:
-            api_base = self.DEFAULT_BASE_URL
+        clean_api_base: Final = (api_base or self.DEFAULT_BASE_URL).rstrip("/")
 
-        if not api_base.endswith("/chat/completions"):
-            api_base = f"{api_base}/chat/completions"
+        if not clean_api_base.endswith("/chat/completions"):
+            return f"{clean_api_base}/chat/completions"
 
-        return api_base
+        return clean_api_base

litellm/llms/moonshot/chat/transformation.py

@@ -93,13 +93,12 @@ class MoonshotChatConfig(OpenAIGPTConfig):
         """
         If api_base is not provided, use the default Moonshot AI /chat/completions endpoint.
         """
-        if not api_base:
-            api_base = "https://api.moonshot.ai/v1"
+        clean_api_base: Final = (api_base or "https://api.moonshot.ai/v1").rstrip("/")
 
-        if not api_base.endswith("/chat/completions"):
-            api_base = f"{api_base}/chat/completions"
+        if not clean_api_base.endswith("/chat/completions"):
+            return f"{clean_api_base}/chat/completions"
 
-        return api_base
+        return clean_api_base
 
     def get_supported_openai_params(self, model: str) -> list:
         """

litellm/llms/ollama/chat/transformation.py

@@ -218,14 +218,10 @@ class OllamaChatConfig(BaseConfig):
 
         Some providers need `model` in `api_base`
         """
-        if api_base is None:
-            api_base = "http://localhost:11434"
-        if api_base.endswith("/api/chat"):
-            url = api_base
-        else:
-            url = f"{api_base}/api/chat"
-
-        return url
+        clean_api_base: Final = (api_base or "http://localhost:11434").rstrip("/")
+        if clean_api_base.endswith("/api/chat"):
+            return clean_api_base
+        return f"{clean_api_base}/api/chat"
 
     def transform_request(
         self,

litellm/llms/ollama/completion/handler.py

@@ -26,6 +26,13 @@ class OllamaEmbeddingResponse(TypedDict):
     prompt_eval_count: ReadOnly[NotRequired[int]]
 
 
+def _ollama_embed_url(api_base: str) -> str:
+    clean_api_base: Final = api_base.rstrip("/")
+    if clean_api_base.endswith("/api/embed"):
+        return clean_api_base
+    return f"{clean_api_base}/api/embed"
+
+
 def _prepare_ollama_embedding_payload(
     model: str, prompts: list[str], optional_params: Mapping[str, object]
 ) -> dict[str, object]:
@@ -90,12 +97,9 @@ async def ollama_aembeddings(
     logging_obj: object,
     encoding: TokenEncoder | None,
 ):
-    if not api_base.endswith("/api/embed"):
-        api_base += "/api/embed"
-
     data: Final = _prepare_ollama_embedding_payload(model, prompts, optional_params)
 
-    response: Final = await litellm.module_level_aclient.post(url=api_base, json=data)
+    response: Final = await litellm.module_level_aclient.post(url=_ollama_embed_url(api_base), json=data)
     response_json: Final[OllamaEmbeddingResponse] = response.json()
 
     return _process_ollama_embedding_response(
@@ -117,12 +121,9 @@ def ollama_embeddings(
     logging_obj: object,
     encoding: TokenEncoder | None = None,
 ):
-    if not api_base.endswith("/api/embed"):
-        api_base += "/api/embed"
-
     data: Final = _prepare_ollama_embedding_payload(model, prompts, optional_params)
 
-    response: Final = litellm.module_level_client.post(url=api_base, json=data)
+    response: Final = litellm.module_level_client.post(url=_ollama_embed_url(api_base), json=data)
     response_json: Final[OllamaEmbeddingResponse] = response.json()
 
     return _process_ollama_embedding_response(

litellm/llms/ollama/completion/transformation.py

@@ -503,14 +503,10 @@ class OllamaConfig(BaseConfig):
 
         Some providers need `model` in `api_base`
         """
-        if api_base is None:
-            api_base = "http://localhost:11434"
-        if api_base.endswith("/api/generate"):
-            url = api_base
-        else:
-            url = f"{api_base}/api/generate"
-
-        return url
+        clean_api_base: Final = (api_base or "http://localhost:11434").rstrip("/")
+        if clean_api_base.endswith("/api/generate"):
+            return clean_api_base
+        return f"{clean_api_base}/api/generate"
 
     def get_model_response_iterator(
         self,

litellm/llms/perplexity/embedding/transformation.py

@@ -66,9 +66,10 @@ class PerplexityEmbeddingConfig(BaseEmbeddingConfig):
         stream: bool | None = None,
     ) -> str:
         if api_base:
-            if not api_base.endswith("/embeddings"):
-                api_base = f"{api_base}/v1/embeddings"
-            return api_base
+            clean_api_base: Final = api_base.rstrip("/")
+            if not clean_api_base.endswith("/embeddings"):
+                return f"{clean_api_base}/v1/embeddings"
+            return clean_api_base
         return "https://api.perplexity.ai/v1/embeddings"
 
     def get_supported_openai_params(self, model: str) -> list:

litellm/llms/perplexity/search/transformation.py

@@ -78,13 +78,11 @@ class PerplexitySearchConfig(BaseSearchConfig):
         """
         Get complete URL for Search endpoint.
         """
-        api_base = api_base or get_secret_str("PERPLEXITY_API_BASE") or self.PERPLEXITY_API_BASE
-
-        # append "/search" to the api base if it's not already there
-        if not api_base.endswith("/search"):
-            api_base = f"{api_base}/search"
-
-        return api_base
+        resolved_api_base: Final = api_base or get_secret_str("PERPLEXITY_API_BASE") or self.PERPLEXITY_API_BASE
+        clean_api_base: Final = resolved_api_base.rstrip("/")
+        if clean_api_base.endswith("/search"):
+            return clean_api_base
+        return f"{clean_api_base}/search"
 
     def transform_search_request(
         self,

litellm/llms/tavily/search/transformation.py

@@ -90,13 +90,11 @@ class TavilySearchConfig(BaseSearchConfig):
         """
         Get complete URL for Search endpoint.
         """
-        api_base = api_base or get_secret_str("TAVILY_API_BASE") or self.TAVILY_API_BASE
-
-        # Append "/search" to the api base if it's not already there
-        if not api_base.endswith("/search"):
-            api_base = f"{api_base}/search"
-
-        return api_base
+        resolved_api_base: Final = api_base or get_secret_str("TAVILY_API_BASE") or self.TAVILY_API_BASE
+        clean_api_base: Final = resolved_api_base.rstrip("/")
+        if clean_api_base.endswith("/search"):
+            return clean_api_base
+        return f"{clean_api_base}/search"
 
     def transform_search_request(
         self,

litellm/llms/topaz/image_variations/transformation.py

@@ -39,8 +39,8 @@ class TopazImageVariationConfig(TopazModelInfo, BaseImageVariationConfig):
         litellm_params: dict,
         stream: bool | None = None,
     ) -> str:
-        api_base = api_base or "https://api.topazlabs.com"
-        return f"{api_base}/image/v1/enhance"
+        clean_api_base: Final = (api_base or "https://api.topazlabs.com").rstrip("/")
+        return f"{clean_api_base}/image/v1/enhance"
 
     def map_openai_params(
         self,

litellm/llms/volcengine/common_utils.py

@@ -32,7 +32,7 @@ def get_volcengine_base_url(api_base: str | None = None) -> str:
         The base URL to use for API calls
     """
     if api_base:
-        return api_base
+        return api_base.rstrip("/")
     return "https://ark.cn-beijing.volces.com"
 
 

litellm/llms/voyage/embedding/transformation.py

@@ -51,9 +51,10 @@ class VoyageEmbeddingConfig(BaseEmbeddingConfig):
         stream: bool | None = None,
     ) -> str:
         if api_base:
-            if not api_base.endswith("/embeddings"):
-                api_base = f"{api_base}/embeddings"
-            return api_base
+            clean_api_base: Final = api_base.rstrip("/")
+            if not clean_api_base.endswith("/embeddings"):
+                return f"{clean_api_base}/embeddings"
+            return clean_api_base
         return f"{get_default_base_url(api_key)}/embeddings"
 
     def get_supported_openai_params(self, model: str) -> list:

litellm/llms/voyage/embedding/transformation_contextual.py

@@ -59,9 +59,10 @@ class VoyageContextualEmbeddingConfig(BaseEmbeddingConfig):
         stream: bool | None = None,
     ) -> str:
         if api_base:
-            if not api_base.endswith("/contextualizedembeddings"):
-                api_base = f"{api_base}/contextualizedembeddings"
-            return api_base
+            clean_api_base: Final = api_base.rstrip("/")
+            if not clean_api_base.endswith("/contextualizedembeddings"):
+                return f"{clean_api_base}/contextualizedembeddings"
+            return clean_api_base
         return f"{get_default_base_url(api_key)}/contextualizedembeddings"
 
     def get_supported_openai_params(self, model: str) -> list:  # mutable-ok: base class signature

litellm/llms/voyage/embedding/transformation_multimodal.py

@@ -60,9 +60,10 @@ class VoyageMultimodalEmbeddingConfig(BaseEmbeddingConfig):
         stream: bool | None = None,
     ) -> str:
         if api_base:
-            if not api_base.endswith("/multimodalembeddings"):
-                api_base = f"{api_base}/multimodalembeddings"
-            return api_base
+            clean_api_base: Final = api_base.rstrip("/")
+            if not clean_api_base.endswith("/multimodalembeddings"):
+                return f"{clean_api_base}/multimodalembeddings"
+            return clean_api_base
         return f"{get_default_base_url(api_key)}/multimodalembeddings"
 
     def get_supported_openai_params(self, model: str) -> list:

tests/integration/providers/_api_base_trailing_slash_support.py

@@ -0,0 +1,313 @@
+import hashlib
+import json
+import re
+import uuid
+from collections.abc import Callable, Mapping
+from dataclasses import dataclass
+from types import MappingProxyType
+from typing import Final
+
+import httpx
+from integration._support.client import Gateway, Scenario, list_value, object_value
+from integration._support.wire import Reply, Request
+from pydantic import JsonValue, TypeAdapter
+
+KEY: Final = "synthetic-trailing-slash-key"
+VECTOR: Final = (0.25, 0.5)
+TRAILING: Final = ("", "/")
+TRAILING_IDS: Final = ("no-slash", "slash")
+
+_MARKER: Final = re.compile(rb"slash-[0-9a-f]{32}")
+_JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue])
+_CREATED: Final = 1700000000
+
+
+@dataclass(frozen=True, slots=True)
+class Row:
+    model: str
+    base: str
+    route: str
+    keyed: bool = True
+    token: bool = False
+    drops_params: bool = False
+
+    @property
+    def target(self) -> str:
+        return f"{self.base}{self.route}"
+
+
+def new_marker() -> str:
+    return f"slash-{uuid.uuid4().hex}"
+
+
+def reply_text(marker: str) -> str:
+    return f"reply {marker}"
+
+
+def result_url(marker: str) -> str:
+    return f"https://result.invalid/{marker}"
+
+
+def file_id(upload: Request) -> str:
+    return f"file-{hashlib.sha256(upload.body).hexdigest()[:24]}"
+
+
+def marker_of(request: Request) -> str:
+    found: Final = _MARKER.search(request.body)
+    return found.group().decode() if found else "unmarked"
+
+
+def request_json(request: Request) -> dict[str, JsonValue]:
+    return _JSON_OBJECT.validate_json(request.body)
+
+
+def _json(payload: Mapping[str, JsonValue]) -> Reply:
+    return Reply(body=json.dumps(payload).encode())
+
+
+def _wants_stream(request: Request) -> bool:
+    return request_json(request).get("stream") is True
+
+
+def _sse(events: tuple[Mapping[str, JsonValue], ...]) -> Reply:
+    return Reply(
+        content_type="text/event-stream",
+        chunks=(*(f"data: {json.dumps(event)}\n\n".encode() for event in events), b"data: [DONE]\n\n"),
+    )
+
+
+def _chunk(marker: str, delta: Mapping[str, JsonValue], finish: str | None) -> dict[str, JsonValue]:
+    return {
+        "id": f"chatcmpl-{marker}",
+        "object": "chat.completion.chunk",
+        "created": _CREATED,
+        "model": "scripted",
+        "choices": [{"index": 0, "delta": dict(delta), "finish_reason": finish}],
+    }
+
+
+def _openai_chat(request: Request) -> Reply:
+    marker: Final = marker_of(request)
+    if _wants_stream(request):
+        return _sse(
+            (
+                _chunk(marker, {"role": "assistant", "content": "reply "}, None),
+                _chunk(marker, {"content": marker}, None),
+                {
+                    **_chunk(marker, {}, "stop"),
+                    "usage": {"prompt_tokens": 3, "completion_tokens": 2, "total_tokens": 5},
+                },
+            )
+        )
+    return _json(
+        {
+            "id": f"chatcmpl-{marker}",
+            "object": "chat.completion",
+            "created": _CREATED,
+            "model": "scripted",
+            "choices": [
+                {"index": 0, "message": {"role": "assistant", "content": reply_text(marker)}, "finish_reason": "stop"}
+            ],
+            "usage": {"prompt_tokens": 3, "completion_tokens": 2, "total_tokens": 5},
+        }
+    )
+
+
+def _anthropic_message(request: Request) -> Reply:
+    marker: Final = marker_of(request)
+    return _json(
+        {
+            "id": f"msg_{marker}",
+            "type": "message",
+            "role": "assistant",
+            "model": "scripted",
+            "content": [{"type": "text", "text": reply_text(marker)}],
+            "stop_reason": "end_turn",
+            "stop_sequence": None,
+            "usage": {"input_tokens": 3, "output_tokens": 2},
+        }
+    )
+
+
+def _deepseek(request: Request) -> Reply:
+    if request.target.endswith("/v1/messages"):
+        return _anthropic_message(request)
+    return _openai_chat(request)
+
+
+def _ollama_line(content: str, done: bool) -> bytes:
+    final: Final = {"done_reason": "stop", "prompt_eval_count": 5, "eval_count": 2} if done else {}
+    line: Final = {
+        "model": "llama3.1",
+        "created_at": "2024-01-01T00:00:00Z",
+        "message": {"role": "assistant", "content": content},
+        "done": done,
+        **final,
+    }
+    return json.dumps(line).encode() + b"\n"
+
+
+def _ollama_chat(request: Request) -> Reply:
+    marker: Final = marker_of(request)
+    if _wants_stream(request):
+        return Reply(
+            content_type="application/x-ndjson",
+            chunks=(_ollama_line("reply ", False), _ollama_line(marker, False), _ollama_line("", True)),
+        )
+    return _json(
+        {
+            "model": "llama3.1",
+            "message": {"role": "assistant", "content": reply_text(marker)},
+            "done": True,
+            "done_reason": "stop",
+            "prompt_eval_count": 5,
+            "eval_count": 2,
+        }
+    )
+
+
+def _ollama(request: Request) -> Reply:
+    if request.target.endswith("/api/chat"):
+        return _ollama_chat(request)
+    if request.target.endswith("/api/generate"):
+        return _json(
+            {
+                "model": "llama3.1",
+                "response": reply_text(marker_of(request)),
+                "done": True,
+                "prompt_eval_count": 5,
+                "eval_count": 2,
+            }
+        )
+    return _json({"embeddings": [list(VECTOR)], "prompt_eval_count": 3})
+
+
+def _embedding_list(model: str) -> Reply:
+    return _json(
+        {
+            "object": "list",
+            "data": [{"object": "embedding", "index": 0, "embedding": list(VECTOR)}],
+            "model": model,
+            "usage": {"prompt_tokens": 2, "total_tokens": 2},
+        }
+    )
+
+
+def _voyage(request: Request) -> Reply:
+    if request.target.endswith("/contextualizedembeddings"):
+        return _json(
+            {
+                "object": "list",
+                "data": [
+                    {
+                        "object": "list",
+                        "index": 0,
+                        "data": [{"object": "embedding", "index": 0, "embedding": list(VECTOR)}],
+                    }
+                ],
+                "model": "voyage-context-3",
+                "usage": {"total_tokens": 4},
+            }
+        )
+    return _embedding_list("voyage-multimodal-3" if request.target.endswith("/multimodalembeddings") else "voyage-3")
+
+
+def _gigachat(request: Request) -> Reply:
+    if request.target.endswith("/files"):
+        return _json({"id": file_id(request), "object": "file", "purpose": "general"})
+    if request.target.endswith("/models"):
+        return _json({"object": "list", "data": [{"id": marker_of(request), "object": "model"}]})
+    if request.target.endswith("/embeddings"):
+        return _json(
+            {
+                "object": "list",
+                "data": [{"object": "embedding", "embedding": list(VECTOR), "index": 0, "usage": {"prompt_tokens": 3}}],
+                "model": "Embeddings",
+            }
+        )
+    return _openai_chat(request)
+
+
+def _search(results: Callable[[str], Mapping[str, JsonValue]]) -> Callable[[Request], Reply]:
+    return lambda request: _json(results(result_url(marker_of(request))))
+
+
+_PERPLEXITY_SEARCH: Final = _search(lambda url: {"results": [{"title": "T", "url": url, "snippet": "S"}]})
+
+
+def _perplexity(request: Request) -> Reply:
+    if request.target.endswith("/search"):
+        return _PERPLEXITY_SEARCH(request)
+    return _embedding_list("pplx-embed-v1-0.6b")
+
+
+def _topaz(request: Request) -> Reply:
+    return Reply(body=f"enhanced {marker_of(request)}".encode(), content_type="image/jpeg")
+
+
+_BY_PREFIX: Final[Mapping[str, Callable[[Request], Reply]]] = MappingProxyType(
+    {
+        "ds": _deepseek,
+        "hk": _openai_chat,
+        "gr": _openai_chat,
+        "gragent": _openai_chat,
+        "ms": _openai_chat,
+        "dc": _openai_chat,
+        "qc": _openai_chat,
+        "qp": _openai_chat,
+        "mo": _openai_chat,
+        "gc": _gigachat,
+        "gcenv": _gigachat,
+        "ol": _ollama,
+        "jn": lambda _: _embedding_list("jina-embeddings-v3"),
+        "vy": _voyage,
+        "px": _perplexity,
+        "pxenv": _PERPLEXITY_SEARCH,
+        "ve": lambda _: _embedding_list("doubao-embedding"),
+        "tv": _search(lambda url: {"results": [{"title": "T", "url": url, "content": "S"}]}),
+        "ex": _search(lambda url: {"results": [{"title": "T", "url": url, "text": "S"}]}),
+        "fc": _search(lambda url: {"data": {"web": [{"title": "T", "url": url, "description": "S"}]}}),
+        "lk": _search(lambda url: {"results": [{"type": "text", "name": "T", "url": url, "content": "S"}]}),
+        "cr": _search(lambda url: {"success": True, "data": [{"title": "T", "url": url, "description": "S"}]}),
+        "tz": _topaz,
+    }
+)
+
+
+def provider_peer(request: Request) -> Reply:
+    return _BY_PREFIX[request.target.split("/")[1]](request)
+
+
+def assert_one_request(received: tuple[Request, ...], target: str, marker: str, keyed: bool = True) -> Request:
+    assert [(request.method, request.target) for request in received] == [("POST", target)], [
+        (request.method, request.target[:200]) for request in received
+    ]
+    (request,) = received
+    assert marker.encode() in request.body, request.body[:500]
+    assert not keyed or any(KEY in value for value in request.headers.values()), sorted(request.headers)
+    return request
+
+
+def of_marker(received: tuple[Request, ...], marker: str) -> tuple[Request, ...]:
+    return tuple(request for request in received if marker_of(request) == marker)
+
+
+def body_of(response: httpx.Response) -> dict[str, JsonValue]:
+    assert response.status_code == 200, response.text
+    return _JSON_OBJECT.validate_json(response.content)
+
+
+def first_item(value: JsonValue) -> dict[str, JsonValue]:
+    return object_value(list_value(value)[0])
+
+
+def deployment(scenario: Scenario, row: Row, api_base: str | None) -> str:
+    if row.token:
+        return scenario.model(model=row.model, api_base=api_base, api_key=KEY, gigachat_access_token=KEY)
+    if row.drops_params:
+        return scenario.model(model=row.model, api_base=api_base, api_key=KEY, drop_params=True)
+    return scenario.model(model=row.model, api_base=api_base, api_key=KEY)
+
+
+def assert_control_answers(gateway: Gateway, scenario: Scenario) -> None:
+    assert first_item(gateway.chat(scenario.model())["choices"])["finish_reason"] == "stop"

tests/integration/providers/test_api_base_trailing_slash_chaos.py

@@ -0,0 +1,144 @@
+import asyncio
+from collections.abc import Awaitable, Callable
+from typing import Final
+
+import httpx
+import openai
+from integration._support.client import Gateway, Scenario, object_value
+from integration._support.wire import Reply, Request, wire_server
+from integration.providers._api_base_trailing_slash_support import (
+    KEY,
+    VECTOR,
+    assert_control_answers,
+    assert_one_request,
+    body_of,
+    marker_of,
+    new_marker,
+    of_marker,
+    provider_peer,
+    reply_text,
+    result_url,
+)
+from pydantic import JsonValue
+
+_PER_KIND: Final = 6
+
+
+def _async_openai(gateway: Gateway) -> openai.AsyncOpenAI:
+    return openai.AsyncOpenAI(base_url=str(gateway.client.base_url) + "/v1", api_key=gateway.key, max_retries=0)
+
+
+def _search_tool(gateway: Gateway, scenario: Scenario, api_base: str) -> str:
+    name: Final = f"slash-tool-{new_marker()}"
+    created: Final = gateway.post(
+        "/search_tools",
+        {
+            "search_tool": {
+                "search_tool_name": name,
+                "litellm_params": {"search_provider": "tavily", "api_key": KEY, "api_base": api_base},
+            }
+        },
+    )
+    scenario.cleanups.callback(gateway.request, "DELETE", f"/search_tools/{created['search_tool_id']}")
+    return name
+
+
+async def _post_json(gateway: Gateway, path: str, body: dict[str, JsonValue]) -> dict[str, JsonValue]:
+    async with httpx.AsyncClient(base_url=str(gateway.client.base_url), timeout=60, trust_env=False) as client:
+        response: Final = await client.post(path, json=body, headers={"Authorization": f"Bearer {gateway.key}"})
+    return body_of(response)
+
+
+def test_mixed_burst_on_slash_bases_lands_each_call_once_on_its_single_slash_url(gateway: Gateway) -> None:
+    with wire_server(provider_peer) as wire, gateway.scenario() as scenario:
+        deepseek: Final = scenario.model(model="deepseek/deepseek-chat", api_base=f"{wire.url}/ds/", api_key=KEY)
+        ollama: Final = scenario.model(model="ollama_chat/llama3.1", api_base=f"{wire.url}/ol/", api_key=KEY)
+        jina: Final = scenario.model(model="jina_ai/jina-embeddings-v3", api_base=f"{wire.url}/jn/v1/", api_key=KEY)
+        tool: Final = _search_tool(gateway, scenario, f"{wire.url}/tv/")
+
+        async def chat(model: str, marker: str) -> bool:
+            client: Final = _async_openai(gateway)
+            try:
+                completion: Final = await client.chat.completions.create(
+                    model=model, messages=[{"role": "user", "content": marker}]
+                )
+            finally:
+                await client.close()
+            return completion.choices[0].message.content == reply_text(marker)
+
+        async def streamed(model: str, marker: str) -> bool:
+            client: Final = _async_openai(gateway)
+            try:
+                stream: Final = await client.chat.completions.create(
+                    model=model, messages=[{"role": "user", "content": marker}], stream=True
+                )
+                parts: Final = [chunk.choices[0].delta.content or "" async for chunk in stream if chunk.choices]
+            finally:
+                await client.close()
+            return "".join(parts) == reply_text(marker)
+
+        async def embed(marker: str) -> bool:
+            payload: Final = await _post_json(gateway, "/v1/embeddings", {"model": jina, "input": marker})
+            return payload["data"] == [{"object": "embedding", "index": 0, "embedding": list(VECTOR)}]
+
+        async def search(marker: str) -> bool:
+            payload: Final = await _post_json(gateway, f"/v1/search/{tool}", {"query": marker})
+            results: Final = payload["results"]
+            assert isinstance(results, list)
+            return [object_value(result)["url"] for result in results] == [result_url(marker)]
+
+        kinds: Final[tuple[tuple[str, Callable[[str], Awaitable[bool]]], ...]] = (
+            ("/ds/chat/completions", lambda marker: chat(deepseek, marker)),
+            ("/ds/chat/completions", lambda marker: streamed(deepseek, marker)),
+            ("/ol/api/chat", lambda marker: chat(ollama, marker)),
+            ("/ol/api/chat", lambda marker: streamed(ollama, marker)),
+            ("/jn/v1/embeddings", embed),
+            ("/tv/search", search),
+        )
+        calls: Final = tuple((target, call, new_marker()) for target, call in kinds for _ in range(_PER_KIND))
+
+        async def burst() -> list[bool]:
+            return await asyncio.gather(*(call(marker) for _, call, marker in calls))
+
+        assert asyncio.run(burst()) == [True] * len(calls)
+        received: Final = wire.drain()
+    assert len(received) == len(calls), sorted(request.target for request in received)
+    for target, _, marker in calls:
+        assert_one_request(of_marker(received, marker), target, marker, keyed=not target.startswith("/ol/"))
+
+
+def test_upstream_dropping_doomed_connections_mid_burst_fails_only_those_calls(gateway: Gateway) -> None:
+    markers: Final = tuple(new_marker() for _ in range(12))
+    doomed: Final = frozenset(markers[::2])
+
+    def respond(request: Request) -> Reply:
+        return Reply(drop_connection=True) if marker_of(request) in doomed else provider_peer(request)
+
+    with wire_server(respond) as wire, gateway.scenario() as scenario:
+        model: Final = scenario.model(model="deepseek/deepseek-chat", api_base=f"{wire.url}/ds/", api_key=KEY)
+
+        async def call(marker: str) -> tuple[int, str]:
+            async with httpx.AsyncClient(base_url=str(gateway.client.base_url), timeout=60, trust_env=False) as client:
+                response: Final = await client.post(
+                    "/v1/chat/completions",
+                    json={"model": model, "messages": [{"role": "user", "content": marker}]},
+                    headers={"Authorization": f"Bearer {gateway.key}"},
+                )
+            return response.status_code, response.text
+
+        async def burst() -> list[tuple[int, str]]:
+            return await asyncio.gather(*(call(marker) for marker in markers))
+
+        answers: Final = dict(zip(markers, asyncio.run(burst()), strict=True))
+        received: Final = wire.drain()
+        assert_control_answers(gateway, scenario)
+    for marker in doomed:
+        _assert_dropped(answers[marker], of_marker(received, marker))
+    for marker in frozenset(markers) - doomed:
+        assert answers[marker][0] == 200 and reply_text(marker) in answers[marker][1], answers[marker]
+        assert_one_request(of_marker(received, marker), "/ds/chat/completions", marker)
+
+
+def _assert_dropped(answer: tuple[int, str], attempts: tuple[Request, ...]) -> None:
+    assert answer[0] >= 500 and "error" in answer[1], answer
+    assert attempts and {request.target for request in attempts} == {"/ds/chat/completions"}, attempts

tests/integration/providers/test_api_base_trailing_slash_owned_proxy.py

@@ -0,0 +1,207 @@
+import base64
+from collections.abc import Iterator
+from typing import Final
+
+import openai
+import pytest
+from integration._support.client import Gateway, eventually, gateway_from_environment, list_value, object_value
+from integration._support.process import OwnedProxy, graceful_stop_seconds, owned_proxy_process
+from integration._support.wire import Wire, wire_server
+from integration.providers._api_base_trailing_slash_support import (
+    KEY,
+    TRAILING,
+    TRAILING_IDS,
+    VECTOR,
+    Row,
+    assert_one_request,
+    body_of,
+    deployment,
+    file_id,
+    first_item,
+    new_marker,
+    of_marker,
+    provider_peer,
+    reply_text,
+    request_json,
+    result_url,
+)
+
+pytestmark: Final = pytest.mark.timeout(2 * graceful_stop_seconds() + 120)
+
+_FLAG_ROWS: Final = (
+    Row("moonshot/kimi-k2-0905-preview", "/ms/v1", "/chat/completions"),
+    Row("dashscope/qwen-plus", "/dc/compatible-mode/v1", "/chat/completions"),
+    Row("qwencloud/qwen-plus", "/qc/compatible-mode/v1", "/chat/completions"),
+    Row("qwen_ai_platform/qwen-plus", "/qp/compatible-mode/v1", "/chat/completions"),
+    Row("modelscope/Qwen/Qwen3-8B", "/mo/v1", "/chat/completions"),
+)
+_MOONSHOT: Final = _FLAG_ROWS[0]
+_PNG_HEADER: Final = b"\x89PNG\r\n\x1a\n"
+
+
+def _overrides(wire_url: str) -> dict[str, str]:
+    return {
+        "EXPERIMENTAL_OPENAI_BASE_LLM_HTTP_HANDLER": "true",
+        "GIGACHAT_ACCESS_TOKEN": KEY,
+        "GIGACHAT_API_BASE": f"{wire_url}/gcenv/v1/",
+        "PERPLEXITY_API_BASE": f"{wire_url}/pxenv/",
+        "GRADIENT_AI_AGENT_ENDPOINT": f"{wire_url}/gragent/",
+    }
+
+
+@pytest.fixture(scope="module")
+def shared_wire() -> Iterator[Wire]:
+    with wire_server(provider_peer) as wire:
+        yield wire
+
+
+@pytest.fixture(scope="module")
+def owned(shared_wire: Wire, tmp_path_factory: pytest.TempPathFactory) -> Iterator[OwnedProxy]:
+    directory: Final = tmp_path_factory.mktemp("api-base-trailing-slash")
+    with gateway_from_environment() as gateway:
+        with owned_proxy_process(gateway, directory, _overrides(shared_wire.url), workers=2) as proxy:
+            yield proxy
+
+
+def _openai(gateway: Gateway) -> openai.OpenAI:
+    return openai.OpenAI(base_url=str(gateway.client.base_url) + "/v1", api_key=gateway.key, max_retries=0)
+
+
+def _chat(gateway: Gateway, model: str, marker: str) -> str | None:
+    completion: Final = _openai(gateway).chat.completions.create(
+        model=model, messages=[{"role": "user", "content": marker}]
+    )
+    return completion.choices[0].message.content
+
+
+@pytest.mark.parametrize("trailing", TRAILING, ids=TRAILING_IDS)
+@pytest.mark.parametrize("row", _FLAG_ROWS, ids=tuple(row.model for row in _FLAG_ROWS))
+def test_flag_gated_chat_reaches_one_single_slash_url(
+    owned: OwnedProxy, shared_wire: Wire, row: Row, trailing: str
+) -> None:
+    marker: Final = new_marker()
+    with owned.gateway.scenario() as scenario:
+        model: Final = deployment(scenario, row, f"{shared_wire.url}{row.base}{trailing}")
+        assert _chat(owned.gateway, model, marker) == reply_text(marker)
+    assert_one_request(of_marker(shared_wire.drain(), marker), row.target, marker)
+
+
+@pytest.mark.parametrize("trailing", TRAILING, ids=TRAILING_IDS)
+def test_flag_gated_streamed_chat_reaches_one_single_slash_url(
+    owned: OwnedProxy, shared_wire: Wire, trailing: str
+) -> None:
+    marker: Final = new_marker()
+    with owned.gateway.scenario() as scenario:
+        model: Final = deployment(scenario, _MOONSHOT, f"{shared_wire.url}{_MOONSHOT.base}{trailing}")
+        stream: Final = _openai(owned.gateway).chat.completions.create(
+            model=model, messages=[{"role": "user", "content": marker}], stream=True
+        )
+        text: Final = "".join(chunk.choices[0].delta.content or "" for chunk in stream if chunk.choices)
+        assert text == reply_text(marker)
+    forwarded: Final = assert_one_request(of_marker(shared_wire.drain(), marker), _MOONSHOT.target, marker)
+    assert request_json(forwarded)["stream"] is True
+
+
+@pytest.mark.parametrize("trailing", TRAILING, ids=TRAILING_IDS)
+def test_gigachat_image_upload_and_chat_reach_single_slash_urls(
+    owned: OwnedProxy, shared_wire: Wire, trailing: str
+) -> None:
+    marker: Final = new_marker()
+    image: Final = base64.b64encode(_PNG_HEADER + marker.encode()).decode()
+    with owned.gateway.scenario() as scenario:
+        model: Final = scenario.model(
+            model="gigachat/GigaChat-Pro", api_base=f"{shared_wire.url}/gc/v1{trailing}", api_key=KEY
+        )
+        completion: Final = _openai(owned.gateway).chat.completions.create(
+            model=model,
+            messages=[
+                {
+                    "role": "user",
+                    "content": [
+                        {"type": "text", "text": marker},
+                        {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{image}"}},
+                    ],
+                }
+            ],
+        )
+        assert completion.choices[0].message.content == reply_text(marker)
+    upload, chat = of_marker(shared_wire.drain(), marker)
+    assert (upload.method, upload.target) == ("POST", "/gc/v1/files"), upload.target
+    assert upload.headers["authorization"] == f"Bearer {KEY}", sorted(upload.headers)
+    assert (chat.method, chat.target) == ("POST", "/gc/v1/chat/completions"), chat.target
+    messages: Final = request_json(chat)["messages"]
+    assert isinstance(messages, list)
+    assert object_value(messages[-1])["attachments"] == [file_id(upload)], chat.body
+
+
+@pytest.mark.parametrize("trailing", TRAILING, ids=TRAILING_IDS)
+def test_gigachat_embedding_reaches_one_single_slash_url(owned: OwnedProxy, shared_wire: Wire, trailing: str) -> None:
+    marker: Final = new_marker()
+    with owned.gateway.scenario() as scenario:
+        model: Final = scenario.model(
+            model="gigachat/Embeddings", api_base=f"{shared_wire.url}/gc/v1{trailing}", api_key=KEY
+        )
+        response: Final = owned.gateway.request("POST", "/v1/embeddings", {"model": model, "input": marker})
+        assert first_item(body_of(response)["data"])["embedding"] == list(VECTOR)
+    assert_one_request(of_marker(shared_wire.drain(), marker), "/gc/v1/embeddings", marker)
+
+
+def test_perplexity_search_with_no_api_base_uses_the_env_base_without_a_double_slash(
+    owned: OwnedProxy, shared_wire: Wire
+) -> None:
+    marker: Final = new_marker()
+    name: Final = f"slash-tool-{new_marker()}"
+    with owned.gateway.scenario() as scenario:
+        created: Final = owned.gateway.post(
+            "/search_tools",
+            {
+                "search_tool": {
+                    "search_tool_name": name,
+                    "litellm_params": {"search_provider": "perplexity", "api_key": KEY},
+                }
+            },
+        )
+        scenario.cleanups.callback(owned.gateway.request, "DELETE", f"/search_tools/{created['search_tool_id']}")
+        response: Final = owned.gateway.request("POST", f"/v1/search/{name}", {"query": marker})
+        assert [object_value(result)["url"] for result in list_value(body_of(response)["results"])] == [
+            result_url(marker)
+        ]
+    assert_one_request(of_marker(shared_wire.drain(), marker), "/pxenv/search", marker)
+
+
+def test_gradient_with_no_api_base_uses_the_agent_endpoint_env_without_a_double_slash(
+    owned: OwnedProxy, shared_wire: Wire
+) -> None:
+    marker: Final = new_marker()
+    with owned.gateway.scenario() as scenario:
+        model: Final = scenario.model(
+            model="gradient_ai/llama3.3-70b-instruct", api_base=None, api_key=KEY, drop_params=True
+        )
+        assert _chat(owned.gateway, model, marker) == reply_text(marker)
+    assert_one_request(of_marker(shared_wire.drain(), marker), "/gragent/api/v1/chat/completions", marker)
+
+
+def test_gigachat_pass_through_without_a_model_joins_the_env_base_once(owned: OwnedProxy, shared_wire: Wire) -> None:
+    marker: Final = new_marker()
+    response: Final = owned.gateway.request("POST", "/gigachat/models", {"probe": marker})
+    assert body_of(response)["data"] == [{"id": marker, "object": "model"}], response.text
+    assert_one_request(of_marker(shared_wire.drain(), marker), "/gcenv/v1/models", marker)
+
+
+def test_worker_kill_leaves_the_survivor_and_the_replacement_building_single_slash_urls(
+    owned: OwnedProxy, shared_wire: Wire
+) -> None:
+    victim: Final = min(eventually(owned.worker_pids, lambda pids: len(pids) == 2, seconds=60))
+    with owned.gateway.scenario() as scenario:
+        model: Final = deployment(scenario, _MOONSHOT, f"{shared_wire.url}{_MOONSHOT.base}/")
+        with owned.replacing_worker(victim) as kill:
+            kill()
+            during: Final = tuple(new_marker() for _ in range(4))
+            assert [_chat(owned.gateway, model, marker) for marker in during] == [reply_text(m) for m in during]
+        after: Final = tuple(new_marker() for _ in range(8))
+        assert [_chat(owned.gateway, model, marker) for marker in after] == [reply_text(m) for m in after]
+    received: Final = shared_wire.drain()
+    for marker in (*during, *after):
+        assert_one_request(of_marker(received, marker), _MOONSHOT.target, marker)
+    assert victim not in owned.worker_pids()
+    assert len(owned.worker_pids()) == 2

tests/integration/providers/test_api_base_trailing_slash_wire.py

@@ -0,0 +1,320 @@
+from typing import Final
+
+import anthropic
+import openai
+import pytest
+from integration._support.client import Gateway, Scenario, list_value, object_value
+from integration._support.wire import Reply, Request, wire_server
+from integration.providers._api_base_trailing_slash_support import (
+    KEY,
+    TRAILING,
+    TRAILING_IDS,
+    VECTOR,
+    Row,
+    assert_control_answers,
+    assert_one_request,
+    body_of,
+    deployment,
+    first_item,
+    new_marker,
+    provider_peer,
+    reply_text,
+    request_json,
+    result_url,
+)
+from pydantic import JsonValue
+
+_CHAT_ROWS: Final = (
+    Row("deepseek/deepseek-chat", "/ds", "/chat/completions"),
+    Row("heroku/claude-4-sonnet", "/hk", "/v1/chat/completions"),
+    Row("gradient_ai/llama3.3-70b-instruct", "/gr", "/api/v1/chat/completions", drops_params=True),
+    Row("gigachat/GigaChat", "/gc/v1", "/chat/completions", token=True),
+    Row("ollama_chat/llama3.1", "/ol", "/api/chat"),
+    Row("ollama/llama3.1", "/ol", "/api/generate"),
+)
+_BRIDGED_ROWS: Final = (
+    Row("deepseek/deepseek-chat", "/ds", "/chat/completions"),
+    Row("ollama_chat/llama3.1", "/ol", "/api/chat"),
+)
+_MESSAGES_ROWS: Final = (
+    Row("deepseek/deepseek-chat", "/ds", "/anthropic/v1/messages"),
+    Row("heroku/claude-4-sonnet", "/hk", "/v1/chat/completions"),
+    Row("ollama_chat/llama3.1", "/ol", "/api/chat"),
+)
+_EMBEDDING_ROWS: Final = (
+    Row("ollama/nomic-embed-text", "/ol", "/api/embed", keyed=False),
+    Row("jina_ai/jina-embeddings-v3", "/jn/v1", "/embeddings"),
+    Row("voyage/voyage-3", "/vy/v1", "/embeddings"),
+    Row("voyage/voyage-multimodal-3", "/vy/v1", "/multimodalembeddings"),
+    Row("perplexity/pplx-embed-v1-0.6b", "/px", "/v1/embeddings"),
+    Row("volcengine/doubao-embedding", "/ve/api/v3", "/embeddings"),
+)
+_SEARCH_ROWS: Final = (
+    Row("perplexity", "/px", "/search"),
+    Row("tavily", "/tv", "/search"),
+    Row("exa_ai", "/ex", "/search"),
+    Row("firecrawl", "/fc", "/search"),
+    Row("linkup", "/lk", "/search"),
+    Row("fastcrw", "/cr", "/search"),
+)
+_FULL_ROUTE_ROWS: Final = (
+    Row("deepseek/deepseek-chat", "/ds/chat/completions", ""),
+    Row("ollama_chat/llama3.1", "/ol/api/chat", ""),
+)
+_FULL_ROUTE_EMBEDDING_ROWS: Final = (
+    Row("jina_ai/jina-embeddings-v3", "/jn/v1/embeddings", ""),
+    Row("voyage/voyage-3", "/vy/v1/embeddings", ""),
+    Row("perplexity/pplx-embed-v1-0.6b", "/px/v1/embeddings", ""),
+)
+
+
+def _ids(rows: tuple[Row, ...]) -> tuple[str, ...]:
+    return tuple(f"{row.model}{row.route}" for row in rows)
+
+
+def _openai(gateway: Gateway) -> openai.OpenAI:
+    return openai.OpenAI(base_url=str(gateway.client.base_url) + "/v1", api_key=gateway.key, max_retries=0)
+
+
+def _chat(gateway: Gateway, model: str, marker: str) -> str | None:
+    completion: Final = _openai(gateway).chat.completions.create(
+        model=model, messages=[{"role": "user", "content": marker}]
+    )
+    return completion.choices[0].message.content
+
+
+def _embedding(gateway: Gateway, model: str, marker: str) -> JsonValue:
+    response: Final = gateway.request("POST", "/v1/embeddings", {"model": model, "input": marker})
+    return first_item(body_of(response)["data"])["embedding"]
+
+
+def _search_tool(gateway: Gateway, scenario: Scenario, provider: str, api_base: str) -> str:
+    name: Final = f"slash-tool-{new_marker()}"
+    created: Final = gateway.post(
+        "/search_tools",
+        {
+            "search_tool": {
+                "search_tool_name": name,
+                "litellm_params": {"search_provider": provider, "api_key": KEY, "api_base": api_base},
+            }
+        },
+    )
+    scenario.cleanups.callback(gateway.request, "DELETE", f"/search_tools/{created['search_tool_id']}")
+    return name
+
+
+@pytest.mark.parametrize("trailing", TRAILING, ids=TRAILING_IDS)
+@pytest.mark.parametrize("row", _CHAT_ROWS, ids=_ids(_CHAT_ROWS))
+def test_chat_completion_reaches_one_single_slash_url(gateway: Gateway, row: Row, trailing: str) -> None:
+    marker: Final = new_marker()
+    with wire_server(provider_peer) as wire, gateway.scenario() as scenario:
+        model: Final = deployment(scenario, row, f"{wire.url}{row.base}{trailing}")
+        assert _chat(gateway, model, marker) == reply_text(marker)
+        assert_one_request(wire.drain(), row.target, marker)
+
+
+@pytest.mark.parametrize("trailing", TRAILING, ids=TRAILING_IDS)
+@pytest.mark.parametrize("row", _BRIDGED_ROWS, ids=_ids(_BRIDGED_ROWS))
+def test_streamed_chat_reaches_one_single_slash_url(gateway: Gateway, row: Row, trailing: str) -> None:
+    marker: Final = new_marker()
+    with wire_server(provider_peer) as wire, gateway.scenario() as scenario:
+        model: Final = deployment(scenario, row, f"{wire.url}{row.base}{trailing}")
+        stream: Final = _openai(gateway).chat.completions.create(
+            model=model, messages=[{"role": "user", "content": marker}], stream=True
+        )
+        text: Final = "".join(chunk.choices[0].delta.content or "" for chunk in stream if chunk.choices)
+        assert text == reply_text(marker)
+        assert request_json(assert_one_request(wire.drain(), row.target, marker))["stream"] is True
+
+
+@pytest.mark.parametrize("trailing", TRAILING, ids=TRAILING_IDS)
+@pytest.mark.parametrize("row", _MESSAGES_ROWS, ids=_ids(_MESSAGES_ROWS))
+def test_messages_reaches_one_single_slash_url(gateway: Gateway, row: Row, trailing: str) -> None:
+    marker: Final = new_marker()
+    with wire_server(provider_peer) as wire, gateway.scenario() as scenario:
+        model: Final = deployment(scenario, row, f"{wire.url}{row.base}{trailing}")
+        client: Final = anthropic.Anthropic(base_url=str(gateway.client.base_url), api_key=gateway.key, max_retries=0)
+        message: Final = client.messages.create(
+            model=model, max_tokens=16, messages=[{"role": "user", "content": marker}]
+        )
+        assert [block.text for block in message.content if block.type == "text"] == [reply_text(marker)]
+        assert_one_request(wire.drain(), row.target, marker)
+
+
+@pytest.mark.parametrize("trailing", TRAILING, ids=TRAILING_IDS)
+@pytest.mark.parametrize("row", _BRIDGED_ROWS, ids=_ids(_BRIDGED_ROWS))
+def test_responses_bridge_reaches_one_single_slash_url(gateway: Gateway, row: Row, trailing: str) -> None:
+    marker: Final = new_marker()
+    with wire_server(provider_peer) as wire, gateway.scenario() as scenario:
+        model: Final = deployment(scenario, row, f"{wire.url}{row.base}{trailing}")
+        response: Final = _openai(gateway).responses.create(model=model, input=marker)
+        assert response.output_text == reply_text(marker)
+        assert_one_request(wire.drain(), row.target, marker)
+
+
+@pytest.mark.parametrize("trailing", TRAILING, ids=TRAILING_IDS)
+@pytest.mark.parametrize("row", _EMBEDDING_ROWS, ids=_ids(_EMBEDDING_ROWS))
+def test_embedding_reaches_one_single_slash_url(gateway: Gateway, row: Row, trailing: str) -> None:
+    marker: Final = new_marker()
+    with wire_server(provider_peer) as wire, gateway.scenario() as scenario:
+        model: Final = deployment(scenario, row, f"{wire.url}{row.base}{trailing}")
+        assert _embedding(gateway, model, marker) == list(VECTOR)
+        assert_one_request(wire.drain(), row.target, marker, keyed=row.keyed)
+
+
+@pytest.mark.parametrize("trailing", TRAILING, ids=TRAILING_IDS)
+def test_voyage_contextual_embedding_reaches_one_single_slash_url(gateway: Gateway, trailing: str) -> None:
+    marker: Final = new_marker()
+    with wire_server(provider_peer) as wire, gateway.scenario() as scenario:
+        model: Final = scenario.model(
+            model="voyage/voyage-context-3", api_base=f"{wire.url}/vy/v1{trailing}", api_key=KEY
+        )
+        response: Final = gateway.request("POST", "/v1/embeddings", {"model": model, "input": [[marker, "chunk b"]]})
+        assert body_of(response)["data"] == [
+            {"object": "list", "index": 0, "data": [{"object": "embedding", "index": 0, "embedding": list(VECTOR)}]}
+        ]
+        assert_one_request(wire.drain(), "/vy/v1/contextualizedembeddings", marker)
+
+
+@pytest.mark.parametrize("trailing", TRAILING, ids=TRAILING_IDS)
+@pytest.mark.parametrize("row", _SEARCH_ROWS, ids=_ids(_SEARCH_ROWS))
+def test_search_reaches_one_single_slash_url(gateway: Gateway, row: Row, trailing: str) -> None:
+    marker: Final = new_marker()
+    with wire_server(provider_peer) as wire, gateway.scenario() as scenario:
+        tool: Final = _search_tool(gateway, scenario, row.model, f"{wire.url}{row.base}{trailing}")
+        response: Final = gateway.request("POST", f"/v1/search/{tool}", {"query": marker})
+        assert [object_value(result)["url"] for result in list_value(body_of(response)["results"])] == [
+            result_url(marker)
+        ]
+        assert_one_request(wire.drain(), row.target, marker)
+
+
+@pytest.mark.parametrize("trailing", TRAILING, ids=TRAILING_IDS)
+def test_gigachat_pass_through_with_a_router_model_reaches_one_single_slash_url(
+    gateway: Gateway, trailing: str
+) -> None:
+    marker: Final = new_marker()
+    with wire_server(provider_peer) as wire, gateway.scenario() as scenario:
+        model: Final = scenario.model(
+            model="gigachat/GigaChat", api_base=f"{wire.url}/gc/v1{trailing}", api_key=KEY, gigachat_access_token=KEY
+        )
+        response: Final = gateway.request(
+            "POST", "/gigachat/chat/completions", {"model": model, "messages": [{"role": "user", "content": marker}]}
+        )
+        assert first_item(body_of(response)["choices"])["message"] == {
+            "role": "assistant",
+            "content": reply_text(marker),
+        }
+        forwarded: Final = assert_one_request(wire.drain(), "/gc/v1/chat/completions", marker)
+        assert request_json(forwarded)["model"] == "GigaChat", forwarded.body
+
+
+@pytest.mark.parametrize(
+    ("row", "trailing"),
+    (
+        (Row("deepseek/deepseek-chat", "/ds", "/chat/completions"), "//"),
+        (Row("deepseek/deepseek-chat", f"/ds/{'p' * 5000}", "/chat/completions"), "/"),
+    ),
+    ids=("two-slashes", "five-kb-prefix"),
+)
+def test_chat_base_with_repeated_slashes_or_a_long_prefix_reaches_one_url(
+    gateway: Gateway, row: Row, trailing: str
+) -> None:
+    marker: Final = new_marker()
+    with wire_server(provider_peer) as wire, gateway.scenario() as scenario:
+        model: Final = deployment(scenario, row, f"{wire.url}{row.base}{trailing}")
+        assert _chat(gateway, model, marker) == reply_text(marker)
+        assert_one_request(wire.drain(), row.target, marker)
+
+
+def test_embedding_and_search_bases_ending_in_two_slashes_reach_one_url(gateway: Gateway) -> None:
+    embed_marker: Final = new_marker()
+    search_marker: Final = new_marker()
+    with wire_server(provider_peer) as wire, gateway.scenario() as scenario:
+        model: Final = scenario.model(model="jina_ai/jina-embeddings-v3", api_base=f"{wire.url}/jn/v1//", api_key=KEY)
+        assert _embedding(gateway, model, embed_marker) == list(VECTOR)
+        assert_one_request(wire.drain(), "/jn/v1/embeddings", embed_marker)
+        tool: Final = _search_tool(gateway, scenario, "tavily", f"{wire.url}/tv//")
+        response: Final = gateway.request("POST", f"/v1/search/{tool}", {"query": search_marker})
+        assert response.status_code == 200, response.text
+        assert_one_request(wire.drain(), "/tv/search", search_marker)
+
+
+@pytest.mark.parametrize("api_base", (12345, ["http://127.0.0.1:9/ds/"]), ids=("int", "list"))
+def test_non_string_api_base_is_rejected_and_other_deployments_keep_answering(
+    gateway: Gateway, api_base: JsonValue
+) -> None:
+    with gateway.scenario() as scenario:
+        response: Final = gateway.request(
+            "POST",
+            "/model/new",
+            {
+                "model_name": f"integration-{new_marker()}",
+                "litellm_params": {"model": "deepseek/deepseek-chat", "api_key": KEY, "api_base": api_base},
+            },
+        )
+        assert response.status_code == 422, response.text
+        assert "api_base" in response.text, response.text
+        assert_control_answers(gateway, scenario)
+
+
+@pytest.mark.parametrize(
+    ("row", "status"),
+    (
+        (Row("deepseek/deepseek-chat", "/ds", "/chat/completions"), 401),
+        (Row("jina_ai/jina-embeddings-v3", "/jn/v1", "/embeddings"), 404),
+    ),
+    ids=("deepseek-401", "jina-404"),
+)
+def test_upstream_error_on_a_slash_base_reaches_the_caller_from_one_url(
+    gateway: Gateway, row: Row, status: int
+) -> None:
+    marker: Final = new_marker()
+    error: Final = f'{{"error": {{"message": "scripted {status} {marker}", "type": "scripted"}}}}'.encode()
+
+    def respond(_: Request) -> Reply:
+        return Reply(status=status, body=error)
+
+    with wire_server(respond) as wire, gateway.scenario() as scenario:
+        model: Final = deployment(scenario, row, f"{wire.url}{row.base}/")
+        body: Final[dict[str, JsonValue]] = (
+            {"model": model, "input": marker}
+            if row.route == "/embeddings"
+            else {"model": model, "messages": [{"role": "user", "content": marker}]}
+        )
+        path: Final = "/v1/embeddings" if row.route == "/embeddings" else "/v1/chat/completions"
+        response: Final = gateway.request("POST", path, body)
+        assert response.status_code == status, response.text
+        assert marker in response.text, response.text
+        assert_one_request(wire.drain(), row.target, marker)
+        assert_control_answers(gateway, scenario)
+
+
+@pytest.mark.parametrize("trailing", TRAILING, ids=TRAILING_IDS)
+@pytest.mark.parametrize("row", _FULL_ROUTE_ROWS, ids=_ids(_FULL_ROUTE_ROWS))
+def test_chat_base_that_already_names_the_route_is_sent_as_is(gateway: Gateway, row: Row, trailing: str) -> None:
+    marker: Final = new_marker()
+    with wire_server(provider_peer) as wire, gateway.scenario() as scenario:
+        model: Final = deployment(scenario, row, f"{wire.url}{row.base}{trailing}")
+        assert _chat(gateway, model, marker) == reply_text(marker)
+        assert_one_request(wire.drain(), row.target, marker)
+
+
+@pytest.mark.parametrize("trailing", TRAILING, ids=TRAILING_IDS)
+@pytest.mark.parametrize("row", _FULL_ROUTE_EMBEDDING_ROWS, ids=_ids(_FULL_ROUTE_EMBEDDING_ROWS))
+def test_embedding_base_that_already_names_the_route_is_sent_as_is(gateway: Gateway, row: Row, trailing: str) -> None:
+    marker: Final = new_marker()
+    with wire_server(provider_peer) as wire, gateway.scenario() as scenario:
+        model: Final = deployment(scenario, row, f"{wire.url}{row.base}{trailing}")
+        assert _embedding(gateway, model, marker) == list(VECTOR)
+        assert_one_request(wire.drain(), row.target, marker)
+
+
+@pytest.mark.parametrize("trailing", TRAILING, ids=TRAILING_IDS)
+def test_search_base_that_already_names_the_route_is_sent_as_is(gateway: Gateway, trailing: str) -> None:
+    marker: Final = new_marker()
+    with wire_server(provider_peer) as wire, gateway.scenario() as scenario:
+        tool: Final = _search_tool(gateway, scenario, "tavily", f"{wire.url}/tv/search{trailing}")
+        response: Final = gateway.request("POST", f"/v1/search/{tool}", {"query": marker})
+        assert response.status_code == 200, response.text
+        assert_one_request(wire.drain(), "/tv/search", marker)

tests/integration/sdk/test_topaz_image_variation_trailing_slash_sdk.py

@@ -0,0 +1,32 @@
+import base64
+import uuid
+from typing import Final
+
+import litellm
+import pytest
+from integration._support.wire import Reply, Request, wire_server
+
+_KEY: Final = "synthetic-topaz-key"
+
+
+@pytest.mark.parametrize("trailing", ("", "/"), ids=("no-slash", "slash"))
+def test_topaz_image_variation_reaches_one_single_slash_url(trailing: str) -> None:
+    marker: Final = f"topaz-{uuid.uuid4().hex}".encode()
+    enhanced: Final = b"enhanced " + marker
+
+    def respond(_: Request) -> Reply:
+        return Reply(body=enhanced, content_type="image/jpeg")
+
+    with wire_server(respond) as wire:
+        response: Final = litellm.image_variation(
+            image=b"\x89PNG\r\n\x1a\n" + marker,
+            model="topaz/Standard V2",
+            api_key=_KEY,
+            api_base=f"{wire.url}/tz{trailing}",
+        )
+        received: Final = wire.drain()
+    assert [(request.method, request.target) for request in received] == [("POST", "/tz/image/v1/enhance")]
+    assert marker in received[0].body
+    assert received[0].headers["x-api-key"] == _KEY, sorted(received[0].headers)
+    assert response.data is not None
+    assert [image.b64_json for image in response.data] == [base64.b64encode(enhanced).decode()]

tests/unit/llms/dashscope/test_dashscope_chat_transformation.py

@@ -198,3 +198,20 @@ class TestDashScopeConfig:
         )
 
         assert transformed_tools[0].get("cache_control") == {"type": "ephemeral"}
+
+
+@pytest.mark.parametrize(
+    "api_base",
+    [
+        "https://gateway.example.com/dashscope/compatible-mode/v1",
+        "https://gateway.example.com/dashscope/compatible-mode/v1/",
+        "https://gateway.example.com/dashscope/compatible-mode/v1/chat/completions/",
+    ],
+)
+def test_get_complete_url_ignores_a_trailing_slash_on_api_base(api_base: str) -> None:
+    assert (
+        DashScopeChatConfig().get_complete_url(
+            api_base=api_base, api_key=None, model="qwen-plus", optional_params={}, litellm_params={}
+        )
+        == "https://gateway.example.com/dashscope/compatible-mode/v1/chat/completions"
+    )

tests/unit/llms/deepseek/chat/test_deepseek_chat_transformation.py

@@ -890,3 +890,20 @@ def test_deepseek_fill_reasoning_content_guard_in_transform_request():
     assert "reasoning_content" not in result["messages"][1], (
         "reasoning_content should not be injected for non-reasoning models"
     )
+
+
+@pytest.mark.parametrize(
+    "api_base",
+    [
+        "https://gateway.example.com/deepseek",
+        "https://gateway.example.com/deepseek/",
+        "https://gateway.example.com/deepseek/chat/completions/",
+    ],
+)
+def test_get_complete_url_ignores_a_trailing_slash_on_api_base(api_base: str):
+    assert (
+        DeepSeekChatConfig().get_complete_url(
+            api_base=api_base, api_key=None, model="deepseek-chat", optional_params={}, litellm_params={}
+        )
+        == "https://gateway.example.com/deepseek/chat/completions"
+    )

tests/unit/llms/exa_ai/search/test_transformation.py

@@ -197,3 +197,18 @@ def test_transform_search_response_rejects_malformed_result_without_naming_its_p
         _transform({"results": results})
 
     assert "429" not in str(exc_info.value)
+
+
+@pytest.mark.parametrize(
+    "api_base",
+    [
+        "https://gateway.example.com/exa",
+        "https://gateway.example.com/exa/",
+        "https://gateway.example.com/exa/search/",
+    ],
+)
+def test_get_complete_url_ignores_a_trailing_slash_on_api_base(api_base: str) -> None:
+    assert (
+        ExaAISearchConfig().get_complete_url(api_base=api_base, optional_params={})
+        == "https://gateway.example.com/exa/search"
+    )

tests/unit/llms/fastcrw/search/test_transformation.py

@@ -99,6 +99,18 @@ def test_get_complete_url_does_not_double_append():
     )
 
 
+@pytest.mark.parametrize(
+    "api_base",
+    [
+        "https://self-hosted.local/api/v1",
+        "https://self-hosted.local/api/v1/",
+        "https://self-hosted.local/api/v1/search/",
+    ],
+)
+def test_get_complete_url_ignores_a_trailing_slash_on_api_base(api_base: str) -> None:
+    assert _config().get_complete_url(api_base, {}) == "https://self-hosted.local/api/v1/search"
+
+
 def test_get_complete_url_reads_env_base():
     with patch.dict(
         os.environ, {"CRW_API_BASE": "https://env-base.local/v1"}, clear=True

tests/unit/llms/firecrawl/search/test_firecrawl_search_transformation.py

@@ -158,3 +158,18 @@ def test_firecrawl_news_entry_with_non_string_date_reports_the_date_next_to_the_
         (("title",), 7),
         (("date",), 1709596800),
     ]
+
+
+@pytest.mark.parametrize(
+    "api_base",
+    [
+        "https://gateway.example.com/firecrawl/v2",
+        "https://gateway.example.com/firecrawl/v2/",
+        "https://gateway.example.com/firecrawl/v2/search/",
+    ],
+)
+def test_get_complete_url_ignores_a_trailing_slash_on_api_base(api_base: str) -> None:
+    assert (
+        FirecrawlSearchConfig().get_complete_url(api_base=api_base, optional_params={})
+        == "https://gateway.example.com/firecrawl/v2/search"
+    )

tests/unit/llms/gigachat/chat/test_gigachat_chat_transformation.py

@@ -88,8 +88,7 @@ class TestGetCompleteUrl:
             litellm_params={},
             stream=False,
         )
-        # get_api_base passes the value through without stripping the slash
-        assert url == "https://custom.example.com//chat/completions"
+        assert url == "https://custom.example.com/chat/completions"
 
     def test_uses_api_base_from_get_api_base_when_none(self):
         url = self.config.get_complete_url(

tests/unit/llms/gigachat/embedding/test_gigachat_embedding_transformation.py

@@ -120,8 +120,7 @@ class TestGetCompleteUrl:
             api_base="https://custom.example.com/", api_key=None, model="GigaChat",
             optional_params={}, litellm_params={},
         )
-        # get_api_base doesn't strip slash, so we get double slash
-        assert url == "https://custom.example.com//embeddings"
+        assert url == "https://custom.example.com/embeddings"
 
 
 class TestTransformEmbeddingRequest:

tests/unit/llms/gigachat/passthrough/test_gigachat_passthrough_transformation.py

@@ -128,6 +128,22 @@ class TestGigaChatPassthroughConfig:
         assert str(complete_url) == "https://custom.gigachat.ru/api/v1/chat/completions"
         assert base_target_url == api_base
 
+    @pytest.mark.parametrize(
+        "api_base",
+        ["https://custom.gigachat.ru/api/v1", "https://custom.gigachat.ru/api/v1/"],
+    )
+    def test_get_complete_url_ignores_a_trailing_slash_on_api_base(self, api_base):
+        complete_url, _ = GigaChatPassthroughConfig().get_complete_url(
+            api_base=api_base,
+            api_key=None,
+            model="GigaChat",
+            endpoint="chat/completions",
+            request_query_params=None,
+            litellm_params={},
+        )
+
+        assert str(complete_url) == "https://custom.gigachat.ru/api/v1/chat/completions"
+
     @patch(  # test-quality-ok: patching litellm internal for unit test isolation
         "litellm.llms.gigachat.passthrough.transformation.get_secret_str"
     )

tests/unit/llms/gradient_ai/chat/test_gradient_ai_chat_transformation.py

@@ -95,3 +95,23 @@ def test_get_openai_compatible_provider_info_default(monkeypatch, config):
     api_base, api_key = config.get_openai_compatible_provider_info(None, None)
     assert api_base == GRADIENT_AI_SERVERLESS_ENDPOINT
     assert api_key == "env-key"
+
+
+@pytest.mark.parametrize(
+    ("api_base", "expected_url"),
+    [
+        (f"{DO_BASE_URL}/", f"{DO_BASE_URL}{DO_ENDPOINT_PATH}"),
+        (f"{GRADIENT_AI_SERVERLESS_ENDPOINT}/", f"{GRADIENT_AI_SERVERLESS_ENDPOINT}/v1/chat/completions"),
+    ],
+)
+def test_get_complete_url_ignores_a_trailing_slash_on_api_base(monkeypatch, config, api_base, expected_url):
+    monkeypatch.delenv("GRADIENT_AI_AGENT_ENDPOINT", raising=False)
+    url = config.get_complete_url(
+        api_base=api_base,
+        api_key="test-key",
+        model="gradient_ai/test-model",
+        optional_params={},
+        litellm_params={},
+        stream=False,
+    )
+    assert url == expected_url

tests/unit/llms/heroku/test_heroku_chat_transformation.py

@@ -179,3 +179,20 @@ class TestHerokuChatConfig:
         assert response.usage.prompt_tokens == 354
         assert response.usage.completion_tokens == 69
         assert response.usage.total_tokens == 423
+
+
+@pytest.mark.parametrize(
+    "api_base",
+    [
+        "https://gateway.example.com/heroku",
+        "https://gateway.example.com/heroku/",
+        "https://gateway.example.com/heroku/v1/chat/completions/",
+    ],
+)
+def test_get_complete_url_ignores_a_trailing_slash_on_api_base(api_base: str):
+    assert (
+        HerokuChatConfig().get_complete_url(
+            api_base=api_base, api_key=None, model="claude-4-sonnet", optional_params={}, litellm_params={}
+        )
+        == "https://gateway.example.com/heroku/v1/chat/completions"
+    )

tests/unit/llms/jina_ai/embedding/test_jina_embedding_transformation.py

@@ -272,3 +272,20 @@ def test_jina_ai_img_embeddings_transforms_mixed_text_and_image_inputs(
         "input": expected_payload_input,
     }
     assert response.data[0]["embedding"] == [0.1, 0.2]
+
+
+@pytest.mark.parametrize(
+    "api_base",
+    [
+        "https://gateway.example.com/jina/v1",
+        "https://gateway.example.com/jina/v1/",
+        "https://gateway.example.com/jina/v1/embeddings/",
+    ],
+)
+def test_get_complete_url_ignores_a_trailing_slash_on_api_base(api_base: str):
+    assert (
+        JinaAIEmbeddingConfig().get_complete_url(
+            api_base=api_base, api_key=None, model="jina-embeddings-v3", optional_params={}, litellm_params={}
+        )
+        == "https://gateway.example.com/jina/v1/embeddings"
+    )

tests/unit/llms/linkup/search/test_linkup_search_transformation.py

@@ -204,3 +204,18 @@ def _vcr_outcome_gate(request, vcr):
     install_live_call_probe(request, vcr)
     yield
     record_vcr_outcome(request, vcr)
+
+
+@pytest.mark.parametrize(
+    "api_base",
+    [
+        "https://gateway.example.com/linkup/v1",
+        "https://gateway.example.com/linkup/v1/",
+        "https://gateway.example.com/linkup/v1/search/",
+    ],
+)
+def test_get_complete_url_ignores_a_trailing_slash_on_api_base(api_base: str) -> None:
+    assert (
+        LinkupSearchConfig().get_complete_url(api_base=api_base, optional_params={})
+        == "https://gateway.example.com/linkup/v1/search"
+    )

tests/unit/llms/modelscope/chat/test_modelscope_chat_transformation.py

@@ -388,3 +388,20 @@ class TestModelScopeConfig:
         assert msg["content"][0] == {"type": "text", "text": "What is in this image?"}
         assert msg["content"][1]["type"] == "image_url"
         assert msg["content"][1]["image_url"]["url"] == "https://example.com/cat.jpg"
+
+
+@pytest.mark.parametrize(
+    "api_base",
+    [
+        "https://gateway.example.com/modelscope/v1",
+        "https://gateway.example.com/modelscope/v1/",
+        "https://gateway.example.com/modelscope/v1/chat/completions/",
+    ],
+)
+def test_get_complete_url_ignores_a_trailing_slash_on_api_base(api_base: str):
+    assert (
+        ModelScopeChatConfig().get_complete_url(
+            api_base=api_base, api_key=None, model="Qwen/Qwen3-8B", optional_params={}, litellm_params={}
+        )
+        == "https://gateway.example.com/modelscope/v1/chat/completions"
+    )

tests/unit/llms/moonshot/test_moonshot_chat_transformation.py

@@ -763,3 +763,20 @@ class TestMoonshotReasoningEffort:
         )
 
         assert "reasoning_effort" not in optional_params
+
+
+@pytest.mark.parametrize(
+    "api_base",
+    [
+        "https://gateway.example.com/moonshot/v1",
+        "https://gateway.example.com/moonshot/v1/",
+        "https://gateway.example.com/moonshot/v1/chat/completions/",
+    ],
+)
+def test_get_complete_url_ignores_a_trailing_slash_on_api_base(api_base: str):
+    assert (
+        MoonshotChatConfig().get_complete_url(
+            api_base=api_base, api_key=None, model="kimi-k3", optional_params={}, litellm_params={}
+        )
+        == "https://gateway.example.com/moonshot/v1/chat/completions"
+    )

tests/unit/llms/ollama/test_ollama_chat_transformation.py

@@ -1091,3 +1091,20 @@ def test_ollama_aembeddings(mock_aembeddings):
         )
     except Exception as e:
         pytest.fail(f"Error occurred: {e}")
+
+
+@pytest.mark.parametrize(
+    "api_base",
+    [
+        "http://gateway.example.com/ollama",
+        "http://gateway.example.com/ollama/",
+        "http://gateway.example.com/ollama/api/chat/",
+    ],
+)
+def test_get_complete_url_ignores_a_trailing_slash_on_api_base(api_base: str):
+    assert (
+        OllamaChatConfig().get_complete_url(
+            api_base=api_base, api_key=None, model="llama3", optional_params={}, litellm_params={}
+        )
+        == "http://gateway.example.com/ollama/api/chat"
+    )

tests/unit/llms/ollama/test_ollama_completion_transformation.py

@@ -973,3 +973,20 @@ def test_ollama_vision_model_sends_the_image_beside_the_user_prompt(respx_mock:
     assert body["images"] == [image_base64]
     assert body["prompt"] == "### User:\nWhats in this image?\n\n"
     assert response.choices[0].message.content == "A white square"
+
+
+@pytest.mark.parametrize(
+    "api_base",
+    [
+        "http://gateway.example.com/ollama",
+        "http://gateway.example.com/ollama/",
+        "http://gateway.example.com/ollama/api/generate/",
+    ],
+)
+def test_get_complete_url_ignores_a_trailing_slash_on_api_base(api_base: str):
+    assert (
+        OllamaConfig().get_complete_url(
+            api_base=api_base, api_key=None, model="llama3", optional_params={}, litellm_params={}
+        )
+        == "http://gateway.example.com/ollama/api/generate"
+    )

tests/unit/llms/ollama/test_ollama_embedding.py

@@ -1,7 +1,11 @@
+from typing import Final
 from unittest.mock import AsyncMock, MagicMock, patch
 
+import httpx
 import pytest
+import respx
 
+import litellm
 from litellm.llms.ollama.completion.handler import ollama_aembeddings, ollama_embeddings
 from litellm.types.utils import EmbeddingResponse
 
@@ -112,3 +116,23 @@ def test_prompt_eval_fallback_when_missing(mock_embedding_response, mock_encodin
         assert response.usage.total_tokens == 5
         assert response.usage.completion_tokens == 0
         assert response.data[0]["embedding"] == [0.1, 0.2, 0.3]
+
+
+@pytest.mark.parametrize(
+    "api_base",
+    [
+        "http://gateway.example.com/ollama",
+        "http://gateway.example.com/ollama/",
+        "http://gateway.example.com/ollama/api/embed/",
+    ],
+)
+@respx.mock
+def test_ollama_embeddings_ignore_a_trailing_slash_on_api_base(api_base: str) -> None:
+    route: Final = respx.post("http://gateway.example.com/ollama/api/embed").mock(
+        return_value=httpx.Response(200, json={"embeddings": [[0.1, 0.2], [0.3, 0.4]], "prompt_eval_count": 2})
+    )
+
+    response: Final = litellm.embedding(model="ollama/all-minilm", api_base=api_base, input=["hello", "world"])
+
+    assert route.called
+    assert [item["embedding"] for item in response.data] == [[0.1, 0.2], [0.3, 0.4]]

tests/unit/llms/perplexity/embedding/test_perplexity_embedding_transformation.py

@@ -397,3 +397,20 @@ def test_perplexity_embedding_usage_that_is_not_an_object_is_rejected(usage):
 def test_perplexity_embedding_token_count_that_is_not_an_integer_is_rejected(usage):
     with pytest.raises(ValidationError):
         _transform({"data": [FLOAT_ROW], "usage": usage})
+
+
+@pytest.mark.parametrize(
+    "api_base",
+    [
+        "https://gateway.example.com/perplexity",
+        "https://gateway.example.com/perplexity/",
+        "https://gateway.example.com/perplexity/v1/embeddings/",
+    ],
+)
+def test_get_complete_url_ignores_a_trailing_slash_on_api_base(api_base: str):
+    assert (
+        PerplexityEmbeddingConfig().get_complete_url(
+            api_base=api_base, api_key=None, model="pplx-embed-v1-0.6b", optional_params={}, litellm_params={}
+        )
+        == "https://gateway.example.com/perplexity/v1/embeddings"
+    )

tests/unit/llms/perplexity/search/test_perplexity_search_transformation.py

@@ -223,3 +223,24 @@ async def test_router_search_with_search_tools(
     assert [(result.title, result.url, result.snippet) for result in response.results] == [
         ("AI news", "https://example.com/news", "Recent AI developments.")
     ]
+
+
+@pytest.mark.parametrize(
+    "api_base",
+    [
+        "https://gateway.example.com/perplexity",
+        "https://gateway.example.com/perplexity/",
+        "https://gateway.example.com/perplexity/search/",
+    ],
+)
+def test_search_ignores_a_trailing_slash_on_api_base(respx_mock: respx.MockRouter, api_base: str) -> None:
+    route: Final = respx_mock.post("https://gateway.example.com/perplexity/search").respond(
+        json={"results": [{"title": "ML", "url": "https://example.com/ml", "snippet": "Machine learning."}]}
+    )
+
+    response: Final = litellm.search(
+        query="machine learning", search_provider="perplexity", api_base=api_base, api_key="test-perplexity-key"
+    )
+
+    assert route.called
+    assert [result.url for result in response.results] == ["https://example.com/ml"]

tests/unit/llms/tavily/search/test_tavily_search_transformation.py

@@ -186,3 +186,18 @@ def _vcr_outcome_gate(request, vcr):
     install_live_call_probe(request, vcr)
     yield
     record_vcr_outcome(request, vcr)
+
+
+@pytest.mark.parametrize(
+    "api_base",
+    [
+        "https://gateway.example.com/tavily",
+        "https://gateway.example.com/tavily/",
+        "https://gateway.example.com/tavily/search/",
+    ],
+)
+def test_get_complete_url_ignores_a_trailing_slash_on_api_base(api_base: str) -> None:
+    assert (
+        TavilySearchConfig().get_complete_url(api_base=api_base, optional_params={})
+        == "https://gateway.example.com/tavily/search"
+    )

tests/unit/llms/topaz/image_variations/test_topaz_image_variations_transformation.py

@@ -0,0 +1,13 @@
+import pytest
+
+from litellm.llms.topaz.image_variations.transformation import TopazImageVariationConfig
+
+
+@pytest.mark.parametrize("api_base", ["https://gateway.example.com/topaz", "https://gateway.example.com/topaz/"])
+def test_get_complete_url_ignores_a_trailing_slash_on_api_base(api_base: str):
+    assert (
+        TopazImageVariationConfig().get_complete_url(
+            api_base=api_base, api_key=None, model="topaz-enhance", optional_params={}, litellm_params={}
+        )
+        == "https://gateway.example.com/topaz/image/v1/enhance"
+    )

tests/unit/llms/volcengine/test_volcengine_embedding.py

@@ -7,6 +7,7 @@ import pytest
 
 import litellm
 from litellm.types.utils import EmbeddingResponse
+from litellm.llms.volcengine.embedding.transformation import VolcEngineEmbeddingConfig
 
 
 class TestVolcEngineEmbedding:
@@ -257,3 +258,21 @@ def test_volcengine_embedding_with_multiple_inputs():
 
 if __name__ == "__main__":
     pytest.main([__file__])
+
+
+@pytest.mark.parametrize(
+    "api_base",
+    [
+        "https://gateway.example.com/ark",
+        "https://gateway.example.com/ark/",
+        "https://gateway.example.com/ark/api/v3",
+        "https://gateway.example.com/ark/api/v3/",
+    ],
+)
+def test_get_complete_url_ignores_a_trailing_slash_on_api_base(api_base: str) -> None:
+    assert (
+        VolcEngineEmbeddingConfig().get_complete_url(
+            api_base=api_base, api_key=None, model="doubao-embedding", optional_params={}, litellm_params={}
+        )
+        == "https://gateway.example.com/ark/api/v3/embeddings"
+    )

tests/unit/llms/voyage/embedding/test_transformation_contextual.py

@@ -562,3 +562,20 @@ def test_voyage_contextual_embedding_usage_that_is_not_an_object_is_rejected(usa
 def test_voyage_contextual_embedding_token_count_that_is_not_an_integer_is_rejected(usage):
     with pytest.raises(ValidationError):
         _transform({"data": [EMBEDDING_ROW], "usage": usage})
+
+
+@pytest.mark.parametrize(
+    "api_base",
+    [
+        "https://gateway.example.com/voyage/v1",
+        "https://gateway.example.com/voyage/v1/",
+        "https://gateway.example.com/voyage/v1/contextualizedembeddings/",
+    ],
+)
+def test_get_complete_url_ignores_a_trailing_slash_on_api_base(api_base: str):
+    assert (
+        VoyageContextualEmbeddingConfig().get_complete_url(
+            api_base=api_base, api_key=None, model="voyage-context-3", optional_params={}, litellm_params={}
+        )
+        == "https://gateway.example.com/voyage/v1/contextualizedembeddings"
+    )

tests/unit/llms/voyage/embedding/test_transformation_multimodal.py

@@ -89,3 +89,20 @@ def test_voyage_multimodal_embedding_usage_that_is_not_an_object_is_rejected(usa
 def test_voyage_multimodal_embedding_token_count_that_is_not_an_integer_is_rejected(usage):
     with pytest.raises(ValidationError):
         _transform({"data": [EMBEDDING_ROW], "usage": usage})
+
+
+@pytest.mark.parametrize(
+    "api_base",
+    [
+        "https://gateway.example.com/voyage/v1",
+        "https://gateway.example.com/voyage/v1/",
+        "https://gateway.example.com/voyage/v1/multimodalembeddings/",
+    ],
+)
+def test_get_complete_url_ignores_a_trailing_slash_on_api_base(api_base: str):
+    assert (
+        VoyageMultimodalEmbeddingConfig().get_complete_url(
+            api_base=api_base, api_key=None, model="voyage-multimodal-3", optional_params={}, litellm_params={}
+        )
+        == "https://gateway.example.com/voyage/v1/multimodalembeddings"
+    )

tests/unit/llms/voyage/embedding/test_voyage_embedding_transformation.py

@@ -89,3 +89,20 @@ def test_voyage_embedding_usage_that_is_not_an_object_is_rejected(usage):
 def test_voyage_embedding_token_count_that_is_not_an_integer_is_rejected(usage):
     with pytest.raises(ValidationError):
         _transform({"data": [EMBEDDING_ROW], "usage": usage})
+
+
+@pytest.mark.parametrize(
+    "api_base",
+    [
+        "https://gateway.example.com/voyage/v1",
+        "https://gateway.example.com/voyage/v1/",
+        "https://gateway.example.com/voyage/v1/embeddings/",
+    ],
+)
+def test_get_complete_url_ignores_a_trailing_slash_on_api_base(api_base: str):
+    assert (
+        VoyageEmbeddingConfig().get_complete_url(
+            api_base=api_base, api_key=None, model="voyage-3.5", optional_params={}, litellm_params={}
+        )
+        == "https://gateway.example.com/voyage/v1/embeddings"
+    )