feat(rate_limit): support per-tag rpm limiting on a single key

litellm/proxy/_types.py

@@ -1045,6 +1045,7 @@ class GenerateRequestBase(LiteLLMPydanticObjectBase):
     model_rpm_limit: Optional[dict] = None
     model_tpm_limit: Optional[dict] = None
     mcp_rpm_limit: Optional[Dict[str, int]] = None
+    tag_rpm_limit: Optional[dict[str, int]] = None
     guardrails: Optional[List[str]] = None
     policies: Optional[List[str]] = None
     prompts: Optional[List[str]] = None
@@ -3869,6 +3870,7 @@ class PassThroughEndpointLoggingTypedDict(TypedDict):
     "model_rpm_limit",
     "model_tpm_limit",
     "mcp_rpm_limit",
+    "tag_rpm_limit",
     "rpm_limit_type",
     "tpm_limit_type",
     "enforced_params",

litellm/proxy/auth/auth_utils.py

@@ -975,6 +975,20 @@ def get_team_mcp_rpm_limit(
     return None
 
 
+def get_key_tag_rpm_limit(
+    user_api_key_dict: UserAPIKeyAuth,
+) -> Optional[dict[str, int]]:
+    """
+    Get the per-request-tag rpm limit configured on a given api key.
+
+    The returned dict is keyed by request tag, so each tag/group tracked on
+    the key gets its own independent RPM counter.
+    """
+    if user_api_key_dict.metadata:
+        return user_api_key_dict.metadata.get("tag_rpm_limit")
+    return None
+
+
 def get_project_model_rpm_limit(
     user_api_key_dict: UserAPIKeyAuth,
 ) -> Optional[Dict[str, int]]:

litellm/proxy/hooks/parallel_request_limiter_v3.py

@@ -31,8 +31,12 @@
     get_str_from_messages,
 )
 from litellm.proxy._types import UserAPIKeyAuth
-from litellm.proxy.auth.auth_utils import get_model_rate_limit_from_metadata
+from litellm.proxy.auth.auth_utils import (
+    get_key_tag_rpm_limit,
+    get_model_rate_limit_from_metadata,
+)
 from litellm.proxy.auth.budget_throttle import throttled_limit
+from litellm.proxy.common_utils.http_parsing_utils import get_tags_from_request_body
 from litellm.proxy.common_utils.proxy_rate_limit_error import (
     ProxyRateLimitError,
     map_v3_rate_limit_type,
@@ -1300,6 +1304,43 @@ def _add_model_per_key_rate_limit_descriptor(
             )
         )
 
