diff --git a/basedpyright-code-budget.json b/basedpyright-code-budget.json index 3180cea2568..9f42fe0fac1 100644 --- a/basedpyright-code-budget.json +++ b/basedpyright-code-budget.json @@ -1,6 +1,6 @@ { "reportAny": { - "limit": 29682 + "limit": 29002 }, "reportArgumentType": { "limit": 2645 @@ -24,7 +24,7 @@ "limit": 42 }, "reportExplicitAny": { - "limit": 9440 + "limit": 9248 }, "reportFunctionMemberAccess": { "limit": 11 @@ -99,19 +99,19 @@ "limit": 0 }, "reportUnknownArgumentType": { - "limit": 45297 + "limit": 45281 }, "reportUnknownLambdaType": { "limit": 113 }, "reportUnknownMemberType": { - "limit": 40411 + "limit": 40379 }, "reportUnknownParameterType": { "limit": 20301 }, "reportUnknownVariableType": { - "limit": 31968 + "limit": 31936 }, "reportUnnecessaryCast": { "limit": 177 diff --git a/litellm/litellm_core_utils/litellm_logging.py b/litellm/litellm_core_utils/litellm_logging.py index 6b05591fb85..03168acc3d8 100644 --- a/litellm/litellm_core_utils/litellm_logging.py +++ b/litellm/litellm_core_utils/litellm_logging.py @@ -175,7 +175,9 @@ from .initialize_dynamic_callback_params import ( from .specialty_caches.dynamic_logging_cache import DynamicLoggingCache if TYPE_CHECKING: + from litellm.integrations.otel.logger import OpenTelemetryV2 from litellm.llms.base_llm.passthrough.transformation import BasePassthroughConfig + from litellm.router import Router try: from litellm_enterprise.enterprise_callbacks.callback_controls import ( EnterpriseCallbackControls, @@ -346,8 +348,12 @@ class Logging(LiteLLMLoggingBaseClass): self.litellm_call_id = litellm_call_id self.litellm_trace_id: str = litellm_trace_id if litellm_trace_id else str(uuid.uuid4()) self.function_id = function_id - self.streaming_chunks: list[Any] = [] # for generating complete stream response - self.sync_streaming_chunks: list[Any] = [] # for generating complete stream response + self.streaming_chunks: list[ + ModelResponse + ] = [] # mutable-ok: accumulates one chunk per stream event over the object's lifetime + self.sync_streaming_chunks: list[ + ModelResponse + ] = [] # mutable-ok: accumulates one chunk per stream event over the object's lifetime self.log_raw_request_response = log_raw_request_response # Initialize dynamic callbacks @@ -392,7 +398,9 @@ class Logging(LiteLLMLoggingBaseClass): self.caching_details: CachingDetails | None = None # Passthrough endpoint guardrails config for field targeting - self.passthrough_guardrails_config: dict[str, Any] | None = None + self.passthrough_guardrails_config: dict[str, object] | None = ( + None # mutable-ok: assigned once from a caller-supplied dict payload + ) self.model_call_details: dict[str, Any] = { "litellm_trace_id": self.litellm_trace_id, @@ -1138,20 +1146,15 @@ class Logging(LiteLLMLoggingBaseClass): self.model_call_details["additional_args"] = additional_args self.model_call_details["log_event_type"] = "post_api_call" - if self.litellm_request_debug: - attr = "warning" - else: - attr = "debug" + callattr = verbose_logger.warning if self.litellm_request_debug else verbose_logger.debug if json_logs: - callattr = getattr(verbose_logger, attr) callattr( "RAW RESPONSE:\n{}\n\n".format( self.model_call_details.get("original_response", self.model_call_details) ), ) else: - callattr = getattr(verbose_logger, attr) callattr( "RAW RESPONSE:\n{}\n\n".format( self.model_call_details.get("original_response", self.model_call_details) @@ -1905,8 +1908,8 @@ class Logging(LiteLLMLoggingBaseClass): def _is_recognized_call_type_for_logging( self, - logging_result: Any, - ): + logging_result: object, + ) -> bool: """ Returns True if the call type is recognized for logging (eg. ModelResponse, ModelResponseStream, etc.) """ @@ -3035,7 +3038,7 @@ class Logging(LiteLLMLoggingBaseClass): return trace_id - def _get_callback_object(self, service_name: Literal["langfuse"]) -> Any | None: + def _get_callback_object(self, service_name: Literal["langfuse"]) -> LangFuseLogger | None: """ Return dynamic callback object. @@ -3548,7 +3551,7 @@ def set_callbacks(callback_list, function_id=None): def _init_custom_logger_compatible_class( logging_integration: _custom_logger_compatible_callbacks_literal, internal_usage_cache: DualCache | None, - llm_router: Any | None, # expect litellm.Router, but typing errors due to circular import + llm_router: "Router | None", custom_logger_init_args: dict | None = {}, ) -> CustomLogger | None: """ @@ -3936,7 +3939,7 @@ def _init_custom_logger_compatible_class( dynamic_rate_limiter_obj = _PROXY_DynamicRateLimitHandler(internal_usage_cache=internal_usage_cache) - if llm_router is not None and isinstance(llm_router, litellm.Router): + if llm_router is not None: dynamic_rate_limiter_obj.update_variables(llm_router=llm_router) _in_memory_loggers.append(dynamic_rate_limiter_obj) return dynamic_rate_limiter_obj # type: ignore @@ -3954,7 +3957,7 @@ def _init_custom_logger_compatible_class( dynamic_rate_limiter_obj_v3 = _PROXY_DynamicRateLimitHandlerV3(internal_usage_cache=internal_usage_cache) - if llm_router is not None and isinstance(llm_router, litellm.Router): + if llm_router is not None: dynamic_rate_limiter_obj_v3.update_variables(llm_router=llm_router) _in_memory_loggers.append(dynamic_rate_limiter_obj_v3) return dynamic_rate_limiter_obj_v3 # type: ignore @@ -4160,7 +4163,9 @@ def _init_custom_logger_compatible_class( return None -def _maybe_construct_otel_v2(callback_name: str, _in_memory_loggers: list) -> Any | None: +def _maybe_construct_otel_v2( + callback_name: str, _in_memory_loggers: list +) -> "OpenTelemetryV2 | None": # mutable-ok: list is only read here; caller owns the mutable in-memory logger registry """If ``LITELLM_OTEL_V2`` is on, build (or reuse) a single ``OpenTelemetryV2`` instance configured via the preset for ``callback_name``. diff --git a/litellm/litellm_core_utils/streaming_handler.py b/litellm/litellm_core_utils/streaming_handler.py index 5cdb5877915..a907b773b80 100644 --- a/litellm/litellm_core_utils/streaming_handler.py +++ b/litellm/litellm_core_utils/streaming_handler.py @@ -70,7 +70,7 @@ def _next_sync_or_exhausted(it: Any) -> Any: return _SYNC_ITER_EXHAUSTED -def is_async_iterable(obj: Any) -> bool: +def is_async_iterable(obj: object) -> bool: """ Check if an object is an async iterable (can be used with 'async for'). @@ -124,9 +124,8 @@ class CustomStreamWrapper: self.sent_last_chunk = False self._stream_created_time: float = time.time() - litellm_params: GenericLiteLLMParams = GenericLiteLLMParams.model_validate( - dict(**self.logging_obj.model_call_details.get("litellm_params", {})) - ) + _init_litellm_params = self.logging_obj.model_call_details.get("litellm_params", {}) + litellm_params: GenericLiteLLMParams = GenericLiteLLMParams.model_validate(dict(**_init_litellm_params)) self.merge_reasoning_content_in_choices: bool = litellm_params.merge_reasoning_content_in_choices or False self.sent_first_thinking_block = False self.sent_last_thinking_block = False @@ -151,7 +150,7 @@ class CustomStreamWrapper: _api_base = get_api_base( model=model or "", - optional_params=self.logging_obj.model_call_details.get("litellm_params", {}), + optional_params=_init_litellm_params, ) self._hidden_params = { @@ -346,9 +345,9 @@ class CustomStreamWrapper: try: if not isinstance(chunk, str): chunk = chunk.decode("utf-8") # DO NOT REMOVE this: This is required for HF inference API + Streaming - text = "" + text: str = "" is_finished = False - finish_reason = "" + finish_reason: str = "" print_verbose(f"chunk: {chunk}") if chunk.startswith("data:"): data_json = json.loads(chunk[5:]) @@ -383,7 +382,7 @@ class CustomStreamWrapper: chunk = chunk.decode("utf-8") data_json = json.loads(chunk) try: - text = data_json["completions"][0]["data"]["text"] + text: str = data_json["completions"][0]["data"]["text"] is_finished = True finish_reason = "stop" return { @@ -398,7 +397,7 @@ class CustomStreamWrapper: chunk = chunk.decode("utf-8") data_json = json.loads(chunk) try: - text = data_json["answer"] + text: str = data_json["answer"] is_finished = True finish_reason = "stop" return { @@ -410,9 +409,9 @@ class CustomStreamWrapper: raise ValueError(f"Unable to parse response. Original response: {chunk}") def handle_nlp_cloud_chunk(self, chunk): - text = "" + text: str = "" is_finished = False - finish_reason = "" + finish_reason: str = "" try: if self.model and "dolphin" in self.model: chunk = self.process_chunk(chunk=chunk) @@ -436,7 +435,7 @@ class CustomStreamWrapper: chunk = chunk.decode("utf-8") data_json = json.loads(chunk) try: - text = data_json["completions"][0]["completion"] + text: str = data_json["completions"][0]["completion"] is_finished = True finish_reason = "stop" return { @@ -449,8 +448,8 @@ class CustomStreamWrapper: def handle_azure_chunk(self, chunk): is_finished = False - finish_reason = "" - text = "" + finish_reason: str = "" + text: str = "" print_verbose(f"chunk: {chunk}") if "data: [DONE]" in chunk: text = "" @@ -548,9 +547,9 @@ class CustomStreamWrapper: def handle_azure_text_completion_chunk(self, chunk): try: - text = "" + text: str = "" is_finished = False - finish_reason = None + finish_reason: str | None = None choices = getattr(chunk, "choices", []) if len(choices) > 0: text = choices[0].text @@ -568,9 +567,9 @@ class CustomStreamWrapper: def handle_openai_text_completion_chunk(self, chunk): try: - text = "" + text: str = "" is_finished = False - finish_reason = None + finish_reason: str | None = None usage = None choices = getattr(chunk, "choices", []) if len(choices) > 0: @@ -589,7 +588,7 @@ class CustomStreamWrapper: except Exception as e: raise e - def handle_baseten_chunk(self, chunk): + def handle_baseten_chunk(self, chunk) -> str: try: chunk = chunk.decode("utf-8") if len(chunk) > 0: diff --git a/litellm/llms/custom_httpx/llm_http_handler.py b/litellm/llms/custom_httpx/llm_http_handler.py index fcb039c9b4f..3398204c565 100644 --- a/litellm/llms/custom_httpx/llm_http_handler.py +++ b/litellm/llms/custom_httpx/llm_http_handler.py @@ -19,6 +19,8 @@ from urllib.parse import parse_qs, urlencode, urlparse, urlunparse import httpx # type: ignore from openai.types.file_deleted import FileDeleted +from pydantic import TypeAdapter +from typing_extensions import TypedDict import litellm import litellm.litellm_core_utils @@ -87,7 +89,11 @@ from litellm.types.containers.main import ( ContainerObject, DeleteContainerResult, ) -from litellm.types.files import StreamingMediaUploadConfig, TwoStepFileUploadConfig +from litellm.types.files import ( + StreamingMediaUploadConfig, + TwoStepFileUploadConfig, + TwoStepFileUploadRequest, +) from litellm.types.integrations.custom_logger import ( AgenticLoopPlan, AgenticLoopRequestPatch, @@ -212,6 +218,45 @@ def _responses_api_optional_request_param_names() -> frozenset[str]: return frozenset(get_type_hints(ResponsesAPIOptionalRequestParams).keys()) +_TWO_STEP_FILE_UPLOAD_REQUEST_ADAPTER = TypeAdapter(TwoStepFileUploadRequest) + + +class _PresignedBatchRequest(TypedDict): + method: str + url: str + headers: dict[str, str] # mutable-ok: TypedDict field mirrors the outgoing HTTP request's header dict + data: str | bytes | dict[str, Any] | None # mutable-ok: TypedDict field mirrors the outgoing HTTP request body dict + + +_PRESIGNED_BATCH_REQUEST_ADAPTER = TypeAdapter(_PresignedBatchRequest) + + +def _send_presigned_batch_request( + client: HTTPHandler, + request: _PresignedBatchRequest, + timeout: float | httpx.Timeout | None, +) -> httpx.Response | None: + method = request["method"].lower() + if method == "get": + return getattr(client, method)(url=request["url"], headers=request["headers"], timeout=timeout) + return getattr(client, method)( + url=request["url"], headers=request["headers"], data=request["data"], timeout=timeout + ) + + +async def _asend_presigned_batch_request( + client: AsyncHTTPHandler, + request: _PresignedBatchRequest, + timeout: float | httpx.Timeout | None, +) -> httpx.Response | None: + method = request["method"].lower() + if method == "get": + return await getattr(client, method)(url=request["url"], headers=request["headers"], timeout=timeout) + return await getattr(client, method)( + url=request["url"], headers=request["headers"], data=request["data"], timeout=timeout + ) + + def _custom_logger_callbacks(logging_obj: LiteLLMLoggingObj) -> list["CustomLogger"]: from litellm.integrations.custom_logger import CustomLogger from litellm.litellm_core_utils.litellm_logging import ( @@ -2007,7 +2052,7 @@ class BaseLLMHTTPHandler: # Also check for extra_headers in kwargs (from config or direct calls) extra_headers_from_kwargs = kwargs.get("extra_headers", None) # Merge all header sources: forwarded < extra_headers < provider_specific - merged_headers = {} + merged_headers: dict[str, str] = {} # mutable-ok: merged in place from multiple header sources below if forwarded_headers: merged_headers.update(forwarded_headers) if extra_headers_from_kwargs: @@ -2370,14 +2415,16 @@ class BaseLLMHTTPHandler: model: str, input: str | ResponseInputParam, custom_llm_provider: str, - response_api_optional_request_params: dict[str, Any], + response_api_optional_request_params: dict[ + str, object + ], # mutable-ok: parameter mirrors the caller's optional-params dict payload litellm_params: GenericLiteLLMParams, logging_obj: LiteLLMLoggingObj, ) -> tuple[ str, str | ResponseInputParam, str, - dict[str, Any], + dict[str, object], GenericLiteLLMParams, ]: if not _has_pre_call_deployment_hook(logging_obj): @@ -2444,7 +2491,9 @@ class BaseLLMHTTPHandler: model: str, input: str | ResponseInputParam, responses_api_provider_config: BaseResponsesAPIConfig, - response_api_optional_request_params: dict[str, Any], + response_api_optional_request_params: dict[ + str, object + ], # mutable-ok: parameter mirrors the caller's optional-params dict payload custom_llm_provider: str, litellm_params: GenericLiteLLMParams, logging_obj: LiteLLMLoggingObj, @@ -2509,8 +2558,9 @@ class BaseLLMHTTPHandler: else: sync_httpx_client = client + request_extra_headers = response_api_optional_request_params.get("extra_headers") headers = responses_api_provider_config.validate_environment( - headers=response_api_optional_request_params.get("extra_headers", {}) or {}, + headers=request_extra_headers if isinstance(request_extra_headers, dict) else {}, model=model, litellm_params=litellm_params, ) @@ -2519,7 +2569,7 @@ class BaseLLMHTTPHandler: headers.update(extra_headers) # Check if streaming is requested - stream = response_api_optional_request_params.get("stream", False) + stream = bool(response_api_optional_request_params.get("stream", False)) api_base = responses_api_provider_config.get_complete_url( api_base=litellm_params.api_base, @@ -2572,8 +2622,6 @@ class BaseLLMHTTPHandler: stream=stream, fake_stream=fake_stream, ) - body_kwargs: dict[str, Any] = {"data": signed_body} if signed_body is not None else {"json": data} - ## LOGGING logging_obj.pre_call( input=input, @@ -2587,13 +2635,18 @@ class BaseLLMHTTPHandler: try: if is_stream_request: - response = sync_httpx_client.post( - url=api_base, - headers=headers, - timeout=timeout or float(response_api_optional_request_params.get("timeout", 0)), - stream=stream, - **body_kwargs, + request_timeout = response_api_optional_request_params.get("timeout", 0) + stream_timeout_value = timeout or float( + request_timeout if isinstance(request_timeout, (int, float)) else 0 ) + if signed_body is not None: + response = sync_httpx_client.post( + url=api_base, headers=headers, timeout=stream_timeout_value, stream=stream, data=signed_body + ) + else: + response = sync_httpx_client.post( + url=api_base, headers=headers, timeout=stream_timeout_value, stream=stream, json=data + ) if fake_stream is True: return MockResponsesAPIStreamingIterator( response=response, @@ -2617,12 +2670,18 @@ class BaseLLMHTTPHandler: call_type=CallTypes.responses.value, ) else: - response = sync_httpx_client.post( - url=api_base, - headers=headers, - timeout=timeout or float(response_api_optional_request_params.get("timeout", 0)), - **body_kwargs, + request_timeout = response_api_optional_request_params.get("timeout", 0) + non_stream_timeout_value = timeout or float( + request_timeout if isinstance(request_timeout, (int, float)) else 0 ) + if signed_body is not None: + response = sync_httpx_client.post( + url=api_base, headers=headers, timeout=non_stream_timeout_value, data=signed_body + ) + else: + response = sync_httpx_client.post( + url=api_base, headers=headers, timeout=non_stream_timeout_value, json=data + ) except Exception as e: raise self._handle_error( e=e, @@ -2687,8 +2746,9 @@ class BaseLLMHTTPHandler: else: async_httpx_client = client + request_extra_headers = response_api_optional_request_params.get("extra_headers") headers = responses_api_provider_config.validate_environment( - headers=response_api_optional_request_params.get("extra_headers", {}) or {}, + headers=request_extra_headers if isinstance(request_extra_headers, dict) else {}, model=model, litellm_params=litellm_params, ) @@ -2697,7 +2757,7 @@ class BaseLLMHTTPHandler: headers.update(extra_headers) # Check if streaming is requested - stream = response_api_optional_request_params.get("stream", False) + stream = bool(response_api_optional_request_params.get("stream", False)) api_base = responses_api_provider_config.get_complete_url( api_base=litellm_params.api_base, @@ -2747,7 +2807,6 @@ class BaseLLMHTTPHandler: stream=stream, fake_stream=fake_stream, ) - body_kwargs: dict[str, Any] = {"data": signed_body} if signed_body is not None else {"json": data} ## LOGGING logging_obj.pre_call( @@ -2762,13 +2821,18 @@ class BaseLLMHTTPHandler: try: if is_stream_request: - response = await async_httpx_client.post( - url=api_base, - headers=headers, - timeout=timeout or float(response_api_optional_request_params.get("timeout", 0)), - stream=stream, - **body_kwargs, + request_timeout = response_api_optional_request_params.get("timeout", 0) + stream_timeout_value = timeout or float( + request_timeout if isinstance(request_timeout, (int, float)) else 0 ) + if signed_body is not None: + response = await async_httpx_client.post( + url=api_base, headers=headers, timeout=stream_timeout_value, stream=stream, data=signed_body + ) + else: + response = await async_httpx_client.post( + url=api_base, headers=headers, timeout=stream_timeout_value, stream=stream, json=data + ) if fake_stream is True: return MockResponsesAPIStreamingIterator( @@ -2794,12 +2858,18 @@ class BaseLLMHTTPHandler: call_type=CallTypes.responses.value, ) else: - response = await async_httpx_client.post( - url=api_base, - headers=headers, - timeout=timeout or float(response_api_optional_request_params.get("timeout", 0)), - **body_kwargs, + request_timeout = response_api_optional_request_params.get("timeout", 0) + non_stream_timeout_value = timeout or float( + request_timeout if isinstance(request_timeout, (int, float)) else 0 ) + if signed_body is not None: + response = await async_httpx_client.post( + url=api_base, headers=headers, timeout=non_stream_timeout_value, data=signed_body + ) + else: + response = await async_httpx_client.post( + url=api_base, headers=headers, timeout=non_stream_timeout_value, json=data + ) except Exception as e: raise self._handle_error( @@ -3355,16 +3425,19 @@ class BaseLLMHTTPHandler: """ if upload_url_location == "headers": # Google Cloud Storage style - URL in X-Goog-Upload-URL header - upload_url = response.headers.get("X-Goog-Upload-URL") - return upload_url, None + upload_url_header = response.headers.get("X-Goog-Upload-URL") + return (upload_url_header if isinstance(upload_url_header, str) else None), None else: # Response body style (e.g., Manus, S3 presigned URLs) try: response_data = response.json() - upload_url = response_data.get(upload_url_key) - return upload_url, response_data if upload_url else None except Exception: return None, None + if not isinstance(response_data, dict): + return None, None + upload_url_raw = response_data.get(upload_url_key) + upload_url = upload_url_raw if isinstance(upload_url_raw, str) else None + return upload_url, response_data if upload_url else None def create_file( self, @@ -3480,7 +3553,7 @@ class BaseLLMHTTPHandler: ): # Handle pre-signed requests (e.g., from Bedrock S3 uploads) # Type narrowing: this is a plain dict, not TwoStepFileUploadConfig - presigned_request = cast(dict[str, Any], transformed_request) + presigned_request = _TWO_STEP_FILE_UPLOAD_REQUEST_ADAPTER.validate_python(transformed_request) upload_response = getattr(sync_httpx_client, presigned_request["method"].lower())( url=presigned_request["url"], headers=presigned_request["headers"], @@ -3526,7 +3599,7 @@ class BaseLLMHTTPHandler: elif isinstance(transformed_request, dict) and "file" in transformed_request: # Handle multipart form-data uploads (e.g., Anthropic Files API) # The dict contains tuples suitable for httpx's `files` parameter - file_request = cast(dict[str, Any], transformed_request) + file_request = transformed_request upload_response = sync_httpx_client.post( url=api_base, headers=headers, @@ -3642,7 +3715,7 @@ class BaseLLMHTTPHandler: ): # Handle pre-signed requests (e.g., from Bedrock S3 uploads) # Type narrowing: this is a plain dict, not TwoStepFileUploadConfig - presigned_request = cast(dict[str, Any], transformed_request) + presigned_request = _TWO_STEP_FILE_UPLOAD_REQUEST_ADAPTER.validate_python(transformed_request) upload_response = await getattr(async_httpx_client, presigned_request["method"].lower())( url=presigned_request["url"], headers=presigned_request["headers"], @@ -3734,13 +3807,12 @@ class BaseLLMHTTPHandler: timeout: float | httpx.Timeout | None, ) -> httpx.Response: headers = {**base_headers, "Content-Type": content_type} - kwargs: dict[str, Any] = { - "headers": headers, - "content": self._iter_in_blocks(body_stream.iter_bytes(), self._MEDIA_UPLOAD_BLOCK_SIZE), - } - if timeout is not None: - kwargs["timeout"] = timeout - resp = client.client.post(url, **kwargs) + resp = client.client.post( + url, + headers=headers, + content=self._iter_in_blocks(body_stream.iter_bytes(), self._MEDIA_UPLOAD_BLOCK_SIZE), + timeout=timeout if timeout is not None else httpx.USE_CLIENT_DEFAULT, + ) self._check_media_upload_response(resp) return resp @@ -3771,10 +3843,12 @@ class BaseLLMHTTPHandler: break yield cast(bytes, block) - kwargs: dict[str, Any] = {"headers": headers, "content": _abody()} - if timeout is not None: - kwargs["timeout"] = timeout - resp = await client.client.post(url, **kwargs) + resp = await client.client.post( + url, + headers=headers, + content=_abody(), + timeout=timeout if timeout is not None else httpx.USE_CLIENT_DEFAULT, + ) await resp.aread() self._check_media_upload_response(resp) return resp @@ -3849,10 +3923,10 @@ class BaseLLMHTTPHandler: try: if isinstance(transformed_request, dict) and "method" in transformed_request: # Handle pre-signed requests (e.g., from Bedrock with AWS auth) - batch_response = getattr(sync_httpx_client, transformed_request["method"].lower())( - url=transformed_request["url"], - headers=transformed_request["headers"], - data=transformed_request["data"], + presigned_batch_request = _PRESIGNED_BATCH_REQUEST_ADAPTER.validate_python(transformed_request) + batch_response = _send_presigned_batch_request( + client=sync_httpx_client, + request=presigned_batch_request, timeout=timeout, ) elif isinstance(transformed_request, dict): @@ -3937,17 +4011,19 @@ class BaseLLMHTTPHandler: try: if isinstance(transformed_request, dict) and "method" in transformed_request: # Handle pre-signed requests (e.g., from Bedrock with AWS auth) - method = transformed_request["method"].lower() - request_kwargs = { - "url": transformed_request["url"], - "headers": transformed_request["headers"], - } - - # Only add data for non-GET requests - if method != "get" and transformed_request.get("data") is not None: - request_kwargs["data"] = transformed_request["data"] - - batch_response = getattr(sync_httpx_client, method)(**request_kwargs) + presigned_batch_request = _PRESIGNED_BATCH_REQUEST_ADAPTER.validate_python(transformed_request) + retrieve_method = presigned_batch_request["method"].lower() + if retrieve_method == "get": + batch_response = getattr(sync_httpx_client, retrieve_method)( + url=presigned_batch_request["url"], + headers=presigned_batch_request["headers"], + ) + else: + batch_response = getattr(sync_httpx_client, retrieve_method)( + url=presigned_batch_request["url"], + headers=presigned_batch_request["headers"], + data=presigned_batch_request["data"], + ) elif isinstance(transformed_request, dict) and api_base: # For other providers that use JSON requests batch_response = sync_httpx_client.get( @@ -4014,10 +4090,10 @@ class BaseLLMHTTPHandler: try: if isinstance(transformed_request, dict) and "method" in transformed_request: # Handle pre-signed requests (e.g., from Bedrock with AWS auth) - batch_response = await getattr(async_httpx_client, transformed_request["method"].lower())( - url=transformed_request["url"], - headers=transformed_request["headers"], - data=transformed_request["data"], + presigned_batch_request = _PRESIGNED_BATCH_REQUEST_ADAPTER.validate_python(transformed_request) + batch_response = await _asend_presigned_batch_request( + client=async_httpx_client, + request=presigned_batch_request, timeout=timeout, ) elif isinstance(transformed_request, dict): @@ -4094,17 +4170,19 @@ class BaseLLMHTTPHandler: try: if isinstance(transformed_request, dict) and "method" in transformed_request: # Handle pre-signed requests (e.g., from Bedrock with AWS auth) - method = transformed_request["method"].lower() - request_kwargs = { - "url": transformed_request["url"], - "headers": transformed_request["headers"], - } - - # Only add data for non-GET requests - if method != "get" and transformed_request.get("data") is not None: - request_kwargs["data"] = transformed_request["data"] - - batch_response = await getattr(async_httpx_client, method)(**request_kwargs) + presigned_batch_request = _PRESIGNED_BATCH_REQUEST_ADAPTER.validate_python(transformed_request) + retrieve_method = presigned_batch_request["method"].lower() + if retrieve_method == "get": + batch_response = await getattr(async_httpx_client, retrieve_method)( + url=presigned_batch_request["url"], + headers=presigned_batch_request["headers"], + ) + else: + batch_response = await getattr(async_httpx_client, retrieve_method)( + url=presigned_batch_request["url"], + headers=presigned_batch_request["headers"], + data=presigned_batch_request["data"], + ) elif isinstance(transformed_request, dict) and api_base: # For other providers that use JSON requests batch_response = await async_httpx_client.get( @@ -4361,7 +4439,6 @@ class BaseLLMHTTPHandler: api_key=litellm_params.api_key, model=model, ) - body_kwargs: dict[str, Any] = {"data": signed_body} if signed_body is not None else {"json": data} ## LOGGING logging_obj.pre_call( @@ -4375,7 +4452,10 @@ class BaseLLMHTTPHandler: ) try: - response = sync_httpx_client.post(url=url, headers=headers, timeout=timeout, **body_kwargs) + if signed_body is not None: + response = sync_httpx_client.post(url=url, headers=headers, timeout=timeout, data=signed_body) + else: + response = sync_httpx_client.post(url=url, headers=headers, timeout=timeout, json=data) except Exception as e: raise self._handle_error( @@ -4453,7 +4533,6 @@ class BaseLLMHTTPHandler: api_key=litellm_params.api_key, model=model, ) - body_kwargs: dict[str, Any] = {"data": signed_body} if signed_body is not None else {"json": data} ## LOGGING logging_obj.pre_call( @@ -4467,7 +4546,10 @@ class BaseLLMHTTPHandler: ) try: - response = await async_httpx_client.post(url=url, headers=headers, timeout=timeout, **body_kwargs) + if signed_body is not None: + response = await async_httpx_client.post(url=url, headers=headers, timeout=timeout, data=signed_body) + else: + response = await async_httpx_client.post(url=url, headers=headers, timeout=timeout, json=data) except Exception as e: raise self._handle_error( @@ -9453,7 +9535,9 @@ class BaseLLMHTTPHandler: litellm_params=dict(litellm_params), extra_body=extra_body, ) - all_optional_params: dict[str, Any] = dict(litellm_params) + all_optional_params: dict[str, object] = dict( + litellm_params + ) # mutable-ok: built once via dict() from litellm_params, matching that dict-based contract all_optional_params.update(vector_store_search_optional_params or {}) headers, signed_json_body = vector_store_provider_config.sign_request( headers=headers, @@ -9549,7 +9633,9 @@ class BaseLLMHTTPHandler: extra_body=extra_body, ) - all_optional_params: dict[str, Any] = dict(litellm_params) + all_optional_params: dict[str, object] = dict( + litellm_params + ) # mutable-ok: built once via dict() from litellm_params, matching that dict-based contract all_optional_params.update(vector_store_search_optional_params or {}) headers, signed_json_body = vector_store_provider_config.sign_request( @@ -9869,7 +9955,7 @@ class BaseLLMHTTPHandler: url = api_base - params: dict[str, Any] = {} + params: dict[str, str | int] = {} # mutable-ok: populated incrementally as query params below if after is not None: params["after"] = after if before is not None: @@ -9947,7 +10033,7 @@ class BaseLLMHTTPHandler: url = api_base - params: dict[str, Any] = {} + params: dict[str, str | int] = {} # mutable-ok: populated incrementally as query params below if after is not None: params["after"] = after if before is not None: @@ -10010,13 +10096,16 @@ class BaseLLMHTTPHandler: encoded_vector_store_id = encode_url_path_segment(vector_store_id, field_name="vector_store_id") url = f"{api_base}/{encoded_vector_store_id}" - request_body: dict[str, Any] = dict(vector_store_update_optional_params) + request_body: dict[str, object] = dict( + vector_store_update_optional_params + ) # mutable-ok: built once via dict() from the optional-params payload # Clean metadata to only include string values (OpenAI requirement) - if "metadata" in request_body and request_body["metadata"] is not None: + metadata_value = request_body.get("metadata") + if isinstance(metadata_value, dict): from litellm.utils import add_openai_metadata - request_body["metadata"] = add_openai_metadata(request_body["metadata"]) + request_body["metadata"] = add_openai_metadata(metadata_value) if extra_body: request_body.update(extra_body) @@ -10088,13 +10177,16 @@ class BaseLLMHTTPHandler: encoded_vector_store_id = encode_url_path_segment(vector_store_id, field_name="vector_store_id") url = f"{api_base}/{encoded_vector_store_id}" - request_body: dict[str, Any] = dict(vector_store_update_optional_params) + request_body: dict[str, object] = dict( + vector_store_update_optional_params + ) # mutable-ok: built once via dict() from the optional-params payload # Clean metadata to only include string values (OpenAI requirement) - if "metadata" in request_body and request_body["metadata"] is not None: + metadata_value = request_body.get("metadata") + if isinstance(metadata_value, dict): from litellm.utils import add_openai_metadata - request_body["metadata"] = add_openai_metadata(request_body["metadata"]) + request_body["metadata"] = add_openai_metadata(metadata_value) if extra_body: request_body.update(extra_body) diff --git a/litellm/main.py b/litellm/main.py index 8b2c3c72f76..c788242fb9e 100644 --- a/litellm/main.py +++ b/litellm/main.py @@ -29,6 +29,7 @@ from typing import ( Any, Literal, Optional, + TypedDict, Union, cast, get_args, @@ -40,6 +41,8 @@ from litellm._uuid import uuid if TYPE_CHECKING: from aiohttp import ClientSession + from litellm.router import Router + import dotenv import httpx import openai @@ -350,7 +353,7 @@ class LiteLLM: class Chat: - def __init__(self, params, router_obj: Any | None): + def __init__(self, params, router_obj: "Router | None"): self.params = params if self.params.get("acompletion", False) is True: self.params.pop("acompletion") @@ -360,7 +363,7 @@ class Chat: class Completions: - def __init__(self, params, router_obj: Any | None): + def __init__(self, params, router_obj: "Router | None"): self.params = params self.router_obj = router_obj @@ -376,7 +379,7 @@ class Completions: class AsyncCompletions: - def __init__(self, params, router_obj: Any | None): + def __init__(self, params, router_obj: "Router | None"): self.params = params self.router_obj = router_obj @@ -1137,13 +1140,18 @@ def _register_custom_pricing_for_request( ) +_CompletionClient = Union[ + openai.OpenAI, openai.AsyncOpenAI, openai.AzureOpenAI, openai.AsyncAzureOpenAI, HTTPHandler, AsyncHTTPHandler +] + + def _complete_azure(ctx: _CompletionDispatchContext) -> _CompletionDispatchResult: _azure_detection_model = ctx._azure_detection_model acompletion = ctx.acompletion api_base = ctx.api_base api_key = ctx.api_key api_version = ctx.api_version - client = ctx.client + client: _CompletionClient | None = ctx.client custom_llm_provider = ctx.custom_llm_provider extra_headers = ctx.extra_headers headers = ctx.headers @@ -1193,6 +1201,9 @@ def _complete_azure(ctx: _CompletionDispatchContext) -> _CompletionDispatchResul if max_retries is not None: optional_params["max_retries"] = max_retries + config: dict[ + str, object + ] # mutable-ok: declared ahead of a branch that assigns it from get_config()'s dict contract if litellm.AzureOpenAIO1Config().is_o_series_model(model=_azure_detection_model): ## LOAD CONFIG - if set config = litellm.AzureOpenAIO1Config.get_config() @@ -1275,7 +1286,7 @@ def _complete_azure_text(ctx: _CompletionDispatchContext) -> _CompletionDispatch api_base = ctx.api_base api_key = ctx.api_key api_version = ctx.api_version - client = ctx.client + client: _CompletionClient | None = ctx.client extra_headers = ctx.extra_headers headers = ctx.headers litellm_params = ctx.litellm_params @@ -1318,7 +1329,9 @@ def _complete_azure_text(ctx: _CompletionDispatchContext) -> _CompletionDispatch optional_params["extra_headers"] = extra_headers ## LOAD CONFIG - if set - config = litellm.AzureOpenAIConfig.get_config() + config: dict[str, object] = ( + litellm.AzureOpenAIConfig.get_config() + ) # mutable-ok: matches get_config()'s plain dict return contract for k, v in config.items(): if ( k not in optional_params @@ -1367,7 +1380,7 @@ def _complete_deepseek(ctx: _CompletionDispatchContext) -> _CompletionDispatchRe acompletion = ctx.acompletion api_base = ctx.api_base api_key = ctx.api_key - client = ctx.client + client: HTTPHandler | AsyncHTTPHandler | None = ctx.client custom_llm_provider = ctx.custom_llm_provider headers = ctx.headers litellm_params = ctx.litellm_params @@ -1418,7 +1431,7 @@ def _complete_azure_ai(ctx: _CompletionDispatchContext) -> _CompletionDispatchRe acompletion = ctx.acompletion api_base = ctx.api_base api_key = ctx.api_key - client = ctx.client + client: HTTPHandler | AsyncHTTPHandler | None = ctx.client custom_llm_provider = ctx.custom_llm_provider extra_headers = ctx.extra_headers headers = ctx.headers @@ -1574,7 +1587,7 @@ def _complete_text_completion_openai( acompletion = ctx.acompletion api_base = ctx.api_base api_key = ctx.api_key - client = ctx.client + client: _CompletionClient | None = ctx.client custom_llm_provider = ctx.custom_llm_provider headers = ctx.headers litellm_params = ctx.litellm_params @@ -1605,7 +1618,9 @@ def _complete_text_completion_openai( headers = headers or litellm.headers ## LOAD CONFIG - if set - config = litellm.OpenAITextCompletionConfig.get_config() + config: dict[str, object] = ( + litellm.OpenAITextCompletionConfig.get_config() + ) # mutable-ok: matches get_config()'s plain dict return contract for k, v in config.items(): if ( k not in optional_params @@ -1656,7 +1671,7 @@ def _complete_fireworks_ai( acompletion = ctx.acompletion api_base = ctx.api_base api_key = ctx.api_key - client = ctx.client + client: HTTPHandler | AsyncHTTPHandler | None = ctx.client custom_llm_provider = ctx.custom_llm_provider headers = ctx.headers litellm_params = ctx.litellm_params @@ -1707,7 +1722,7 @@ def _complete_heroku(ctx: _CompletionDispatchContext) -> _CompletionDispatchResu acompletion = ctx.acompletion api_base = ctx.api_base api_key = ctx.api_key - client = ctx.client + client: HTTPHandler | AsyncHTTPHandler | None = ctx.client custom_llm_provider = ctx.custom_llm_provider headers = ctx.headers litellm_params = ctx.litellm_params @@ -1757,7 +1772,7 @@ def _complete_ragflow(ctx: _CompletionDispatchContext) -> _CompletionDispatchRes acompletion = ctx.acompletion api_base = ctx.api_base api_key = ctx.api_key - client = ctx.client + client: HTTPHandler | AsyncHTTPHandler | None = ctx.client custom_llm_provider = ctx.custom_llm_provider headers = ctx.headers litellm_params = ctx.litellm_params @@ -1807,7 +1822,7 @@ def _complete_xai(ctx: _CompletionDispatchContext) -> _CompletionDispatchResult: acompletion = ctx.acompletion api_base = ctx.api_base api_key = ctx.api_key - client = ctx.client + client: HTTPHandler | AsyncHTTPHandler | None = ctx.client custom_llm_provider = ctx.custom_llm_provider headers = ctx.headers litellm_params = ctx.litellm_params @@ -1858,7 +1873,7 @@ def _complete_groq(ctx: _CompletionDispatchContext) -> _CompletionDispatchResult acompletion = ctx.acompletion api_base = ctx.api_base api_key = ctx.api_key - client = ctx.client + client: HTTPHandler | AsyncHTTPHandler | None = ctx.client custom_llm_provider = ctx.custom_llm_provider headers = ctx.headers litellm_params = ctx.litellm_params @@ -1889,7 +1904,9 @@ def _complete_groq(ctx: _CompletionDispatchContext) -> _CompletionDispatchResult headers = headers or litellm.headers ## LOAD CONFIG - if set - config = litellm.GroqChatConfig.get_config() + config: dict[str, object] = ( + litellm.GroqChatConfig.get_config() + ) # mutable-ok: matches get_config()'s plain dict return contract for k, v in config.items(): if ( k not in optional_params @@ -1922,7 +1939,7 @@ def _complete_bedrock_mantle( acompletion = ctx.acompletion api_base = ctx.api_base api_key = ctx.api_key - client = ctx.client + client: HTTPHandler | AsyncHTTPHandler | None = ctx.client custom_llm_provider = ctx.custom_llm_provider headers = ctx.headers litellm_params = ctx.litellm_params @@ -1938,7 +1955,9 @@ def _complete_bedrock_mantle( api_base = api_base or litellm.api_base or get_secret("BEDROCK_MANTLE_API_BASE") api_key = api_key or litellm.api_key or get_secret("BEDROCK_MANTLE_API_KEY") headers = headers or litellm.headers - config = litellm.BedrockMantleChatConfig.get_config() + config: dict[str, object] = ( + litellm.BedrockMantleChatConfig.get_config() + ) # mutable-ok: matches get_config()'s plain dict return contract for k, v in config.items(): if k not in optional_params: optional_params[k] = v @@ -1966,7 +1985,7 @@ def _complete_a2a(ctx: _CompletionDispatchContext) -> _CompletionDispatchResult: acompletion = ctx.acompletion api_base = ctx.api_base api_key = ctx.api_key - client = ctx.client + client: HTTPHandler | AsyncHTTPHandler | None = ctx.client custom_llm_provider = ctx.custom_llm_provider headers = ctx.headers litellm_params = ctx.litellm_params @@ -2029,7 +2048,7 @@ def _complete_gigachat(ctx: _CompletionDispatchContext) -> _CompletionDispatchRe acompletion = ctx.acompletion api_base = ctx.api_base api_key = ctx.api_key - client = ctx.client + client: HTTPHandler | AsyncHTTPHandler | None = ctx.client custom_llm_provider = ctx.custom_llm_provider headers = ctx.headers litellm_params = ctx.litellm_params @@ -2091,7 +2110,7 @@ def _complete_sap(ctx: _CompletionDispatchContext) -> _CompletionDispatchResult: acompletion = ctx.acompletion api_base = ctx.api_base api_key = ctx.api_key - client = ctx.client + client: HTTPHandler | AsyncHTTPHandler | None = ctx.client custom_llm_provider = ctx.custom_llm_provider headers = ctx.headers litellm_params = ctx.litellm_params @@ -2106,7 +2125,9 @@ def _complete_sap(ctx: _CompletionDispatchContext) -> _CompletionDispatchResult: headers = headers or litellm.headers ## LOAD CONFIG - if set - config = litellm.GenAIHubOrchestrationConfig.get_config() + config: dict[str, object] = ( + litellm.GenAIHubOrchestrationConfig.get_config() + ) # mutable-ok: matches get_config()'s plain dict return contract for k, v in config.items(): if ( k not in optional_params @@ -2139,7 +2160,7 @@ def _complete_aiohttp_openai( acompletion = ctx.acompletion api_base = ctx.api_base api_key = ctx.api_key - client = ctx.client + client: HTTPHandler | AsyncHTTPHandler | ClientSession | None = ctx.client custom_llm_provider = ctx.custom_llm_provider extra_headers = ctx.extra_headers headers = ctx.headers @@ -2194,7 +2215,7 @@ def _complete_cometapi(ctx: _CompletionDispatchContext) -> _CompletionDispatchRe acompletion = ctx.acompletion api_base = ctx.api_base api_key = ctx.api_key - client = ctx.client + client: HTTPHandler | AsyncHTTPHandler | None = ctx.client custom_llm_provider = ctx.custom_llm_provider headers = ctx.headers litellm_params = ctx.litellm_params @@ -2243,7 +2264,7 @@ def _complete_minimax(ctx: _CompletionDispatchContext) -> _CompletionDispatchRes acompletion = ctx.acompletion api_base = ctx.api_base api_key = ctx.api_key - client = ctx.client + client: HTTPHandler | AsyncHTTPHandler | None = ctx.client custom_llm_provider = ctx.custom_llm_provider headers = ctx.headers litellm_params = ctx.litellm_params @@ -2289,7 +2310,7 @@ def _complete_hosted_vllm(ctx: _CompletionDispatchContext) -> _CompletionDispatc acompletion = ctx.acompletion api_base = ctx.api_base api_key = ctx.api_key - client = ctx.client + client: HTTPHandler | AsyncHTTPHandler | None = ctx.client custom_llm_provider = ctx.custom_llm_provider headers = ctx.headers litellm_params = ctx.litellm_params @@ -2335,7 +2356,7 @@ def _complete_custom_openai( acompletion = ctx.acompletion api_base = ctx.api_base api_key = ctx.api_key - client = ctx.client + client: HTTPHandler | AsyncHTTPHandler | None = ctx.client custom_llm_provider = ctx.custom_llm_provider custom_prompt_dict = ctx.custom_prompt_dict extra_headers = ctx.extra_headers @@ -2401,7 +2422,9 @@ def _complete_custom_openai( optional_params["metadata"] = openai_metadata ## LOAD CONFIG - if set - config = litellm.OpenAIConfig.get_config() + config: dict[str, object] = ( + litellm.OpenAIConfig.get_config() + ) # mutable-ok: matches get_config()'s plain dict return contract for k, v in config.items(): if ( k not in optional_params @@ -2479,7 +2502,7 @@ def _complete_mistral(ctx: _CompletionDispatchContext) -> _CompletionDispatchRes acompletion = ctx.acompletion api_base = ctx.api_base api_key = ctx.api_key - client = ctx.client + client: HTTPHandler | AsyncHTTPHandler | None = ctx.client custom_llm_provider = ctx.custom_llm_provider headers = ctx.headers litellm_params = ctx.litellm_params @@ -2630,7 +2653,7 @@ def _complete_anthropic(ctx: _CompletionDispatchContext) -> _CompletionDispatchR acompletion = ctx.acompletion api_base = ctx.api_base api_key = ctx.api_key - client = ctx.client + client: _CompletionClient | None = ctx.client custom_llm_provider = ctx.custom_llm_provider custom_prompt_dict = ctx.custom_prompt_dict headers = ctx.headers @@ -2929,7 +2952,7 @@ def _complete_huggingface(ctx: _CompletionDispatchContext) -> _CompletionDispatc acompletion = ctx.acompletion api_base = ctx.api_base api_key = ctx.api_key - client = ctx.client + client: HTTPHandler | AsyncHTTPHandler | None = ctx.client custom_llm_provider = ctx.custom_llm_provider headers = ctx.headers litellm_params = ctx.litellm_params @@ -2972,7 +2995,7 @@ def _complete_oci(ctx: _CompletionDispatchContext) -> _CompletionDispatchResult: acompletion = ctx.acompletion api_base = ctx.api_base api_key = ctx.api_key - client = ctx.client + client: HTTPHandler | AsyncHTTPHandler | None = ctx.client custom_llm_provider = ctx.custom_llm_provider headers = ctx.headers litellm_params = ctx.litellm_params @@ -3007,7 +3030,7 @@ def _complete_compactifai(ctx: _CompletionDispatchContext) -> _CompletionDispatc acompletion = ctx.acompletion api_base = ctx.api_base api_key = ctx.api_key - client = ctx.client + client: HTTPHandler | AsyncHTTPHandler | None = ctx.client custom_llm_provider = ctx.custom_llm_provider headers = ctx.headers litellm_params = ctx.litellm_params @@ -3083,7 +3106,7 @@ def _complete_databricks(ctx: _CompletionDispatchContext) -> _CompletionDispatch acompletion = ctx.acompletion api_base = ctx.api_base api_key = ctx.api_key - client = ctx.client + client: HTTPHandler | AsyncHTTPHandler | None = ctx.client headers = ctx.headers litellm_params = ctx.litellm_params logging = ctx.logging @@ -3155,7 +3178,7 @@ def _complete_datarobot(ctx: _CompletionDispatchContext) -> _CompletionDispatchR acompletion = ctx.acompletion api_base = ctx.api_base api_key = ctx.api_key - client = ctx.client + client: HTTPHandler | AsyncHTTPHandler | None = ctx.client custom_llm_provider = ctx.custom_llm_provider headers = ctx.headers litellm_params = ctx.litellm_params @@ -3192,7 +3215,7 @@ def _complete_openrouter(ctx: _CompletionDispatchContext) -> _CompletionDispatch acompletion = ctx.acompletion api_base = ctx.api_base api_key = ctx.api_key - client = ctx.client + client: HTTPHandler | AsyncHTTPHandler | None = ctx.client headers = ctx.headers litellm_params = ctx.litellm_params logging = ctx.logging @@ -3229,7 +3252,9 @@ def _complete_openrouter(ctx: _CompletionDispatchContext) -> _CompletionDispatch headers = openrouter_headers ## Load Config - config = litellm.OpenrouterConfig.get_config() + config: dict[str, object] = ( + litellm.OpenrouterConfig.get_config() + ) # mutable-ok: matches get_config()'s plain dict return contract for k, v in config.items(): if k == "extra_body": # we use openai 'extra_body' to pass openrouter specific params - transforms, route, models @@ -3271,7 +3296,7 @@ def _complete_vercel_ai_gateway( acompletion = ctx.acompletion api_base = ctx.api_base api_key = ctx.api_key - client = ctx.client + client: HTTPHandler | AsyncHTTPHandler | None = ctx.client headers = ctx.headers litellm_params = ctx.litellm_params logging = ctx.logging @@ -3307,7 +3332,9 @@ def _complete_vercel_ai_gateway( headers = vercel_headers ## Load Config - config = litellm.VercelAIGatewayConfig.get_config() + config: dict[str, object] = ( + litellm.VercelAIGatewayConfig.get_config() + ) # mutable-ok: matches get_config()'s plain dict return contract for k, v in config.items(): if k == "extra_body": # we use openai 'extra_body' to pass vercel specific params - providerOptions @@ -3349,7 +3376,7 @@ def _complete_vertex_ai_beta( acompletion = ctx.acompletion api_base = ctx.api_base api_key = ctx.api_key - client = ctx.client + client: HTTPHandler | AsyncHTTPHandler | None = ctx.client custom_llm_provider = ctx.custom_llm_provider headers = ctx.headers litellm_params = ctx.litellm_params @@ -3414,7 +3441,7 @@ def _complete_vertex_ai_beta( def _complete_vertex_ai(ctx: _CompletionDispatchContext) -> _CompletionDispatchResult: acompletion = ctx.acompletion api_base = ctx.api_base - client = ctx.client + client: HTTPHandler | AsyncHTTPHandler | None = ctx.client custom_llm_provider = ctx.custom_llm_provider custom_prompt_dict = ctx.custom_prompt_dict headers = ctx.headers @@ -3711,7 +3738,7 @@ def _complete_text_completion_inception( acompletion = ctx.acompletion api_base = ctx.api_base api_key = ctx.api_key - client = ctx.client + client: _CompletionClient | None = ctx.client headers = ctx.headers litellm_params = ctx.litellm_params logger_fn = ctx.logger_fn @@ -3775,7 +3802,7 @@ def _complete_sagemaker_chat( acompletion = ctx.acompletion api_base = ctx.api_base api_key = ctx.api_key - client = ctx.client + client: HTTPHandler | AsyncHTTPHandler | None = ctx.client custom_llm_provider = ctx.custom_llm_provider headers = ctx.headers litellm_params = ctx.litellm_params @@ -3838,7 +3865,7 @@ def _complete_bedrock(ctx: _CompletionDispatchContext) -> _CompletionDispatchRes acompletion = ctx.acompletion api_base = ctx.api_base api_key = ctx.api_key - client = ctx.client + client: HTTPHandler | AsyncHTTPHandler | None = ctx.client custom_prompt_dict = ctx.custom_prompt_dict headers = ctx.headers litellm_params = ctx.litellm_params @@ -3962,7 +3989,7 @@ def _complete_watsonx(ctx: _CompletionDispatchContext) -> _CompletionDispatchRes acompletion = ctx.acompletion api_base = ctx.api_base api_key = ctx.api_key - client = ctx.client + client: HTTPHandler | AsyncHTTPHandler | None = ctx.client custom_prompt_dict = ctx.custom_prompt_dict headers = ctx.headers litellm_params = ctx.litellm_params @@ -4001,7 +4028,7 @@ def _complete_watsonx_text( acompletion = ctx.acompletion api_base = ctx.api_base api_key = ctx.api_key - client = ctx.client + client: HTTPHandler | AsyncHTTPHandler | None = ctx.client headers = ctx.headers litellm_params = ctx.litellm_params logging = ctx.logging @@ -4113,7 +4140,7 @@ def _complete_ollama(ctx: _CompletionDispatchContext) -> _CompletionDispatchResu acompletion = ctx.acompletion api_base = ctx.api_base api_key = ctx.api_key - client = ctx.client + client: HTTPHandler | AsyncHTTPHandler | None = ctx.client headers = ctx.headers litellm_params = ctx.litellm_params logging = ctx.logging @@ -4153,7 +4180,7 @@ def _complete_ollama_chat(ctx: _CompletionDispatchContext) -> _CompletionDispatc acompletion = ctx.acompletion api_base = ctx.api_base api_key = ctx.api_key - client = ctx.client + client: HTTPHandler | AsyncHTTPHandler | None = ctx.client headers = ctx.headers litellm_params = ctx.litellm_params logging = ctx.logging @@ -4268,7 +4295,7 @@ def _complete_cloudflare(ctx: _CompletionDispatchContext) -> _CompletionDispatch def _complete_petals(ctx: _CompletionDispatchContext) -> _CompletionDispatchResult: api_base = ctx.api_base - client = ctx.client + client: HTTPHandler | AsyncHTTPHandler | None = ctx.client litellm_params = ctx.litellm_params logger_fn = ctx.logger_fn logging = ctx.logging @@ -4310,7 +4337,7 @@ def _complete_snowflake(ctx: _CompletionDispatchContext) -> _CompletionDispatchR acompletion = ctx.acompletion api_base = ctx.api_base api_key = ctx.api_key - client = ctx.client + client: _CompletionClient | None = ctx.client custom_llm_provider = ctx.custom_llm_provider headers = ctx.headers litellm_params = ctx.litellm_params @@ -4398,7 +4425,7 @@ def _complete_gdc(ctx: _CompletionDispatchContext) -> _CompletionDispatchResult: acompletion = ctx.acompletion api_base = ctx.api_base api_key = ctx.api_key - client = ctx.client + client: HTTPHandler | AsyncHTTPHandler | None = ctx.client custom_llm_provider = ctx.custom_llm_provider headers = ctx.headers litellm_params = ctx.litellm_params @@ -4437,7 +4464,7 @@ def _complete_bytez(ctx: _CompletionDispatchContext) -> _CompletionDispatchResul acompletion = ctx.acompletion api_base = ctx.api_base api_key = ctx.api_key - client = ctx.client + client: HTTPHandler | AsyncHTTPHandler | None = ctx.client custom_llm_provider = ctx.custom_llm_provider headers = ctx.headers litellm_params = ctx.litellm_params @@ -4477,7 +4504,7 @@ def _complete_lemonade(ctx: _CompletionDispatchContext) -> _CompletionDispatchRe acompletion = ctx.acompletion api_base = ctx.api_base api_key = ctx.api_key - client = ctx.client + client: HTTPHandler | AsyncHTTPHandler | None = ctx.client custom_llm_provider = ctx.custom_llm_provider headers = ctx.headers litellm_params = ctx.litellm_params @@ -4517,7 +4544,7 @@ def _complete_ovhcloud(ctx: _CompletionDispatchContext) -> _CompletionDispatchRe acompletion = ctx.acompletion api_base = ctx.api_base api_key = ctx.api_key - client = ctx.client + client: HTTPHandler | AsyncHTTPHandler | None = ctx.client custom_llm_provider = ctx.custom_llm_provider headers = ctx.headers litellm_params = ctx.litellm_params @@ -4697,7 +4724,7 @@ def _complete_langgraph(ctx: _CompletionDispatchContext) -> _CompletionDispatchR acompletion = ctx.acompletion api_base = ctx.api_base api_key = ctx.api_key - client = ctx.client + client: HTTPHandler | AsyncHTTPHandler | None = ctx.client custom_llm_provider = ctx.custom_llm_provider headers = ctx.headers litellm_params = ctx.litellm_params @@ -4746,7 +4773,7 @@ def _complete_langflow(ctx: _CompletionDispatchContext) -> _CompletionDispatchRe acompletion = ctx.acompletion api_base = ctx.api_base api_key = ctx.api_key - client = ctx.client + client: HTTPHandler | AsyncHTTPHandler | None = ctx.client custom_llm_provider = ctx.custom_llm_provider headers = ctx.headers litellm_params = ctx.litellm_params @@ -4904,7 +4931,7 @@ def completion( # type: ignore thinking = validate_and_fix_thinking_param(thinking=thinking) ######### unpacking kwargs ##################### - args = locals() + args: dict[str, object] = locals() # mutable-ok: locals() itself returns a plain mutable dict # Set by the responses->completion fallback so completion() does not bridge # back to the Responses API: that round-trip mutually recurses forever for a @@ -7144,7 +7171,9 @@ def text_completion( tokenizer = tiktoken.encoding_for_model("text-davinci-003") ## if it's a 2d list - each element in the list is a text_completion() request if len(prompt) > 0 and isinstance(prompt[0], list): - responses = [None for x in prompt] # init responses + responses: list[TextChoices | None] = [ + None for x in prompt + ] # mutable-ok: entries are overwritten in place by the thread pool loop below def process_prompt(i, individual_prompt): decoded_prompt = tokenizer.decode(individual_prompt) @@ -7161,7 +7190,7 @@ def text_completion( text_completion_response["object"] = "text_completion" text_completion_response["created"] = response.get("created", None) text_completion_response["model"] = response.get("model", None) - return response["choices"][0] + return response.choices[0] with concurrent.futures.ThreadPoolExecutor() as executor: completed_futures = [ @@ -8442,7 +8471,7 @@ def stream_chunk_builder( if isinstance(delta_obj, dict): delta = delta_obj elif hasattr(delta_obj, "model_dump"): - delta = cast(dict[str, Any], delta_obj.model_dump()) + delta = cast(dict[str, object], delta_obj.model_dump()) else: delta = {} @@ -8618,7 +8647,9 @@ def stream_chunk_builder( ] if len(provider_specific_chunks) > 0: - combined_provider_fields: dict[str, Any] = {} + combined_provider_fields: dict[ + str, object + ] = {} # mutable-ok: populated incrementally from provider field lookups below for chunk in provider_specific_chunks: fields = chunk["choices"][0]["delta"]["provider_specific_fields"] if isinstance(fields, dict): diff --git a/litellm/proxy/_lazy_openapi_snapshot.py b/litellm/proxy/_lazy_openapi_snapshot.py index 56c08d4176b..2e3671a7caa 100644 --- a/litellm/proxy/_lazy_openapi_snapshot.py +++ b/litellm/proxy/_lazy_openapi_snapshot.py @@ -118,9 +118,11 @@ def generate_snapshot() -> dict[str, dict]: break op["tags"] = [feat.name] full = ensure_unique_openapi_operation_ids(full, used_operation_ids) + components = full.get("components", {}) + schemas = components.get("schemas", {}) if isinstance(components, dict) else {} fragments[feat.name] = { "paths": paths, - "components": {"schemas": full.get("components", {}).get("schemas", {})}, + "components": {"schemas": schemas}, } return fragments diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index 1aeb8814585..a4d056aabf5 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -23,6 +23,7 @@ from typing import ( Any, Literal, Optional, + TypeAlias, TypedDict, Union, cast, @@ -34,7 +35,8 @@ from typing import ( import anyio import websockets import websockets.exceptions -from pydantic import BaseModel, Json, JsonValue +from pydantic import BaseModel, Json, JsonValue, TypeAdapter +from pydantic_core import ErrorDetails from typing_extensions import NotRequired, assert_never from litellm._uuid import uuid @@ -130,9 +132,11 @@ if TYPE_CHECKING: from litellm.integrations.opentelemetry import OpenTelemetry Span = Union[_Span, Any] + _OtelSpanType: TypeAlias = _Span else: Span = Any OpenTelemetry = Any + _OtelSpanType = Any REALTIME_REQUEST_SCOPE_TEMPLATE: dict[str, Any] = { "type": "http", @@ -653,7 +657,7 @@ from fastapi.responses import ( RedirectResponse, StreamingResponse, ) -from fastapi.routing import APIRouter +from fastapi.routing import APIRoute, APIRouter from fastapi.security import OAuth2PasswordBearer from fastapi.security.api_key import APIKeyHeader from fastapi.staticfiles import StaticFiles @@ -720,7 +724,7 @@ def _redact_worker_config_for_logging(worker_config: str | dict[str, JsonValue] return None if isinstance(worker_config, dict): return _redact_secret_values_in_obj(worker_config) - parsed = safe_json_loads(worker_config, default=None) + parsed = TypeAdapter(JsonValue | None).validate_python(safe_json_loads(worker_config, default=None)) if isinstance(parsed, dict): return safe_dumps(_redact_secret_values_in_obj(parsed)) return worker_config @@ -844,21 +848,14 @@ async def _initialize_shared_aiohttp_session(): _build_aiohttp_keepalive_socket_factory, ) - connector_kwargs: dict[str, Any] = { - "keepalive_timeout": AIOHTTP_KEEPALIVE_TIMEOUT, - "ttl_dns_cache": AIOHTTP_TTL_DNS_CACHE, - } - if AIOHTTP_NEEDS_CLEANUP_CLOSED: - connector_kwargs["enable_cleanup_closed"] = True - if AIOHTTP_CONNECTOR_LIMIT > 0: - connector_kwargs["limit"] = AIOHTTP_CONNECTOR_LIMIT - if AIOHTTP_CONNECTOR_LIMIT_PER_HOST > 0: - connector_kwargs["limit_per_host"] = AIOHTTP_CONNECTOR_LIMIT_PER_HOST - socket_factory = _build_aiohttp_keepalive_socket_factory() - if socket_factory is not None: - connector_kwargs["socket_factory"] = socket_factory - - connector = TCPConnector(**connector_kwargs) + connector = TCPConnector( + keepalive_timeout=AIOHTTP_KEEPALIVE_TIMEOUT, + ttl_dns_cache=AIOHTTP_TTL_DNS_CACHE, + enable_cleanup_closed=AIOHTTP_NEEDS_CLEANUP_CLOSED, + limit=AIOHTTP_CONNECTOR_LIMIT if AIOHTTP_CONNECTOR_LIMIT > 0 else 100, + limit_per_host=max(0, AIOHTTP_CONNECTOR_LIMIT_PER_HOST), + socket_factory=_build_aiohttp_keepalive_socket_factory(), + ) session = ClientSession(connector=connector) verbose_proxy_logger.info( @@ -911,7 +908,9 @@ async def proxy_startup_event(app: FastAPI): ) _module_path, _func_name = _hook_spec.rsplit(":", 1) _module = importlib.import_module(_module_path) - _hook_fn = getattr(_module, _func_name) + _hook_fn: object = getattr(_module, _func_name) + if not callable(_hook_fn): + raise TypeError(f"Hook '{_hook_spec}' is not callable") if inspect.iscoroutinefunction(_hook_fn): await _hook_fn() else: @@ -959,9 +958,9 @@ async def proxy_startup_event(app: FastAPI): await initialize(**worker_config) else: # if not, assume it's a json string - worker_config = json.loads(worker_config) - if isinstance(worker_config, dict): - await initialize(**worker_config) + parsed_worker_config = TypeAdapter(JsonValue).validate_python(json.loads(worker_config)) + if isinstance(parsed_worker_config, dict): + await initialize(**parsed_worker_config) # check if DATABASE_URL in environment - load from there if prisma_client is None: @@ -1162,7 +1161,7 @@ async def proxy_startup_event(app: FastAPI): await proxy_shutdown_event() # type: ignore[reportGeneralTypeIssues] -def _generate_stable_operation_id(route: Any) -> str: +def _generate_stable_operation_id(route: APIRoute) -> str: operation_id = re.sub(r"\W", "_", f"{route.name}{route.path_format}") route_methods = sorted(route.methods or []) if len(route_methods) == 1: @@ -1199,22 +1198,26 @@ def _strip_operation_id_method_suffix(operation_id: str) -> str: def ensure_unique_openapi_operation_ids( - openapi_schema: dict[str, Any], + openapi_schema: dict[str, JsonValue], # mutable-ok: parameter mirrors the caller's OpenAPI schema dict payload reserved_operation_ids: set[str] | None = None, -) -> dict[str, Any]: - operation_entries = [] +) -> dict[str, JsonValue]: # mutable-ok: mirrors the OpenAPI schema dict payload this function returns + operation_entries: list[ + tuple[str, dict[str, JsonValue], str] + ] = [] # mutable-ok: accumulates operation entries in the loop below operation_id_counts: dict[str, int] = {} - for path_item in openapi_schema.get("paths", {}).values(): - if not isinstance(path_item, dict): - continue - for method, operation in path_item.items(): - if method not in _OPENAPI_HTTP_METHODS or not isinstance(operation, dict): + paths = openapi_schema.get("paths", {}) + if isinstance(paths, dict): + for path_item in paths.values(): + if not isinstance(path_item, dict): continue - operation_id = operation.get("operationId") - if not isinstance(operation_id, str): - continue - operation_entries.append((method, operation, operation_id)) - operation_id_counts[operation_id] = operation_id_counts.get(operation_id, 0) + 1 + for method, operation in path_item.items(): + if method not in _OPENAPI_HTTP_METHODS or not isinstance(operation, dict): + continue + operation_id = operation.get("operationId") + if not isinstance(operation_id, str): + continue + operation_entries.append((method, operation, operation_id)) + operation_id_counts[operation_id] = operation_id_counts.get(operation_id, 0) + 1 used_operation_ids = set(reserved_operation_ids or set()) seen_operation_ids: set[str] = set() @@ -1421,7 +1424,7 @@ async def openai_exception_handler(request: Request, exc: ProxyException): def _close_dangling_otel_server_span(request: Request, status_code: int, exc: Exception | None = None) -> None: - parent_otel_span = getattr(request.state, "parent_otel_span", None) + parent_otel_span: _OtelSpanType | None = getattr(request.state, "parent_otel_span", None) if parent_otel_span is None: return if open_telemetry_logger is None: @@ -1468,13 +1471,16 @@ async def management_problem_exception_handler(request: Request, exc: Management async def otel_request_validation_exception_handler(request: Request, exc: RequestValidationError): if request.url.path.startswith(MANAGEMENT_V1_PREFIX): _close_dangling_otel_server_span(request, 400, exc=exc) + validation_errors: list[ErrorDetails] = list( + exc.errors() + ) # mutable-ok: collects validation errors in the loop below return problem_response( ProblemDetail( type=f"{PROBLEM_TYPE_BASE}invalid-query-parameter", title="Invalid query parameter", status=400, detail="; ".join( - f"{'.'.join(str(part) for part in error['loc'][1:])}: {error['msg']}" for error in exc.errors() + f"{'.'.join(str(part) for part in error['loc'][1:])}: {error['msg']}" for error in validation_errors ) or "The request query parameters are invalid.", ) @@ -2033,7 +2039,9 @@ use_queue = False health_check_interval = None health_check_concurrency = None health_check_details = None -health_check_results: dict[str, int | list[dict[str, Any]]] = {} +health_check_results: dict[ + str, int | list[dict[str, object]] +] = {} # mutable-ok: populated incrementally per model in the health check loop below background_health_check_loop_active = False background_health_check_cycle_seq = 0 queue: list = [] @@ -2340,7 +2348,7 @@ async def _read_spend_counter_estimate(counter_key: str, fallback_spend: float) redis_clean_miss = False if spend_counter_cache.redis_cache is not None: try: - val = await spend_counter_cache.redis_cache.async_get_cache(key=counter_key) + val: str | float | int | None = await spend_counter_cache.redis_cache.async_get_cache(key=counter_key) if val is not None: return float(val), True redis_clean_miss = True @@ -2706,7 +2714,7 @@ async def _get_source_cache_base_spend( ) -> float: source_cache_keys = [source_cache_key] if isinstance(source_cache_key, str) else source_cache_key for cache_key in source_cache_keys: - source = await user_api_key_cache.async_get_cache(key=cache_key) + source = TypeAdapter(object).validate_python(await user_api_key_cache.async_get_cache(key=cache_key)) if source is None: continue if isinstance(source, dict): @@ -2745,7 +2753,7 @@ async def _ensure_window_spend_counter_initialized( async def _is_spend_counter_cache_warm(counter_key: str) -> bool: if spend_counter_cache.redis_cache is not None: try: - current_value = await spend_counter_cache.redis_cache.async_get_cache( + current_value: str | float | int | None = await spend_counter_cache.redis_cache.async_get_cache( key=counter_key, ) if current_value is None: @@ -2817,7 +2825,7 @@ async def update_cache( Put any alerting logic in here. """ - values_to_update_in_cache: list[tuple[Any, Any]] = [] + values_to_update_in_cache: list[tuple[str, object]] = [] # mutable-ok: accumulates cache updates in the loop below ### UPDATE KEY SPEND ### async def _update_key_cache(token: str, response_cost: float): @@ -2883,7 +2891,7 @@ async def update_cache( # Fetch the existing cost for the given user if _id is None: continue - cached_user = await user_api_key_cache.async_get_cache(key=_id) + cached_user = TypeAdapter(object).validate_python(await user_api_key_cache.async_get_cache(key=_id)) if cached_user is None: # do nothing if there is no cache value return @@ -2906,7 +2914,10 @@ async def update_cache( ) ) ## UPDATE GLOBAL PROXY ## - global_proxy_spend = await user_api_key_cache.async_get_cache(key=GLOBAL_PROXY_SPEND_CACHE_KEY) + global_proxy_spend = TypeAdapter(float | int | None).validate_python( + await user_api_key_cache.async_get_cache(key=GLOBAL_PROXY_SPEND_CACHE_KEY), + strict=True, + ) if global_proxy_spend is None: # do nothing if not in cache return @@ -2932,7 +2943,7 @@ async def update_cache( _id = f"end_user_id:{end_user_id}" try: # Fetch the existing cost for the given user - cached_end_user = await user_api_key_cache.async_get_cache(key=_id) + cached_end_user = TypeAdapter(object).validate_python(await user_api_key_cache.async_get_cache(key=_id)) if cached_end_user is None: # if user does not exist in LiteLLM_UserTable, create a new user # do nothing if end-user not in api key cache @@ -2973,7 +2984,7 @@ async def update_cache( _id = f"team_id:{team_id}" try: - cached_team = await user_api_key_cache.async_get_cache(key=_id) + cached_team = TypeAdapter(object).validate_python(await user_api_key_cache.async_get_cache(key=_id)) if cached_team is None: # do nothing if team not in api key cache return @@ -3023,7 +3034,9 @@ async def update_cache( cache_key = f"tag:{tag_name}" # Fetch the existing tag object from cache - cached_tag = await user_api_key_cache.async_get_cache(key=cache_key) + cached_tag = TypeAdapter(object).validate_python( + await user_api_key_cache.async_get_cache(key=cache_key) + ) if cached_tag is None: # do nothing if tag not in api key cache continue @@ -3527,11 +3540,13 @@ _DB_OVERLAY_REMOTE_MODULE_LIST_FIELDS: dict[str, tuple[str, ...]] = { } -def _is_remote_module_url(value: Any) -> bool: +def _is_remote_module_url(value: object) -> bool: return isinstance(value, str) and (value.startswith("s3://") or value.startswith("gcs://")) -def _scrub_guardrail_inner(inner: dict[str, Any]) -> None: +def _scrub_guardrail_inner( + inner: dict[str, JsonValue], +) -> None: # mutable-ok: parameter mirrors the caller's guardrail config dict payload """Strip remote-URL entries from a guardrail's ``callbacks`` list and ``guardrail`` (v2 module-path) field. Mutates in place.""" cbs = inner.get("callbacks") @@ -3551,7 +3566,7 @@ def _scrub_guardrail_inner(inner: dict[str, Any]) -> None: inner["guardrail"] = None -def _scrub_db_overlay_remote_module_loads(section: str, db_value: Any) -> Any: +def _scrub_db_overlay_remote_module_loads(section: str, db_value: JsonValue) -> JsonValue: """Strip ``s3://`` / ``gcs://`` entries from the DB-overlay value for fields whose contents reach ``get_instance_fn``. The same scheme is allowed from a YAML config (the documented operator flow) but a @@ -3699,7 +3714,7 @@ def _build_redis_usage_cache(redis_params: Mapping[str, object]) -> RedisCache: if startup_nodes is None: env_cluster_nodes = get_secret_str("REDIS_CLUSTER_NODES") if env_cluster_nodes is not None: - startup_nodes = json.loads(env_cluster_nodes) + startup_nodes = TypeAdapter(object).validate_python(json.loads(env_cluster_nodes)) non_node_params = {key: value for key, value in redis_params.items() if key != "startup_nodes"} if startup_nodes: return RedisClusterCache(startup_nodes=startup_nodes, **non_node_params) @@ -15236,7 +15251,7 @@ def _general_settings_ui_litellm_default( return False if spec["type"] == "Boolean" else None -def _validate_general_settings_ui_litellm_value(field_name: str, value: Any) -> GeneralSettingsUILiteLLMValue: +def _validate_general_settings_ui_litellm_value(field_name: str, value: JsonValue) -> GeneralSettingsUILiteLLMValue: spec = _GENERAL_SETTINGS_UI_LITELLM_FIELDS[field_name] field_type = spec["type"] if value is None or value == "": @@ -15276,7 +15291,7 @@ def _validate_general_settings_ui_litellm_value(field_name: str, value: Any) -> async def _persist_general_settings_ui_litellm_field( - field_name: str, value: Any, user_api_key_dict: UserAPIKeyAuth + field_name: str, value: JsonValue, user_api_key_dict: UserAPIKeyAuth ) -> dict: validated = _validate_general_settings_ui_litellm_value(field_name, value) config = await proxy_config.get_config() diff --git a/litellm/proxy/utils.py b/litellm/proxy/utils.py index 51c9b300b6e..69b3c7f55c2 100644 --- a/litellm/proxy/utils.py +++ b/litellm/proxy/utils.py @@ -21,6 +21,7 @@ from typing import ( ClassVar, Literal, Optional, + TypedDict, Union, cast, overload, @@ -160,6 +161,7 @@ from litellm.repositories.verification_token_repository import ( ) from litellm.secret_managers.main import str_to_bool from litellm.types.integrations.slack_alerting import DEFAULT_ALERT_TYPES +from litellm.types.llms.base import HiddenParams from litellm.types.mcp import ( MCPDuringCallResponseObject, MCPPreCallRequestObject, @@ -345,7 +347,7 @@ def _accepts_litellm_call_info(cb: CustomLogger) -> bool: return _CALLBACK_ACCEPTS_CALL_INFO[key] -def _enrich_http_exception_with_guardrail_context(exc: BaseException, callback: Any) -> None: +def _enrich_http_exception_with_guardrail_context(exc: BaseException, callback: CustomLogger) -> None: """ If `exc` is an HTTPException with a dict `detail`, mutate it in place to add `guardrail_name` and `guardrail_mode` taken from the callback instance. @@ -377,6 +379,15 @@ def _exception_changes_request_flow(exc: BaseException) -> bool: return isinstance(exc, (SensitiveDataRouteException, ModifyResponseException)) +class _MCPPreCallHookResult(TypedDict): + should_proceed: bool + modified_arguments: ( + dict[str, object] | None + ) # mutable-ok: parameter mirrors the caller's request-arguments dict payload + error_message: str | None + hidden_params: HiddenParams + + @dataclass(frozen=True) class _CallbackCapabilities: """Cached per-hook capability flags derived from ``litellm.callbacks``. @@ -394,11 +405,11 @@ class _CallbackCapabilities: # Tuple[(resolved_callback, "override" | "apply_guardrail"), ...] # Ordered the same as ``litellm.callbacks``; used to build the streaming # iterator chain without re-scanning per request. - iterator_overrides: tuple[tuple[Any, str], ...] = field(default_factory=tuple) + iterator_overrides: tuple[tuple[CustomLogger, str], ...] = field(default_factory=tuple) # Resolved CustomLogger callbacks in original order. Pre-resolving once # avoids the per-request ``get_custom_logger_compatible_class`` walk for # every string entry in ``litellm.callbacks``. - resolved_callbacks: tuple[Any, ...] = field(default_factory=tuple) + resolved_callbacks: tuple[CustomLogger, ...] = field(default_factory=tuple) class ProxyLogging: @@ -676,12 +687,10 @@ class ProxyLogging: return synthetic_data - def _convert_llm_result_to_mcp_response(self, llm_result, request_obj) -> Any | None: + def _convert_llm_result_to_mcp_response(self, llm_result, request_obj) -> MCPPreCallResponseObject | None: """ Convert LLM guardrail result back to MCP response format. """ - from litellm.types.mcp import MCPPreCallResponseObject - # If result is an exception, it means the guardrail blocked the request if isinstance(llm_result, Exception): return MCPPreCallResponseObject( @@ -802,7 +811,7 @@ class ProxyLogging: verbose_proxy_logger.error("Error in manual argument parsing: %s", e) return None - def _convert_llm_result_to_mcp_during_response(self, llm_result, request_obj) -> Any | None: + def _convert_llm_result_to_mcp_during_response(self, llm_result, request_obj) -> MCPDuringCallResponseObject | None: """ Convert LLM guardrail result back to MCP during call response format. """ @@ -848,7 +857,7 @@ class ProxyLogging: self, response: MCPPreCallResponseObject, original_request: MCPPreCallRequestObject, - ) -> dict[str, Any]: + ) -> _MCPPreCallHookResult: """ Parse the response from the pre_mcp_tool_call_hook @@ -856,7 +865,7 @@ class ProxyLogging: 2. Apply any argument modifications 3. Handle validation errors """ - result = { + result: _MCPPreCallHookResult = { "should_proceed": response.should_proceed, "modified_arguments": response.modified_arguments or original_request.arguments, "error_message": response.error_message, @@ -868,9 +877,6 @@ class ProxyLogging: """ Helper function to create MCPPreCallRequestObject from kwargs for standard pre_call_hook. """ - from litellm.types.llms.base import HiddenParams - from litellm.types.mcp import MCPPreCallRequestObject - user_api_key_auth_dict = self._convert_user_api_key_auth_to_dict(kwargs.get("user_api_key_auth")) return MCPPreCallRequestObject( @@ -1031,7 +1037,7 @@ class ProxyLogging: selected_guardrail = llm_router.get_available_guardrail(guardrail_name=guardrail_name) callback = selected_guardrail.get("callback") - if callback is None: + if not isinstance(callback, CustomGuardrail): raise ValueError(f"No callback found for guardrail: {guardrail_name}") return await self._execute_guardrail_hook( @@ -1142,8 +1148,8 @@ class ProxyLogging: self, data: dict, litellm_logging_obj: Any, - prompt_id: Any, - prompt_version: Any, + prompt_id: str, + prompt_version: int | None, call_type: CallTypesLiteral, ) -> None: """Process prompt template if applicable.""" @@ -1438,9 +1444,7 @@ class ProxyLogging: data = result elif ( - _callback is not None - and isinstance(_callback, CustomLogger) - and "async_pre_call_hook" in vars(_callback.__class__) + "async_pre_call_hook" in vars(_callback.__class__) and _callback.__class__.async_pre_call_hook != CustomLogger.async_pre_call_hook ): if call_type == "call_mcp_tool" and user_api_key_dict is None: @@ -1614,7 +1618,7 @@ class ProxyLogging: break @staticmethod - async def _run_guardrail_with_metrics(callback: Any, coro: Awaitable[Any], hook_type: str) -> Any: + async def _run_guardrail_with_metrics(callback: CustomGuardrail, coro: Awaitable[Any], hook_type: str) -> Any: """ Await `coro`, recording its latency and status to the `litellm_guardrail_latency_seconds` metric under `hook_type`, and @@ -1646,7 +1650,7 @@ class ProxyLogging: @staticmethod async def _wrap_streaming_iterator_with_enrichment( - callback: Any, gen: AsyncGenerator[Any, None] + callback: CustomLogger, gen: AsyncGenerator[Any, None] ) -> AsyncGenerator[Any, None]: """ Yield from `gen`; if iteration raises an HTTPException with dict detail, @@ -1691,12 +1695,14 @@ class ProxyLogging: has_streaming_chunk_override = False has_guardrail = False has_pre_call_override = False - iterator_overrides: list[tuple[Any, str]] = [] # (callback, kind) - resolved_callbacks: list[Any] = [] + iterator_overrides: list[ + tuple[CustomLogger, str] + ] = [] # mutable-ok: accumulates (callback, kind) overrides in the loop below + resolved_callbacks: list[CustomLogger] = [] # mutable-ok: accumulates resolved callbacks in the loop below for callback in callbacks: if isinstance(callback, str): - resolved: Any = litellm.litellm_core_utils.litellm_logging.get_custom_logger_compatible_class( + resolved = litellm.litellm_core_utils.litellm_logging.get_custom_logger_compatible_class( cast(_custom_logger_compatible_callbacks_literal, callback) ) else: @@ -2096,7 +2102,7 @@ class ProxyLogging: """ litellm_debug_info = getattr(original_exception, "litellm_debug_info", None) exception_str = str(original_exception) - if litellm_debug_info is not None: + if isinstance(litellm_debug_info, str): exception_str += litellm_debug_info asyncio.create_task( @@ -2765,7 +2771,7 @@ class ProxyLogging: ), ) else: - # kind == "apply_guardrail": route through unified_guardrail + assert isinstance(resolved_callback, CustomGuardrail) current_response = self._wrap_streaming_iterator_with_enrichment( resolved_callback, unified_guardrail.async_post_call_streaming_iterator_hook( @@ -6557,7 +6563,7 @@ def model_dump_with_preserved_fields( model_dump(exclude_none=True) strips them. Args: - obj: The Pydantic BaseModel instance to serialize + obj: The ModelResponse / ModelResponseStream instance to serialize preserve_fields: Deprecated, kept for backward compatibility. exclude_unset: Whether to exclude fields that were not explicitly set diff --git a/litellm/responses/streaming_iterator.py b/litellm/responses/streaming_iterator.py index 3bcc19822a6..a98967bf42b 100644 --- a/litellm/responses/streaming_iterator.py +++ b/litellm/responses/streaming_iterator.py @@ -9,7 +9,7 @@ from collections.abc import Mapping from datetime import datetime from functools import lru_cache from types import MappingProxyType -from typing import Any, Literal +from typing import TYPE_CHECKING, Any, Literal, Protocol, runtime_checkable import httpx from openai._streaming import SSEDecoder @@ -34,6 +34,40 @@ from litellm.types.llms.openai import ResponsesAPIStreamEvents from litellm.types.utils import CallTypes from litellm.utils import async_post_call_success_deployment_hook +if TYPE_CHECKING: + from fastapi import WebSocket + + from litellm.types.llms.openai import ( + ContentPartDoneEvent, + ContentPartDonePartOutputText, + ContentPartDonePartReasoningText, + ContentPartDonePartRefusal, + ResponsesAPIResponse, + ResponsesAPIStreamingResponse, + ) + + +@runtime_checkable +class _ModelDumpable(Protocol): + def model_dump( + self, *, exclude_none: bool = ... + ) -> dict[ + str, object + ]: ... # mutable-ok: Protocol mirrors Pydantic's model_dump(), which returns a plain mutable dict + + +@runtime_checkable +class _ModelDumpJsonable(Protocol): + def model_dump_json(self, *, exclude_none: bool = ...) -> str: ... + + +class _BackendWebSocketLike(Protocol): + async def recv(self, decode: bool = ...) -> str | bytes: ... + + async def send(self, message: str) -> None: ... + + async def close(self) -> None: ... + @lru_cache(maxsize=1) def _get_openai_response_types(): @@ -42,7 +76,7 @@ def _get_openai_response_types(): return openai_types -def _log_background_task_failure(task: asyncio.Task[Any], *, task_name: str) -> None: +def _log_background_task_failure(task: asyncio.Task[object], *, task_name: str) -> None: if task.cancelled(): return exception = task.exception() @@ -131,7 +165,7 @@ class BaseResponsesAPIStreamingIterator: self.logging_obj = logging_obj self.finished = False self.responses_api_provider_config = responses_api_provider_config - self.completed_response: Any | None = None + self.completed_response: ResponsesAPIStreamingResponse | None = None self.start_time = getattr(logging_obj, "start_time", datetime.now()) self._failure_handled = False # Track if failure handler has been called self._yielded_first_chunk = False @@ -176,7 +210,7 @@ class BaseResponsesAPIStreamingIterator: llm_provider=self.custom_llm_provider or "", ) - def _process_chunk(self, chunk) -> Any | None: + def _process_chunk(self, chunk: str | None) -> ResponsesAPIStreamingResponse | None: """Process a single chunk of data from the stream""" if not chunk: return None @@ -196,7 +230,7 @@ class BaseResponsesAPIStreamingIterator: try: # Parse the JSON chunk - parsed_chunk = json.loads(chunk) + parsed_chunk: object = json.loads(chunk) # Format as ResponsesAPIStreamingResponse if isinstance(parsed_chunk, dict): @@ -401,7 +435,7 @@ class BaseResponsesAPIStreamingIterator: ) self._handle_failure(exception) - def _record_failed_response_usage(self, response_obj: Any | None) -> None: + def _record_failed_response_usage(self, response_obj: ResponsesAPIResponse | None) -> None: if response_obj is None or self.logging_obj is None: return usage_obj = getattr(response_obj, "usage", None) @@ -451,7 +485,7 @@ class BaseResponsesAPIStreamingIterator: is_pre_first_chunk=not self._yielded_first_chunk, ) - def _get_completed_response_object(self) -> Any | None: + def _get_completed_response_object(self) -> ResponsesAPIResponse | None: openai_types = _get_openai_response_types() completed_response = self.completed_response if isinstance(completed_response, openai_types.ResponsesAPIResponse): @@ -577,7 +611,7 @@ class BaseResponsesAPIStreamingIterator: if self.completed_response is None: return - request_payload: dict[str, Any] = {} + request_payload: dict[str, object] = {} # mutable-ok: populated incrementally from the request payload below if isinstance(self.request_data, dict): request_payload.update(self.request_data) try: @@ -707,7 +741,7 @@ class ResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator): def __aiter__(self): return self - async def __anext__(self) -> Any: + async def __anext__(self) -> ResponsesAPIStreamingResponse: try: self._check_max_streaming_duration() while True: @@ -880,7 +914,7 @@ class MockResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator): def _set_events_from_response( self, - transformed: Any, + transformed: ResponsesAPIResponse, logging_obj: LiteLLMLoggingObj, ) -> None: self._events = _build_synthetic_response_events( @@ -894,7 +928,7 @@ class MockResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator): def __aiter__(self): return self - async def __anext__(self) -> Any: + async def __anext__(self) -> ResponsesAPIStreamingResponse: if self._idx >= len(self._events): raise StopAsyncIteration evt = self._events[self._idx] @@ -908,7 +942,7 @@ class MockResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator): def __iter__(self): return self - def __next__(self) -> Any: + def __next__(self) -> ResponsesAPIStreamingResponse: if self._idx >= len(self._events): raise StopIteration evt = self._events[self._idx] @@ -923,7 +957,7 @@ class MockResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator): class CachedResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator): def __init__( self, - response: Any, + response: ResponsesAPIResponse, logging_obj: LiteLLMLoggingObj, request_data: dict[str, Any] | None = None, call_type: str | None = None, @@ -941,13 +975,15 @@ class CachedResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator): ) self._completed_response_cache_hit = True self._persist_completed_response_before_logging = False - self._events: list[Any] = [] + self._events: list[ + ResponsesAPIStreamingResponse + ] = [] # mutable-ok: accumulates one event per stream update over the object's lifetime self._idx = 0 self._set_events_from_response(transformed=response, logging_obj=logging_obj) def _set_events_from_response( self, - transformed: Any, + transformed: ResponsesAPIResponse, logging_obj: LiteLLMLoggingObj, ) -> None: self._events = _build_synthetic_response_events( @@ -961,7 +997,7 @@ class CachedResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator): def __aiter__(self): return self - async def __anext__(self) -> Any: + async def __anext__(self) -> ResponsesAPIStreamingResponse: if self._idx >= len(self._events): raise StopAsyncIteration evt = self._events[self._idx] @@ -975,7 +1011,7 @@ class CachedResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator): def __iter__(self): return self - def __next__(self) -> Any: + def __next__(self) -> ResponsesAPIStreamingResponse: if self._idx >= len(self._events): raise StopIteration evt = self._events[self._idx] @@ -987,8 +1023,10 @@ class CachedResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator): return evt -def _dump_response_object(obj: Any) -> dict[str, Any]: - if hasattr(obj, "model_dump"): +def _dump_response_object( + obj: object, +) -> dict[str, Any]: # mutable-ok: returns a plain mutable dict to match model_dump()'s contract + if isinstance(obj, _ModelDumpable): return obj.model_dump() if isinstance(obj, dict): return obj @@ -1000,8 +1038,8 @@ def _build_response_status_event( "response.created", "response.in_progress", ], - transformed: Any, -) -> Any: + transformed: ResponsesAPIResponse, +) -> ResponsesAPIStreamingResponse: openai_types = _get_openai_response_types() in_progress_response = transformed.model_copy( deep=True, @@ -1018,10 +1056,10 @@ def _build_content_part_done_event( output_index: int, content_index: int, part_payload: dict[str, Any], -) -> Any | None: +) -> ContentPartDoneEvent | None: openai_types = _get_openai_response_types() part_type = part_payload.get("type") - part: Any + part: ContentPartDonePartOutputText | ContentPartDonePartRefusal | ContentPartDonePartReasoningText if part_type == "output_text": annotations = [ openai_types.BaseLiteLLMOpenAIResponseObject(**annotation) @@ -1057,7 +1095,7 @@ def _build_content_part_done_event( def _add_text_like_part_events( *, - events: list[Any], + events: list[ResponsesAPIStreamingResponse], # mutable-ok: parameter mirrors the caller's mutable event list item_id: str, output_index: int, content_index: int, @@ -1123,13 +1161,15 @@ def _add_text_like_part_events( def _build_synthetic_response_events( *, - transformed: Any, + transformed: ResponsesAPIResponse, logging_obj: LiteLLMLoggingObj, chunk_size: int, -) -> list[Any]: +) -> list[ + ResponsesAPIStreamingResponse +]: # mutable-ok: returns a plain mutable event list to match the caller's accumulator openai_types = _get_openai_response_types() if litellm.include_cost_in_streaming_usage and logging_obj is not None: - usage_obj: Any | None = getattr(transformed, "usage", None) + usage_obj = transformed.usage if usage_obj is not None: try: cost: float | None = logging_obj._response_cost_calculator(result=transformed) @@ -1138,13 +1178,15 @@ def _build_synthetic_response_events( except Exception: pass - events: list[Any] = [ + events: list[ + ResponsesAPIStreamingResponse + ] = [ # mutable-ok: built once as a mutable event list for the caller to extend _build_response_status_event(openai_types.ResponsesAPIStreamEvents.RESPONSE_CREATED, transformed), _build_response_status_event(openai_types.ResponsesAPIStreamEvents.RESPONSE_IN_PROGRESS, transformed), ] sequence_number = 0 - for output_index, output_item in enumerate(getattr(transformed, "output", []) or []): + for output_index, output_item in enumerate(transformed.output or []): output_item_payload = _dump_response_object(output_item) item_id = str(output_item_payload.get("id") or transformed.id) item_type = output_item_payload.get("type") @@ -1292,8 +1334,8 @@ class ResponsesWebSocketStreaming: def __init__( self, - websocket: Any, - backend_ws: Any, + websocket: WebSocket, + backend_ws: _BackendWebSocketLike, logging_obj: LiteLLMLoggingObj, user_api_key_dict: Any | None = None, request_data: dict | None = None, @@ -1319,12 +1361,16 @@ class ResponsesWebSocketStreaming: def _should_store_event(self, event_obj: dict) -> bool: return event_obj.get("type") in RESPONSES_WS_LOGGED_EVENT_TYPES - def _store_event(self, event: Any) -> None: + def _store_event( + self, event: str | bytes | dict[str, object] + ) -> None: # mutable-ok: parameter mirrors the caller's event payload, which may be a mutable dict if isinstance(event, bytes): event = event.decode("utf-8") if isinstance(event, str): try: - event_obj = json.loads(event) + event_obj: dict[str, object] = json.loads( + event + ) # mutable-ok: parsed once via json.loads(), which returns a plain mutable dict except (json.JSONDecodeError, TypeError): return else: @@ -1333,15 +1379,17 @@ class ResponsesWebSocketStreaming: if self._should_store_event(event_obj): self.messages.append(event_obj) - def _collect_input_from_client_event(self, message: Any) -> None: + def _collect_input_from_client_event( + self, message: str | dict[str, object] + ) -> None: # mutable-ok: parameter mirrors the caller's message payload, which may be a mutable dict """Extract user input content from response.create for logging.""" try: if isinstance(message, str): - msg_obj = json.loads(message) - elif isinstance(message, dict): - msg_obj = message + msg_obj: dict[str, object] = json.loads( + message + ) # mutable-ok: parsed once via json.loads(), which returns a plain mutable dict else: - return + msg_obj = message if msg_obj.get("type") != "response.create": return @@ -1368,7 +1416,9 @@ class ResponsesWebSocketStreaming: except (json.JSONDecodeError, AttributeError, TypeError): pass - def _store_input(self, message: Any) -> None: + def _store_input( + self, message: str | dict[str, object] + ) -> None: # mutable-ok: parameter mirrors the caller's message payload, which may be a mutable dict self._collect_input_from_client_event(message) if self.logging_obj: self.logging_obj.pre_call(input=message, api_key="") @@ -1407,7 +1457,10 @@ class ResponsesWebSocketStreaming: # masked response.completed. if self.output_guardrail_callbacks: try: - _evt_type = json.loads(response_str).get("type") + _evt_parsed: dict[str, object] = json.loads( + response_str + ) # mutable-ok: parsed once via json.loads(), which returns a plain mutable dict + _evt_type = _evt_parsed.get("type") except (json.JSONDecodeError, TypeError): _evt_type = None if _evt_type in self._DELTA_EVENT_TYPES or _evt_type in self._OUTPUT_DONE_EVENT_TYPES: @@ -1793,7 +1846,7 @@ class ManagedResponsesWebSocketHandler: def __init__( self, - websocket: Any, + websocket: WebSocket, model: str, logging_obj: LiteLLMLoggingObj, user_api_key_dict: Any | None = None, @@ -1832,12 +1885,12 @@ class ManagedResponsesWebSocketHandler: # ------------------------------------------------------------------ @staticmethod - def _serialize_chunk(chunk: Any) -> str | None: + def _serialize_chunk(chunk: object) -> str | None: """Serialize a streaming chunk to a JSON string for WebSocket transmission.""" try: - if hasattr(chunk, "model_dump_json"): + if isinstance(chunk, _ModelDumpJsonable): return chunk.model_dump_json(exclude_none=True) - if hasattr(chunk, "model_dump"): + if isinstance(chunk, _ModelDumpable): return json.dumps(chunk.model_dump(exclude_none=True), default=str) if isinstance(chunk, dict): return json.dumps(chunk, default=str) @@ -1925,7 +1978,9 @@ class ManagedResponsesWebSocketHandler: return messages @staticmethod - def _input_to_messages(input_val: Any) -> list[dict[str, Any]]: + def _input_to_messages( + input_val: object, + ) -> list[dict[str, Any]]: # mutable-ok: returns a plain mutable list to match callers that append further messages """ Normalise the ``input`` field of a ``response.create`` event to a list of Responses API message dicts. diff --git a/litellm/router.py b/litellm/router.py index 7d6499cf7d2..676fa932eec 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -249,7 +249,7 @@ if TYPE_CHECKING: ResponsesAPIResponse, ) - Span = Union[_Span, Any] + Span = _Span else: Span = Any AutoRouter = Any @@ -463,6 +463,13 @@ class Router: ``` """ + self._override_selectors: dict[ + str, Any + ] = {} # mutable-ok: populated on demand and mutated by the selector-registration helpers below + self._group_selectors: dict[ + str, dict[str, Any] + ] = {} # mutable-ok: populated on demand and mutated by the selector-registration helpers below + self.set_verbose = set_verbose self.ignore_invalid_deployments = ignore_invalid_deployments self.debug_level = debug_level @@ -679,7 +686,7 @@ class Router: routing_strategy_args=routing_strategy_args, ) self._init_routing_groups(self._routing_groups_input) - self._override_selectors: dict[str, Any] = {} + self._override_selectors = {} self._override_selectors_lock = threading.Lock() self.access_groups = None ## USAGE TRACKING ## @@ -949,7 +956,7 @@ class Router: self._unregister_router_selectors( [getattr(self, attr, None) for attr in self._DEFAULT_SELECTOR_ATTR_BY_STRATEGY.values()] - + list(getattr(self, "_override_selectors", {}).values()) + + list(self._override_selectors.values()) ) self._override_selectors = {} @@ -985,12 +992,12 @@ class Router: attributes set up in `routing_strategy_init`. """ self._unregister_router_selectors( - [sel for selectors in getattr(self, "_group_selectors", {}).values() for sel in selectors.values()] + [sel for selectors in self._group_selectors.values() for sel in selectors.values()] ) self._routing_groups: dict[str, RoutingGroup] = {} self._model_to_group: dict[str, str] = {} - self._group_selectors: dict[str, dict[str, Any]] = {} + self._group_selectors = {} if not groups_input: return @@ -1910,7 +1917,7 @@ class Router: async def _run_silent_completion(): await self.acompletion( model=silent_model, - messages=cast(list[AllMessageValues], messages), + messages=messages, **silent_kwargs, ) # Drain any fire-and-forget tasks (e.g. alerting hooks) @@ -2753,7 +2760,7 @@ class Router: # Trigger the silent request await self.acompletion( model=silent_model, - messages=cast(list[AllMessageValues], messages), + messages=messages, **silent_kwargs, ) except Exception as e: @@ -3276,7 +3283,11 @@ class Router: ) ) responses = await asyncio.gather(*_tasks) - final_responses: list[list[Any]] = [[] for _ in range(len(messages))] + final_responses: list[ + list[ModelResponse | CustomStreamWrapper | Exception] + ] = [ # mutable-ok: populated incrementally by the thread pool loop below + [] for _ in range(len(messages)) + ] for response in responses: if isinstance(response, tuple): final_responses[response[1]].append(response[0]) @@ -4583,7 +4594,7 @@ class Router: # fallback to the original reference for any non-picklable value. # The original_generic_function is preserved so the per-attempt # helper knows which underlying API to call on fallback. - fallback_kwargs: dict[str, Any] = kwargs.copy() + fallback_kwargs = kwargs.copy() if isinstance(fallback_kwargs.get("litellm_metadata"), dict): fallback_kwargs["litellm_metadata"] = safe_deep_copy(fallback_kwargs["litellm_metadata"]) if isinstance(fallback_kwargs.get("metadata"), dict): @@ -7079,7 +7090,7 @@ class Router: except Exception as e: raise e - async def async_deployment_callback_on_failure(self, kwargs, completion_response: Any | None, start_time, end_time): + async def async_deployment_callback_on_failure(self, kwargs, completion_response, start_time, end_time): """ Update RPM usage for a deployment """ @@ -10999,7 +11010,7 @@ class Router: self, model: str, request_kwargs: dict, - messages: list[dict[str, Any]] | None, + messages: list[dict[str, str]] | None, # mutable-ok: parameter mirrors the caller's message-dict payload ) -> RoutingContext: """ Build a RoutingContext for `model`, run it through `self.routing_plugins` @@ -11098,7 +11109,7 @@ class Router: self, model: str, request_kwargs: dict, - messages: list[dict[str, Any]] | None = None, + messages: list[dict[str, str]] | None = None, # mutable-ok: parameter mirrors the caller's message-dict payload input: str | list | None = None, specific_deployment: bool | None = False, ) -> PreRoutingHookResponse | None: @@ -11181,7 +11192,7 @@ class Router: @staticmethod def _redact_prompt_text_if_needed( - request_kwargs: Mapping[str, Any], + request_kwargs: Mapping[str, object], routing_decision: StandardLoggingRoutingDecision, ) -> StandardLoggingRoutingDecision: """Drop verbatim prompt text from the record when message logging is redacted. diff --git a/ruff-strict-budget.json b/ruff-strict-budget.json index d27b168d6ca..3a0e4c861b2 100644 --- a/ruff-strict-budget.json +++ b/ruff-strict-budget.json @@ -9,10 +9,10 @@ "limit": 831 }, "ANN201": { - "limit": 2137 + "limit": 2135 }, "ANN202": { - "limit": 941 + "limit": 939 }, "ANN204": { "limit": 724 @@ -24,7 +24,7 @@ "limit": 130 }, "ANN401": { - "limit": 1848 + "limit": 1756 }, "ASYNC230": { "limit": 14 @@ -135,7 +135,7 @@ "limit": 30 }, "PERF401": { - "limit": 142 + "limit": 138 }, "PERF402": { "limit": 9 @@ -222,7 +222,7 @@ "limit": 0 }, "RET504": { - "limit": 702 + "limit": 700 }, "RUF010": { "limit": 0 @@ -321,7 +321,7 @@ "limit": 121 }, "TRY300": { - "limit": 879 + "limit": 877 }, "UP006": { "limit": 0 diff --git a/type-discipline-budget.json b/type-discipline-budget.json index 289c0a0afd6..023a4f4d658 100644 --- a/type-discipline-budget.json +++ b/type-discipline-budget.json @@ -1,9 +1,9 @@ { "LIT001": { - "limit": 23349 + "limit": 23301 }, "LIT002": { - "limit": 27252 + "limit": 27199 }, "LIT003": { "limit": 292 @@ -15,7 +15,7 @@ "limit": 0 }, "LIT006": { - "limit": 1105 + "limit": 1095 }, "LIT007": { "limit": 0