Merge 995045d0cf into aeede53112

.github/workflows/test-linting.yml

@@ -191,10 +191,10 @@ jobs:
       - name: Check tests/e2e basedpyright (zero errors)
         if: steps.changes.outputs.decision != 'skip'
         run: |
-          if git diff --name-only --diff-filter=ACMRD "$GATE_BASE_SHA" HEAD -- ':(glob)tests/e2e/**/*.py' ':(glob)tests/e2e_harness/**/*.py' pyrightconfig.json | grep -q .; then
+          if git diff --name-only --diff-filter=ACMRD "$GATE_BASE_SHA" HEAD -- ':(glob)tests/e2e/**/*.py' ':(glob)tests/e2e_harness/**/*.py' ':(glob)litellm/**/*.py' pyrightconfig.json pyproject.toml uv.lock | grep -q .; then
             uv run --no-sync basedpyright tests/e2e tests/e2e_harness
           else
-            echo "No changed tests/e2e Python files; skipping."
+            echo "No changed tests/e2e, litellm Python, or dependency files; skipping."
           fi
 
       - name: Run the e2e harness tests
@@ -623,6 +623,9 @@ jobs:
       - name: check_e2e_no_raw_requests
         run: uv run --no-sync python ./tests/code_coverage_tests/check_e2e_no_raw_requests.py
 
+      - name: check_router_constructor_stub
+        run: uv run --no-sync python ./tests/code_coverage_tests/check_router_constructor_stub.py
+
       - name: check_migrations_no_data_rewrites
         run: uv run --no-sync python ./tests/code_coverage_tests/check_migrations_no_data_rewrites.py
 

AGENTS.md