+    def _add_tag_per_key_rate_limit_descriptor(
+        self,
+        user_api_key_dict: UserAPIKeyAuth,
+        data: dict,
+        descriptors: list[RateLimitDescriptor],
+    ) -> None:
+        """
+        Add per-request-tag rpm limit descriptors for the API key.
+
+        Each tag carried on the request that has a configured limit gets its own
+        ``{api_key}:{tag}`` counter, so a burst on one tag/group never consumes
+        another's budget. Tags without a configured limit fall through to the
+        key-level descriptor.
+        """
+        if not user_api_key_dict.api_key:
+            return
+
+        tag_rpm_limit = get_key_tag_rpm_limit(user_api_key_dict) or {}
+        if not tag_rpm_limit:
+            return
+
+        for tag in dict.fromkeys(get_tags_from_request_body(data)):
+            rpm_limit = tag_rpm_limit.get(tag)
+            if rpm_limit is None:
+                continue
+            descriptors.append(
+                RateLimitDescriptor(
+                    key="tag_per_key",
+                    value=f"{user_api_key_dict.api_key}:{tag}",
+                    rate_limit={
+                        "requests_per_unit": rpm_limit,
+                        "tokens_per_unit": None,
+                        "window_size": self.window_size,
+                    },
+                )
+            )
+
     def _add_mcp_per_key_rate_limit_descriptor(
         self,
         user_api_key_dict: UserAPIKeyAuth,
@@ -1645,6 +1686,13 @@ def _create_rate_limit_descriptors(
             descriptors=descriptors,
         )
 
+        # Per-request-tag rate limits scoped to this key
+        self._add_tag_per_key_rate_limit_descriptor(
+            user_api_key_dict=user_api_key_dict,
+            data=data,
+            descriptors=descriptors,
+        )
+
         # REST MCP calls pass the raw body through this hook before server
         # resolution; only the later synthetic hook payload may carry this key.
         if call_type == CallTypes.call_mcp_tool.value and "server_id" not in data:
@@ -1961,6 +2009,7 @@ async def async_pre_call_hook(
 
         # Org Level Rate Limits
         descriptors.extend(self.create_organization_rate_limit_descriptor(user_api_key_dict, requested_model))
+
         # Only check rate limits if we have descriptors with actual limits
         if descriptors:
             # First pass: RPM and max_parallel_requests sliding-window check.

litellm/proxy/management_endpoints/internal_user_endpoints.py

@@ -377,6 +377,7 @@ async def new_user(
     - budget_fallbacks: Optional[Dict[str, List[str]]] - Per-model fallback chain tried in order when that model's own `model_max_budget` is exceeded, e.g. {"gpt-4o": ["gpt-4o-mini"]}.
     - model_rpm_limit: Optional[float] - Model-specific rpm limit for user. [Docs](https://docs.litellm.ai/docs/proxy/users#add-model-specific-limits-to-keys)
     - mcp_rpm_limit: Optional[dict] - Per-MCP-server rpm limit, keyed by MCP server name {"github": 100, "slack": 200}. Enforced for keys and teams only; values set on a user are stored but not enforced per user.
+    - tag_rpm_limit: Optional[dict] - Per-request-tag rpm limit, keyed by request tag {"cell-1": 1000, "cell-2": 500}. Enforced for keys only; values set on a user are stored but not enforced per user.
     - model_tpm_limit: Optional[float] - Model-specific tpm limit for user. [Docs](https://docs.litellm.ai/docs/proxy/users#add-model-specific-limits-to-keys)
     - spend: Optional[float] - Amount spent by user. Default is 0. Will be updated by proxy whenever user is used. You can set duration as seconds ("30s"), minutes ("30m"), hours ("30h"), days ("30d"), months ("1mo").
     - agent_id: Optional[str] - The agent id associated with the user.
@@ -1379,6 +1380,7 @@ async def user_update(
         - budget_fallbacks: Optional[Dict[str, List[str]]] - Per-model fallback chain tried in order when that model's own `model_max_budget` is exceeded, e.g. {"gpt-4o": ["gpt-4o-mini"]}.
         - model_rpm_limit: Optional[float] - Model-specific rpm limit for user. [Docs](https://docs.litellm.ai/docs/proxy/users#add-model-specific-limits-to-keys)
         - mcp_rpm_limit: Optional[dict] - Per-MCP-server rpm limit, keyed by MCP server name {"github": 100, "slack": 200}. Enforced for keys and teams only; values set on a user are stored but not enforced per user.
+        - tag_rpm_limit: Optional[dict] - Per-request-tag rpm limit, keyed by request tag {"cell-1": 1000, "cell-2": 500}. Enforced for keys only; values set on a user are stored but not enforced per user.
         - model_tpm_limit: Optional[float] - Model-specific tpm limit for user. [Docs](https://docs.litellm.ai/docs/proxy/users#add-model-specific-limits-to-keys)
         - spend: Optional[float] - Amount spent by user. Default is 0. Will be updated by proxy whenever user is used. You can set duration as seconds ("30s"), minutes ("30m"), hours ("30h"), days ("30d"), months ("1mo").
         - agent_id: Optional[str] - The agent id associated with the user.

litellm/proxy/management_endpoints/key_management_endpoints.py

@@ -1496,6 +1496,7 @@ async def generate_key_fn(
     - model_rpm_limit: Optional[dict] - key-specific model rpm limit. Example - {"text-davinci-002": 1000, "gpt-3.5-turbo": 1000}. IF null or {} then no model specific rpm limit.
     - model_tpm_limit: Optional[dict] - key-specific model tpm limit. Example - {"text-davinci-002": 1000, "gpt-3.5-turbo": 1000}. IF null or {} then no model specific tpm limit.
     - mcp_rpm_limit: Optional[dict] - key-specific per-MCP-server rpm limit, keyed by MCP server name (alias if set, else the configured name). Example - {"github": 100, "slack": 200}. IF null or {} then no MCP-specific rpm limit.
+    - tag_rpm_limit: Optional[dict] - key-specific per-request-tag rpm limit, keyed by request tag. Example - {"cell-1": 1000, "cell-2": 500}. Each tag gets an independent counter; requests whose tag is absent fall back to the key-level rpm limit.
     - tpm_limit_type: Optional[str] - Type of tpm limit. Options: "best_effort_throughput" (no error if we're overallocating tpm), "guaranteed_throughput" (raise an error if we're overallocating tpm), "dynamic" (dynamically exceed limit when no 429 errors). Defaults to "best_effort_throughput".
     - rpm_limit_type: Optional[str] - Type of rpm limit. Options: "best_effort_throughput" (no error if we're overallocating rpm), "guaranteed_throughput" (raise an error if we're overallocating rpm), "dynamic" (dynamically exceed limit when no 429 errors). Defaults to "best_effort_throughput".
     - allowed_cache_controls: Optional[list] - List of allowed cache control values. Example - ["no-cache", "no-store"]. See all values - https://docs.litellm.ai/docs/proxy/caching#turn-on--off-caching-per-request
@@ -2514,6 +2515,7 @@ async def update_key_fn(
     - rpm_limit: Optional[int] - Requests per minute limit
     - model_rpm_limit: Optional[dict] - Model-specific RPM limits {"gpt-4": 100, "claude-v1": 200}
     - mcp_rpm_limit: Optional[dict] - Per-MCP-server RPM limits, keyed by MCP server name {"github": 100, "slack": 200}
+    - tag_rpm_limit: Optional[dict] - Per-request-tag RPM limits, keyed by request tag {"cell-1": 1000, "cell-2": 500}. Each tag gets an independent counter; absent tags fall back to the key-level rpm limit.
     - model_tpm_limit: Optional[dict] - Model-specific TPM limits {"gpt-4": 100000, "claude-v1": 200000}
     - tpm_limit_type: Optional[str] - TPM rate limit type - "best_effort_throughput", "guaranteed_throughput", or "dynamic"
     - rpm_limit_type: Optional[str] - RPM rate limit type - "best_effort_throughput", "guaranteed_throughput", or "dynamic"
@@ -3551,6 +3553,7 @@ async def generate_key_helper_fn(
     model_rpm_limit: Optional[dict] = None,
     model_tpm_limit: Optional[dict] = None,
     mcp_rpm_limit: Optional[dict] = None,
+    tag_rpm_limit: Optional[dict] = None,
     guardrails: Optional[list] = None,
     policies: Optional[list] = None,
     prompts: Optional[list] = None,
@@ -3624,6 +3627,9 @@ async def generate_key_helper_fn(
     if mcp_rpm_limit is not None:
         metadata = metadata or {}
         metadata["mcp_rpm_limit"] = mcp_rpm_limit
+    if tag_rpm_limit is not None:
+        metadata = metadata or {}
+        metadata["tag_rpm_limit"] = tag_rpm_limit
     if guardrails is not None:
         metadata = metadata or {}
         metadata["guardrails"] = guardrails

tests/test_litellm/proxy/auth/test_auth_utils.py

@@ -19,6 +19,7 @@
     get_key_mcp_rpm_limit,
     get_key_model_rpm_limit,
     get_key_model_tpm_limit,
+    get_key_tag_rpm_limit,
     get_model_from_request,
     get_project_model_rpm_limit,
     get_project_model_tpm_limit,
@@ -2393,3 +2394,17 @@ def test_normal_body_still_passes(self):
             )
             is True
         )
+
+
+class TestGetKeyTagRateLimits:
+    """Tests for get_key_tag_rpm_limit."""
+
+    def test_reads_tag_rpm_limit_from_metadata(self):
+        key = UserAPIKeyAuth(
+            api_key="sk-123", metadata={"tag_rpm_limit": {"cell-1": 5}}
+        )
+        assert get_key_tag_rpm_limit(key) == {"cell-1": 5}
+
+    def test_returns_none_when_unset(self):
+        key = UserAPIKeyAuth(api_key="sk-123")
+        assert get_key_tag_rpm_limit(key) is None

tests/test_litellm/proxy/hooks/test_parallel_request_limiter_v3.py

@@ -3573,3 +3573,136 @@ async def spy_reserve(*args, **kwargs):
     )
 
     assert TPM_RESERVED_TOKENS_KEY not in (data.get("metadata") or {})
+
+
+@pytest.mark.asyncio
+async def test_per_tag_rate_limit_independent_counters_v3(monkeypatch):
+    """
+    A single key with per-tag RPM limits tracks each tag independently: a tag
+    at its limit returns 429 while a different (unlimited) tag keeps flowing,
+    governed only by the generous key-level limit.
+    """
+    monkeypatch.setenv("LITELLM_RATE_LIMIT_WINDOW_SIZE", "60")
+    _api_key = hash_token("sk-per-tag-rpm")
+    user_api_key_dict = UserAPIKeyAuth(
+        api_key=_api_key,
+        rpm_limit=100,
+        metadata={"tag_rpm_limit": {"cell-1": 2}},
+    )
+    local_cache = DualCache()
+    handler = _PROXY_MaxParallelRequestsHandler(
+        internal_usage_cache=InternalUsageCache(local_cache)
+    )
+
+    async def call(tag: str) -> None:
+        await handler.async_pre_call_hook(
+            user_api_key_dict=user_api_key_dict,
+            cache=local_cache,
+            data={"model": "gpt-3.5-turbo", "metadata": {"tags": [tag]}},
+            call_type="",
+        )
+
+    await call("cell-1")
+    await call("cell-1")
+    with pytest.raises(HTTPException) as exc_info:
+        await call("cell-1")
+    assert exc_info.value.status_code == 429
+    assert "tag_per_key" in str(exc_info.value.detail)
+
+    # cell-2 has no configured tag limit, so cell-1's exhausted counter must
+    # not block it; only the generous key-level limit applies.
+    for _ in range(5):
+        await call("cell-2")
+
+
+@pytest.mark.asyncio
+async def test_per_tag_descriptor_creation_v3():
+    """
+    _create_rate_limit_descriptors emits a tag_per_key descriptor carrying the
+    configured RPM limit only for request tags present in the configured map.
+    """
+    _api_key = hash_token("sk-per-tag-desc")
+    user_api_key_dict = UserAPIKeyAuth(
+        api_key=_api_key,
+        metadata={"tag_rpm_limit": {"cell-1": 5}},
+    )
+    handler = _PROXY_MaxParallelRequestsHandler(
+        internal_usage_cache=InternalUsageCache(DualCache())
+    )
+
+    descriptors = handler._create_rate_limit_descriptors(
+        user_api_key_dict=user_api_key_dict,
+        data={"model": "gpt-3.5-turbo", "metadata": {"tags": ["cell-1", "cell-2"]}},
+        rpm_limit_type=None,
+        tpm_limit_type=None,
+        model_has_failures=False,
+    )
+
+    tag_descriptors = [d for d in descriptors if d["key"] == "tag_per_key"]
+    assert len(tag_descriptors) == 1, "only the configured tag yields a descriptor"
+    descriptor = tag_descriptors[0]
+    assert descriptor["value"] == f"{_api_key}:cell-1"
+    assert descriptor["rate_limit"]["requests_per_unit"] == 5
+
+
+@pytest.mark.asyncio
+async def test_per_tag_descriptor_absent_without_config_v3():
+    """No tag_per_key descriptor is created when the key has no tag limits."""
+    user_api_key_dict = UserAPIKeyAuth(
+        api_key=hash_token("sk-no-tag"),
+        rpm_limit=10,
+    )
+    handler = _PROXY_MaxParallelRequestsHandler(
+        internal_usage_cache=InternalUsageCache(DualCache())
+    )
+
+    descriptors = handler._create_rate_limit_descriptors(
+        user_api_key_dict=user_api_key_dict,
+        data={"model": "gpt-3.5-turbo", "metadata": {"tags": ["cell-1"]}},
+        rpm_limit_type=None,
+        tpm_limit_type=None,
+        model_has_failures=False,
+    )
+
+    assert not [d for d in descriptors if d["key"] == "tag_per_key"]
+
+
+@pytest.mark.asyncio
+async def test_per_tag_untagged_request_governed_by_key_limit_v3(monkeypatch):
+    """
+    Per-tag limits are opt-in sub-limits under the key-level ceiling, not a
+    standalone enforcement boundary: a request that carries no tag (or a tag
+    without a configured limit) is not rejected by any tag counter, but it is
+    still bounded by the key-level rpm_limit. This pins the documented
+    untagged-fallback behavior so a future "fail closed on missing tag" change
+    would fail here instead of silently breaking it.
+    """
+    monkeypatch.setenv("LITELLM_RATE_LIMIT_WINDOW_SIZE", "60")
+    _api_key = hash_token("sk-untagged-fallback")
+    user_api_key_dict = UserAPIKeyAuth(
+        api_key=_api_key,
+        rpm_limit=3,
+        metadata={"tag_rpm_limit": {"cell-1": 2}},
+    )
+    local_cache = DualCache()
+    handler = _PROXY_MaxParallelRequestsHandler(
+        internal_usage_cache=InternalUsageCache(local_cache)
+    )
+
+    async def call(metadata: dict) -> None:
+        await handler.async_pre_call_hook(
+            user_api_key_dict=user_api_key_dict,
+            cache=local_cache,
+            data={"model": "gpt-3.5-turbo", "metadata": metadata},
+            call_type="",
+        )
+
+    # Untagged and unconfigured-tag requests share the key-level budget of 3
+    # and never hit a tag_per_key counter.
+    await call({})
+    await call({"tags": ["cell-99"]})
+    await call({})
+    with pytest.raises(HTTPException) as exc_info:
+        await call({"tags": ["cell-99"]})
+    assert exc_info.value.status_code == 429
+    assert "tag_per_key" not in str(exc_info.value.detail)

tests/test_litellm/proxy/management_endpoints/test_key_management_endpoints.py

@@ -15,8 +15,11 @@
 
 from fastapi import HTTPException
 
+import inspect
+
 from litellm.proxy._types import (
     GenerateKeyRequest,
+    NewUserRequest,
     LiteLLM_BudgetTable,
     LiteLLM_OrganizationTable,
     LiteLLM_TeamTableCachedObj,
@@ -14480,3 +14483,23 @@ async def test_regenerate_key_non_admin_permissions_rejected_before_enterprise_g
     assert int(exc.value.code) == 403
     assert "permissions" in str(exc.value.message)
     assert "Enterprise" not in str(exc.value.message)
+
+
+def test_generate_key_helper_fn_accepts_per_tag_rate_limits():
+    """
+    Regression: new_user / SSO sign-in forward NewUserRequest fields to
+    generate_key_helper_fn via `**data_json`. The per-tag limit field must be
+    an accepted kwarg, otherwise user creation 500s with
+    "generate_key_helper_fn() got an unexpected keyword argument 'tag_rpm_limit'".
+    """
+    params = inspect.signature(generate_key_helper_fn).parameters
+    assert "tag_rpm_limit" in params
+
+    # The field exists on the request model that new_user forwards via **data_json.
+    assert "tag_rpm_limit" in NewUserRequest.model_fields
+
+    # Binding the per-tag kwarg must not raise an unexpected-keyword TypeError.
+    inspect.signature(generate_key_helper_fn).bind_partial(
+        request_type="user",
+        tag_rpm_limit={"cell-1": 5},
+    )