Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/common_utils/model_listing_utils.py: 57%
154 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1"""Team-scoped (BYOK) model-name translation for the model listing endpoints.
3`/v1/models`, `/models`, and `GET /v1/models/{id}` should surface the public
4`team_public_model_name` rather than the internal routing key
5`model_name_{team_id}_{uuid}`, consistent with `/v1/model/info`. The internal
6key still routes regardless; this is a presentation-layer swap only and does not
7touch access-group or auth semantics (see issue #28382). Operators can pin the
8legacy internal names with `general_settings.use_team_public_model_name: false`.
9"""
11from __future__ import annotations
13import re
14from collections.abc import Container, Mapping, Sequence
15from dataclasses import dataclass
16from functools import reduce
17from types import MappingProxyType
18from typing import TYPE_CHECKING, Final, cast
20from pydantic import TypeAdapter, ValidationError
22import litellm
24if TYPE_CHECKING: 24 ↛ 25line 24 didn't jump to line 25 because the condition on line 24 was never true
25 from litellm.router import Router
26 from litellm.types.proxy.model_listing import ModelInfoResponse
28CLAUDE_CODE_PICKER_PATTERN: Final = re.compile(r"claude|anthropic", re.IGNORECASE)
29GATEWAY_CLIENT_HEADER: Final = "x-gateway-client"
30CLAUDE_CODE_CLIENT: Final = "claude-code"
31_CLAUDE_CODE_ALIAS_PREFIX: Final = "claude-router-"
32_ONE_MILLION_SUFFIX: Final = "[1m]"
33_ONE_MILLION_TOKENS: Final = 1_000_000
34_ALIAS_ENTRIES: Final = TypeAdapter(Mapping[object, object])
35_NO_ALIASES: Final[Mapping[str, str]] = MappingProxyType({})
38def configured_display_names(
39 entries: Sequence[tuple[str, str]],
40 llm_router: Router | None,
41) -> Mapping[str, str]:
42 """response_id -> configured `model_info.display_name` for the listing entries
43 that have one.
45 Metadata is looked up by each entry's internal lookup id (so team-scoped rows
46 resolve), while the returned map is keyed by the public response id the
47 Anthropic-shaped listing is built from. Entries without a configured name are
48 omitted so the listing falls back to the id itself.
49 """
50 if llm_router is None:
51 return MappingProxyType({})
52 resolved: Final = (
53 (response_id, llm_router.get_configured_display_name(lookup_id)) for response_id, lookup_id in entries
54 )
55 return MappingProxyType(
56 {response_id: display_name for response_id, display_name in resolved if display_name is not None}
57 )
60def _unmarked(name: str) -> str:
61 return name[: -len(_ONE_MILLION_SUFFIX)] if name.lower().endswith(_ONE_MILLION_SUFFIX) else name
64def _compatibility_id(model_id: str) -> str:
65 return f"{_CLAUDE_CODE_ALIAS_PREFIX}{model_id.encode().hex()}"
68def _decoded_compatibility_id(view_id: str) -> str | None:
69 encoded: Final = _unmarked(view_id).removeprefix(_CLAUDE_CODE_ALIAS_PREFIX)
70 if encoded == _unmarked(view_id):
71 return None
72 try:
73 model_id: Final = bytes.fromhex(encoded).decode()
74 except (ValueError, UnicodeDecodeError):
75 return None
76 return model_id if _compatibility_id(model_id) == _unmarked(view_id) else None
79def claude_code_model_id(
80 model_id: str,
81 max_input_tokens: float | None,
82 routing_names: Container[str],
83) -> str:
84 """The collision-free id Claude Code's picker lists a model under."""
85 if "*" in model_id:
86 return model_id
87 shaped: Final = model_id if CLAUDE_CODE_PICKER_PATTERN.search(model_id) else _compatibility_id(model_id)
88 one_million: Final = max_input_tokens is not None and max_input_tokens >= _ONE_MILLION_TOKENS
89 marked: Final = (
90 f"{shaped}{_ONE_MILLION_SUFFIX}" if one_million and not shaped.lower().endswith(_ONE_MILLION_SUFFIX) else shaped
91 )
92 return next(
93 (
94 name
95 for name in (marked, shaped)
96 if name == model_id or claude_code_group_name(name, routing_names) == model_id
97 ),
98 model_id,
99 )
102def claude_code_group_name(view_id: str, routing_names: Container[str]) -> str | None:
103 """Decode a canonical compatibility id only when no configured route claims it."""
104 if view_id in routing_names:
105 return None
106 unmarked: Final = _unmarked(view_id)
107 if unmarked != view_id and unmarked in routing_names:
108 return unmarked
109 model_id: Final = _decoded_compatibility_id(view_id)
110 return model_id if model_id and model_id in routing_names else None
113def is_claude_code_client(headers: Mapping[str, str]) -> bool:
114 """Claude Code itself, or a client asking for its view of the listing the way Ramp Router's does"""
115 from litellm.llms.anthropic.common_utils import is_claude_code_user_agent
117 return (
118 is_claude_code_user_agent(headers.get("user-agent", ""))
119 or headers.get(GATEWAY_CLIENT_HEADER, "").lower() == CLAUDE_CODE_CLIENT
120 )
123def claude_code_view_ids(
124 rows: Sequence[ModelInfoResponse],
125 headers: Mapping[str, str],
126 routing_names: Container[str],
127) -> Mapping[str, str]:
128 """served id -> Claude Code id for the requested listing view"""
129 if not is_claude_code_client(headers):
130 return MappingProxyType({})
131 return MappingProxyType(
132 {row["id"]: claude_code_model_id(row["id"], row.get("max_input_tokens"), routing_names) for row in rows}
133 )
136@dataclass(frozen=True, slots=True)
137class ClaudeCodeRoutingNames:
138 """Existing routes always own their names, including aliases and wildcard routes."""
140 llm_router: Router | None
141 team_id: str | None = None
142 alias_maps: tuple[object, ...] = ()
144 def __contains__(self, name: object) -> bool:
145 if not isinstance(name, str):
146 return False
147 if name in litellm.model_alias_map or any(
148 isinstance(aliases, Mapping) and name in aliases for aliases in self.alias_maps
149 ):
150 return True
151 if self.llm_router is None:
152 return False
153 return (
154 name in self.llm_router.model_group_alias
155 or self.llm_router.has_model_id(name)
156 or bool(self.llm_router.get_candidate_model_ids_for_route(name, self.team_id))
157 )
160@dataclass(frozen=True, slots=True)
161class CallerAliases:
162 """`own` are the caller's key and team alias maps, the names `/v1/models` lists for it.
163 `rewrite` are the maps `/chat/completions` rewrites its model through, in the order it
164 applies them: the team's, the key's in `add_litellm_data_to_request`, then the global
165 `model_alias_map` and the key's again in `common_processing_pre_call_logic`."""
167 own: tuple[object, ...]
168 rewrite: tuple[object, ...]
171def caller_alias_maps(
172 key_aliases: object,
173 team_aliases: object,
174 key_team_id: str | None,
175 listed_team_id: str | None,
176) -> CallerAliases:
177 """Team aliases count only when listing the team the key authenticated as."""
178 if listed_team_id is not None and listed_team_id != key_team_id:
179 return CallerAliases((key_aliases,), (key_aliases, litellm.model_alias_map, key_aliases))
180 return CallerAliases((team_aliases, key_aliases), (team_aliases, key_aliases, litellm.model_alias_map, key_aliases))
183def alias_map(aliases: object) -> Mapping[str, str]:
184 try:
185 entries: Final = _ALIAS_ENTRIES.validate_python(aliases, strict=True)
186 except ValidationError:
187 return _NO_ALIASES
188 return MappingProxyType(
189 {alias: target for alias, target in entries.items() if isinstance(alias, str) and isinstance(target, str)}
190 )
193def _alias_names(alias_maps: Sequence[Mapping[str, str]]) -> tuple[str, ...]:
194 return tuple(dict.fromkeys(alias for aliases in alias_maps for alias in aliases))
197def _rewrite(model_id: str, alias_maps: Sequence[Mapping[str, str]]) -> str | None:
198 target: Final = reduce(lambda name, aliases: aliases.get(name, name), alias_maps, model_id)
199 return None if target == model_id else target
202def alias_target(model_id: str, aliases: CallerAliases, listed: Container[str] = frozenset()) -> str | None:
203 """The model group `/chat/completions` rewrites `model_id` to, else None. A `model_id`
204 already `listed` keeps its own row, so it is never rewritten."""
205 if model_id in listed: 205 ↛ 206line 205 didn't jump to line 206 because the condition on line 205 was never true
206 return None
207 return _rewrite(model_id, tuple(alias_map(raw) for raw in aliases.rewrite))
210def alias_listing_entries(
211 entries: Sequence[tuple[str, str]],
212 aliases: CallerAliases,
213) -> tuple[tuple[str, str], ...]:
214 """`entries` plus one `(alias, lookup_id)` row per key or team alias whose target is
215 listed. An alias colliding with a listed id keeps the listed entry."""
216 maps: Final = tuple(alias_map(raw) for raw in aliases.rewrite)
217 own: Final = tuple(alias_map(raw) for raw in aliases.own)
218 lookup_by_response: Final = MappingProxyType(dict(entries))
219 lookup_ids: Final = frozenset(lookup_by_response.values())
220 targets: Final = MappingProxyType(
221 {alias: _rewrite(alias, maps) for alias in _alias_names(own) if alias not in lookup_by_response}
222 )
223 added: Final = tuple(
224 (alias, lookup_by_response.get(target, target))
225 for alias, target in targets.items()
226 if target is not None and (target in lookup_by_response or target in lookup_ids)
227 )
228 return (*entries, *added)
231def claude_code_requested_group(
232 requested: str,
233 llm_router: Router,
234 team_id: str | None,
235 alias_maps: tuple[object, ...] = (),
236) -> str | None:
237 return claude_code_group_name(requested, ClaudeCodeRoutingNames(llm_router, team_id, alias_maps))
240class TeamModelNameTranslator:
241 """Translates internal team routing keys to their public names for the model
242 listing/retrieve responses. Stateless; the live router and general_settings
243 are injected per call so the unit tests can drive it without globals.
244 """
246 @staticmethod
247 def _internal_public_pair(model: object) -> tuple[str, str] | None:
248 """`(internal_routing_key, public_name)` for a team-scoped row, else None."""
249 if not isinstance(model, dict): 249 ↛ 250line 249 didn't jump to line 250 because the condition on line 249 was never true
250 return None
251 model_dict: Final = cast(dict[str, object], model) # any-ok: checked
252 model_info_raw: Final[object] = model_dict.get("model_info")
253 if not isinstance(model_info_raw, Mapping): 253 ↛ 254line 253 didn't jump to line 254 because the condition on line 253 was never true
254 return None
255 model_info: Final = cast(Mapping[str, object], model_info_raw) # any-ok: checked
256 team_id: Final = model_info.get("team_id")
257 team_public: Final = model_info.get("team_public_model_name")
258 name: Final = model_dict.get("model_name")
259 if ( 259 ↛ 267line 259 didn't jump to line 267 because the condition on line 259 was never true
260 isinstance(team_id, str)
261 and isinstance(team_public, str)
262 and isinstance(name, str)
263 and team_id
264 and team_public
265 and name.startswith(f"model_name_{team_id}_")
266 ):
267 return name, team_public
268 return None
270 @staticmethod
271 def _is_enabled(general_settings: Mapping[str, object]) -> bool:
272 return general_settings.get("use_team_public_model_name", True) is not False
274 @staticmethod
275 def build_internal_to_public_map(
276 llm_router: Router | None,
277 general_settings: Mapping[str, object],
278 ) -> dict[str, str]:
279 """Internal team routing key -> public `team_public_model_name`.
281 Empty when disabled via the legacy flag, the router is absent, or the
282 router model list is malformed.
283 """
284 if llm_router is None or not TeamModelNameTranslator._is_enabled(general_settings): 284 ↛ 285line 284 didn't jump to line 285 because the condition on line 284 was never true
285 return {}
286 router_model_list: Final = llm_router.get_model_list()
287 if not isinstance(router_model_list, list): 287 ↛ 288line 287 didn't jump to line 288 because the condition on line 287 was never true
288 return {}
289 return dict(
290 pair
291 for pair in (TeamModelNameTranslator._internal_public_pair(model) for model in router_model_list)
292 if pair is not None
293 )
295 @staticmethod
296 def _response_to_lookup_map(
297 model_names: Sequence[str],
298 internal_to_public: dict[str, str],
299 ) -> dict[str, str]:
300 """Map each public response id to the first internal lookup id seen in
301 `model_names`, preserving first-occurrence order. First-wins keeps list
302 and retrieve in agreement on which accessible deployment a shared public
303 id resolves to: a global iterated before a colliding team alias stays
304 the listed entry, and sibling team rows collapse to their first
305 occurrence.
306 """
307 result: Final[dict[str, str]] = {}
308 for name in model_names:
309 result.setdefault(internal_to_public.get(name, name), name)
310 return result
312 @staticmethod
313 def listing_entries(
314 model_names: Sequence[str],
315 llm_router: Router | None,
316 general_settings: Mapping[str, object],
317 ) -> list[tuple[str, str]]:
318 """`(response_id, metadata_lookup_id)` for each listed model, de-duplicated
319 by response_id while preserving order.
321 For team-scoped rows `response_id` is the public name shown to the client,
322 while `metadata_lookup_id` stays the internal routing key so downstream
323 metadata/fallback lookups (keyed by the routing name) still resolve. The
324 lookup id is always one of `model_names` (the caller's accessible set), so
325 a public name shared across teams never resolves to another team's
326 internal key. Both ids are identical for unmapped names (globals,
327 access-group keys).
328 """
329 internal_to_public: Final = TeamModelNameTranslator.build_internal_to_public_map(llm_router, general_settings)
330 if not internal_to_public: 330 ↛ 332line 330 didn't jump to line 332 because the condition on line 330 was always true
331 return [(name, name) for name in model_names]
332 return list(TeamModelNameTranslator._response_to_lookup_map(model_names, internal_to_public).items())
334 @staticmethod
335 def translate_listing(
336 model_names: list[str],
337 llm_router: Router | None,
338 general_settings: Mapping[str, object],
339 ) -> list[str]:
340 """Public-name view of `model_names` (the `response_id` of each listing
341 entry). Sibling deployments sharing a public name collapse to one entry
342 while preserving order; unmapped names pass through.
343 """
344 return [
345 entry[0] for entry in TeamModelNameTranslator.listing_entries(model_names, llm_router, general_settings)
346 ]
348 @staticmethod
349 def resolve_public_name(
350 model_id: str,
351 available_models: list[str],
352 llm_router: Router | None,
353 general_settings: Mapping[str, object],
354 ) -> str:
355 """Resolve a public team name back to the internal routing key the router
356 indexes by, so `GET /v1/models/{id}` accepts the name the listing returns.
358 Resolution is restricted to `available_models` (the caller's accessible
359 set) so colliding public names across teams never resolve across an access
360 boundary. Uses the same first-occurrence dedup as `listing_entries` so a
361 public id advertised by `/v1/models` resolves to the same internal
362 deployment that the listing's metadata was built from. Returns `model_id`
363 unchanged when it is not an accessible public team name (already-internal
364 names and globals pass through).
365 """
366 internal_to_public: Final = TeamModelNameTranslator.build_internal_to_public_map(llm_router, general_settings)
367 if not internal_to_public: 367 ↛ 369line 367 didn't jump to line 369 because the condition on line 367 was always true
368 return model_id
369 return TeamModelNameTranslator._response_to_lookup_map(available_models, internal_to_public).get(
370 model_id, model_id
371 )