@@ -101,7 +101,7 @@ Follow these coding conventions for new/updated code (a three-line fix in a lega
 - Never-nester: early returns over deep nesting
 - Don't throw. Model failures as values (One function (e.g., raise_public) maps error union to existing public exception contracts via exhaustive match + assert_never)
 - No mutation. Don't reassign variables, global or local. Instead of mutable lists and dicts, prefer tuples, frozen dataclasses (with slots=True), `MappingProxyType`, etc.
-  - Annotate every variable with `: Final` (LIT010). Unpacking and walrus targets cannot carry the annotation, so they are implicitly final. Don't rebind them. Never rebind or mutate function parameters (LIT011). `self`/`cls` attribute stores are the exception. If rebinding or in-place mutation is truly unavoidable, suppress with `# rebind-ok: <reason>`
+  - Annotate every variable with `: Final` (LIT010). A `TypeVar`, `ParamSpec`, or `TypeVarTuple` is the exception and stays bare, since mypy reads a Final one as `Any`. Unpacking and walrus targets cannot carry the annotation, so they are implicitly final. Don't rebind them. Never rebind or mutate function parameters (LIT011). `self`/`cls` attribute stores are the exception. If rebinding or in-place mutation is truly unavoidable, suppress with `# rebind-ok: <reason>`
   - Qualify every TypedDict field with `ReadOnly[...]` (LIT012), which nests freely with `Required` / `NotRequired` / `Annotated` in any order. If making the key writable is truly unavoidable, suppress with `# writable-ok: <reason>`
   - Comprehensions take at most one `for` clause and one `if` clause (LIT014). Split stacked clauses into a helper generator, a named intermediate, or a plain loop. Suppress with `# comprehension-ok: <reason>` only when unavoidable
   - Every pydantic model must be frozen (LIT015), set via `model_config = ConfigDict(frozen=True)`, a dict-literal `model_config`, an inner `class Config`, or the class keywords. Subclasses inherit it unless they override it. Replace in-place field writes with `model_copy(update=...)`. If making the model mutable is truly unavoidable, suppress with `# frozen-ok: <reason>` on the `class` line

litellm/caching/redis_batch.py

@@ -43,7 +43,7 @@ if TYPE_CHECKING:
     from litellm.rust_bridge._native import NativeRequestRedis, NativeSend
 
 _T = TypeVar("_T")
-_T_co: Final = TypeVar("_T_co", covariant=True)
+_T_co = TypeVar("_T_co", covariant=True)
 _ScriptArg = str | bytes | int | float
 SettledHook = Callable[[asyncio.Future[_T]], Awaitable[None] | None]
 POST_CALL_FLUSH_DEADLINE_SECONDS: Final = 1.0

litellm/caching/redis_cluster_cache.py

@@ -31,7 +31,7 @@ else:
     Span = Any
 
 
-_T: Final = TypeVar("_T")
+_T = TypeVar("_T")
 
 
 def _connection_limit(max_connections: object) -> int | None:

litellm/google_genai/adapters/transformation.py

@@ -118,7 +118,7 @@ _JSON_MIME_TYPE: Final = "application/json"
 _GEMINI_ONLY_SCHEMA_KEYS: Final = frozenset({"propertyOrdering", "property_ordering"})
 _CONFIG_FIELDS: Final = TypeAdapter(Mapping[str, object])
 _JSON_OBJECT_SCHEMA: Final = TypeAdapter(dict[str, JsonValue])
-_Validated: Final = TypeVar("_Validated")
+_Validated = TypeVar("_Validated")
 _GENAI_CONTENT_PARTS: Final = TypeAdapter(Sequence[object])
 _GENAI_INLINE_DATA: Final = TypeAdapter(_GenAIInlineData)
 _GENAI_FILE_DATA: Final = TypeAdapter(_GenAIFileData)

litellm/integrations/anthropic_cache_control_hook.py

@@ -140,7 +140,7 @@ def _validated_object_list(value: object) -> list[object] | None:
         return None
 
 
-_MessageT: Final = TypeVar("_MessageT")
+_MessageT = TypeVar("_MessageT")
 _CacheControlRewrite: TypeAlias = Callable[[object], ChatCompletionCachedContent | None]
 _MARK_CARRYING_MESSAGE_FIELDS: Final = ("content", "tool_calls")
 

litellm/integrations/prometheus.py

@@ -111,8 +111,8 @@ class _BudgetedCustomerFilter(TypedDict):
     budget_id: ReadOnly[_IsNotNull]
 
 
-_BudgetRowT: Final = TypeVar("_BudgetRowT")
-_TableRowT: Final = TypeVar("_TableRowT", bound=BaseModel)
+_BudgetRowT = TypeVar("_BudgetRowT")
+_TableRowT = TypeVar("_TableRowT", bound=BaseModel)
 
 _DEFAULT_BUDGET_METRICS_PER_REQUEST_TIMEOUT: Final = 5.0
 
@@ -263,7 +263,7 @@ class _LabeledMetric:
 
 _MetricLike: TypeAlias = "NoOpMetric | _LabeledMetric | MetricWrapperBase"
 
-_SeriesLimitT: Final = TypeVar("_SeriesLimitT", int, float)
+_SeriesLimitT = TypeVar("_SeriesLimitT", int, float)
 _REQUEST_METADATA_ADAPTER: Final[TypeAdapter[Mapping[str, object]]] = TypeAdapter(Mapping[str, object])
 _POSITIVE_SERIES_CAP: Final[TypeAdapter[int]] = TypeAdapter(Annotated[int, Field(gt=0)])
 _POSITIVE_SERIES_TTL: Final[TypeAdapter[float]] = TypeAdapter(Annotated[float, Field(gt=0)])

litellm/integrations/websearch_interception/handler.py

@@ -102,7 +102,7 @@ _RESPONSE_CONTENT_FIELD: Final = "content"
 
 _CLAUDE_CODE_WEB_SEARCH_PROMPT_PREFIX: Final = "Perform a web search for the query:"
 
-_ResponseT: Final = TypeVar("_ResponseT")
+_ResponseT = TypeVar("_ResponseT")
 
 
 class _PlanMetadataView(TypedDict):

litellm/litellm_core_utils/asyncify.py

@@ -8,7 +8,7 @@ import anyio
 import anyio.to_thread
 from typing_extensions import ParamSpec, TypeVar
 
-T_ParamSpec: Final = ParamSpec("T_ParamSpec")
+T_ParamSpec = ParamSpec("T_ParamSpec")
 T_Retval = TypeVar("T_Retval")
 
 

litellm/litellm_core_utils/batch_utils.py

@@ -32,7 +32,7 @@ BATCH_COST_REQUEST_ID_SUFFIX: Final = "_batch_cost"
 _AZURE_PROVIDERS: Final = frozenset({LlmProviders.AZURE.value, LlmProviders.AZURE_AI.value})
 
 _batch_create_routing: Final[ContextVar[bool]] = ContextVar("batch_create_routing", default=False)
-_DeploymentT: Final = TypeVar("_DeploymentT", bound=Mapping[str, object])
+_DeploymentT = TypeVar("_DeploymentT", bound=Mapping[str, object])
 
 
 class InMemoryFile(io.BytesIO):

litellm/litellm_core_utils/chat_completion_agentic_loop.py

@@ -37,7 +37,7 @@ from litellm.types.integrations.custom_logger import (
 from litellm.types.utils import ModelResponse
 from litellm.utils import CustomStreamWrapper
 
-_DispatchResult: Final = TypeVar("_DispatchResult")
+_DispatchResult = TypeVar("_DispatchResult")
 
 _FOLLOWUP_INTERNAL_PARAMS: Final = frozenset(
     (

litellm/litellm_core_utils/core_helpers.py

@@ -140,7 +140,7 @@ def attribute_or_none(source: object, name: str) -> object:
     return getattr(source, name, None)
 
 
-_Shape: Final = TypeVar("_Shape")
+_Shape = TypeVar("_Shape")
 
 
 def has_attributes(

litellm/litellm_core_utils/env_utils.py

@@ -6,7 +6,7 @@ import logging
 import os
 from typing import Final, TypeVar
 
-_NumberT: Final = TypeVar("_NumberT", int, float)
+_NumberT = TypeVar("_NumberT", int, float)
 
 # Marked before logging: the LiteLLM loggers' own filters read env knobs through these helpers, so
 # warning first would re-enter the helper from inside the log call and recurse until RecursionError.

litellm/litellm_core_utils/litellm_logging.py

@@ -601,7 +601,7 @@ _LITELLM_PARAMS: Final = TypeAdapter(dict[str, object])
 _SERVED_ANTHROPIC_BODY: Final = TypeAdapter(dict[str, object])
 _GOOGLE_USAGE_FIELDS: Final = frozenset(UsageMetadata.__annotations__)
 _REDACTED_RESULT: Final = TypeAdapter(object)
-_LoggedResponse: Final = TypeVar("_LoggedResponse")
+_LoggedResponse = TypeVar("_LoggedResponse")
 _GUARDRAIL_INFORMATION_KEY: Final = "standard_logging_guardrail_information"
 
 

litellm/litellm_core_utils/prompt_templates/factory.py

@@ -2868,7 +2868,7 @@ def _unsigned_thinking_identity(kind: str, thinking: object, signature: object)
             return None
 
 
-_Replayed: Final = TypeVar("_Replayed")
+_Replayed = TypeVar("_Replayed")
 
 
 def _with_missing_thinking_blocks(

litellm/litellm_core_utils/prompt_templates/mid_conversation_system.py

@@ -63,7 +63,7 @@ _RENDERED_ASSISTANT_PART_TYPES: Final = frozenset({"text", "server_tool_use"})
 _THINKING_BLOCK_TYPES: Final = frozenset({"thinking", "redacted_thinking"})
 
 _MessageKind: TypeAlias = Literal["system", "tool", "user", "other"]
-_Message: Final = TypeVar("_Message")
+_Message = TypeVar("_Message")
 _TextPart: TypeAlias = tuple[str, ChatCompletionCachedContent | None]
 
 

litellm/litellm_core_utils/prompt_templates/prompt_cache_breakpoint_utils.py

@@ -11,7 +11,7 @@ from litellm.types.llms.openai import (
     PromptCacheBreakpoint,
 )
 
-_MarkedT: Final = TypeVar("_MarkedT", bound=Mapping[str, object])
+_MarkedT = TypeVar("_MarkedT", bound=Mapping[str, object])
 
 
 PROMPT_CACHE_BREAKPOINT_DROP_PARAMS_HINT: Final = (
@@ -33,7 +33,7 @@ _PROMPT_CACHE_BREAKPOINT: Final = TypeAdapter(PromptCacheBreakpoint)
 _CHAT_COMPLETION_PROMPT_CACHE_BREAKPOINT: Final = TypeAdapter(ChatCompletionPromptCacheBreakpoint)
 
 
-_AcceptedBreakpointT: Final = TypeVar("_AcceptedBreakpointT", bound=Mapping[str, object])
+_AcceptedBreakpointT = TypeVar("_AcceptedBreakpointT", bound=Mapping[str, object])
 
 
 def _accepted_prompt_cache_breakpoint(

litellm/litellm_core_utils/ptu_shares.py

@@ -20,7 +20,7 @@ from litellm.llms.azure.ptu_capacity import (
     is_str_keyed_mapping,
 )
 
-_DeploymentT: Final = TypeVar("_DeploymentT", bound=Mapping[str, object])
+_DeploymentT = TypeVar("_DeploymentT", bound=Mapping[str, object])
 WildcardRoute: TypeAlias = Callable[[str], Sequence[Mapping[str, object]]]
 
 

litellm/litellm_core_utils/single_flight.py

@@ -5,8 +5,8 @@ from dataclasses import dataclass, field
 from functools import lru_cache
 from typing import Final, Generic, ParamSpec, TypeVar
 
-_P: Final = ParamSpec("_P")
-_R: Final = TypeVar("_R")
+_P = ParamSpec("_P")
+_R = TypeVar("_R")
 
 
 @dataclass(slots=True)

litellm/litellm_core_utils/streaming_chunk_builder_utils.py

@@ -1327,7 +1327,7 @@ def _as_text_completion_chunk(
     return _text_completion_chunk_from_mapping(chunk)
 
 
-_LogprobsValue: Final = TypeVar("_LogprobsValue")
+_LogprobsValue = TypeVar("_LogprobsValue")
 
 
 def _joined_or_none(fields: Iterable[Sequence[_LogprobsValue] | None]) -> list[_LogprobsValue] | None:

litellm/litellm_core_utils/token_counter.py

@@ -611,7 +611,7 @@ def estimate_image_tokens(
 
 
 _MEDIA_FILE_CHARS_PER_TOKEN: Final = 4
-_EmbeddingInputT: Final = TypeVar("_EmbeddingInputT")
+_EmbeddingInputT = TypeVar("_EmbeddingInputT")
 
 
 def _max_input_tokens(model: str) -> int | None:
@@ -766,7 +766,7 @@ class _PdfReader(Protocol):
 
 
 EXTRAPOLATION_SAMPLES: Final = 16
-T_ParamSpec: Final = ParamSpec("T_ParamSpec")
+T_ParamSpec = ParamSpec("T_ParamSpec")
 T_Retval = TypeVar("T_Retval")
 _COUNT_OFFLOAD_LIMITER: Final = anyio.lowlevel.RunVar[anyio.CapacityLimiter]("litellm_count_offload_limiter")
 _ANTHROPIC_TOOL_KEYS_MAPPED_BY_BRIDGE: Final = frozenset(

litellm/litellm_core_utils/tool_fields.py

@@ -180,7 +180,7 @@ def resolve_deployment_tool_fields(
     return resolve_tool_fields(kwargs, lambda: (drop_policy(kwargs),))
 
 
-_ToolT: Final = TypeVar("_ToolT")
+_ToolT = TypeVar("_ToolT")
 
 HOSTED_MCP_DROP_WARNING: Final = (
     "Dropping the hosted MCP tool(s) %s for model=%s on %s (drop_params=True) because this route cannot run a "
@@ -295,7 +295,7 @@ def tools_without_mcp_servers(
 
 
 _TOOL_SELECTION_PARAMS: Final = frozenset({"tools", "tool_choice", "parallel_tool_calls"})
-_ParamsT: Final = TypeVar("_ParamsT", bound=dict[str, object])
+_ParamsT = TypeVar("_ParamsT", bound=dict[str, object])
 
 
 def tool_params_without_hosted_mcp(

litellm/llms/anthropic/messages/adapters/transformation.py

@@ -427,7 +427,7 @@ class AnthropicAdapter:
         return anthropic_wrapper.anthropic_sse_wrapper()
 
 
-_BlockT: Final = TypeVar("_BlockT", bound=Mapping[str, object])
+_BlockT = TypeVar("_BlockT", bound=Mapping[str, object])
 
 
 _CLAUDE_OVER_REASONING_EFFORT_PROVIDERS: Final = frozenset({LlmProviders.KYMA.value})

litellm/llms/anthropic/messages/context_management/editors/compact.py

@@ -65,7 +65,7 @@ from ..result import PolyfillResult
 
 _SUMMARY_TAG_RE: Final = re.compile(r"<summary>(.*?)</summary>", re.IGNORECASE | re.DOTALL)
 
-_MsgT: Final = TypeVar("_MsgT", bound=Mapping[str, object])
+_MsgT = TypeVar("_MsgT", bound=Mapping[str, object])
 
 
 def _as_object(value: object) -> object:

litellm/llms/anthropic/messages/utils.py

@@ -27,7 +27,7 @@ _EFFORT_DEGRADATION_CHAIN: Final[Mapping[str, tuple[str, ...]]] = MappingProxyTy
 )
 _THINKING_OFF: Final = "none"
 
-_JsonValue: Final = TypeVar("_JsonValue")
+_JsonValue = TypeVar("_JsonValue")
 
 
 def json_copy_or_original(value: _JsonValue) -> _JsonValue:

litellm/llms/bedrock/chat/converse_to_openai.py

@@ -21,7 +21,7 @@ from litellm.types.llms.openai import (
 )
 
 JsonPath: TypeAlias = tuple[str | int, ...]
-_Node: Final = TypeVar("_Node")
+_Node = TypeVar("_Node")
 _OBJECT: Final = TypeAdapter(Mapping[str, object])
 _ITEMS: Final = TypeAdapter(Sequence[object])
 

litellm/llms/bedrock/passthrough/guardrail_translation/handler.py

@@ -849,7 +849,7 @@ class InvokeStreamFrame:
 
 
 _JSON_OBJECT: Final = TypeAdapter(Mapping[str, object])
-_Frame: Final = TypeVar("_Frame", ConverseStreamFrame, InvokeStreamFrame)
+_Frame = TypeVar("_Frame", ConverseStreamFrame, InvokeStreamFrame)
 
 
 def _event_message(raw: bytes) -> tuple[str, bytes] | None:

litellm/llms/custom_httpx/http_handler.py

@@ -76,7 +76,7 @@ _RequestContent: TypeAlias = str | bytes | Iterable[bytes] | AsyncIterable[bytes
 
 _IPV4_LOCAL_ADDRESS: Final = "0.0.0.0"
 
-_HttpxTransportT: Final = TypeVar("_HttpxTransportT", httpx.BaseTransport, httpx.AsyncBaseTransport)
+_HttpxTransportT = TypeVar("_HttpxTransportT", httpx.BaseTransport, httpx.AsyncBaseTransport)
 
 HTTPEventHooks: TypeAlias = Mapping[str, list[Callable[..., object]]]
 

litellm/llms/minimax/videos/transformation.py

@@ -176,7 +176,7 @@ _CREATE_REQUEST_PARAMS: Final = frozenset({*_CREATE_BODY_KEYS, "content", "input
 
 _TASK_NOT_FOUND_MESSAGE: Final = "record not found"
 
-_ParsedT: Final = TypeVar("_ParsedT")
+_ParsedT = TypeVar("_ParsedT")
 
 
 def _minimax_error(status_code: int, message: str) -> MinimaxVideoError:

litellm/llms/openai/responses/guardrail_translation/handler.py

@@ -610,7 +610,7 @@ def tool_output_texts(item: object) -> tuple[tuple[int | None, str], ...]:
     return ((None, tool_content),) if isinstance(tool_content, str) and tool_content else ()
 
 
-_Grouped: Final = TypeVar("_Grouped")
+_Grouped = TypeVar("_Grouped")
 
 
 def _pair_call_id(pair: tuple[str, object]) -> str:

litellm/llms/vertex_ai/common_utils.py

@@ -754,7 +754,7 @@ def _integer_enums_fit(schema: JsonValue) -> bool:
     )
 
 
-_GeminiRequestPart: Final = TypeVar("_GeminiRequestPart")
+_GeminiRequestPart = TypeVar("_GeminiRequestPart")
 
 
 def with_integer_enums_that_fit(

litellm/llms/vertex_ai/context_caching/rejected_cache_retry.py

@@ -13,8 +13,8 @@ from litellm.utils import is_cached_message
 
 from .known_cached_contents import names_rejected_cache
 
-Built: Final = TypeVar("Built")
-Sent: Final = TypeVar("Sent")
+Built = TypeVar("Built")
+Sent = TypeVar("Sent")
 
 
 def _build_from_copies(

litellm/llms/vertex_ai/gemini/long_decode.py

@@ -63,7 +63,7 @@ _CANDIDATE_COUNT_KEYS: Final = ("candidate_count", "candidateCount")
 LongDecodeHops: TypeAlias = tuple[GenerateContentResponseBody, ...]
 AsyncHopPost: TypeAlias = Callable[[RequestBody], Awaitable[httpx.Response]]
 SyncHopPost: TypeAlias = Callable[[RequestBody], httpx.Response]
-_Metadata: Final = TypeVar("_Metadata", bound=Mapping[str, object])
+_Metadata = TypeVar("_Metadata", bound=Mapping[str, object])
 
 
 @dataclass(frozen=True, slots=True)

litellm/llms/vertex_ai/gemini/rejected_thought_signature_retry.py

@@ -11,8 +11,8 @@ from litellm.types.llms.vertex_ai import ContentType, PartType, RequestBody
 
 from ..context_caching.rejected_cache_retry import status_and_message
 
-Body: Final = TypeVar("Body", bound=Mapping[str, object])
-Sent: Final = TypeVar("Sent")
+Body = TypeVar("Body", bound=Mapping[str, object])
+Sent = TypeVar("Sent")
 
 _REJECTED_SIGNATURE_MARKER: Final = "thought signature"
 _SIGNATURE_KEYS: Final = frozenset({"thoughtSignature", "thought_signature"})

litellm/llms/volcengine/videos/transformation.py

@@ -36,7 +36,7 @@ if TYPE_CHECKING:
     from litellm.litellm_core_utils.litellm_logging import Logging
 
 
-_ParsedT: Final = TypeVar("_ParsedT")
+_ParsedT = TypeVar("_ParsedT")
 
 
 class VolcEngineVideoError(BaseLLMException):

litellm/proxy/auth/model_checks.py

@@ -44,7 +44,7 @@ _PROVIDERS_LISTED_ONLY_THROUGH_A_DEPLOYMENT: Final = frozenset(
     {LlmProviders.OPENROUTER.value, LlmProviders.HOSTED_VLLM.value, LlmProviders.LITELLM_PROXY.value}
 )
 
-_Listed: Final = TypeVar("_Listed")
+_Listed = TypeVar("_Listed")
 
 
 async def run_model_listing(listing: Callable[[], _Listed]) -> _Listed:

litellm/proxy/bug_report_config.py

@@ -108,7 +108,7 @@ def _litellm_settings_keys() -> frozenset[str]:
 def _router_settings_keys() -> frozenset[str]:
     from litellm.router import Router
 
-    return frozenset(name for name in inspect.signature(Router.__init__).parameters if name != "self")  # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType]  # untyped params, only names are read
+    return frozenset(inspect.signature(Router).parameters)
 
 
 @functools.cache

litellm/proxy/client/cli/commands/agents.py

@@ -540,7 +540,7 @@ class ModelUpstream:
 
 
 _NO_UPSTREAMS: Final[Mapping[str, ModelUpstream]] = MappingProxyType({})
-_Pinned: Final = TypeVar("_Pinned", bound=Hashable)
+_Pinned = TypeVar("_Pinned", bound=Hashable)
 
 
 def _public_name(deployment: _Deployment) -> str:
@@ -1230,7 +1230,7 @@ def _launch(
         raise click.ClickException(str(e))
 
 
-_Decorated: Final = TypeVar("_Decorated", bound=Callable[..., object] | click.Command)
+_Decorated = TypeVar("_Decorated", bound=Callable[..., object] | click.Command)
 
 
 def _subscription_option(binary: str) -> Callable[[_Decorated], _Decorated]:

litellm/proxy/client/cli/commands/auth.py

@@ -138,7 +138,7 @@ class CliAuthResult(TypedDict):
     team_id: str | None
 
 
-_TeamMapping: Final = TypeVar("_TeamMapping", bound=Mapping[str, object])
+_TeamMapping = TypeVar("_TeamMapping", bound=Mapping[str, object])
 
 _JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
 

litellm/proxy/common_request_processing.py

@@ -160,8 +160,8 @@ from litellm.types.router_weights import RouterWeights, validate_router_weights
 
 _LateResponseT = TypeVar("_LateResponseT", bound=Response)
 _LlmCallT = TypeVar("_LlmCallT")
-_ResponseT: Final = TypeVar("_ResponseT")
-_ChunkT: Final = TypeVar("_ChunkT")
+_ResponseT = TypeVar("_ResponseT")
+_ChunkT = TypeVar("_ChunkT")
 
 KNOWN_PROXY_ROUTES: Final = frozenset(
     route for member in LiteLLMRoutes for route in member.value if route.startswith("/")

litellm/proxy/common_utils/openapi_route.py

@@ -12,7 +12,7 @@ from starlette.routing import BaseRoute, Route
 from litellm.proxy.common_utils.custom_openapi_spec import JsonObject
 
 OpenAPISchema: TypeAlias = JsonObject
-Inputs: Final = TypeVar("Inputs")
+Inputs = TypeVar("Inputs")
 _SERVERS: Final = TypeAdapter(list[dict[str, object]])
 
 

litellm/proxy/common_utils/token_count_log.py

@@ -25,9 +25,9 @@ def _logfmt_value(value: str | None) -> str:
     return json.dumps(value).translate(_JSON_ESCAPED_BRACES)
 
 
-_Counted: Final = TypeVar("_Counted")
-_Rejection: Final = TypeVar("_Rejection")
-_ProviderCount: Final = TypeVar("_ProviderCount", bound=Mapping[str, object])
+_Counted = TypeVar("_Counted")
+_Rejection = TypeVar("_Rejection")
+_ProviderCount = TypeVar("_ProviderCount", bound=Mapping[str, object])
 
 
 @dataclass(frozen=True, slots=True)

litellm/proxy/db/exception_handler.py

@@ -602,7 +602,7 @@ def _coerce_timeout(value: object, fallback: float) -> float:
     return fallback
 
 
-_ReadResultT: Final = TypeVar("_ReadResultT")
+_ReadResultT = TypeVar("_ReadResultT")
 
 
 class _DBReconnectClient(Protocol):

litellm/proxy/db/prisma_client.py

@@ -105,7 +105,7 @@ class _PrismaDrainTracker:
         await self._drained.wait()
 
 
-_EngineResponseT: Final = TypeVar("_EngineResponseT")
+_EngineResponseT = TypeVar("_EngineResponseT")
 
 
 class _TrackedPrismaEngine:

litellm/proxy/guardrails/guardrail_hooks/akto/akto_attachments.py

@@ -42,7 +42,7 @@ _FILE_SOURCE_FIELDS: Final = frozenset(("file_data", "file_id"))
 _ATTACHMENT_BLOCK_TYPES: Final = frozenset(_FILE_CHECKED_FIELDS)
 _OBJECT_MAPPING: Final[TypeAdapter[dict[str, object]]] = TypeAdapter(dict[str, object])
 
-_T: Final = TypeVar("_T")
+_T = TypeVar("_T")
 
 
 @dataclass(frozen=True, slots=True)

litellm/proxy/guardrails/guardrail_hooks/custom_code/bounded_execution.py

@@ -22,7 +22,7 @@ from typing import Final, Generic, TypeVar
 
 from litellm._logging import verbose_proxy_logger
 
-T: Final = TypeVar("T")
+T = TypeVar("T")
 
 _INTERRUPT_GRACE_SECONDS: Final = 1.0
 _INTERRUPT_POLL_SECONDS: Final = 0.05

litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/content_filter.py

@@ -82,7 +82,7 @@ from .patterns import PATTERN_EXTRA_CONFIG, get_compiled_pattern
 
 MAX_KEYWORD_VALUE_GAP_WORDS: Final = 1
 _DECODED_ARGUMENTS: Final = TypeAdapter(object)
-_StreamItemT: Final = TypeVar("_StreamItemT")
+_StreamItemT = TypeVar("_StreamItemT")
 _JSON_ENTRIES: Final = TypeAdapter(list[object])
 GAP_WORD_TOKENIZER: Final = re.compile(r"\b\w+\b")
 SENTENCE_TERMINATORS: Final = re.compile(r"[.!?]+")

litellm/proxy/guardrails/guardrail_hooks/noma/noma.py

@@ -52,7 +52,7 @@ SENSITIVE_DATA_DETECTOR_KEYS: Final[list[str]] = ["sensitiveData", "dataDetector
 # Type aliases
 MessageRole = Literal["user", "assistant"]
 LLMResponse = object
-_LLMResponseT: Final = TypeVar("_LLMResponseT")
+_LLMResponseT = TypeVar("_LLMResponseT")
 _LEGACY_NOMA_DEPRECATION_WARNED = False
 
 if TYPE_CHECKING:

litellm/proxy/guardrails/guardrail_hooks/tool_permission.py

@@ -80,7 +80,7 @@ _DECODED_ARGUMENTS: Final = TypeAdapter(object)
 _REQUEST_DATA: Final = TypeAdapter(object)
 _ToolTarget: TypeAlias = tuple[str | None, str | None]
 _JudgedCalls: TypeAlias = Mapping[str, tuple[tuple[str, str], ...]]
-_GroupedT: Final = TypeVar("_GroupedT")
+_GroupedT = TypeVar("_GroupedT")
 
 
 def _snake_case(name: str) -> str:

litellm/proxy/health_check.py

@@ -19,7 +19,7 @@ if TYPE_CHECKING:
     from litellm.router import Router
 
 logger: Final = logging.getLogger(__name__)
-_DeploymentT: Final = TypeVar("_DeploymentT", bound=Mapping[str, object])
+_DeploymentT = TypeVar("_DeploymentT", bound=Mapping[str, object])
 from litellm.constants import (
     BACKGROUND_HEALTH_CHECK_MAX_TOKENS,
     BACKGROUND_HEALTH_CHECK_MAX_TOKENS_REASONING,

litellm/proxy/list_api/in_memory.py

@@ -25,7 +25,7 @@ from litellm.proxy.list_api.list_framework import (
     Within,
 )
 
-TRow: Final = TypeVar("TRow")
+TRow = TypeVar("TRow")
 
 Cell: TypeAlias = str | int | float | datetime | None
 # A tuple-valued cell is a row's repeated field (a model group's providers, say). A predicate

litellm/proxy/management_endpoints/customer_endpoints.py

@@ -56,7 +56,7 @@ from litellm.types.proxy.management_endpoints.customer_endpoints import (
     UnblockUsersResponse,
 )
 
-_RowT_co: Final = TypeVar("_RowT_co", covariant=True)
+_RowT_co = TypeVar("_RowT_co", covariant=True)
 _STR_OBJECT_DICT: Final = TypeAdapter(dict[str, object])
 _CLEARABLE_LIST_FIELDS: Final = frozenset({"models"})
 

litellm/proxy/management_endpoints/key_management_endpoints.py

@@ -978,7 +978,7 @@ _KEY_TYPE_PRESET_ROUTES: Final = MappingProxyType(
 )
 
 
-_KeyJsonT: Final = TypeVar("_KeyJsonT", bound=dict[str, object])
+_KeyJsonT = TypeVar("_KeyJsonT", bound=dict[str, object])
 
 
 def handle_key_type(data: GenerateKeyRequest, data_json: _KeyJsonT) -> _KeyJsonT:

litellm/proxy/management_endpoints/router_settings_endpoints.py

@@ -65,9 +65,8 @@ def _router_setting_source(
 
 def _get_routing_strategies_from_router_class() -> list[str]:
     """
-    Dynamically extract routing strategies from the Router class __init__ method.
+    Dynamically extract routing strategies from the Router constructor's signature.
     """
-    # Get the __init__ signature
     sig: Final = inspect.signature(Router)
 
     # Get the routing_strategy parameter

litellm/proxy/management_endpoints/tool_management_endpoints.py

@@ -59,7 +59,7 @@ from litellm.types.tool_management import (
     ToolUsageLogsResponse,
 )
 
-_RowT_co: Final = TypeVar("_RowT_co", covariant=True)
+_RowT_co = TypeVar("_RowT_co", covariant=True)
 
 if TYPE_CHECKING:
 

litellm/proxy/management_endpoints/user_agent_analytics_endpoints.py

@@ -139,7 +139,7 @@ _DISTINCT_TAG_ROWS: Final = TypeAdapter(list[_DistinctTagRow])
 _ACTIVE_USERS_ROWS: Final = TypeAdapter(list[_ActiveUsersRow])
 _TAG_SUMMARY_ROWS: Final = TypeAdapter(list[_TagSummaryRow])
 
-_RowT_co: Final = TypeVar("_RowT_co", covariant=True)
+_RowT_co = TypeVar("_RowT_co", covariant=True)
 
 if TYPE_CHECKING:
 

litellm/proxy/management_helpers/audit_logs.py

@@ -135,7 +135,7 @@ def _audit_log_task_done_callback(task: asyncio.Task) -> None:
         verbose_proxy_logger.error("Audit log callback task failed: %s", exc, exc_info=exc)
 
 
-_T: Final = TypeVar("_T")
+_T = TypeVar("_T")
 
 
 class AuditTaskRegistry:

litellm/proxy/management_helpers/bulk_user_creation.py

@@ -80,7 +80,7 @@ BULK_NEW_USER_CONCURRENCY: Final = 10
 
 TeamRole: TypeAlias = Literal["user", "admin"]
 KeyGenerator: TypeAlias = Callable[..., Awaitable[object]]
-_T: Final = TypeVar("_T")
+_T = TypeVar("_T")
 
 
 @dataclass(frozen=True, slots=True)

litellm/proxy/management_helpers/utils.py

@@ -769,8 +769,8 @@ async def _emit_management_endpoint_otel_span(
         )
 
 
-_P: Final = ParamSpec("_P")
-_R: Final = TypeVar("_R")
+_P = ParamSpec("_P")
+_R = TypeVar("_R")
 
 
 def management_endpoint_wrapper(func: Callable[_P, Awaitable[_R]]) -> Callable[_P, Awaitable[_R]]:

litellm/proxy/mcp_server/faults/list_outcomes.py

@@ -258,7 +258,7 @@ def rate_limit_retry_delay(
     return min(max_delay, base_delay * 2.0**attempt * (1 + jitter))
 
 
-T: Final = TypeVar("T")
+T = TypeVar("T")
 
 
 async def retry_upstream_rate_limit(

litellm/proxy/mcp_server/interactions.py

@@ -193,7 +193,7 @@ def seal_continuation(
     return InputRequiredResult(input_requests=result.input_requests, request_state=sealed.ok, _meta=result.meta)
 
 
-_Terminal: Final = TypeVar("_Terminal")
+_Terminal = TypeVar("_Terminal")
 
 
 @dataclass(frozen=True, slots=True)

litellm/proxy/mcp_server/semantic_tool_filter.py

@@ -21,7 +21,7 @@ if TYPE_CHECKING:
 
     from litellm.router import Router
 
-_Tool: Final = TypeVar("_Tool")
+_Tool = TypeVar("_Tool")
 
 
 class SemanticToolFilterContextWindowError(Exception):

litellm/proxy/native_compaction.py

@@ -20,7 +20,7 @@ from litellm.router_strategy.complexity_router.context_compaction import (
     native_compaction_call,
 )
 
-_ResultT: Final = TypeVar("_ResultT")
+_ResultT = TypeVar("_ResultT")
 _ASGI_APP: Final = TypeAdapter[ASGIApp](ASGIApp)
 _ROOT_PATH: Final = TypeAdapter(str)
 _JSON_OBJECT: Final = TypeAdapter(Mapping[str, object])

litellm/proxy/pass_through_endpoints/pass_through_endpoints.py

@@ -2671,7 +2671,7 @@ WebsocketUpstreamFrameClassifier: TypeAlias = Callable[[Mapping[str, object]], W
 _SESSION_FRAMES_EVERY_ROW_CARRIES: Final[frozenset[WebsocketUpstreamFrameKind]] = frozenset(
     {"opens_session", "updates_session"}
 )
-_FrameT: Final = TypeVar("_FrameT", bound=Mapping[str, object])
+_FrameT = TypeVar("_FrameT", bound=Mapping[str, object])
 
 
 @dataclass(frozen=True, slots=True)

litellm/proxy/roi_calculator/github.py

@@ -18,7 +18,7 @@ from litellm.types.llms.custom_http import httpxSpecialProvider
 from litellm.types.roi_calculator import ROIPullCommit, ROIPullEvidence, ROIPullFile, ROISettings
 from litellm.types.roi_observed import ObservedIssue
 
-_T: Final = TypeVar("_T")
+_T = TypeVar("_T")
 
 
 class SourceError(Exception):

litellm/proxy/roi_calculator/gitlab.py

@@ -19,7 +19,7 @@ from litellm.types.llms.custom_http import httpxSpecialProvider
 from litellm.types.roi_calculator import ROIPullCommit, ROIPullEvidence, ROIPullFile, ROISettings
 from litellm.types.roi_observed import ObservedIssue
 
-_T: Final = TypeVar("_T", bound=BaseModel)
+_T = TypeVar("_T", bound=BaseModel)
 
 
 class _User(LiteLLMBaseModel):

litellm/proxy/utils.py

@@ -336,7 +336,7 @@ if TYPE_CHECKING:
 else:
     Span = Any
 
-_T: Final = TypeVar("_T")
+_T = TypeVar("_T")
 
 
 class _ViewCountRow(TypedDict):

litellm/repositories/chunked_in.py

@@ -21,8 +21,8 @@ IN_LIST_CHUNK_SIZE: Final = 5_000
 MAX_IN_LIST_CHUNK_SIZE: Final = 30_000
 LOGICAL_KEYS: Final = frozenset({"AND", "OR", "NOT"})
 
-RowT: Final = TypeVar("RowT")
-ResultT: Final = TypeVar("ResultT")
+RowT = TypeVar("RowT")
+ResultT = TypeVar("ResultT")
 
 Atomicity: TypeAlias = Literal["caller_transaction", "per_chunk_ok"]
 """More than `chunk_size` values means more than one statement. `caller_transaction`

litellm/responses/litellm_completion_transformation/compaction_items.py

@@ -17,7 +17,7 @@ from .tool_search import chat_config_builds_anthropic_request, provider_chat_con
 
 _JSON_OBJECT: Final = TypeAdapter(dict[str, object])
 _OBJECT_LIST: Final = TypeAdapter(list[object])
-_Item: Final = TypeVar("_Item")
+_Item = TypeVar("_Item")
 _INSTRUCTION_ROLES: Final = frozenset(("system", "developer"))
 
 

litellm/responses/litellm_completion_transformation/transformation.py

@@ -282,8 +282,8 @@ _AUDIO_PART_ADAPTER: Final = TypeAdapter(ChatCompletionAudioObject)
 _IMAGE_URL_ADAPTER: Final = TypeAdapter(ChatCompletionImageUrlObject)
 _FILE_PART_ADAPTER: Final = TypeAdapter(_ResponsesFilePart)
 _MessageT = TypeVar("_MessageT")
-_MediaT: Final = TypeVar("_MediaT")
-_InputItemT: Final = TypeVar("_InputItemT")
+_MediaT = TypeVar("_MediaT")
+_InputItemT = TypeVar("_InputItemT")
 _CACHED_CONTENT_ADAPTER: Final = TypeAdapter(ChatCompletionCachedContent)
 _DICT_ITEMS_LIST_ADAPTER: Final = TypeAdapter(list[dict[object, object]])
 _TEXT_ADAPTER: Final = TypeAdapter(str)

litellm/responses/utils.py

@@ -131,7 +131,7 @@ def _as_input_text_part(part: object) -> object:
     return part
 
 
-_RequestInputT: Final = TypeVar("_RequestInputT")
+_RequestInputT = TypeVar("_RequestInputT")
 
 
 def _provider_mapped_params(

litellm/router.py

@@ -63,7 +63,7 @@ def _router_init(
     ## CACHING ##
     redis_url: str | None = None,
     redis_host: str | None = None,
-    redis_port: int | None = None,
+    redis_port: int | str | None = None,
     redis_password: str | None = None,
     redis_db: int | None = None,
     cache_responses: bool | None = False,
@@ -120,7 +120,6 @@ def _router_init(
     fallback_budget_check: FallbackBudgetCheck | None = None,
     auto_router_capability_limit: AutoRouterCapabilityLimit | None = None,
     session_affinity_repin_on_failover: bool = False,
-    _host: PythonHost = python_host,
 ) -> None:
     """
     Initialize the Router class with the given parameters for caching, reliability, and routing strategy.
@@ -129,7 +128,7 @@ def _router_init(
         model_list (Optional[list]): List of models to be used. Defaults to None.
         redis_url (Optional[str]): URL of the Redis server. Defaults to None.
         redis_host (Optional[str]): Hostname of the Redis server. Defaults to None.
-        redis_port (Optional[int]): Port of the Redis server. Defaults to None.
+        redis_port (int | str | None): Port of the Redis server. Defaults to None.
         redis_password (Optional[str]): Password of the Redis server. Defaults to None.
         cache_responses (Optional[bool]): Flag to enable caching of responses. Defaults to False.
         cache_kwargs (dict): Additional kwargs to pass to RedisCache. Defaults to {}.
@@ -228,7 +227,7 @@ if TYPE_CHECKING:
     from litellm.types.llms.openai import AllMessageValues, OpenAIFileObject
     from litellm.types.utils import EmbeddingResponse, LiteLLMBatch, ModelResponse
 
-    _P: Final = ParamSpec("_P")
+    _P = ParamSpec("_P")
 
     def _takes_parameters_of(source: Callable[_P, None]) -> Callable[[Callable[..., None]], Callable[_P, None]]: ...
 
@@ -432,9 +431,7 @@ else:
             core: Final = self.__dict__.get("_core")
             return sorted({*super().__dir__(), *(dir(core) if core is not None else ())})
 
-    Router.__init__.__signature__ = _CORE_SIGNATURE.replace(  # pyright: ignore[reportFunctionMemberAccess]  # inspect reads it
-        parameters=[parameter for parameter in _CORE_SIGNATURE.parameters.values() if parameter.name != "_host"]
-    )
+    Router.__init__.__signature__ = _CORE_SIGNATURE  # pyright: ignore[reportFunctionMemberAccess]  # inspect reads it
     Router.__init__.__doc__ = _router_init.__doc__
 
 

litellm/router_deprecated.py

@@ -60,8 +60,8 @@ if TYPE_CHECKING:
     )
     from litellm.types.utils import ModelInfo as ModelMapInfo
 
-_T: Final = TypeVar("_T")
-_P: Final = ParamSpec("_P")
+_T = TypeVar("_T")
+_P = ParamSpec("_P")
 
 DEPRECATED_ROUTER_MEMBER_USES: Final[Counter[str]] = Counter()  # mutable-ok: the live tally _note increments
 
@@ -831,7 +831,7 @@ class DeprecatedRouterMembers:
 # Router-family functions and classes that took the Router (``litellm_router_instance=``,
 # ``router=``, ...) take its RoutingContext now. These keep the old spellings working.
 
-_RoutingCallable: Final = TypeVar("_RoutingCallable", bound=Callable[..., object])
+_RoutingCallable = TypeVar("_RoutingCallable", bound=Callable[..., object])
 
 
 @functools.cache

litellm/router_rust/invoke/streaming.py

@@ -82,7 +82,7 @@ MAX_HELD_PRE_OUTPUT_RESPONSES_EVENTS: Final = 200
 
 # The hidden params and headers a fallback's response carries, as dicts the proxy writes into.
 _Prepared: TypeAlias = "tuple[dict[str, object], dict[str, object]]"
-_RestT: Final = TypeVar("_RestT", Iterator[object], AsyncIterator[object])
+_RestT = TypeVar("_RestT", Iterator[object], AsyncIterator[object])
 
 
 # --- the attempt a stream came from -------------------------------------------------------------
@@ -289,7 +289,7 @@ def _record_now(recording: Callable[[], object]) -> None:
         verbose_router_logger.warning("a stream's outcome was not recorded: %s", error)
 
 
-_ItemT: Final = TypeVar("_ItemT")
+_ItemT = TypeVar("_ItemT")
 
 
 async def _areported(

litellm/router_utils/endpoint_eligibility.py

@@ -38,7 +38,7 @@ def _endpoint_path(endpoint: str) -> str:
 
 
 CHAT_ENDPOINTS: Final = frozenset(_endpoint_path(endpoint) for endpoint in CHAT_ENDPOINT_BY_CALL.values())
-_Deployment: Final = TypeVar("_Deployment", bound=Mapping[str, object])
+_Deployment = 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)

litellm/router_utils/kubernetes_pod_discovery.py

@@ -41,7 +41,7 @@ _SyncLookup: TypeAlias = Future[list[_AddressInfo]]
 _AsyncLookup: TypeAlias = asyncio.Future[list[_AddressInfo]]
 _CacheKey: TypeAlias = tuple[str, int | None]
 _PodSelection: TypeAlias = tuple[str, int, Literal["round_robin", "session_affinity", "session_affinity_retry"]]
-_DeploymentT: Final = TypeVar("_DeploymentT")
+_DeploymentT = TypeVar("_DeploymentT")
 
 
 @dataclass(frozen=True, slots=True)
@@ -564,9 +564,9 @@ class _HasPodDiscovery(Protocol):
     kubernetes_pod_discovery: KubernetesPodDiscovery
 
 
-_P: Final = ParamSpec("_P")
-_R: Final = TypeVar("_R")
-_S: Final = TypeVar("_S", bound=_HasPodDiscovery)
+_P = ParamSpec("_P")
+_R = TypeVar("_R")
+_S = TypeVar("_S", bound=_HasPodDiscovery)
 
 
 def _request_kwargs_from_call(

litellm/rust_bridge/diagnostics.py

@@ -6,7 +6,7 @@ from typing import Final, Protocol, TypeVar, cast
 
 from litellm.rust_bridge.bindings import NativeBinding
 
-ResultT: Final = TypeVar("ResultT")
+ResultT = TypeVar("ResultT")
 
 
 class NativeDiagnosticProcessor(Protocol):

litellm/rust_bridge/dispatch.py

@@ -9,9 +9,9 @@ from litellm.rust_bridge import catalog, runtime
 from litellm.rust_bridge.bindings import NativeBinding
 from litellm.rust_bridge.catalog import Route, RouteContext, Rules
 
-RequestT: Final = TypeVar("RequestT")
-NativeT: Final = TypeVar("NativeT")
-ResultT: Final = TypeVar("ResultT")
+RequestT = TypeVar("RequestT")
+NativeT = TypeVar("NativeT")
+ResultT = TypeVar("ResultT")
 
 NativeHook: TypeAlias = Callable[[RequestT, tuple[object, ...], Mapping[str, object]], ResultT]
 

litellm/rust_bridge/execution.py

@@ -5,7 +5,7 @@ from contextlib import ExitStack
 from dataclasses import dataclass
 from typing import Final, Protocol, TypeVar, cast  # noqa: TID251  # trusts the native stub at the binding boundary
 
-_Call: Final = TypeVar("_Call")
+_Call = TypeVar("_Call")
 
 
 @dataclass(frozen=True, slots=True)

litellm/rust_bridge/public_call.py

@@ -105,7 +105,7 @@ def native_call(legacy: inspect.Signature, args: tuple[object, ...], kwargs: Map
     return NativeCall(args=args, kwargs=kwargs, base=_without_bags(legacy, positional.arguments))
 
 
-NativeResultT: Final = TypeVar("NativeResultT")
+NativeResultT = TypeVar("NativeResultT")
 
 
 def native_call_hook(

litellm/rust_bridge/response_metadata.py

@@ -5,7 +5,7 @@ from litellm.litellm_core_utils.add_retry_fallback_headers import (
     get_hidden_params_dict,
 )
 
-ResultT: Final = TypeVar("ResultT")
+ResultT = TypeVar("ResultT")
 
 
 def mark_rust_response(response: ResultT) -> ResultT:

litellm/telemetry/aggregate.py

@@ -12,8 +12,8 @@ QUIET_RETRIES: Final = 3
 
 _LOGGER: Final = logging.getLogger(__name__)
 
-_Key: Final = TypeVar("_Key")
-_Row: Final = TypeVar("_Row")
+_Key = TypeVar("_Key")
+_Row = TypeVar("_Row")
 
 
 class AggregatingSink:

litellm/tracing/queries.py

@@ -27,8 +27,8 @@ class ClickHouseSQLEnvelope(LiteLLMBaseModel):
     statistics: TraceQueryStatistics
 
 
-ParamsT: Final = TypeVar("ParamsT", bound=BaseModel)
-RowT: Final = TypeVar("RowT")
+ParamsT = TypeVar("ParamsT", bound=BaseModel)
+RowT = TypeVar("RowT")
 
 
 class QueryResponse(LiteLLMBaseModel, Generic[RowT]):

litellm/tracing/storage.py

@@ -16,7 +16,7 @@ _TRACE_PAGE: Final = TypeAdapter(TracePage)
 _TRACE: Final[TypeAdapter[Trace | None]] = TypeAdapter(Trace | None)
 _SPAN_DETAIL: Final[TypeAdapter[SpanDetail | None]] = TypeAdapter(SpanDetail | None)
 _SPAN_ERROR_PAGE: Final[TypeAdapter[SpanErrorPage | None]] = TypeAdapter(SpanErrorPage | None)
-_ResponseT: Final = TypeVar("_ResponseT")
+_ResponseT = TypeVar("_ResponseT")
 
 
 def _decode_query_response(adapter: TypeAdapter[_ResponseT], body: str) -> _ResponseT:

litellm/types/router.py

@@ -547,7 +547,7 @@ _RESERVED_INIT_KEYS: Final = frozenset({"self", "params", "__class__"})
 
 _STRICT_LITELLM_PARAMS: Final = MappingProxyType({"strict_litellm_params": True})
 
-_StrictlyValidatedParams: Final = TypeVar("_StrictlyValidatedParams", bound=BaseModel)
+_StrictlyValidatedParams = TypeVar("_StrictlyValidatedParams", bound=BaseModel)
 
 
 def validate_litellm_params_strictly(

litellm/utils.py

@@ -813,7 +813,7 @@ _OIDC_TOKEN_FETCH_EXECUTOR: Final = ThreadPoolExecutor(thread_name_prefix="litel
 _PASS_THROUGH_CALL_TYPES: Final = frozenset(
     {CallTypes.allm_passthrough_route.value, CallTypes.llm_passthrough_route.value}
 )
-_Resolved: Final = TypeVar("_Resolved")
+_Resolved = TypeVar("_Resolved")
 
 
 async def _resolved_off_the_event_loop(resolve: Callable[[OidcResolver], _Resolved], api_key: object) -> _Resolved:
@@ -4074,7 +4074,7 @@ def is_generalized_model_info(model_info: ModelInfoBase) -> bool:
 
 _REGISTRATION_ATTEMPTS: Final = 3
 
-_ModelInfoT: Final = TypeVar("_ModelInfoT", bound=ModelInfoBase)
+_ModelInfoT = TypeVar("_ModelInfoT", bound=ModelInfoBase)
 
 
 def _lookup_mapped_model_info(lookup: Callable[[], _ModelInfoT]) -> _ModelInfoT | ModelInfoLookupMiss:

scripts/check_type_discipline.py

@@ -54,8 +54,11 @@ LIT010  Variable assignment without a `Final` declaration. Every local and modul
         that namespace is the SDK's runtime-settable config surface (users follow
         the documented `litellm.api_key = ...` pattern and the proxy rebinds these
         via setattr), and the package ships py.typed, so a Final there turns every
-        documented downstream assignment into a mypy error. Suppress a deliberately
-        rebindable name with `# rebind-ok: <reason>` on each offending line.
+        documented downstream assignment into a mypy error. Type variables are the
+        reverse: a `TypeVar`, `ParamSpec`, or `TypeVarTuple` stays bare, and a Final one
+        trips, since mypy reads it as a plain variable and every signature using it as
+        Any. Suppress a deliberately rebindable name with `# rebind-ok: <reason>` on
+        each offending line.
 LIT011  Function-argument mutation: a parameter that is re-bound (`param = ...`,
         `param += ...`, a `for`/`with`/unpacking/walrus target, `del param`, or a
         re-bind in a nested function under `nonlocal param`) or mutated in place
@@ -458,6 +461,7 @@ def iter_guard_violations(path: Path, tree: ast.AST) -> Iterator[Violation]:
 # --------------------------------------------------------------------------- #
 
 CONSTANT_DECLARATIONS = frozenset(("Final", "TypeAlias"))
+TYPE_VARIABLE_FACTORIES = frozenset(("TypeVar", "ParamSpec", "TypeVarTuple"))
 CONFIG_SURFACE_PARTS = ("litellm", "__init__.py")
 SELF_PARAMS = frozenset(("self", "cls"))
 NESTED_SCOPES = (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef, ast.Lambda)
@@ -480,6 +484,14 @@ def _declares_constant(annotation: ast.expr) -> bool:
     return isinstance(root, ast.Attribute) and root.attr in CONSTANT_DECLARATIONS
 
 
+def _creates_type_variable(value: ast.expr | None) -> bool:
+    match value:
+        case ast.Call(func=ast.Name(id=name) | ast.Attribute(attr=name)):
+            return name in TYPE_VARIABLE_FACTORIES
+        case _:
+            return False
+
+
 def _defaults(args: ast.arguments) -> tuple[ast.expr, ...]:
     return (*args.defaults, *(d for d in args.kw_defaults if d is not None))
 
@@ -561,6 +573,8 @@ def _bound_names(target: ast.expr) -> Iterator[tuple[str, int]]:
 
 def _node_bindings(node: ast.AST, in_loop: bool) -> Iterator[Binding]:
     match node:
+        case ast.Assign(targets=[ast.Name(id=name, lineno=line)], value=value) if _creates_type_variable(value):
+            yield Binding(name, line, "other", in_loop)
         case ast.Assign(targets=targets):
             for target in targets:
                 if isinstance(target, ast.Name):
@@ -569,7 +583,7 @@ def _node_bindings(node: ast.AST, in_loop: bool) -> Iterator[Binding]:
                     yield from (Binding(n, line, "unpack", in_loop) for n, line in _bound_names(target))
         case ast.AnnAssign(target=ast.Name(id=name, lineno=line), annotation=annotation, value=value):
             if _declares_constant(annotation):
-                yield Binding(name, line, "declared", in_loop)
+                yield Binding(name, line, "final_type_variable" if _creates_type_variable(value) else "declared", in_loop)
             elif value is not None:
                 yield Binding(name, line, "annassign", in_loop)
         case ast.AugAssign(target=ast.Name(id=name, lineno=line)):
@@ -646,6 +660,15 @@ def iter_final_violations(path: Path, tree: ast.AST) -> Iterator[Violation]:
         declared = frozenset(b.name for b in bindings if b.form == "declared")
         first = _first_binding_index(bindings)
         for i, b in enumerate(bindings):
+            if b.form == "final_type_variable":
+                yield Violation(
+                    path,
+                    b.line,
+                    "LIT010",
+                    f"`{b.name}` is a type variable declared Final: mypy then reads it as a plain "
+                    f"variable and every signature using it as Any, so remove the Final",
+                )
+                continue
             if b.name in declared or b.name in params or b.in_loop:
                 continue
             if _exempt_final_name(b.name):

tests/code_coverage_tests/check_router_constructor_stub.py

@@ -0,0 +1,57 @@
+#!/usr/bin/env python3
+"""Fail unless mypy reads the Router constructor stub in litellm/router.py.
+
+litellm ships py.typed, so a stub that mypy reads as Any accepts every argument name in SDK users' code,
+while basedpyright, which `make lint-e2e-basedpyright` runs, can still read it.
+"""
+
+import subprocess
+import sys
+import tempfile
+from pathlib import Path
+from typing import Final
+
+PROBE: Final = """\
+from litellm.router import Router
+
+Router(model_list=[], num_retries=2)
+Router(model_list=[], redis_host="localhost", redis_port="6379")
+Router(model_list=[], not_a_parameter=1)
+"""
+EXPECTED_ERROR: Final = 'Unexpected keyword argument "not_a_parameter" for "Router"'
+
+
+def mypy_errors(probe: Path) -> tuple[str, ...]:
+    completed: Final = subprocess.run(
+        (
+            sys.executable,
+            "-I",
+            "-m",
+            "mypy",
+            "--config-file=",
+            "--follow-imports=skip",
+            "--ignore-missing-imports",
+            "litellm/router.py",
+            str(probe),
+        ),
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    return tuple(line for line in completed.stdout.splitlines() if ": error: " in line)
+
+
+def main() -> int:
+    with tempfile.TemporaryDirectory() as directory:
+        probe: Final = Path(directory) / "router_constructor_probe.py"
+        probe.write_text(PROBE)
+        errors: Final = mypy_errors(probe)
+    if len(errors) == 1 and EXPECTED_ERROR in errors[0]:
+        print("mypy reads the Router constructor stub")
+        return 0
+    print(f"expected only {EXPECTED_ERROR!r} from mypy, got: {errors}")
+    return 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tests/e2e/AGENTS.md

@@ -131,7 +131,7 @@ Current limits: Bedrock cannot be mounted in record or replay (SigV4 signs the H
 
 ## Typing
 
-The harness is fully typed with no error budget: `make lint-e2e-basedpyright` must report zero basedpyright errors, and CI enforces that on any PR touching `tests/e2e/**/*.py`. When a response field is untyped, model it in `models.py` (just the fields you read) and let pydantic validate it, rather than threading a `dict` or `Any` through the test
+The harness is fully typed with no error budget: `make lint-e2e-basedpyright` must report zero basedpyright errors, and CI enforces that on any PR touching `tests/e2e/**/*.py`, `tests/e2e_harness/**/*.py`, `litellm/**/*.py`, `pyrightconfig.json`, `pyproject.toml` or `uv.lock`. When a response field is untyped, model it in `models.py` (just the fields you read) and let pydantic validate it, rather than threading a `dict` or `Any` through the test
 
 ## Typed test metadata
 

tests/e2e/CONTRIBUTING.md

@@ -224,7 +224,7 @@ Mark live tests with `@pytest.mark.e2e` (on the class or the module). Pure cover
 
 Before you push
 
-1. Run `make lint-e2e-basedpyright` (or `make check` with your changes staged); the harness is fully typed and the gate allows zero basedpyright errors, enforced in CI on any PR touching `tests/e2e/**/*.py` or `tests/e2e_harness/**/*.py`
+1. Run `make lint-e2e-basedpyright` (or `make check` with your changes staged); the harness is fully typed and the gate allows zero basedpyright errors, enforced in CI on any PR touching `tests/e2e/**/*.py`, `tests/e2e_harness/**/*.py`, `litellm/**/*.py`, `pyrightconfig.json`, `pyproject.toml`, or `uv.lock`
 
 2. Add the models your test needs to the config your local proxy loads
 

tests/integration/compatibility/test_openai_sdk_client_paths.py

@@ -78,7 +78,7 @@ SPEECH_UPSTREAM_MODEL: Final = "gpt-4o-mini-tts"
 PAYMENT_REQUIRED_DEPLOYMENTS: Final[Mapping[str, str]] = MappingProxyType(
     {"openai": "openai/gpt-4o-mini", "anthropic": "anthropic/claude-sonnet-4-5"}
 )
-_R: Final = TypeVar("_R")
+_R = TypeVar("_R")
 
 
 def _identity(prefix: str) -> str:

tests/integration/sdk/test_openai_sdk_client_sessions_wire.py

@@ -21,7 +21,7 @@ from litellm import Router
 from litellm.llms.base_llm.chat.transformation import BaseLLMException
 from litellm.types.utils import ImageResponse, ModelResponse, TextCompletionResponse
 
-_R: Final = TypeVar("_R")
+_R = TypeVar("_R")
 
 _PROVIDER_KEY: Final = "sk-scripted-provider"
 _API_VERSION: Final = "2024-10-21"

tests/unit/test_check_type_discipline.py

@@ -303,6 +303,24 @@ def test_single_leading_underscore_is_not_exempt(tmp_path):
     assert "LIT010" in _codes(tmp_path, "_x = 1\n")
 
 
+@pytest.mark.parametrize(
+    "declaration",
+    ("TypeVar('_T')", "ParamSpec('_T')", "TypeVarTuple('_T')", "typing.TypeVar('_T')", "typing_extensions.ParamSpec('_T')"),
+)
+def test_a_type_variable_stays_bare_and_a_final_one_trips(tmp_path, declaration):
+    assert "LIT010" not in _codes(tmp_path, f"_T = {declaration}\n")
+    assert "LIT010" in _codes(tmp_path, f"_T: Final = {declaration}\n")
+    assert "LIT010" in _codes(tmp_path, f"_T: Final[object] = {declaration}\n")
+
+
+def test_rebinding_a_bare_type_variable_still_trips(tmp_path):
+    assert "LIT010" in _codes(tmp_path, "_T = TypeVar('_T')\n_T = 1\n")
+
+
+def test_a_final_call_to_any_other_factory_stays_clean(tmp_path):
+    assert "LIT010" not in _codes(tmp_path, "limit: Final = int('1')\n")
+
+
 def test_rebind_ok_suppresses_a_repeated_unpack(tmp_path):
     src = "a, b = (1, 2)\na, b = (b, a)  # rebind-ok: swap in place\n"
     assert "LIT010" not in _codes(tmp_path, src)