feat(router): tag routing denylist support via ! prefix

litellm/router_strategy/tag_based_routing.py

@@ -7,7 +7,7 @@
 """
 
 import re
-from typing import TYPE_CHECKING, Any, Dict, List, Literal, Optional, Union
+from typing import TYPE_CHECKING, Any, Literal, Optional, Union
 
 from litellm._logging import verbose_logger
 from litellm.types.router import RouterErrors
@@ -21,8 +21,8 @@
 
 
 def _is_valid_deployment_tag_regex(
-    tag_regexes: List[str],
-    header_strings: List[str],
+    tag_regexes: list[str],
+    header_strings: list[str],
 ) -> Optional[str]:
     """
     Test compiled regex patterns against "Header-Name: value" strings.
@@ -44,7 +44,7 @@ def _is_valid_deployment_tag_regex(
 
 
 def is_valid_deployment_tag(
-    deployment_tags: List[str], request_tags: List[str], match_any: bool = True
+    deployment_tags: list[str], request_tags: list[str], match_any: bool = True
 ) -> bool:
     """
     Check if a tag is valid, the matching can be either any or all based on `match_any` flag
@@ -73,10 +73,10 @@ def is_valid_deployment_tag(
 
 def _match_deployment(
     deployment: Any,
-    request_tags: Optional[List[str]],
-    header_strings: List[str],
+    request_tags: Optional[list[str]],
+    header_strings: list[str],
     match_any: bool,
-) -> Optional[Dict[str, str]]:
+) -> Optional[dict[str, str]]:
     """
     Determine whether *deployment* matches the current request.
 
@@ -89,8 +89,8 @@ def _match_deployment(
          ran and failed, so the regex cannot override strict-tag policy.
     """
     litellm_params = deployment.get("litellm_params", {})
-    deployment_tags: Optional[List[str]] = litellm_params.get("tags")
-    deployment_tag_regex: Optional[List[str]] = litellm_params.get("tag_regex")
+    deployment_tags: Optional[list[str]] = litellm_params.get("tags")
+    deployment_tag_regex: Optional[list[str]] = litellm_params.get("tag_regex")
 
     # 1. Exact tag match (existing behaviour).
     if deployment_tags and request_tags:
@@ -118,11 +118,55 @@ def _match_deployment(
     return None
 
 
+def _split_tags(tags: list[str]) -> tuple[list[str], list[str]]:
+    positive = [t for t in tags if not t.startswith("!")]
+    excluded = [tag[1:] for tag in tags if tag.startswith("!") and len(tag) > 1]
+    return positive, excluded
+
+
+def _exclude_deployments(
+    deployments: Union[list[Any], dict[Any, Any]],
+    excluded_set: frozenset[str],
+) -> list[Any]:
+    if not excluded_set:
+        return list(deployments)
+    return [
+        d
+        for d in deployments
+        if not excluded_set.intersection(d.get("litellm_params", {}).get("tags") or [])
+    ]
+
+
+def _require_candidates(
+    candidates: list[Any],
+    model: str,
+    request_tags: Any,
+) -> list[Any]:
+    if not candidates:
+        raise ValueError(
+            f"{RouterErrors.no_deployments_with_tag_routing.value}."
+            f" Passed model={model} and tags={request_tags}"
+        )
+    return candidates
+
+
+def _ban_only_base_pool(
+    deployments: Union[list[Any], dict[Any, Any]],
+) -> list[Any]:
+    # Mirrors untagged-request semantics so callers can't use !tags to escape the default pool.
+    defaults = [
+        d
+        for d in deployments
+        if "default" in (d.get("litellm_params", {}).get("tags") or [])
+    ]
+    return defaults if defaults else list(deployments)
+
+
 async def get_deployments_for_tag(
     llm_router_instance: LitellmRouter,
     model: str,  # used to raise the correct error
-    healthy_deployments: Union[List[Any], Dict[Any, Any]],
-    request_kwargs: Optional[Dict[Any, Any]] = None,
+    healthy_deployments: Union[list[Any], dict[Any, Any]],
+    request_kwargs: Optional[dict[Any, Any]] = None,
     metadata_variable_name: Literal["metadata", "litellm_metadata"] = "metadata",
 ):
     """
@@ -140,16 +184,9 @@ async def get_deployments_for_tag(
         )
         return healthy_deployments
 
-    if healthy_deployments is None:
-        verbose_logger.debug(
-            "get_deployments_for_tag: healthy_deployments is None returning healthy_deployments"
-        )
-        return healthy_deployments
-
-    # Tag filtering applies only when there is at least one deployment to evaluate.
-    if isinstance(healthy_deployments, list) and len(healthy_deployments) == 0:
+    if not healthy_deployments:
         verbose_logger.debug(
-            "get_deployments_for_tag: empty candidate set; skipping tag filter"
+            "get_deployments_for_tag: empty or None healthy_deployments; skipping tag filter"
         )
         return healthy_deployments
 
@@ -164,34 +201,42 @@ async def get_deployments_for_tag(
         # Build header strings for regex matching from what the proxy already stores.
         # Currently we match against User-Agent; format matches "^User-Agent: claude-code/..."
         user_agent = metadata.get("user_agent", "")
-        header_strings: List[str] = [f"User-Agent: {user_agent}"] if user_agent else []
+        header_strings: list[str] = [f"User-Agent: {user_agent}"] if user_agent else []
 
-        new_healthy_deployments: List[Any] = []
-        default_deployments: List[Any] = []
+        positive_tags, excluded_patterns = _split_tags(request_tags or [])
+
+        excluded_set = frozenset(excluded_patterns)
+        candidates = _exclude_deployments(healthy_deployments, excluded_set)
 
-        # Only activate header-based regex filtering when at least one deployment in
-        # the candidate set has tag_regex configured.  This preserves existing
-        # behaviour for operators who use plain tags: a request that carries a
-        # User-Agent (all proxy requests do) but targets deployments with no
-        # tag_regex will continue to use the original tag-only code path.
         has_regex_deployments = any(
-            d.get("litellm_params", {}).get("tag_regex") for d in healthy_deployments
+            d.get("litellm_params", {}).get("tag_regex") for d in candidates
         )
-        has_tag_filter = bool(request_tags) or (
+        has_tag_filter = bool(positive_tags) or (
             bool(header_strings) and has_regex_deployments
         )
+        ban_only = bool(excluded_set) and not has_tag_filter
+
+        if ban_only:
+            pool = _exclude_deployments(
+                _ban_only_base_pool(healthy_deployments), excluded_set
+            )
+            return _require_candidates(pool, model, request_tags)
+
+        new_healthy_deployments: list[Any] = []
+        default_deployments: list[Any] = []
+
         if has_tag_filter:
             verbose_logger.debug(
                 "get_deployments_for_tag routing: request_tags=%s user_agent=%s",
                 request_tags,
                 user_agent,
             )
-            for deployment in healthy_deployments:
+            for deployment in candidates:
                 deployment_tags = deployment.get("litellm_params", {}).get("tags")
 
                 match_result = _match_deployment(
                     deployment=deployment,
-                    request_tags=request_tags,
+                    request_tags=positive_tags,
                     header_strings=header_strings,
                     match_any=match_any,
                 )
@@ -203,10 +248,6 @@ async def get_deployments_for_tag(
                         match_result["matched_via"],
                         match_result["matched_value"],
                     )
-                    # Record provenance in metadata so it flows to SpendLogs.
-                    # Written only for the first match — load balancer selects one
-                    # deployment from new_healthy_deployments, so overwriting on
-                    # subsequent matches would produce misleading observability data.
                     if "tag_routing" not in metadata:
                         metadata["tag_routing"] = {
                             "matched_deployment": deployment.get("model_name"),
@@ -222,7 +263,8 @@ async def get_deployments_for_tag(
 
             if len(new_healthy_deployments) == 0 and len(default_deployments) == 0:
                 raise ValueError(
-                    f"{RouterErrors.no_deployments_with_tag_routing.value}. Passed model={model} and tags={request_tags}"
+                    f"{RouterErrors.no_deployments_with_tag_routing.value}."
+                    f" Passed model={model} and tags={request_tags}"
                 )
 
             return (
@@ -249,9 +291,9 @@ async def get_deployments_for_tag(
 
 
 def _get_tags_from_request_kwargs(
-    request_kwargs: Optional[Dict[Any, Any]] = None,
+    request_kwargs: Optional[dict[Any, Any]] = None,
     metadata_variable_name: Literal["metadata", "litellm_metadata"] = "metadata",
-) -> List[str]:
+) -> list[str]:
     """
     Helper to get tags from request kwargs