feat(router): opt in to non-chat deployment filtering (#46257)

* feat(proxy): reject chat routes on models declared non-chat via model_info.supported_endpoints

A model whose every deployment sets a non-chat model_info.mode and a supported_endpoints list with no chat endpoint now answers /v1/chat/completions, /v1/messages and /v1/responses with a 400 before any provider call. Models without both fields keep today's behavior.

* fix(proxy): let fallbacks answer and cover more chat routes in the non-chat model check

The check now steps aside when a request, key, team or router fallback would answer, skips such models in the rate-limit fallback loop, ignores mode and supported_endpoints copied from the cost map by old Admin UI saves, treats a blank mode as unset, and also covers /v1/completions, the Responses WebSocket, Gemini generateContent and /v1beta/interactions

* fix(proxy): only step aside for fallbacks that can answer chat, gate the Responses WebSocket after alias resolution, and compare declarations to the loaded catalog

* fix(proxy): gate the Responses WebSocket with the router after aliases resolve, and ignore client-set internal fallback flags

* fix(proxy): walk fallback groups breadth first and release reservations on a rejected WebSocket

Each fallback group is looked up once however often the request repeats it, and a WebSocket closed by the endpoint gate runs the failure hook so rate-limit reservations are released

* fix(proxy): walk fallback layers without recursion and gate the compact Responses route

The fallback walk now folds one layer at a time, so the recursion check passes, and /v1/responses/compact on a model declared non-chat gets the same 400 as the other chat-shaped routes

* fix(proxy): keep chat on rows whose cost map entry serves chat

A row that carries the cost map key of an entry that serves chat counts as declaring nothing, so a catalog edit that adds a chat path never turns old Admin UI copies into 400s

* fix(proxy): check chat sent to a deployment id or backend model name

* fix(proxy): check a model the read-through loads from the database

* fix(proxy): step aside for fallbacks the check cannot follow and wildcards that serve a backend name

A fallback target with no deployments of its own, a client routing stamp, and a wildcard that does not match the requested backend name no longer turn a working request into a 400

* fix(proxy): ignore a client model_group that is not a string when building rate-limit headers

A client metadata.model_group sent as an int or a list crashed the error handler with a 500 once the embedding-only check rejected the call before routing. The header helper now treats a non-string stamp as no model group. Drops the pattern clause in the backend-name lookup, since the group lookup already returns wildcard matches. Adds the third audit round's routing-stamp cells.

* fix(proxy): keep a non-string client model_group out of spend rows

* fix(proxy): validate standard fallbacks before the typed fallback-group lookup

* feat(router): opt in to non-chat deployment filtering

* fix(router): skip unresolved local rate-limit fallbacks

* fix(router): ignore caller-provided endpoint selector values

* fix(router): normalize declared chat endpoint prefixes

* test(router): cover preflight error-value behavior

* test(router): derive replay metadata defaults from model info

* fix(router): preserve fallback aliases and check synchronous text calls

* test(routing): rewrite the WebSocket model with a pre-call callback, since main no longer runs guardrails at socket admission

* test(routing): keep the endpoint gate's owned proxies on the worker's database

* fix(router): keep the endpoint gate on model_info patches that omit it and on same-group stream retries

A partial PATCH of model_info reset a stored enforce_supported_endpoints true back to false, because the
field defaulted to False and the update merges model_dump(exclude_none=True). The field now defaults to None
so an omitted flag keeps the stored value

The same-group retry of a stream dropped before content called the provider without the endpoint selection
context, so it could land on an enforced embedding-only deployment of the same group. It now runs inside
endpoint_selection like every other router attempt

* fix(router): probe rate-limit fallbacks through get_routable_deployments so the router surface does not grow

Main's router fence lets code outside the router use only listed Router members and
router_utils modules. The proxy's fallback probe now calls the listed
get_routable_deployments with the request's call type, and the gate's 400 moves to
litellm.exceptions as EndpointNotSupportedByModelError. It stays separate from main's
UnsupportedEndpointError, which the router retries on the group's other deployments,
while the gate refuses the whole group

* test(responses): cover the WebSocket error frame for a model that refuses the endpoint

* fix(proxy): let the rate-limit fallback probe try a fallback this worker has not loaded yet

The probe now skips a fallback only when the endpoint check removed every deployment it has, so a model added on another worker still reaches route_request and its database lookup

* fix(proxy): keep key router settings off the Responses WebSocket, read unrooted or single-string endpoint lists, and add status to the refusal frame

* fix(router): warn when enforce_supported_endpoints has no usable list and match chat endpoints regardless of case, trailing slash, or Gemini path

* test(router): type the load-warning test's caplog and router

* test(routing): match the endpoint gate's integration tests to base fallback behavior on unloaded fallbacks and the Responses WebSocket

---------

Co-authored-by: Mateo <mateo@Mateos-MBP-2.localdomain>

litellm/exceptions.py

@@ -323,6 +323,17 @@ class UnsupportedEndpointError(BadRequestError):
     pass
 
 
+class EndpointNotSupportedByModelError(BadRequestError):
+    def __init__(self, message: str, model: str, llm_provider: str = "") -> None:
+        super().__init__(  # pyright: ignore[reportUnknownMemberType]  # inherited public exception constructor is untyped
+            message=message, model=model, llm_provider=llm_provider
+        )
+        self.message = message
+        self.type = "invalid_request_error"
+        self.param = "model"
+        self.code = "400"
+
+
 class UnprocessableEntityError(openai.UnprocessableEntityError):
     def __init__(
         self,

litellm/proxy/common_request_processing.py

@@ -129,6 +129,7 @@ from litellm.proxy.guardrails.auto_router_compression import arm_pre_call as _ar
 from litellm.proxy.hooks.parallel_request_limiter_v3 import get_or_create_request_stash
 from litellm.proxy.native_compaction import with_proxy_compaction_executor
 from litellm.proxy.route_llm_request import (
+    endpoint_unsupported_by_model,
     route_request,
 )
 from litellm.proxy.telemetry.request_context import bind_call_id as bind_telemetry_call_id
@@ -871,6 +872,17 @@ _DECODED_JSON: Final = TypeAdapter(object)
 _ROUTER_SETTINGS_ADAPTER: Final[TypeAdapter[Mapping[str, object] | None]] = TypeAdapter(Mapping[str, object] | None)
 
 
+def _fallback_probe_data(
+    pristine: object, fallback_model: str, key_aliases: object, group_alias: object
+) -> Mapping[str, object]:
+    return {
+        **_REQUEST_FIELDS.validate_python(pristine),
+        "model": resolve_served_model(
+            fallback_model, _DECODED_JSON.validate_python(litellm.model_alias_map), key_aliases, group_alias
+        ),
+    }
+
+
 def _request_model(data: Mapping[str, object]) -> str | list[str] | None:
     try:
         return _REQUEST_MODEL.validate_python(data.get("model"), strict=True)
@@ -2676,6 +2688,17 @@ class ProxyBaseLLMRequestProcessing:
                 original_model,
                 fallback_models,
             )
+            raw_router_settings: Final = _DECODED_JSON.validate_python(
+                _REQUEST_FIELDS.validate_python(rate_limited_data).get("router_settings_override")
+            )
+            group_alias: Final = _DECODED_JSON.validate_python(
+                _REQUEST_FIELDS.validate_python(raw_router_settings).get("model_group_alias")
+                if isinstance(raw_router_settings, Mapping)
+                else None
+            )
+            key_aliases: Final[object] = cast(  # cast-ok: auth aliases are dynamically typed at runtime
+                object, user_api_key_dict.aliases
+            )
             try:
                 for fallback_model in fallback_models:
                     if fallback_model == original_model:
@@ -2691,6 +2714,13 @@ class ProxyBaseLLMRequestProcessing:
                     self.data = independent_snapshot(pristine)
                     self.data["model"] = fallback_model
                     try:
+                        if endpoint_unsupported_by_model(
+                            route_type,
+                            _fallback_probe_data(pristine, fallback_model, key_aliases, group_alias),
+                            llm_router,
+                            user_api_key_dict.team_id,
+                        ):
+                            continue
                         return await self.common_processing_pre_call_logic(
                             request=request,
                             general_settings=general_settings,

litellm/proxy/common_utils/callback_utils.py

@@ -436,10 +436,7 @@ def get_model_group_from_litellm_kwargs(kwargs: dict) -> str | None:
 def get_model_group_from_request_data(data: dict) -> str | None:
     _metadata: Final = data.get("metadata", None) or {}
     _model_group: Final = _metadata.get("model_group", None)
-    if _model_group is not None:
-        return _model_group
-
-    return None
+    return _model_group if isinstance(_model_group, str) else None
 
 
 def get_remaining_tokens_and_requests_from_request_data(data: dict) -> dict[str, str]:

litellm/proxy/response_api_endpoints/endpoints.py

@@ -34,6 +34,7 @@ from litellm._logging import verbose_proxy_logger
 from litellm.anthropic_interface.exceptions.exception_mapping_utils import AnthropicExceptionMapping
 from litellm.caching.dual_cache import DualCache
 from litellm.constants import EMPTY_MAPPING, MANAGED_BACKGROUND_RESPONSE_METADATA_KEY, TERMINAL_RESPONSE_STATUSES
+from litellm.exceptions import EndpointNotSupportedByModelError
 from litellm.integrations.custom_guardrail import ModifyResponseException
 from litellm.litellm_core_utils.core_helpers import as_str_mapping, get_metadata_variable_name_from_kwargs
 from litellm.litellm_core_utils.internal_call_metadata import is_background_response, requests_background
@@ -2086,6 +2087,14 @@ def _routing_hints_from_first_ws_frame(first_message: str) -> Mapping[str, objec
 
 
 def _responses_ws_failure_frame(failure: Exception) -> str:
+    if isinstance(failure, EndpointNotSupportedByModelError):
+        return json.dumps(
+            {
+                "type": "error",
+                "status": 400,
+                "error": {"type": "invalid_request_error", "message": failure.message.split("\n\n", 1)[0]},
+            }
+        )
     proxy_exception: Final = handle_exception_on_proxy(failure)
     status: Final = int(proxy_exception.code) if str(proxy_exception.code).isdigit() else 500
     error_type: Final = (
@@ -2444,7 +2453,12 @@ async def _responses_websocket_session(
             with contextlib.suppress(Exception):
                 await websocket.send_text(_responses_ws_failure_frame(e))
             with contextlib.suppress(Exception):
-                await websocket.close(code=1011, reason="Internal server error")
+                await websocket.close(
+                    code=1008 if isinstance(e, EndpointNotSupportedByModelError) else 1011,
+                    reason="Pre-call error"
+                    if isinstance(e, EndpointNotSupportedByModelError)
+                    else "Internal server error",
+                )
             return crash_row
         return bill_socket_failure(outcome) if isinstance(outcome, Exception) else None
 

litellm/proxy/route_llm_request.py

@@ -431,6 +431,28 @@ def raise_if_required_body_param_missing(
     )
 
 
+def endpoint_unsupported_by_model(
+    route_type: str,
+    data: Mapping[str, object],
+    llm_router: LitellmRouter | None,
+    team_id: str | None,
+) -> bool:
+    model: Final = data.get("model")
+    if not isinstance(model, str) or llm_router is None:
+        return False
+    metadata: Final = data.get("metadata")
+    request_kwargs: Final = {
+        **data,
+        "metadata": {
+            **(metadata if isinstance(metadata, Mapping) else {}),
+            "user_api_key_team_id": team_id,
+        },
+    }
+    return bool(llm_router.get_routable_deployments(model, request_kwargs)) and not llm_router.get_routable_deployments(
+        model, request_kwargs, call_type=route_type
+    )
+
+
 class MockTestingParamsDisabledError(HTTPException):
     def __init__(self, params: tuple[str, ...]):
         super().__init__(

litellm/proxy/spend_tracking/spend_tracking_utils.py

@@ -464,7 +464,11 @@ def _sl_attribution_fallback(
 ) -> str:
     if standard_logging_payload is None:
         return ""
-    return standard_logging_payload.get(field) or ""
+    return _string_or_empty(standard_logging_payload.get(field))
+
+
+def _string_or_empty(value: object) -> str:
+    return value if isinstance(value, str) else ""
 
 
 def _deployment_provider(deployment: DeploymentTypedDict) -> str | None:
@@ -682,7 +686,7 @@ def get_logging_payload(
     _model_id: Final = metadata.get("model_info", {}).get("id", "") or _sl_attribution_fallback(
         standard_logging_payload, "model_id"
     )
-    _model_group: Final = metadata.get("model_group", "") or _sl_attribution_fallback(
+    _model_group: Final = _string_or_empty(metadata.get("model_group")) or _sl_attribution_fallback(
         standard_logging_payload, "model_group"
     )
     _api_base: Final = litellm_params.get("api_base", "") or _sl_attribution_fallback(

litellm/router.py

@@ -287,6 +287,11 @@ from litellm.router_utils.embedding_batch_size_check import (
     embedding_deployment_selection,
     log_rejection_before_provider_call,
 )
+from litellm.router_utils.endpoint_eligibility import (
+    eligible_deployments,
+    endpoint_selection,
+    warn_on_unenforced_endpoint_flag,
+)
 from litellm.router_utils.fallback_event_handlers import (
     MID_STREAM_FALLBACK_CONTROLS_KEY,
     AttemptedFallbackTargets,
@@ -1449,7 +1454,7 @@ class Router:
         self.assistants_config = assistants_config
         self.search_tools = search_tools or []
         self.guardrail_list = guardrail_list or []
-        self.deployment_names: list = []  # names of models under litellm_params. ex. azure/chatgpt-v-2
+        self.deployment_names: list[str] = []  # names of models under litellm_params. ex. azure/chatgpt-v-2
         self.deployment_latency_map = {}
         ### CACHING ###
         cache_type: Literal["local", "redis", "redis-semantic", "s3", "disk"] = "local"  # default to an in-memory cache
@@ -6054,12 +6059,13 @@ class Router:
             kwargs.setdefault("metadata", {}).update({"model_group": model})
 
             # pick the one that is available (lowest TPM/RPM)
-            deployment: Final = self.get_available_deployment(
-                model=model,
-                messages=messages,
-                specific_deployment=kwargs.pop("specific_deployment", None),
-                request_kwargs=kwargs,
-            )
+            with endpoint_selection("text_completion", model=model):
+                deployment: Final = self.get_available_deployment(
+                    model=model,
+                    messages=messages,
+                    specific_deployment=kwargs.pop("specific_deployment", None),
+                    request_kwargs=kwargs,
+                )
 
             data: Final = deployment["litellm_params"].copy()
             for k, v in self.default_litellm_params.items():
@@ -6915,6 +6921,7 @@ class Router:
         """
         model_group: Final = cast(str, initial_kwargs.get("model"))  # cast-ok: kwargs always carries the model group
         retry_kwargs: Final = mid_stream_retry_kwargs(initial_kwargs)
+        handler: Final = retry_kwargs.get("original_generic_function")
         healthy_deployments, all_deployments = await self._async_get_healthy_deployments(
             model=model_group,
             parent_otel_span=get_parent_otel_span_from_kwargs(retry_kwargs),
@@ -6945,7 +6952,10 @@ class Router:
             )
             carry_superseded_attempt_usage(retry_kwargs)
             try:
-                response = await attempt_function(**retry_kwargs)
+                with endpoint_selection(
+                    getattr(handler, "__name__", None), getattr(handler, "__module__", None), model=model_group
+                ):
+                    response = await attempt_function(**retry_kwargs)
             except Exception as retry_error:  # noqa: BLE001  # the fallback chain judges the failed retry
                 wrapped = _stream_fallback_error_for_raised(retry_error, model_group, False, e.llm_provider)
                 if wrapped is None:
@@ -9315,9 +9325,18 @@ class Router:
         Handler for making a call to the .completion()/.embeddings()/etc. functions.
         """
         model_group: Final = kwargs.get("model")
-        response = original_function(*args, **kwargs)
-        if coroutine_checker.is_async_callable(response) or inspect.isawaitable(response):
-            response = await response
+        generic_function: Final = _OPAQUE_VALUE_ADAPTER.validate_python(kwargs.get("original_generic_function"))
+        endpoint_function: Final = (
+            generic_function if callable(generic_function) else _OPAQUE_VALUE_ADAPTER.validate_python(original_function)
+        )
+        with endpoint_selection(
+            getattr(endpoint_function, "__name__", None),
+            getattr(endpoint_function, "__module__", None),
+            model=_OPAQUE_VALUE_ADAPTER.validate_python(model_group),
+        ):
+            response = original_function(*args, **kwargs)
+            if coroutine_checker.is_async_callable(response) or inspect.isawaitable(response):
+                response = await response
         await self.increment_deployment_usage_for_response(response=response, request_kwargs=kwargs)
         ## PROCESS RESPONSE HEADERS
         response = await self.set_response_headers(response=response, model_group=model_group, request_kwargs=kwargs)
@@ -10638,6 +10657,7 @@ class Router:
             declared_params: Final = _LITELLM_PARAMS_ADAPTER.validate_python(_litellm_params)
             warn_on_provider_credential_mismatch(model_name=_model_name, litellm_params=declared_params)
             warn_on_ignored_model_capability_flags(model_name=_model_name, litellm_params=declared_params)
+            warn_on_unenforced_endpoint_flag(model_name=_model_name, model_info=_model_info)
             deployment = without_model_capability_flags(
                 Deployment(
                     **deployment_info,
@@ -11441,6 +11461,9 @@ class Router:
         declared_params: Final = deployment.litellm_params.model_dump(exclude_none=True)
         warn_on_provider_credential_mismatch(model_name=deployment.model_name, litellm_params=declared_params)
         warn_on_ignored_model_capability_flags(model_name=deployment.model_name, litellm_params=declared_params)
+        warn_on_unenforced_endpoint_flag(
+            model_name=deployment.model_name, model_info=deployment.model_info.model_dump()
+        )
         served: Final = with_admin_api_key_secret_reference(without_model_capability_flags(deployment))
 
         # add to model list
@@ -14797,9 +14820,15 @@ class Router:
         """
         return [m for m in self.model_list if m["litellm_params"]["model"] == model]
 
-    def get_routable_deployments(self, model: str) -> tuple[DeploymentTypedDict, ...]:
+    def get_routable_deployments(
+        self,
+        model: str,
+        request_kwargs: Mapping[str, object] = MappingProxyType({}),
+        call_type: str | None = None,
+    ) -> tuple[DeploymentTypedDict, ...]:
         try:
-            _, candidates = self._common_checks_available_deployment(model=model, request_kwargs={})  # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType]  # its signature leaves input and the result untyped
+            with endpoint_selection(call_type, model=model):
+                _, candidates = self._common_checks_available_deployment(model=model, request_kwargs=request_kwargs)  # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType]  # its signature leaves input and the result untyped
         except (litellm.BadRequestError, ValueError):
             return ()
         if isinstance(candidates, dict):
@@ -15075,7 +15104,10 @@ class Router:
                 model=model,
                 llm_provider="",
             )
-        return selectable
+        eligible: Final = eligible_deployments(model, selectable)
+        if isinstance(eligible, tuple):
+            return list(eligible)
+        raise eligible
 
     def _filter_reserved_deployments(
         self,

litellm/router_utils/endpoint_eligibility.py

@@ -0,0 +1,120 @@
+from collections.abc import Generator, Mapping, Sequence
+from contextlib import contextmanager
+from contextvars import ContextVar
+from types import MappingProxyType
+from typing import Final, TypeVar
+
+from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError, field_validator
+
+from litellm._logging import verbose_router_logger
+from litellm.exceptions import EndpointNotSupportedByModelError
+
+CHAT_ENDPOINT_BY_CALL: Final[Mapping[str, str]] = MappingProxyType(
+    {
+        "completion": "/v1/chat/completions",
+        "acompletion": "/v1/chat/completions",
+        "_completion": "/v1/chat/completions",
+        "_acompletion": "/v1/chat/completions",
+        "text_completion": "/v1/completions",
+        "atext_completion": "/v1/completions",
+        "_atext_completion": "/v1/completions",
+        "anthropic_messages": "/v1/messages",
+        "aresponses": "/v1/responses",
+        "agenerate_content": "generateContent",
+        "agenerate_content_stream": "streamGenerateContent",
+        "acreate_interaction": "/v1beta/interactions",
+        "litellm.interactions.main.acreate": "/v1beta/interactions",
+        "litellm.interactions.main.create": "/v1beta/interactions",
+        "_aresponses_websocket": "/v1/responses",
+        "acompact_responses": "/v1/responses/compact",
+    }
+)
+
+
+def _endpoint_path(endpoint: str) -> str:
+    rooted: Final = "/" + endpoint.strip().strip("/").casefold()
+    path: Final = rooted.removeprefix("/openai").removeprefix("/v1beta").removeprefix("/v1")
+    return "/" + path.rsplit(":", 1)[-1] if ":" in path else path
+
+
+CHAT_ENDPOINTS: Final = frozenset(_endpoint_path(endpoint) for endpoint in CHAT_ENDPOINT_BY_CALL.values())
+_Deployment: Final = TypeVar("_Deployment", bound=Mapping[str, object])
+_ENFORCE_FLAG: Final = TypeAdapter(bool | None)
+_REQUEST_SELECTION: Final[ContextVar[tuple[str | None, str | None]]] = ContextVar(
+    "router_request_endpoint", default=(None, None)
+)
+
+
+class _EndpointDeclaration(BaseModel):
+    model_config = ConfigDict(frozen=True)
+
+    enforce_supported_endpoints: bool | None = None
+    supported_endpoints: tuple[str, ...] | None = None
+
+    @field_validator("supported_endpoints", mode="before")
+    @classmethod
+    def _one_endpoint_as_a_list(cls, value: object) -> object:
+        return (value,) if isinstance(value, str) else value
+
+
+@contextmanager
+def endpoint_selection(call_type: object, module: object = None, model: object = None) -> Generator[None]:
+    qualified_call: Final = f"{module}.{call_type}" if module == "litellm.interactions.main" else call_type
+    endpoint: Final = CHAT_ENDPOINT_BY_CALL.get(qualified_call) if isinstance(qualified_call, str) else None
+    token: Final = _REQUEST_SELECTION.set((endpoint, model if isinstance(model, str) else None))
+    try:
+        yield
+    finally:
+        _REQUEST_SELECTION.reset(token)
+
+
+def _enforced_endpoints(model_info: object) -> tuple[str, ...] | None:
+    try:
+        declaration: Final = _EndpointDeclaration.model_validate(model_info)
+    except ValidationError:
+        return None
+    return declaration.supported_endpoints if declaration.enforce_supported_endpoints else None
+
+
+def warn_on_unenforced_endpoint_flag(model_name: str, model_info: Mapping[str, object]) -> str | None:
+    try:
+        flagged: Final = _ENFORCE_FLAG.validate_python(model_info.get("enforce_supported_endpoints"))
+    except ValidationError:
+        return None
+    if not flagged or _enforced_endpoints(model_info) is not None:
+        return None
+    warning: Final = (
+        f"Deployment '{model_name}' sets enforce_supported_endpoints without a supported_endpoints list of "
+        'strings, so litellm enforces nothing for it. Set supported_endpoints, e.g. ["/v1/embeddings"].'
+    )
+    verbose_router_logger.warning(warning)
+    return warning
+
+
+def non_chat_endpoints(deployment: Mapping[str, object]) -> frozenset[str] | None:
+    declared: Final = _enforced_endpoints(deployment.get("model_info"))
+    if declared is None:
+        return None
+    endpoints: Final = frozenset(declared)
+    normalized: Final = frozenset(_endpoint_path(endpoint) for endpoint in endpoints)
+    return endpoints if normalized.isdisjoint(CHAT_ENDPOINTS) else None
+
+
+def eligible_deployments(
+    model: str, deployments: Sequence[_Deployment]
+) -> tuple[_Deployment, ...] | EndpointNotSupportedByModelError:
+    endpoint, requested_model = _REQUEST_SELECTION.get()
+    if endpoint is None or not deployments:
+        return tuple(deployments)
+    eligible: Final = tuple(deployment for deployment in deployments if non_chat_endpoints(deployment) is None)
+    if eligible:
+        return eligible
+    requested: Final = requested_model or model
+    supported: Final = ", ".join(
+        sorted(frozenset[str]().union(*(non_chat_endpoints(deployment) or frozenset() for deployment in deployments)))
+    )
+    return EndpointNotSupportedByModelError(
+        message=f"{endpoint} is not supported by model={requested}. Supported endpoints: {supported or 'none'}",
+        model=model,
+        llm_provider="",
+    )

litellm/types/router.py

@@ -270,6 +270,7 @@ def _whole_float_as_int(value: object) -> object:
 
 
 class ModelInfo(MirroredPricingParams):
+    enforce_supported_endpoints: bool | None = None
     id: str | None  # Allow id to be optional on input, but it will always be present as a str in the model instance
     db_model: bool = False  # used for proxy - to separate models which are stored in the db vs. config.
     updated_at: datetime.datetime | None = None

tests/e2e/coverage_registry/llm_nonconversational.yaml

@@ -2,6 +2,7 @@
 - {id: llm.completions.openai.basic.nonstream.works, module: llm, tier: P0, subject_endpoint: completions, route: openai, capability: basic, streaming: nonstream, assertions: [works], source: "test_completions_endpoint_e2e.py", rationale: "Legacy text /completions endpoint, second-highest production request volume"}
 - {id: llm.embeddings.openai.basic.nonstream.works, module: llm, tier: P0, subject_endpoint: embeddings, route: openai, capability: basic, streaming: nonstream, assertions: [works], source: "test_embeddings_endpoint_e2e.py:23", rationale: "Core endpoint, live vector response"}
 - {id: llm.embeddings.openai.input_validation.nonstream.works, module: llm, tier: P1, subject_endpoint: embeddings, route: openai, capability: input_validation, streaming: nonstream, assertions: [works], source: "vendor strategy §9.3 / LIT-4778", rationale: "Missing model/input on /embeddings return client errors"}
+- {id: llm.embeddings.openai.supported_endpoints_gate.nonstream.works, module: llm, tier: P1, subject_endpoint: embeddings, route: openai, capability: input_validation, streaming: nonstream, assertions: [works], fail_before_fix: proven, source: "test_embedding_only_supported_endpoints_e2e.py", rationale: "Explicit endpoint restrictions preserve real embeddings, refuse chat with a gateway 400, and route chat through a live fallback; zero-provider-call contracts live in integration/routing/test_non_chat_model_endpoint_gate.py"}
 - {id: llm.embeddings.openai.basic.nonstream.cost_logged, module: llm, tier: P0, subject_endpoint: embeddings, route: openai, capability: basic, streaming: nonstream, assertions: [cost_logged], source: "SPEND_TRACKING_COVERAGE_MATRIX.md:34", rationale: "Cost tracking on embeddings"}
 - {id: llm.embeddings.openai.passthrough.nonstream.cost_logged, module: llm, tier: P0, subject_endpoint: embeddings, route: openai, capability: basic, streaming: nonstream, assertions: [cost_logged], source: "test_passthrough_e2e.py", rationale: "POST /openai_passthrough/v1/embeddings is costed; the route wrote no spend row at all, so budgets never saw traffic OpenAI was billing for (GitHub issue #36646)"}
 - {id: llm.embeddings.azure_openai.basic.nonstream.works, module: llm, tier: P0, subject_endpoint: embeddings, route: azure_openai, capability: basic, streaming: nonstream, assertions: [works], source: "llms/azure/azure.py", rationale: "Azure embeddings via translation"}

tests/e2e/llm_translation/test_embedding_only_supported_endpoints_e2e.py

@@ -0,0 +1,119 @@
+"""Live provider checks for embeddings, the chat refusal, and chat fallback from an explicitly restricted model."""
+
+from __future__ import annotations
+
+import dataclasses
+from typing import Final
+
+import pytest
+from e2e_config import unique_marker
+from e2e_metadata import Domain, Mode, Provider, Route, Subject, meta
+from lifecycle import ResourceManager
+from models import LiteLLMParamsBody, ModelInfoBody, ModelMode, ModelNewBody
+from openai import BadRequestError, OpenAI
+from proxy_client import ProxyClient
+from sdk_clients import NO_PROXY_CACHE, SdkClients
+
+pytestmark = pytest.mark.e2e
+
+EMBEDDING_BACKEND: Final = "openai/text-embedding-3-small"
+CHAT_FALLBACK_BACKEND: Final = "openai/gpt-5.4-nano"
+EMBEDDINGS_ONLY: Final = ("/v1/embeddings",)
+PROMPT: Final = "Say this is a test!"
+
+
+def _subject(route: Route) -> Subject:
+    return Subject(
+        domain=Domain.LLM_TRANSLATION,
+        route=route,
+        providers=(Provider.OPENAI,),
+        models=(EMBEDDING_BACKEND,),
+        mode=Mode.NONSTREAM,
+    )
+
+
+def _register(
+    proxy: ProxyClient,
+    resources: ResourceManager,
+    prefix: str,
+    supported_endpoints: tuple[str, ...] | None,
+    backend: str = EMBEDDING_BACKEND,
+    mode: ModelMode = "embedding",
+) -> str:
+    model: Final = f"{prefix}-{unique_marker()}"
+    model_id: Final = proxy.register_model(
+        ModelNewBody(
+            model_name=model,
+            litellm_params=LiteLLMParamsBody(model=backend, api_key="os.environ/OPENAI_API_KEY"),
+            model_info=ModelInfoBody(
+                enforce_supported_endpoints=supported_endpoints is not None,
+                mode=mode,
+                supported_endpoints=None if supported_endpoints is None else list(supported_endpoints),
+            ),
+        )
+    )
+    resources.defer(lambda: proxy.delete_model(model_id))
+    return model
+
+
+def _openai_at(sdk: SdkClients, api_prefix: str, key: str) -> OpenAI:
+    return dataclasses.replace(sdk, base_url=f"{sdk.base_url}{api_prefix}").openai(key)
+
+
+class TestEmbeddingOnlySupportedEndpoints:
+    @pytest.mark.covers("llm.embeddings.openai.supported_endpoints_gate.nonstream.works")
+    @meta(_subject(Route.EMBEDDINGS))
+    @pytest.mark.parametrize(
+        ("prefix", "supported_endpoints"),
+        (
+            pytest.param("e2e-embed-only", EMBEDDINGS_ONLY, id="embed-only"),
+            pytest.param("e2e-embed-control", None, id="control"),
+        ),
+    )
+    def test_embeddings_return_a_vector(
+        self,
+        proxy: ProxyClient,
+        resources: ResourceManager,
+        sdk: SdkClients,
+        prefix: str,
+        supported_endpoints: tuple[str, ...] | None,
+    ) -> None:
+        model: Final = _register(proxy, resources, prefix, supported_endpoints)
+        embeddings: Final = sdk.openai(resources.key()).embeddings.create(
+            model=model, input=PROMPT, extra_body=NO_PROXY_CACHE
+        )
+        assert embeddings.data, f"/embeddings returned no data for {model}: {embeddings!r}"
+        assert any(component != 0.0 for component in embeddings.data[0].embedding), (
+            f"/embeddings returned an empty or all-zero vector for {model}"
+        )
+
+    @pytest.mark.covers("llm.embeddings.openai.supported_endpoints_gate.nonstream.works")
+    @meta(dataclasses.replace(_subject(Route.CHAT_COMPLETIONS), models=(EMBEDDING_BACKEND, CHAT_FALLBACK_BACKEND)))
+    def test_chat_completions_with_a_fallback_are_answered_by_the_fallback(
+        self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
+    ) -> None:
+        model: Final = _register(proxy, resources, "e2e-embed-only", EMBEDDINGS_ONLY)
+        fallback: Final = _register(proxy, resources, "e2e-chat-fallback", None, CHAT_FALLBACK_BACKEND, "chat")
+        completion: Final = _openai_at(sdk, "/v1", resources.key()).chat.completions.create(
+            model=model,
+            messages=[{"role": "user", "content": PROMPT}],
+            extra_body={**NO_PROXY_CACHE, "fallbacks": [fallback]},
+        )
+        assert completion.choices and completion.choices[0].message.content, (
+            f"the fallback {fallback} should answer a chat call to {model}, got {completion!r}"
+        )
+
+    @pytest.mark.covers("llm.embeddings.openai.supported_endpoints_gate.nonstream.works")
+    @meta(_subject(Route.CHAT_COMPLETIONS))
+    def test_chat_completions_without_a_fallback_are_refused_by_the_gateway(
+        self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
+    ) -> None:
+        model: Final = _register(proxy, resources, "e2e-embed-only", EMBEDDINGS_ONLY)
+        with pytest.raises(BadRequestError) as refused:
+            _ = _openai_at(sdk, "/v1", resources.key()).chat.completions.create(
+                model=model, messages=[{"role": "user", "content": PROMPT}], extra_body=NO_PROXY_CACHE
+            )
+        assert (
+            f"/v1/chat/completions is not supported by model={model}. Supported endpoints: /v1/embeddings"
+            in refused.value.message
+        ), f"chat on {model} should be refused by the gateway before any provider call, got {refused.value!r}"

tests/e2e/models.py

@@ -1416,7 +1416,7 @@ class LiteLLMParamsBody(BaseModel):
     order: int | None = None
 
 
-ModelMode = Literal["chat", "batch", "realtime", "image_generation"]
+ModelMode = Literal["chat", "embedding", "batch", "realtime", "image_generation"]
 
 
 class TokenCountModelBody(BaseModel, frozen=True):
@@ -1425,6 +1425,7 @@ class TokenCountModelBody(BaseModel, frozen=True):
 
 
 class ModelInfoBody(BaseModel):
+    enforce_supported_endpoints: bool | None = None
     # id is left unset so the proxy assigns a unique model_id per deployment.
     # Pinning it to the model_name made re-registrations of a fixed-name model
     # (e.g. the batch suite's openai-batch) collide on the model_id unique

tests/unit/proxy/common_utils/test_callback_utils.py

@@ -49,6 +49,19 @@ def test_get_remaining_tokens_and_requests_from_request_data():
     }
 
 
+@pytest.mark.parametrize("client_model_group", [7, ["embed-only"], {"name": "embed-only"}])
+def test_remaining_limit_headers_ignore_a_model_group_that_is_not_a_string(client_model_group: object):
+    casedata: Final = {
+        "metadata": {
+            "model_group": client_model_group,
+            f"litellm-key-remaining-requests-{client_model_group}": 100,
+            f"litellm-key-remaining-tokens-{client_model_group}": 200,
+        }
+    }
+
+    assert get_remaining_tokens_and_requests_from_request_data(casedata) == {}
+
+
 @patch(
     "litellm.proxy.common_utils.callback_utils.CustomLogger.get_callback_env_vars",
     return_value=["API_KEY", "MISSING_VAR"],

tests/unit/proxy/management_endpoints/test_model_management_endpoints.py

@@ -3617,6 +3617,33 @@ class TestDeleteTeamModels:
         assert any(e[0] == "delete_many" for e in prisma.events)
 
 
+@pytest.mark.parametrize(
+    ("model_info_patch", "expected_gate"),
+    [
+        pytest.param({"tpm": 5}, True, id="unrelated-field"),
+        pytest.param({"supported_endpoints": ["/v1/embeddings", "/v1/rerank"]}, True, id="endpoint-list-only"),
+        pytest.param({"enforce_supported_endpoints": False}, False, id="explicit-false"),
+    ],
+)
+def test_model_info_patch_changes_the_endpoint_gate_only_when_it_sends_the_flag(
+    model_info_patch: dict[str, object], expected_gate: bool
+):
+    from litellm.proxy.management_endpoints.model_management_endpoints import update_db_model
+
+    result: Final = update_db_model(
+        db_model=Deployment(
+            model_name="embedding-only",
+            litellm_params=LiteLLM_Params(model="openai/text-embedding-3-small"),
+            model_info=ModelInfo(
+                id="embedding-only", enforce_supported_endpoints=True, supported_endpoints=["/v1/embeddings"]
+            ),
+        ),
+        updated_patch=updateDeployment.model_validate({"model_info": model_info_patch}),
+    )
+
+    assert json.loads(result["model_info"]).get("enforce_supported_endpoints") is expected_gate
+
+
 def _build_db_model_for_blocked_test():
     from litellm.types.router import Deployment, LiteLLM_Params, ModelInfo
 

tests/unit/proxy/response_api_endpoints/test_endpoints.py

@@ -2055,6 +2055,22 @@ def test_responses_ws_failure_frame_uses_the_http_status_for_the_same_failure(
     assert frame["error"]["message"] == handle_exception_on_proxy(failure).message
 
 
+def test_responses_ws_failure_frame_shows_only_the_refusal_for_a_model_that_refuses_the_endpoint():
+    from litellm.exceptions import EndpointNotSupportedByModelError
+    from litellm.proxy.response_api_endpoints.endpoints import _responses_ws_failure_frame
+
+    refusal: Final = "/v1/responses is not supported by model=embed-only. Supported endpoints: /v1/embeddings"
+    failure: Final = EndpointNotSupportedByModelError(
+        message=f"{refusal}\n\nLiteLLM: model group 'embed-only' failed with the error above", model="embed-only"
+    )
+
+    assert json.loads(_responses_ws_failure_frame(failure)) == {
+        "type": "error",
+        "status": 400,
+        "error": {"type": "invalid_request_error", "message": refusal},
+    }
+
+
 class TestWSSessionCostTracking:
     @pytest.mark.asyncio
     async def test_router_budget_limiter_skips_arealtime_call_type(self):

tests/unit/proxy/spend_tracking/test_spend_tracking_utils.py

@@ -5384,6 +5384,28 @@ def test_get_logging_payload_router_rejected_request_takes_provider_from_model_g
     assert payload["custom_llm_provider"] == expected_provider
 
 
+@pytest.mark.parametrize("client_model_group", [7, ["embed-only"], {"name": "embed-only"}])
+def test_get_logging_payload_drops_a_client_model_group_that_is_not_a_string(client_model_group: object):
+    payload: Final = get_logging_payload(
+        kwargs={
+            "model": "embed-only",
+            "litellm_params": {
+                "metadata": {"user_api_key": "test-key", "model_group": client_model_group, "status": "failure"}
+            },
+            "standard_logging_object": {
+                **_make_failed_request_standard_logging_payload(),
+                "model_group": client_model_group,
+            },
+        },
+        response_obj={},
+        start_time=datetime.datetime.now(timezone.utc),
+        end_time=datetime.datetime.now(timezone.utc),
+        llm_router=None,
+    )
+
+    assert payload["model_group"] == ""
+
+
 def test_get_logging_payload_router_rejected_request_without_router_leaves_provider_empty():
     assert _router_rejected_failure_payload("openai-group", None)["custom_llm_provider"] == ""
 

tests/unit/proxy/test_common_request_processing.py

@@ -456,7 +456,7 @@ class TestProxyBaseLLMRequestProcessing:
             mock_add_litellm_data_to_request,
         )
         mock_general_settings = {}
-        mock_user_api_key_dict = MagicMock(spec=UserAPIKeyAuth)
+        mock_user_api_key_dict = ProxyUserAPIKeyAuth()
         mock_proxy_config = MagicMock(spec=ProxyConfig)
         route_type = "acompletion"
 
@@ -740,7 +740,7 @@ class TestProxyBaseLLMRequestProcessing:
         returned_data, logging_obj = await processing_obj.common_processing_pre_call_logic(
             request=mock_request,
             general_settings={},
-            user_api_key_dict=MagicMock(spec=UserAPIKeyAuth),
+            user_api_key_dict=ProxyUserAPIKeyAuth(),
             proxy_logging_obj=mock_proxy_logging_obj,
             proxy_config=MagicMock(spec=ProxyConfig),
             route_type=route_type,
@@ -1186,7 +1186,7 @@ class TestProxyBaseLLMRequestProcessing:
         )
 
         mock_general_settings = {}
-        mock_user_api_key_dict = MagicMock(spec=UserAPIKeyAuth)
+        mock_user_api_key_dict = ProxyUserAPIKeyAuth()
         mock_proxy_config = MagicMock(spec=ProxyConfig)
 
         mock_router_settings = {
@@ -2208,7 +2208,7 @@ class TestProxyBaseLLMRequestProcessing:
             mock_add_litellm_data_to_request,
         )
         mock_general_settings = {}
-        mock_user_api_key_dict = MagicMock(spec=UserAPIKeyAuth)
+        mock_user_api_key_dict = ProxyUserAPIKeyAuth()
         mock_proxy_config = MagicMock(spec=ProxyConfig)
         route_type = "acompletion"
 
@@ -7706,6 +7706,56 @@ class TestPreCallWithFallbacksOnLocalRateLimit:
         assert processor.data["model"] == fallback_model
         assert call_count == 2
 
+    @pytest.mark.asyncio
+    async def test_local_rate_limit_fallback_skips_an_aliased_embedding_only_model(self):
+        from litellm.proxy.common_request_processing import ProxyBaseLLMRequestProcessing
+        from litellm.proxy.common_utils.proxy_rate_limit_error import ProxyRateLimitError
+
+        router: Final = Router(
+            model_list=[
+                {
+                    "model_name": "embed-only",
+                    "litellm_params": {"model": "openai/text-embedding-3-small", "api_key": "sk-synthetic"},
+                    "model_info": {"enforce_supported_endpoints": True, "supported_endpoints": ["/v1/embeddings"]},
+                },
+                {"model_name": "chat-model", "litellm_params": {"model": "openai/gpt-5.4-nano", "api_key": "sk-synthetic"}},
+            ],
+            fallbacks=[{"rate-limited": ["embed-alias", "chat-model"]}],
+        )
+        processor: Final = ProxyBaseLLMRequestProcessing(
+            data={
+                "model": "rate-limited",
+                "router_settings_override": {"model_group_alias": {"embed-alias": "embed-only"}},
+            }
+        )
+        attempted: Final[list[object]] = []
+
+        async def pre_call_logic(**_: object):
+            attempted.append(processor.data.get("model"))
+            if processor.data.get("model") == "rate-limited":
+                raise ProxyRateLimitError(detail="key rpm limit exceeded", headers={"retry-after": "30"})
+            return processor.data, MagicMock()
+
+        with patch.object(processor, "common_processing_pre_call_logic", side_effect=pre_call_logic):
+            await processor._pre_call_with_fallbacks(
+                request=MagicMock(),
+                general_settings={},
+                proxy_logging_obj=MagicMock(),
+                user_api_key_dict=ProxyUserAPIKeyAuth(),
+                version=None,
+                proxy_config=MagicMock(),
+                user_model=None,
+                user_temperature=None,
+                user_request_timeout=None,
+                user_max_tokens=None,
+                user_api_base=None,
+                model="rate-limited",
+                route_type="acompletion",
+                llm_router=router,
+            )
+
+        assert attempted == ["rate-limited", "chat-model"]
+
     @pytest.mark.asyncio
     async def test_raises_when_no_fallbacks_configured(self):
         from litellm.proxy.common_utils.proxy_rate_limit_error import ProxyRateLimitError

tests/unit/proxy/test_route_llm_request.py

@@ -28,6 +28,7 @@ from litellm.proxy.route_llm_request import (
     JSON_OBJECT_ADAPTER,
     ProxyModelNotFoundError,
     UserConfigRouters,
+    endpoint_unsupported_by_model,
     route_request,
     user_config_routers,
 )
@@ -3027,3 +3028,27 @@ async def test_user_config_routers_discard_the_least_recently_used_router_past_c
     assert first_handler() is not None
     assert second_handler() is None
     routers.clear()
+
+
+def test_fallback_probe_skips_only_a_loaded_model_that_refuses_the_route() -> None:
+    router: Final = litellm.Router(
+        model_list=[
+            {
+                "model_name": "chat-fallback",
+                "litellm_params": {"model": "openai/gpt-5.4-nano", "api_key": "unit-provider-key"},
+            },
+            {
+                "model_name": "embed-only",
+                "litellm_params": {"model": "openai/text-embedding-3-small", "api_key": "unit-provider-key"},
+                "model_info": {"enforce_supported_endpoints": True, "supported_endpoints": ["/v1/embeddings"]},
+            },
+        ],
+        num_retries=0,
+    )
+
+    def skipped(model: str) -> bool:
+        return endpoint_unsupported_by_model("acompletion", {"model": model}, router, None)
+
+    assert skipped("embed-only") is True
+    assert skipped("chat-fallback") is False
+    assert skipped("added-on-another-worker") is False

tests/unit/router_utils/test_endpoint_eligibility.py

@@ -0,0 +1,139 @@
+from collections.abc import Mapping
+from typing import Final
+
+import pytest
+from pydantic import TypeAdapter
+
+from litellm.exceptions import EndpointNotSupportedByModelError
+from litellm.router_utils.endpoint_eligibility import (
+    eligible_deployments,
+    endpoint_selection,
+    warn_on_unenforced_endpoint_flag,
+)
+
+_DEPLOYMENT: Final = TypeAdapter(list[dict[str, object]])
+
+
+def _deployment(info: Mapping[str, object], identity: str = "embedding") -> dict[str, object]:
+    return _DEPLOYMENT.validate_python(
+        [
+            {
+                "model_name": "shared",
+                "litellm_params": {"model": "openai/test-model"},
+                "model_info": {"id": identity, **info},
+            }
+        ]
+    )[0]
+
+
+@pytest.mark.parametrize("call_type", ["acompletion", "anthropic_messages", "aresponses", "acreate_interaction"])
+def test_explicit_restriction_excludes_an_embedding_deployment_from_a_mixed_group(call_type: str) -> None:
+    embedding: Final = _deployment({"enforce_supported_endpoints": True, "supported_endpoints": ["/v1/embeddings"]})
+    chat: Final = _deployment({"supported_endpoints": ["/v1/chat/completions"]}, "chat")
+    with endpoint_selection(call_type):
+        result: Final = eligible_deployments("shared", (embedding, chat))
+    assert result == (chat,)
+
+
+@pytest.mark.parametrize(
+    "info",
+    [
+        {"mode": "audio_speech", "supported_endpoints": ["/v1/audio/speech"]},
+        {"enforce_supported_endpoints": False, "supported_endpoints": ["/v1/embeddings"]},
+        {"enforce_supported_endpoints": True},
+        {"enforce_supported_endpoints": True, "supported_endpoints": None},
+        {"enforce_supported_endpoints": True, "supported_endpoints": 7},
+        {"enforce_supported_endpoints": True, "supported_endpoints": ["/v1/messages"]},
+    ],
+)
+def test_advisory_or_incomplete_metadata_preserves_routing(info: Mapping[str, object]) -> None:
+    deployment: Final = _deployment(info)
+    with endpoint_selection("acompletion"):
+        result: Final = eligible_deployments("shared", (deployment,))
+    assert result == (deployment,)
+
+
+def test_a_single_endpoint_string_is_enforced_like_a_one_item_list() -> None:
+    deployment: Final = _deployment({"enforce_supported_endpoints": True, "supported_endpoints": "/v1/embeddings"})
+    with endpoint_selection("acompletion"):
+        result: Final = eligible_deployments("shared", (deployment,))
+    assert isinstance(result, EndpointNotSupportedByModelError)
+    assert result.message.endswith("Supported endpoints: /v1/embeddings")
+
+
+def test_unsupported_error_explains_the_requested_operation_and_available_endpoints() -> None:
+    deployment: Final = _deployment({"enforce_supported_endpoints": True, "supported_endpoints": ["/v1/embeddings"]})
+    with endpoint_selection("acreate_interaction"):
+        result: Final = eligible_deployments("shared", (deployment,))
+    assert isinstance(result, EndpointNotSupportedByModelError)
+    assert result.status_code == 400
+    assert "/v1beta/interactions" in result.message
+    assert "/v1/embeddings" in result.message
+
+
+@pytest.mark.parametrize("call_type", ["aembedding", "aspeech", "_arealtime", "aget_responses"])
+def test_chat_restriction_does_not_block_other_operations(call_type: str) -> None:
+    deployment: Final = _deployment({"enforce_supported_endpoints": True, "supported_endpoints": ["/v1/embeddings"]})
+    with endpoint_selection(call_type):
+        result: Final = eligible_deployments("shared", (deployment,))
+    assert result == (deployment,)
+
+
+def test_nested_non_chat_calls_do_not_inherit_the_outer_chat_restriction() -> None:
+    deployment: Final = _deployment({"enforce_supported_endpoints": True, "supported_endpoints": ["/v1/embeddings"]})
+    with endpoint_selection("acompletion"):
+        with endpoint_selection("aembedding"):
+            embedding_result: Final = eligible_deployments("shared", (deployment,))
+        chat_result: Final = eligible_deployments("shared", (deployment,))
+    assert embedding_result == (deployment,)
+    assert isinstance(chat_result, EndpointNotSupportedByModelError)
+
+
+@pytest.mark.parametrize(
+    "endpoint",
+    [
+        "/messages",
+        "/interactions",
+        "/responses/compact",
+        "/openai/v1/responses/compact",
+        "/openai/v1/chat/completions",
+        "chat/completions",
+        "v1/responses",
+        " /v1/chat/completions ",
+        "/v1/chat/completions/",
+        "/V1/Chat/Completions",
+        "/v1beta/models/gemini-3-flash:generateContent",
+    ],
+)
+def test_chat_endpoint_declarations_keep_their_meaning_across_prefixes(endpoint: str) -> None:
+    deployment: Final = _deployment({"enforce_supported_endpoints": True, "supported_endpoints": [endpoint]})
+    with endpoint_selection("acompletion"):
+        result: Final = eligible_deployments("shared", (deployment,))
+    assert result == (deployment,)
+
+
+@pytest.mark.parametrize(
+    "info",
+    [
+        {"enforce_supported_endpoints": True},
+        {"enforce_supported_endpoints": True, "supported_endpoints": ["/v1/embeddings", None]},
+        {"enforce_supported_endpoints": "true", "supported_endpoints": 7},
+    ],
+)
+def test_a_flag_without_a_usable_endpoint_list_is_reported(info: Mapping[str, object]) -> None:
+    warning: Final = warn_on_unenforced_endpoint_flag("embed-only", info)
+    assert warning is not None
+    assert "'embed-only'" in warning
+
+
+@pytest.mark.parametrize(
+    "info",
+    [
+        {"enforce_supported_endpoints": True, "supported_endpoints": ["/v1/embeddings"]},
+        {"enforce_supported_endpoints": True, "supported_endpoints": "/v1/embeddings"},
+        {"enforce_supported_endpoints": False},
+        {"supported_endpoints": ["/v1/embeddings", None]},
+    ],
+)
+def test_an_enforceable_or_unflagged_deployment_is_not_reported(info: Mapping[str, object]) -> None:
+    assert warn_on_unenforced_endpoint_flag("embed-only", info) is None

tests/unit/test_router/test_router.py

@@ -12270,6 +12270,67 @@ def test_is_deployment_blocked_static_helper_reflects_blocked_flag():
     )
 
 
+def test_synchronous_text_completion_rejects_a_restricted_model_before_the_provider() -> None:
+    model: Final = "text-completion-openai/unit-text-model"
+    router: Final = Router(
+        model_list=[
+            {
+                "model_name": model,
+                "litellm_params": {
+                    "model": model,
+                    "api_base": "https://sync-router.invalid/v1",
+                    "api_key": "unit-provider-key",
+                },
+                "model_info": {"enforce_supported_endpoints": True, "supported_endpoints": ["/v1/embeddings"]},
+            }
+        ],
+        num_retries=0,
+    )
+    with respx.mock(assert_all_called=False) as upstream:
+        provider: Final = upstream.post("https://sync-router.invalid/v1/completions").respond(
+            200,
+            json={
+                "id": "cmpl-unit",
+                "object": "text_completion",
+                "created": 1,
+                "model": "unit-text-model",
+                "choices": [{"text": "provider called", "index": 0, "finish_reason": "stop", "logprobs": None}],
+                "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2},
+            },
+        )
+        with pytest.raises(litellm.BadRequestError, match="/v1/completions is not supported"):
+            router.text_completion(model=model, prompt="test")
+        assert provider.call_count == 0
+
+
+def test_routable_deployments_leave_out_models_that_refuse_the_call_type() -> None:
+    router: Final = Router(
+        model_list=[
+            {
+                "model_name": "available",
+                "litellm_params": {"model": "openai/gpt-5.4-nano", "api_key": "unit-provider-key"},
+            },
+            {
+                "model_name": "embed-only",
+                "litellm_params": {"model": "openai/text-embedding-3-small", "api_key": "unit-provider-key"},
+                "model_info": {"enforce_supported_endpoints": True, "supported_endpoints": ["/v1/embeddings"]},
+            },
+        ],
+        num_retries=0,
+    )
+
+    def routable(model: str, call_type: str | None) -> tuple[str, ...]:
+        return tuple(
+            deployment["model_name"] for deployment in router.get_routable_deployments(model, {}, call_type=call_type)
+        )
+
+    assert routable("unregistered-fallback", "acompletion") == ()
+    assert routable("available", "acompletion") == ("available",)
+    assert routable("embed-only", "acompletion") == ()
+    assert routable("embed-only", "aembedding") == ("embed-only",)
+    assert routable("embed-only", None) == ("embed-only",)
+
+
 def test_routing_groups_legacy_property_forwards_to_public_attribute():
     router = _router_with_two_deployments([False, False])
 
@@ -18803,6 +18864,35 @@ async def test_anthropic_messages_stream_dropped_before_content_is_retried_withi
     assert [(attempted, budget) for _, attempted, budget in provider.calls] == [(0, 2), (1, 2)]
 
 
+@pytest.mark.asyncio
+async def test_anthropic_messages_same_group_retry_skips_an_embedding_only_deployment():
+    """The retry of a /v1/messages stream dropped before content must not land on an enforced embedding-only
+    deployment of the same group, even when that deployment is the group's preferred order."""
+    router: Final = Router(
+        model_list=[
+            {"model_name": "glm", "litellm_params": {"model": "anthropic/glm-chat", "api_key": "sk-test", "order": 2}},
+            {
+                "model_name": "glm",
+                "litellm_params": {"model": "anthropic/glm-embed", "api_key": "sk-test", "order": 1},
+                "model_info": {"enforce_supported_endpoints": True, "supported_endpoints": ["/v1/embeddings"]},
+            },
+        ],
+        num_retries=1,
+    )
+    provider: Final = _AnthropicMessagesScriptedProvider(
+        _anthropic_messages_dropped_before_content, _anthropic_messages_retried_stream
+    )
+
+    async def anthropic_messages(**kwargs: object) -> object:
+        return await provider(**kwargs)
+
+    stream: Final = await _anthropic_messages_stream_through_router(router, anthropic_messages)
+    body: Final = [chunk async for chunk in stream]
+
+    assert body == [_anthropic_messages_message_start_chunk(), _anthropic_messages_content_chunk("pong")]
+    assert [model for model, _, _ in provider.calls] == ["anthropic/glm-chat", "anthropic/glm-chat"]
+
+
 class _AnthropicMessagesRetryStashRecorder:
     def __init__(self) -> None:
         self.observed: tuple[Usage | None, ...] = ()
@@ -31431,3 +31521,26 @@ def test_router_keeps_supported_fallback_shapes(
     router: Final = Router(model_list=list(_FALLBACK_VALIDATION_MODEL_LIST), **router_kwargs)
 
     assert router.fallbacks == expected_fallbacks
+
+
+def test_router_warns_when_a_deployment_enforces_endpoints_without_a_list(caplog: pytest.LogCaptureFixture) -> None:
+    with caplog.at_level(logging.WARNING, logger="LiteLLM Router"):
+        router: Final = Router(
+            model_list=[
+                {
+                    "model_name": "embed-from-config",
+                    "litellm_params": {"model": "openai/text-embedding-3-small", "api_key": "sk-fake"},
+                    "model_info": {"enforce_supported_endpoints": True},
+                }
+            ]
+        )
+        router.add_deployment(
+            Deployment(
+                model_name="embed-from-api",
+                litellm_params=LiteLLM_Params(model="openai/text-embedding-3-small", api_key="sk-fake"),
+                model_info=ModelInfo(enforce_supported_endpoints=True, supported_endpoints=["/v1/embeddings", 7]),
+            )
+        )
+
+    assert "Deployment 'embed-from-config' sets enforce_supported_endpoints without" in caplog.text
+    assert "Deployment 'embed-from-api' sets enforce_supported_endpoints without" in caplog.text

ui/litellm-dashboard/src/lib/http/schema.d.ts

@@ -50433,6 +50433,8 @@ export interface components {
             discoverable?: boolean | null;
             /** Enable Tag Filtering */
             enable_tag_filtering?: boolean | null;
+            /** Enforce Supported Endpoints */
+            enforce_supported_endpoints?: boolean | null;
             /** Explicit Cache Read Input Token Cost */
             explicit_cache_read_input_token_cost?: number | null;
             /** Id */