diff --git a/litellm/interactions/litellm_responses_transformation/handler.py b/litellm/interactions/litellm_responses_transformation/handler.py index f95b5611831..628f35edd50 100644 --- a/litellm/interactions/litellm_responses_transformation/handler.py +++ b/litellm/interactions/litellm_responses_transformation/handler.py @@ -2,8 +2,8 @@ Handler for transforming interactions API requests to litellm.responses requests. """ -from collections.abc import AsyncIterator, Callable, Coroutine, Iterator -from typing import Any, Final +from collections.abc import AsyncIterator, Callable, Coroutine, Iterator, Mapping +from typing import Final import litellm from litellm.interactions.litellm_responses_transformation.streaming_iterator import ( @@ -72,6 +72,7 @@ class LiteLLMResponsesInteractionsHandler: model=model, input=input, optional_params=optional_params, + custom_llm_provider=custom_llm_provider, **kwargs, ) @@ -108,10 +109,11 @@ class LiteLLMResponsesInteractionsHandler: async def async_interactions_api_handler( self, - responses_request: dict[str, Any], + responses_request: Mapping[str, object], model: str, input: InteractionInput | None, optional_params: InteractionsAPIOptionalRequestParams, + custom_llm_provider: str | None = None, **kwargs, ) -> InteractionsAPIResponse | AsyncIterator[InteractionsAPIStreamingResponse]: """Async handler for interactions API requests.""" @@ -128,7 +130,7 @@ class LiteLLMResponsesInteractionsHandler: litellm_custom_stream_wrapper=responses_response, request_input=input, optional_params=optional_params, - custom_llm_provider=responses_request.get("custom_llm_provider"), + custom_llm_provider=custom_llm_provider, litellm_metadata=kwargs.get("litellm_metadata", {}), ) @@ -139,5 +141,5 @@ class LiteLLMResponsesInteractionsHandler: return LiteLLMResponsesInteractionsConfig.transform_responses_response_to_interactions_response( responses_response=responses_api_response, model=model, - custom_llm_provider=responses_request.get("custom_llm_provider"), + custom_llm_provider=custom_llm_provider, ) diff --git a/litellm/litellm_core_utils/json_validation_rule.py b/litellm/litellm_core_utils/json_validation_rule.py index b6b16aff7a7..030bee48530 100644 --- a/litellm/litellm_core_utils/json_validation_rule.py +++ b/litellm/litellm_core_utils/json_validation_rule.py @@ -1,9 +1,11 @@ import json -from typing import Any, Final +from collections.abc import Mapping +from typing import Final from pydantic import TypeAdapter from litellm.constants import DEFAULT_MAX_RECURSE_DEPTH +from litellm.litellm_core_utils.llm_cost_calc.guardrail_cost import is_str_keyed_dict _DECODED_RESPONSE: Final = TypeAdapter(object) @@ -76,7 +78,7 @@ def normalize_json_schema_types( return schema -def normalize_tool_schema(tool: dict[str, Any]) -> dict[str, Any]: +def normalize_tool_schema(tool: Mapping[str, object]) -> Mapping[str, object]: """ Normalize a tool's parameter schema to use standard JSON Schema lowercase types. @@ -92,10 +94,14 @@ def normalize_tool_schema(tool: dict[str, Any]) -> dict[str, Any]: normalized_tool: Final = tool.copy() # Normalize function parameters if present - if "function" in tool and isinstance(tool["function"], dict): - normalized_tool["function"] = tool["function"].copy() - if "parameters" in tool["function"]: - normalized_tool["function"]["parameters"] = normalize_json_schema_types(tool["function"]["parameters"]) + if "function" not in tool: + return normalized_tool + function: Final = tool["function"] + if is_str_keyed_dict(function): + normalized_function: Final = function.copy() + normalized_tool["function"] = normalized_function + if "parameters" in function: + normalized_function["parameters"] = normalize_json_schema_types(function["parameters"]) return normalized_tool diff --git a/litellm/litellm_core_utils/litellm_logging.py b/litellm/litellm_core_utils/litellm_logging.py index 72d61b5b487..8b51cf97cd0 100644 --- a/litellm/litellm_core_utils/litellm_logging.py +++ b/litellm/litellm_core_utils/litellm_logging.py @@ -96,6 +96,7 @@ from litellm.litellm_core_utils.core_helpers import ( decoded_json_object, finite_cost, get_provider_response_headers_from_hidden_params, + has_attributes, is_expected_client_error, native_messages_reported_cost, provider_reported_usage_cost, @@ -704,6 +705,10 @@ class _MetadataDefaults(Protocol): setdefault: Callable[[str, Mapping[str, object]], object] +class _MetadataSlot(Protocol): + __setitem__: Callable[[str, object], None] + + def _callback_control_params(litellm_params: object) -> dict[str, object]: mapping: Final = as_str_mapping(litellm_params) if isinstance(mapping, dict): @@ -3115,9 +3120,13 @@ class Logging(LiteLLMLoggingBaseClass): if metadata_hidden_params.get("response_cost") is None and response_cost is not None: metadata_hidden_params["response_cost"] = response_cost - litellm_params: Final = self.model_call_details["litellm_params"] - metadata: Final = litellm_params.get("metadata") or {} + litellm_params: Final = require_attributes( + self._model_call_details_view["litellm_params"], _HiddenParamsSlot, "get" + ) + metadata: Final = litellm_params.get("metadata") or dict[str, object]() litellm_params["metadata"] = metadata + if not has_attributes(metadata, _MetadataSlot, "__setitem__"): + raise TypeError(f"'{type(metadata).__name__}' object does not support item assignment") metadata["hidden_params"] = metadata_hidden_params def _process_hidden_params_and_response_cost( @@ -7175,7 +7184,7 @@ class StandardLoggingPayloadSetup: @staticmethod def get_usage_as_dict( - response_obj: dict[str, Any] | None, + response_obj: Mapping[str, object] | None, combined_usage_object: Usage | None = None, ) -> dict[str, Any]: """ @@ -7199,19 +7208,23 @@ class StandardLoggingPayloadSetup: reported: Final = response_obj.get("usage_info") if isinstance(reported, BaseModel): return reported.model_dump(exclude_none=True) + reported_fields: Final = as_str_mapping(reported) return ( - {key: value for key, value in reported.items() if value is not None} - if isinstance(reported, dict) + {key: value for key, value in reported_fields.items() if value is not None} + if isinstance(reported_fields, dict) else dict(get_rerank_usage(_RESPONSE_MAPPING.validate_python(response_obj)) or _empty) ) if isinstance(_raw, ResponseAPIUsage): return ResponseAPILoggingUtils.transform_response_api_usage_to_chat_usage(_raw).model_dump() - if isinstance(_raw, dict): - if ResponseAPILoggingUtils.is_response_api_usage(_raw): - return ResponseAPILoggingUtils.transform_response_api_usage_to_chat_usage(_raw).model_dump() - if InteractionsUsageObjectTransformation.is_interactions_usage_object(_raw): - return InteractionsUsageObjectTransformation.transform_interactions_usage_object(_raw).model_dump() - return _raw + raw_fields: Final = as_str_mapping(_raw) + if isinstance(raw_fields, dict): + if ResponseAPILoggingUtils.is_response_api_usage(raw_fields): + return ResponseAPILoggingUtils.transform_response_api_usage_to_chat_usage(raw_fields).model_dump() + if InteractionsUsageObjectTransformation.is_interactions_usage_object(raw_fields): + return InteractionsUsageObjectTransformation.transform_interactions_usage_object( + raw_fields + ).model_dump() + return raw_fields if isinstance(_raw, Usage): return _raw.model_dump() return _empty diff --git a/litellm/litellm_core_utils/realtime_streaming.py b/litellm/litellm_core_utils/realtime_streaming.py index 7a5f7c1526a..6d40d963d76 100644 --- a/litellm/litellm_core_utils/realtime_streaming.py +++ b/litellm/litellm_core_utils/realtime_streaming.py @@ -26,7 +26,7 @@ from litellm.constants import ( REALTIME_SESSION_SUCCESS_LOGGED_KEY, ) from litellm.litellm_core_utils.audio_utils.utils import normalized_audio_duration_seconds -from litellm.litellm_core_utils.core_helpers import as_str_mapping, decoded_json_object +from litellm.litellm_core_utils.core_helpers import as_str_mapping, decoded_json_object, json_contains from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER from litellm.llms.base_llm.realtime.transcription_protocol import RealtimeTranscriptionProtocolError from litellm.llms.base_llm.realtime.transformation import ( @@ -220,9 +220,13 @@ def _unscreened_audio_error_event(client_event_id: object) -> Mapping[str, objec } +def _decode_json(payload: str) -> object: + return _DECODED_JSON.validate_python(json.loads(payload)) + + def _decode_json_object(payload: str) -> Mapping[str, object]: """Decode a realtime frame into its top-level field mapping.""" - return json.loads(payload) + return decoded_json_object(_decode_json(payload)) _GUARDRAIL_TURN_DETECTION_GATE: Final[Mapping[str, object]] = MappingProxyType( @@ -1020,7 +1024,7 @@ class RealTimeStreaming: sent = True continue try: - msg_obj = _decode_json_object(msg) + msg_obj = as_str_mapping(_decode_json(msg)) except (json.JSONDecodeError, TypeError): msg_obj = None if isinstance(msg_obj, dict) and self.provider_config.is_setup_message(msg_obj): @@ -1162,13 +1166,15 @@ class RealTimeStreaming: return _session_update_frame(self._held_session_update or _NO_SESSION_FIELDS) async def _hold_session_update(self, message: str) -> None: - client_event: Final = _decode_json_object(message) + client_event: Final = _decode_json(message) if self._held_session_update_bytes + len(message.encode("utf-8")) > RealTimeStreaming._MAX_BUFFERED_BYTES: await self.websocket.send_text( - json.dumps(_HELD_SESSION_UPDATE_TOO_LARGE.error_event(client_event.get("event_id"))) + json.dumps( + _HELD_SESSION_UPDATE_TOO_LARGE.error_event(decoded_json_object(client_event).get("event_id")) + ) ) return - session: Final = client_event.get("session") + session: Final = decoded_json_object(client_event).get("session") update: Final = _SESSION_FIELDS.validate_python(session) if isinstance(session, Mapping) else _NO_SESSION_FIELDS held: Final = _merge_session_fields(self._held_session_update or _NO_SESSION_FIELDS, update) self._held_session_update = held @@ -1182,7 +1188,7 @@ class RealTimeStreaming: _session_update_frame(session), self.model, None ) return next( - (frame for frame in preview if isinstance(frame, str) and "setup" in _decode_json_object(frame)), None + (frame for frame in preview if isinstance(frame, str) and json_contains(_decode_json(frame), "setup")), None ) def _leaves_applied_setup_unchanged(self, update: Mapping[str, object]) -> bool: @@ -1194,7 +1200,7 @@ class RealTimeStreaming: return ( applied_setup is not None and merged_setup is not None - and _decode_json_object(applied_setup) == _decode_json_object(merged_setup) + and _decode_json(applied_setup) == _decode_json(merged_setup) ) async def _acknowledge_session_setup(self, session: Mapping[str, object]) -> None: @@ -1522,8 +1528,8 @@ class RealTimeStreaming: ``return_new_content_delta_events`` modality lookup, ...). """ try: - message_obj: Final = _decode_json_object(transformed_message) - if "setup" in message_obj: + message_obj: Final = _decode_json(transformed_message) + if json_contains(message_obj, "setup"): self.session_configuration_request = transformed_message except (json.JSONDecodeError, TypeError): return @@ -1852,7 +1858,7 @@ class RealTimeStreaming: def _parse_backend_event(raw_response: str) -> dict[str, object] | None: """Parse a backend frame once. Returns None for non-JSON or non-object frames.""" try: - event: Final = _decode_json_object(raw_response) + event: Final = as_str_mapping(_decode_json(raw_response)) except (json.JSONDecodeError, TypeError): return None return event if isinstance(event, dict) else None diff --git a/litellm/litellm_core_utils/streaming_handler.py b/litellm/litellm_core_utils/streaming_handler.py index 24e4e1625ff..f72a25428b6 100644 --- a/litellm/litellm_core_utils/streaming_handler.py +++ b/litellm/litellm_core_utils/streaming_handler.py @@ -30,9 +30,11 @@ from litellm.litellm_core_utils.core_helpers import ( attribute_or_none, chunk_usage, decoded_json_object, + json_contains, provider_reported_usage_cost, require_attributes, stamp_litellm_bill_on_usage, + subscript, without, ) from litellm.litellm_core_utils.hidden_params import ( @@ -288,12 +290,8 @@ class _PredibaseStreamData(TypedDict): error: str | None -class _NlpCloudStreamData(TypedDict): - generated_text: str - - -class _AlephAlphaStreamData(TypedDict): - completions: Sequence[Mapping[str, str]] +class _Replaceable(Protocol): + def replace(self, old: str, new: str, /) -> object: ... class _DeltaDumpDict(TypedDict): @@ -833,11 +831,11 @@ class CustomStreamWrapper: if self.model and "dolphin" in self.model: chunk = self.process_chunk(chunk=chunk) else: - data_json: Final[_NlpCloudStreamData] = json.loads(chunk) - chunk = data_json["generated_text"] + data_json: Final = _DECODED_JSON.validate_python(json.loads(chunk)) + chunk = subscript(data_json, "generated_text") text = chunk - if "[DONE]" in text: - text = text.replace("[DONE]", "") + if json_contains(text, "[DONE]"): + text = require_attributes(text, _Replaceable, "replace").replace("[DONE]", "") is_finished = True finish_reason = "stop" return { @@ -850,9 +848,9 @@ class CustomStreamWrapper: def handle_aleph_alpha_chunk(self, chunk): chunk = chunk.decode("utf-8") - data_json: Final[_AlephAlphaStreamData] = json.loads(chunk) + data_json: Final = _DECODED_JSON.validate_python(json.loads(chunk)) try: - text: Final = data_json["completions"][0]["completion"] + text: Final = subscript(subscript(subscript(data_json, "completions"), 0), "completion") is_finished: Final = True finish_reason: Final = "stop" return { @@ -1110,7 +1108,8 @@ class CustomStreamWrapper: provider_specific_fields: Final = getattr(original_chunk, "provider_specific_fields", None) if provider_specific_fields is not None: model_response.provider_specific_fields = provider_specific_fields - _apply_provider_fields(model_response, provider_specific_fields) + if model_response.provider_specific_fields is not None: + _apply_provider_fields(model_response, model_response.provider_specific_fields) return model_response def is_chunk_non_empty( diff --git a/litellm/llms/anthropic/chat/transformation.py b/litellm/llms/anthropic/chat/transformation.py index e0d313175b7..fe4b7eb2039 100644 --- a/litellm/llms/anthropic/chat/transformation.py +++ b/litellm/llms/anthropic/chat/transformation.py @@ -6,7 +6,7 @@ from collections.abc import Callable, Iterable, Iterator, Mapping, Sequence from functools import partial from itertools import chain from types import MappingProxyType -from typing import TYPE_CHECKING, Annotated, Any, ClassVar, Final, NoReturn, cast +from typing import TYPE_CHECKING, Annotated, Any, ClassVar, Final, NoReturn, Protocol, cast, runtime_checkable import httpx from openai.types.responses.response_function_web_search import ResponseFunctionWebSearch @@ -31,6 +31,7 @@ from litellm.constants import ( ) from litellm.litellm_core_utils.asyncify import is_event_loop_running from litellm.litellm_core_utils.core_helpers import ( + as_object_list, coerce_token_limit, drops_unsupported_params, map_finish_reason, @@ -436,6 +437,46 @@ def _enum_conflicts_with_declared_type(schema: Mapping[str, object]) -> bool: return check is not None and not all(check(value) for value in enum_values) +def _as_json_object(value: object) -> Mapping[str, object] | None: + return value if isinstance(value, dict) else None + + +def _filtered_output_subschema(value: object) -> object: + subschema: Final = _as_json_object(value) + return value if subschema is None else AnthropicConfig.filter_anthropic_output_schema(subschema) + + +def _filtered_output_subschemas(value: object) -> object: + subschemas: Final = _as_json_object(value) + if subschemas is None: + return value + return {name: _filtered_output_subschema(subschema) for name, subschema in subschemas.items()} + + +def _filtered_output_branches(branches: Sequence[object]) -> Sequence[object]: + return [_filtered_output_subschema(branch) for branch in branches] + + +@runtime_checkable +class _SupportsAdd(Protocol): + def __add__(self, other: object, /) -> object: ... + + +def _concatenated(left: object, right: str | Sequence[object]) -> object: + if isinstance(left, _SupportsAdd): + return left + right + raise TypeError(f"unsupported operand type(s) for +: '{type(left).__name__}' and '{type(right).__name__}'") + + +def _extended_any_of(result: Mapping[str, object], branches: Sequence[object]) -> object: + filtered: Final = _filtered_output_branches(branches) + return _concatenated(result["anyOf"], filtered) if "anyOf" in result else filtered + + +def _described_with_note(description: object, note: str) -> object: + return _concatenated(_concatenated(description, " "), note) if description else note + + # Single, internal-only key on ``litellm_params`` used to thread the per- # request reverse map (sanitized -> original) from request build to response # parsing. ``litellm_params`` is never serialized to a provider; ``optional_ @@ -913,7 +954,7 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig): return params @staticmethod - def filter_anthropic_output_schema(schema: dict[str, Any]) -> dict[str, Any]: + def filter_anthropic_output_schema(schema: Mapping[str, object]) -> Mapping[str, object]: """ Filter out unsupported fields from JSON schema for Anthropic's output_format API. @@ -1005,16 +1046,12 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig): note_value = json.dumps(value) if isinstance(value, (dict, list)) else value constraint_descriptions.append(label.format(note_value)) - result: Final[dict[str, Any]] = {} + result: Final[dict[str, object]] = {} # Update description with removed constraint info if constraint_descriptions: - existing_desc: Final = schema.get("description", "") constraint_note: Final = "Note: " + ", ".join(constraint_descriptions) + "." - if existing_desc: - result["description"] = existing_desc + " " + constraint_note - else: - result["description"] = constraint_note + result["description"] = _described_with_note(schema.get("description"), constraint_note) drops_conflicting_type: Final = _enum_conflicts_with_declared_type(schema) @@ -1027,26 +1064,21 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig): # Already handled above continue - if key in ("properties", "$defs", "definitions") and isinstance(value, dict): - nested_schemas: dict[str, Any] = value # mutable-ok: filter_anthropic_output_schema takes a dict - result[key] = {k: AnthropicConfig.filter_anthropic_output_schema(v) for k, v in nested_schemas.items()} - elif key == "items" and isinstance(value, dict): - nested_schema: dict[str, Any] = value # mutable-ok: filter_anthropic_output_schema takes a dict - result[key] = AnthropicConfig.filter_anthropic_output_schema(nested_schema) - elif key == "anyOf" and isinstance(value, list): - result["anyOf"] = result.get("anyOf", []) + [ - AnthropicConfig.filter_anthropic_output_schema(item) for item in value - ] - elif key == "allOf" and isinstance(value, list): - result[key] = [AnthropicConfig.filter_anthropic_output_schema(item) for item in value] + branches = as_object_list(value) + if key in ("properties", "$defs", "definitions"): + result[key] = _filtered_output_subschemas(value) + elif key == "items": + result[key] = _filtered_output_subschema(value) + elif key == "anyOf" and branches is not None: + result["anyOf"] = _extended_any_of(result, branches) + elif key == "allOf" and branches is not None: + result[key] = _filtered_output_branches(branches) elif key == "$ref" and isinstance(value, str): result[key] = schema_fragment_ref(value) - elif key == "oneOf" and isinstance(value, list): + elif key == "oneOf" and branches is not None: # Anthropic rejects oneOf ("Schema type 'oneOf' is not supported"); # the Anthropic SDK rewrites it to anyOf, so do the same. - result["anyOf"] = result.get("anyOf", []) + [ - AnthropicConfig.filter_anthropic_output_schema(item) for item in value - ] + result["anyOf"] = _extended_any_of(result, branches) else: result[key] = value diff --git a/litellm/llms/anthropic/messages/adapters/transformation.py b/litellm/llms/anthropic/messages/adapters/transformation.py index 2cbbb759582..81f36bec5f6 100644 --- a/litellm/llms/anthropic/messages/adapters/transformation.py +++ b/litellm/llms/anthropic/messages/adapters/transformation.py @@ -3,12 +3,12 @@ import hashlib import json from collections.abc import AsyncIterator, Iterator, Mapping, Sequence from types import MappingProxyType -from typing import TYPE_CHECKING, Any, Final, Literal, TypeAlias, TypeVar, cast +from typing import TYPE_CHECKING, Any, Final, Literal, Protocol, TypeAlias, TypeVar, cast from pydantic import JsonValue, TypeAdapter import litellm -from litellm.litellm_core_utils.core_helpers import as_str_mapping, normalize_drop_params +from litellm.litellm_core_utils.core_helpers import as_str_mapping, normalize_drop_params, require_attributes from litellm.litellm_core_utils.hidden_params import get_hidden_params from litellm.llms.anthropic.messages.convertible_blocks import tool_result_text_part from litellm.llms.anthropic.messages.utils import ( @@ -252,6 +252,11 @@ if TYPE_CHECKING: ToolResultContent: TypeAlias = str | list[ToolMessageContentPart] +class _CarriesUsage(Protocol): + @property + def usage(self) -> Usage: ... + + def target_supports_mid_conversation_system(model: str | None, custom_llm_provider: str | None) -> bool: if not model: return False @@ -1738,7 +1743,7 @@ class LiteLLMAnthropicMessagesAdapter: else translated_finish_reason ) # extract usage - usage: Final[Usage] = getattr(response, "usage") + usage: Final = require_attributes(response, _CarriesUsage, "usage").usage message_usage: Final = self.translate_openai_usage_to_anthropic_usage(usage) polyfill_iterations: Final = polyfill_result.iterations_usage if polyfill_result is not None else None anthropic_usage: Final[AnthropicUsage] = ( diff --git a/litellm/llms/anthropic/messages/native/mcp_handler.py b/litellm/llms/anthropic/messages/native/mcp_handler.py index 8fe18d79f24..cefd6c6a6d9 100644 --- a/litellm/llms/anthropic/messages/native/mcp_handler.py +++ b/litellm/llms/anthropic/messages/native/mcp_handler.py @@ -46,6 +46,10 @@ def _get_stop_reason(response: AnthropicMessagesResponse) -> str | None: return stop_reason if isinstance(stop_reason, str) else None +def _requests_stream(kwargs: Mapping[str, object]) -> bool: + return bool(kwargs.get("stream", False)) + + def _build_tool_result_message(tool_results: Sequence[Mapping[str, object]]) -> AnthropicMessagesUserMessageParam: """Turn executed tool results into the user message Anthropic expects.""" return AnthropicMessagesUserMessageParam( @@ -120,7 +124,8 @@ async def anthropic_messages_with_mcp( should_auto_execute: Final = LiteLLM_Proxy_MCP_Handler.should_auto_execute_tools( mcp_tools_with_litellm_proxy=mcp_references ) - stream: Final = bool(kwargs.pop("stream", False)) + stream: Final = _requests_stream(kwargs) + kwargs.pop("stream", None) base_call_args: Final[Mapping[str, object]] = { "max_tokens": max_tokens, diff --git a/litellm/llms/anthropic/messages/native/streaming_iterator.py b/litellm/llms/anthropic/messages/native/streaming_iterator.py index 52b3f1042f8..1195652dfd2 100644 --- a/litellm/llms/anthropic/messages/native/streaming_iterator.py +++ b/litellm/llms/anthropic/messages/native/streaming_iterator.py @@ -28,6 +28,7 @@ from litellm.llms.anthropic.messages.utils import anthropic_sse_frame from litellm.proxy.pass_through_endpoints.success_handler import ( PassThroughEndpointLogging, ) +from litellm.responses.mcp.parked_stream_dispatch import DeferredStreamRelease from litellm.types.llms.anthropic_messages.anthropic_response import AnthropicMessagesResponse from litellm.types.passthrough_endpoints.pass_through_endpoints import EndpointType from litellm.types.utils import GenericStreamingChunk, ModelResponseStream @@ -647,8 +648,12 @@ class BaseAnthropicMessagesStreamingIterator: disconnect (GeneratorExit / CancelledError) re-raises past it. Without this dispatch that window loses the spend row entirely. """ - deferred_cb: Final = getattr(self.litellm_logging_obj, "_on_deferred_stream_complete", None) - deferred_args: Final = getattr(self.litellm_logging_obj, "_deferred_stream_complete_args", None) + deferred_cb: Final[DeferredStreamRelease | None] = getattr( + self.litellm_logging_obj, "_on_deferred_stream_complete", None + ) + deferred_args: Final[tuple[object, ...] | None] = getattr( + self.litellm_logging_obj, "_deferred_stream_complete_args", None + ) if deferred_cb is None or deferred_args is None: return self.litellm_logging_obj._on_deferred_stream_complete = None diff --git a/litellm/llms/bedrock/chat/invoke_transformations/amazon_twelvelabs_pegasus_transformation.py b/litellm/llms/bedrock/chat/invoke_transformations/amazon_twelvelabs_pegasus_transformation.py index dc14ef005b5..f9e834e9d37 100644 --- a/litellm/llms/bedrock/chat/invoke_transformations/amazon_twelvelabs_pegasus_transformation.py +++ b/litellm/llms/bedrock/chat/invoke_transformations/amazon_twelvelabs_pegasus_transformation.py @@ -7,10 +7,11 @@ https://docs.twelvelabs.io/docs/models/pegasus import json import time +from collections.abc import Mapping from typing import TYPE_CHECKING, Any, Final import httpx -from pydantic import ConfigDict, TypeAdapter, with_config +from pydantic import BaseModel, ConfigDict, TypeAdapter, with_config from typing_extensions import NotRequired, ReadOnly, TypedDict import litellm @@ -43,6 +44,7 @@ class _PegasusResponse(TypedDict): _PEGASUS_RESPONSE: Final = TypeAdapter(_PegasusResponse) +_OPTIONAL_PARAM: Final = TypeAdapter(object) class AmazonTwelveLabsPegasusConfig(AmazonInvokeConfig, BaseConfig): @@ -65,7 +67,7 @@ class AmazonTwelveLabsPegasusConfig(AmazonInvokeConfig, BaseConfig): def map_openai_params( self, - non_default_params: dict, + non_default_params: Mapping[str, object], optional_params: dict, model: str, drop_params: bool, @@ -79,7 +81,7 @@ class AmazonTwelveLabsPegasusConfig(AmazonInvokeConfig, BaseConfig): optional_params["responseFormat"] = self._normalize_response_format(value) return optional_params - def _normalize_response_format(self, value: Any) -> object: + def _normalize_response_format(self, value: object) -> object: """Normalize response_format to TwelveLabs format. TwelveLabs expects: @@ -110,7 +112,11 @@ class AmazonTwelveLabsPegasusConfig(AmazonInvokeConfig, BaseConfig): return value # Otherwise return the dict as is return value - return type_to_response_format_param(response_format=value) or value + if value is None: + return None + if isinstance(value, type) and issubclass(value, BaseModel): + return type_to_response_format_param(response_format=value) or value + raise TypeError(f"Unsupported response_format type - {value}") def transform_request( self, @@ -134,7 +140,7 @@ class AmazonTwelveLabsPegasusConfig(AmazonInvokeConfig, BaseConfig): # Handle responseFormat - transform to TwelveLabs format if "responseFormat" in optional_params: - response_format: Final = optional_params["responseFormat"] + response_format: Final = _OPTIONAL_PARAM.validate_python(optional_params["responseFormat"]) transformed_format: Final = self._normalize_response_format(response_format) if transformed_format: request_data["responseFormat"] = transformed_format diff --git a/litellm/llms/cohere/embed/v1_transformation.py b/litellm/llms/cohere/embed/v1_transformation.py index 056b5bce742..ca096e369e1 100644 --- a/litellm/llms/cohere/embed/v1_transformation.py +++ b/litellm/llms/cohere/embed/v1_transformation.py @@ -2,12 +2,14 @@ Legacy /v1/embedding transformation logic for Bedrock Cohere. """ -from collections.abc import Sized +from collections.abc import Mapping, Sized from typing import Final, Protocol import httpx +from pydantic import TypeAdapter from litellm import COHERE_DEFAULT_EMBEDDING_INPUT_TYPE +from litellm.litellm_core_utils.core_helpers import as_str_mapping, subscript from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj from litellm.llms.custom_httpx.http_handler import header_value from litellm.types.llms.bedrock import ( @@ -24,6 +26,17 @@ class _SupportsEncode(Protocol): def encode(self, text: str, /) -> Sized: ... +_DECODED_JSON: Final = TypeAdapter(object) + + +def _embed_response_body(response_json: object) -> Mapping[str, object]: + body: Final = as_str_mapping(response_json) + if body is None: + subscript(response_json, "embeddings") + raise TypeError(f"'{type(response_json).__name__}' object is not subscriptable") + return body + + def _bedrock_billed_input_tokens(response: httpx.Response) -> int | None: count: Final = header_value(response.headers, "x-amzn-bedrock-input-token-count") return int(count) if count is not None and count.isdigit() else None @@ -119,7 +132,7 @@ class CohereEmbeddingConfig: encoding: _SupportsEncode, input: list, ) -> EmbeddingResponse: - response_json: Final = response.json() + response_json: Final = _DECODED_JSON.validate_python(response.json()) ## LOGGING logging_obj.post_call( input=input, @@ -127,13 +140,14 @@ class CohereEmbeddingConfig: additional_args={"complete_input_dict": data}, original_response=response_json, ) + billed_input_tokens: Final = _bedrock_billed_input_tokens(response) if "images" in data else None return self._populate_embedding_response( - response_json=response_json, + response_json=dict(_embed_response_body(response_json)), model_response=model_response, model=model, encoding=encoding, input=input, - billed_input_tokens=_bedrock_billed_input_tokens(response) if "images" in data else None, + billed_input_tokens=billed_input_tokens, ) def transform_response( diff --git a/litellm/llms/custom_httpx/llm_http_handler.py b/litellm/llms/custom_httpx/llm_http_handler.py index 477ce845508..eb91744127e 100644 --- a/litellm/llms/custom_httpx/llm_http_handler.py +++ b/litellm/llms/custom_httpx/llm_http_handler.py @@ -623,6 +623,7 @@ _OPTIONAL_PARAMS: Final = TypeAdapter(Mapping[str, object]) _JSON_OBJECT: Final = TypeAdapter(dict[str, object]) _JSON_OBJECTS: Final = TypeAdapter(list[dict[str, object]]) _DECODED_JSON: Final = TypeAdapter(object) +_AGENTIC_HOOK_RESULT: Final = TypeAdapter(object) def _decoded_body_headers(response: httpx.Response) -> httpx.Headers: @@ -6481,21 +6482,23 @@ class BaseLLMHTTPHandler: callback.__class__.async_build_agentic_loop_plan is not CustomLogger.async_build_agentic_loop_plan ) if not build_plan_overridden: - agentic_result: object = await bill_first_round_on_failure( - callback.async_run_agentic_loop( - tools=tool_calls, - model=model, - messages=messages, - response=response, - anthropic_messages_provider_config=anthropic_messages_provider_config, - anthropic_messages_optional_request_params=anthropic_messages_optional_request_params, + agentic_result: object = _AGENTIC_HOOK_RESULT.validate_python( + await bill_first_round_on_failure( + callback.async_run_agentic_loop( + tools=tool_calls, + model=model, + messages=messages, + response=response, + anthropic_messages_provider_config=anthropic_messages_provider_config, + anthropic_messages_optional_request_params=anthropic_messages_optional_request_params, + logging_obj=logging_obj, + stream=stream, + kwargs=kwargs_with_provider, + ), logging_obj=logging_obj, + response=response, stream=stream, - kwargs=kwargs_with_provider, - ), - logging_obj=logging_obj, - response=response, - stream=stream, + ) ) return self._maybe_wrap_in_fake_stream(agentic_result, logging_obj, api_surface) @@ -6526,10 +6529,12 @@ class BaseLLMHTTPHandler: plan.stop_reason, ) return self._maybe_wrap_in_fake_stream( - await callback.async_post_agentic_loop_response_hook( - response=self._finalize_refused_agentic_response(response=response, tool_calls=tool_calls), - plan=plan, - kwargs=kwargs_with_provider, + _AGENTIC_HOOK_RESULT.validate_python( + await callback.async_post_agentic_loop_response_hook( + response=self._finalize_refused_agentic_response(response=response, tool_calls=tool_calls), + plan=plan, + kwargs=kwargs_with_provider, + ) ), logging_obj, api_surface, diff --git a/litellm/llms/huggingface/embedding/transformation.py b/litellm/llms/huggingface/embedding/transformation.py index 4d2d9786111..e3210482b2a 100644 --- a/litellm/llms/huggingface/embedding/transformation.py +++ b/litellm/llms/huggingface/embedding/transformation.py @@ -55,6 +55,12 @@ def _json_len(value: object) -> int: raise TypeError(f"object of type '{type(value).__name__}' has no len()") +def _generated_text(value: object) -> str: + if isinstance(value, str): + return value + raise AttributeError(f"'{type(value).__name__}' object has no attribute 'strip'", name="strip", obj=value) + + def _add_logprob(total: float, logprob: object) -> float: if isinstance(logprob, (int, float)): return total + logprob @@ -425,7 +431,9 @@ class HuggingFaceEmbeddingConfig(BaseConfig): ) if _json_len(_DECODED_JSON.validate_python(completion_response[0]["generated_text"])) > 0: - model_response.choices[0].message.content = output_parser(completion_response[0]["generated_text"]) + model_response.choices[0].message.content = output_parser( + _generated_text(_DECODED_JSON.validate_python(completion_response[0]["generated_text"])) + ) ## GETTING LOGPROBS + FINISH REASON if "details" in completion_response[0] and "tokens" in completion_response[0]["details"]: model_response.choices[0].finish_reason = completion_response[0]["details"]["finish_reason"] @@ -438,21 +446,25 @@ class HuggingFaceEmbeddingConfig(BaseConfig): if "best_of" in optional_params and optional_params["best_of"] > 1: if "details" in completion_response[0] and "best_of_sequences" in completion_response[0]["details"]: choices_list: Final = [] - for idx, item in enumerate(completion_response[0]["details"]["best_of_sequences"]): + for idx, item in enumerate( + iter_items( + _DECODED_JSON.validate_python(completion_response[0]["details"]["best_of_sequences"]) + ) + ): sum_logprob = 0 - for token in iter_items(_DECODED_JSON.validate_python(item["tokens"])): + for token in iter_items(subscript(item, "tokens")): logprob = subscript(token, "logprob") if logprob is not None: sum_logprob = _add_logprob(sum_logprob, logprob) - if _json_len(_DECODED_JSON.validate_python(item["generated_text"])) > 0: + if _json_len(subscript(item, "generated_text")) > 0: message_obj = Message( - content=output_parser(item["generated_text"]), + content=output_parser(_generated_text(subscript(item, "generated_text"))), logprobs=sum_logprob, ) else: message_obj = Message(content=None) choice_obj = Choices( - finish_reason=item["finish_reason"], + finish_reason=subscript(item, "finish_reason"), index=idx + 1, message=message_obj, ) @@ -465,7 +477,9 @@ class HuggingFaceEmbeddingConfig(BaseConfig): isinstance(completion_response, list) and _json_len(_DECODED_JSON.validate_python(completion_response[0]["generated_text"])) > 0 ): - model_response.choices[0].message.content = output_parser(completion_response[0]["generated_text"]) + model_response.choices[0].message.content = output_parser( + _generated_text(_DECODED_JSON.validate_python(completion_response[0]["generated_text"])) + ) ## CALCULATING USAGE prompt_tokens = 0 try: diff --git a/litellm/llms/ragflow/vector_stores/transformation.py b/litellm/llms/ragflow/vector_stores/transformation.py index ef44b6e4c0f..421bd769778 100644 --- a/litellm/llms/ragflow/vector_stores/transformation.py +++ b/litellm/llms/ragflow/vector_stores/transformation.py @@ -41,6 +41,11 @@ class _RAGFlowCreateResponseRoot(RootModel[_RAGFlowCreateResponse]): model_config = ConfigDict(frozen=True, hide_input_in_errors=True) +class _RAGFlowStatus(TypedDict): + code: ReadOnly[NotRequired[int]] + message: ReadOnly[NotRequired[str]] + + _RAGFLOW_EMPTY_DATA: Final[_RAGFlowDataset] = {} @@ -197,7 +202,7 @@ class RAGFlowVectorStoreConfig(BaseVectorStoreConfig): } """ try: - response_json: Final = response.json() + response_json: Final[_RAGFlowStatus] = response.json() # Check for RAGFlow error response if response_json.get("code") != 0: diff --git a/litellm/llms/replicate/chat/transformation.py b/litellm/llms/replicate/chat/transformation.py index 62dfac43bd7..4553a02e451 100644 --- a/litellm/llms/replicate/chat/transformation.py +++ b/litellm/llms/replicate/chat/transformation.py @@ -4,6 +4,7 @@ from typing import TYPE_CHECKING, Any, Final import httpx from pydantic import ConfigDict, TypeAdapter +from typing_extensions import NotRequired, ReadOnly, TypedDict import litellm from litellm.constants import REPLICATE_MODEL_NAME_WITH_ID_LENGTH @@ -33,6 +34,10 @@ _TEXTS: Final = TypeAdapter(Iterable[str], config=ConfigDict(strict=True, hide_i _JSON_DICT: Final = TypeAdapter(dict[str, object], config=ConfigDict(hide_input_in_errors=True)) +class _ReplicatePrediction(TypedDict): + urls: ReadOnly[NotRequired[Mapping[str, str]]] + + class ReplicateConfig(BaseConfig): """ Reference: https://replicate.com/meta/llama-2-70b-chat/api @@ -292,7 +297,7 @@ class ReplicateConfig(BaseConfig): "urls":{"cancel":"https://api.replicate.com/v1/predictions/gqsmqmp1pdrj00cknr08dgmvb4/cancel","get":"https://api.replicate.com/v1/predictions/gqsmqmp1pdrj00cknr08dgmvb4","stream":"https://stream-b.svc.rno2.c.replicate.net/v1/streams/eot4gbydowuin4snhncydwxt57dfwgsc3w3snycx5nid7oef7jga"} } """ - response_json: Final = response.json() + response_json: Final[_ReplicatePrediction] = response.json() prediction_url: Final = response_json.get("urls", {}).get("get") if prediction_url is None: raise ReplicateError( diff --git a/litellm/llms/stability/image_edit/transformations.py b/litellm/llms/stability/image_edit/transformations.py index 974996c5e48..65277973d39 100644 --- a/litellm/llms/stability/image_edit/transformations.py +++ b/litellm/llms/stability/image_edit/transformations.py @@ -6,13 +6,13 @@ Handles transformation between OpenAI-compatible format and Stability AI API for API Reference: https://platform.stability.ai/docs/api-reference """ -from collections.abc import Sequence from typing import TYPE_CHECKING, Any, Final, cast import httpx from httpx._types import RequestFiles -from typing_extensions import ReadOnly, TypedDict +from pydantic import TypeAdapter +from litellm.litellm_core_utils.core_helpers import decoded_json_object, json_contains, subscript from litellm.llms.base_llm.image_edit.transformation import BaseImageEditConfig from litellm.secret_managers.main import get_secret_str from litellm.types.images.main import ImageEditOptionalRequestParams @@ -31,12 +31,7 @@ if TYPE_CHECKING: else: LiteLLMLoggingObj = Any - -class _StabilityImageEditResponse(TypedDict, total=False): - image: ReadOnly[str] - finish_reason: ReadOnly[str] - seed: ReadOnly[int] - errors: ReadOnly[Sequence[str]] +_DECODED_JSON: Final = TypeAdapter(object) class StabilityImageEditConfig(BaseImageEditConfig): @@ -268,7 +263,7 @@ class StabilityImageEditConfig(BaseImageEditConfig): OpenAI expects: {"data": [{"b64_json": "base64..."}], "created": timestamp} """ try: - response_data: Final[_StabilityImageEditResponse] = raw_response.json() + response_json: Final = _DECODED_JSON.validate_python(raw_response.json()) except Exception as e: raise self.get_error_class( error_message=f"Error parsing Stability AI response: {e}", @@ -277,14 +272,15 @@ class StabilityImageEditConfig(BaseImageEditConfig): ) # Check for errors in response - if "errors" in response_data: + if json_contains(response_json, "errors"): raise self.get_error_class( - error_message=f"Stability AI error: {response_data['errors']}", + error_message=f"Stability AI error: {subscript(response_json, 'errors')}", status_code=raw_response.status_code, headers=raw_response.headers, ) # Check finish_reason + response_data: Final = decoded_json_object(response_json) finish_reason: Final = response_data.get("finish_reason", "") if finish_reason == "CONTENT_FILTERED": raise self.get_error_class( diff --git a/litellm/llms/you_com/search/transformation.py b/litellm/llms/you_com/search/transformation.py index 9228deda26b..b56b547a949 100644 --- a/litellm/llms/you_com/search/transformation.py +++ b/litellm/llms/you_com/search/transformation.py @@ -8,7 +8,9 @@ OpenAPI spec: https://you.com/specs/openapi_search_v1.yaml from typing import Final, TypedDict import httpx +from pydantic import TypeAdapter +from litellm.litellm_core_utils.core_helpers import decoded_json_object, iter_items, subscript from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj from litellm.litellm_core_utils.url_utils import join_api_base from litellm.llms.base_llm.search.transformation import ( @@ -18,6 +20,8 @@ from litellm.llms.base_llm.search.transformation import ( ) from litellm.secret_managers.main import get_secret_str +_DECODED_JSON: Final = TypeAdapter(object) + class _YouComSearchRequestRequired(TypedDict): """Required fields for You.com Search API request.""" @@ -169,23 +173,26 @@ class YouComSearchConfig(BaseSearchConfig): - snippets[0] → SearchResult.snippet (falls back to `description`) - page_age → SearchResult.date """ - response_json: Final = raw_response.json() - raw_results: Final = response_json.get("results") or {} + response_json: Final = decoded_json_object(_DECODED_JSON.validate_python(raw_response.json())) + raw_results: Final = decoded_json_object(response_json.get("results") or {}) web_results: Final = raw_results.get("web") or [] news_results: Final = raw_results.get("news") or [] results: Final[list[SearchResult]] = [] - for item in list(web_results) + list(news_results): + for entry in list(iter_items(web_results)) + list(iter_items(news_results)): + item = decoded_json_object(entry) snippets = item.get("snippets") or [] - snippet = snippets[0] if snippets else item.get("description", "") + snippet = subscript(snippets, 0) if snippets else item.get("description", "") results.append( - SearchResult( - title=item.get("title", ""), - url=item.get("url", ""), - snippet=snippet, - date=item.get("page_age"), - last_updated=None, + SearchResult.model_validate( + { + "title": item.get("title", ""), + "url": item.get("url", ""), + "snippet": snippet, + "date": item.get("page_age"), + "last_updated": None, + } ) ) diff --git a/litellm/proxy/_lazy_openapi_snapshot.py b/litellm/proxy/_lazy_openapi_snapshot.py index c5a63b0f990..e428017badd 100644 --- a/litellm/proxy/_lazy_openapi_snapshot.py +++ b/litellm/proxy/_lazy_openapi_snapshot.py @@ -13,7 +13,7 @@ the JSON, then run `npm run gen:api` in ui/litellm-dashboard and commit schema.d import json import re import sys -from collections.abc import Callable, Mapping +from collections.abc import Callable, Mapping, Sequence from dataclasses import dataclass from pathlib import Path from types import MappingProxyType @@ -43,7 +43,7 @@ def _stabilize_multi_method_route_ids(routes) -> None: """FastAPI derives route IDs from a set of methods; make snapshots stable.""" for route in routes: - methods = sorted(getattr(route, "methods", None) or []) + methods: Sequence[str] = sorted(getattr(route, "methods", None) or []) if len(methods) <= 1 or not getattr(route, "path_format", None): continue diff --git a/litellm/proxy/agent_endpoints/a2a_endpoints.py b/litellm/proxy/agent_endpoints/a2a_endpoints.py index f0a4d47f1e4..bff270b8983 100644 --- a/litellm/proxy/agent_endpoints/a2a_endpoints.py +++ b/litellm/proxy/agent_endpoints/a2a_endpoints.py @@ -23,7 +23,7 @@ from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError import litellm from litellm._logging import verbose_proxy_logger -from litellm.litellm_core_utils.core_helpers import as_str_mapping +from litellm.litellm_core_utils.core_helpers import as_str_mapping, keyword_arguments from litellm.litellm_core_utils.url_utils import SSRFError, validate_url from litellm.llms.a2a.common_utils import resolve_a2a_hop_auth_header from litellm.llms.azure.common_utils import caused_by_entra_rejection @@ -145,19 +145,20 @@ def _to_jsonrpc_object(chunk: object) -> object: return chunk -def _build_message_send_params(params: dict[str, Any]) -> "MessageSendParams": +def _build_message_send_params(params: object) -> "MessageSendParams": """Build MessageSendParams from wire (0.3) or A2A 1.0 JSON-RPC params.""" from a2a.compat.v0_3.types import MessageSendParams + fields: Final = dict(**keyword_arguments(params, MessageSendParams)) try: - return MessageSendParams(**params) + return MessageSendParams.model_validate(fields) except ValidationError: from a2a.compat.v0_3.conversions import pb2_v10, to_compat_send_message_request from google.protobuf.json_format import ParseDict, ParseError pb: Final = pb2_v10.SendMessageRequest() try: - ParseDict(params, pb, ignore_unknown_fields=True) + ParseDict(fields, pb, ignore_unknown_fields=True) except ParseError as e: raise ValueError(f"Invalid message/send params: {e}") from e return to_compat_send_message_request(pb, "").params @@ -1027,7 +1028,7 @@ async def invoke_agent_a2a( from a2a.compat.v0_3.types import SendMessageRequest try: - message_send_params: Final = _build_message_send_params(params) + message_send_params: Final = _build_message_send_params(_DECODED_JSON.validate_python(params)) except (ValidationError, ValueError) as e: return _jsonrpc_error(request_id, -32602, f"Invalid params: {e}") diff --git a/litellm/proxy/openai_files_endpoints/common_utils.py b/litellm/proxy/openai_files_endpoints/common_utils.py index 1ae0d475d8c..e3e46a0c872 100644 --- a/litellm/proxy/openai_files_endpoints/common_utils.py +++ b/litellm/proxy/openai_files_endpoints/common_utils.py @@ -23,7 +23,7 @@ from typing_extensions import NotRequired, ReadOnly, TypedDict from litellm.constants import MAX_FILE_LIST_LIMIT from litellm.litellm_core_utils.batch_utils import batch_cost_is_final -from litellm.litellm_core_utils.core_helpers import as_str_mapping +from litellm.litellm_core_utils.core_helpers import as_str_mapping, attribute_or_none from litellm.proxy._types import ProxyException from litellm.repositories.managed_file_repository import ManagedFileRepository from litellm.repositories.table_repositories import ( @@ -1374,7 +1374,7 @@ async def resolve_output_file_ids_to_unified(response, prisma_client) -> None: if not prisma_client: return for attr in ("output_file_id", "error_file_id"): - raw_id = getattr(response, attr, None) + raw_id = attribute_or_none(response, attr) if not raw_id or is_base64_encoded_unified_file_id(raw_id): continue try: diff --git a/litellm/proxy/pass_through_endpoints/llm_provider_handlers/anthropic_passthrough_logging_handler.py b/litellm/proxy/pass_through_endpoints/llm_provider_handlers/anthropic_passthrough_logging_handler.py index c1fc6479421..dafbf6f74b6 100644 --- a/litellm/proxy/pass_through_endpoints/llm_provider_handlers/anthropic_passthrough_logging_handler.py +++ b/litellm/proxy/pass_through_endpoints/llm_provider_handlers/anthropic_passthrough_logging_handler.py @@ -11,6 +11,7 @@ import litellm from litellm._logging import verbose_proxy_logger from litellm.constants import ANTHROPIC_BATCHES_ROUTE from litellm.litellm_core_utils.core_helpers import ( + json_contains, native_messages_reported_cost, override_stamped_response_cost, response_cost_in_hidden_params, @@ -58,6 +59,7 @@ else: EndpointType = Any _JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True)) +_DECODED_JSON: Final = TypeAdapter(object) def _event_payload(event: str) -> Mapping[str, object] | None: @@ -609,10 +611,10 @@ class AnthropicPassthroughLoggingHandler: from litellm.types.utils import Choices, SpecialEnums try: - _json_response: Final = httpx_response.json() + _json_response: Final = _DECODED_JSON.validate_python(httpx_response.json()) # Only handle successful batch job creation (POST requests with 201 status) - if httpx_response.status_code == 200 and "id" in _json_response: + if httpx_response.status_code == 200 and json_contains(_json_response, "id"): # Transform Anthropic response to LiteLLM batch format anthropic_batches_config: Final = AnthropicBatchesConfig() litellm_batch_response: Final = anthropic_batches_config.transform_retrieve_batch_response( @@ -626,7 +628,7 @@ class AnthropicPassthroughLoggingHandler: litellm_batch_response.status = "validating" # Extract batch ID from the response - batch_id: Final = _json_response.get("id", "") + batch_id: Final = litellm_batch_response.id # Get model from request body (batch response doesn't include model) request_body = request_body or {} diff --git a/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py b/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py index 53317010af4..93615cf92fb 100644 --- a/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py +++ b/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py @@ -6,7 +6,16 @@ import posixpath import re import traceback from base64 import b64encode -from collections.abc import AsyncGenerator, AsyncIterator, Callable, Iterable, Mapping, MutableMapping, Sequence +from collections.abc import ( + AsyncGenerator, + AsyncIterator, + Callable, + Coroutine, + Iterable, + Mapping, + MutableMapping, + Sequence, +) from dataclasses import dataclass from datetime import datetime from functools import reduce @@ -270,6 +279,11 @@ async def set_env_variables_in_header(custom_headers: dict | None) -> dict | Non return headers +class _AdapterCompletionRouter(Protocol): + @property + def aadapter_completion(self) -> Callable[..., Coroutine[object, object, object]]: ... + + async def chat_completion_pass_through_endpoint( fastapi_response: Response, request: Request, @@ -360,7 +374,11 @@ async def chat_completion_pass_through_endpoint( and data["model"] is not None and llm_router.registry().resolve(data["model"], kind="recognized") ): - llm_response = asyncio.create_task(llm_router.aadapter_completion(**data)) + llm_response = asyncio.create_task( + require_attributes(llm_router, _AdapterCompletionRouter, "aadapter_completion").aadapter_completion( + **data + ) + ) elif ( llm_router is not None and data["model"] not in router_model_names @@ -369,11 +387,19 @@ async def chat_completion_pass_through_endpoint( or len(llm_router.registry().patterns().patterns) > 0 ) ): # check for wildcard routes or default deployment before checking deployment_names - llm_response = asyncio.create_task(llm_router.aadapter_completion(**data)) + llm_response = asyncio.create_task( + require_attributes(llm_router, _AdapterCompletionRouter, "aadapter_completion").aadapter_completion( + **data + ) + ) elif llm_router is not None and data["model"] in llm_router.registry().names( kind="deployment_names" ): # model in router deployments, calling a specific deployment on the router (lowest priority) - llm_response = asyncio.create_task(llm_router.aadapter_completion(**data, specific_deployment=True)) + llm_response = asyncio.create_task( + require_attributes(llm_router, _AdapterCompletionRouter, "aadapter_completion").aadapter_completion( + **data, specific_deployment=True + ) + ) elif user_model is not None: # `litellm --model ` llm_response = asyncio.create_task(litellm.aadapter_completion(**data)) else: diff --git a/litellm/rag/ingestion/base_ingestion.py b/litellm/rag/ingestion/base_ingestion.py index 8b2cf6e0a4e..3742d06557f 100644 --- a/litellm/rag/ingestion/base_ingestion.py +++ b/litellm/rag/ingestion/base_ingestion.py @@ -14,12 +14,13 @@ from __future__ import annotations import base64 from abc import ABC, abstractmethod -from typing import TYPE_CHECKING, Any, Final, cast +from typing import TYPE_CHECKING, Any, Final, Protocol, cast import litellm from litellm._logging import verbose_logger from litellm._uuid import uuid4 from litellm.constants import DEFAULT_CHUNK_OVERLAP, DEFAULT_CHUNK_SIZE +from litellm.litellm_core_utils.core_helpers import has_attributes, require_attributes from litellm.litellm_core_utils.url_utils import async_safe_get from litellm.llms.custom_httpx.http_handler import ( get_async_httpx_client, @@ -31,6 +32,8 @@ from litellm.rag.text_splitters import RecursiveCharacterTextSplitter from litellm.types.rag import RAGIngestOptions, RAGIngestResponse if TYPE_CHECKING: + from collections.abc import Awaitable, Callable, Iterable + from litellm import Router @@ -38,6 +41,21 @@ class RAGIngestBadRequestError(ValueError): status_code: Final = 400 +class _OcrRouter(Protocol): + @property + def aocr(self) -> Callable[..., Awaitable[object]]: ... + + +class _OcrPages(Protocol): + @property + def pages(self) -> Iterable[object]: ... + + +class _OcrPage(Protocol): + @property + def markdown(self) -> str: ... + + class BaseRAGIngestion(ABC): """ Base class for RAG ingestion. @@ -168,7 +186,7 @@ class BaseRAGIngestion(ABC): # Use router if available if self.router is not None: - ocr_response = await self.router.aocr( + ocr_response = await require_attributes(self.router, _OcrRouter, "aocr").aocr( model=ocr_model, document={"type": doc_type, url_key: data_url}, ) @@ -179,8 +197,10 @@ class BaseRAGIngestion(ABC): ) # Extract text from pages - if hasattr(ocr_response, "pages") and ocr_response.pages: - return "\n\n".join(page.markdown for page in ocr_response.pages if hasattr(page, "markdown")) + if has_attributes(ocr_response, _OcrPages, "pages") and ocr_response.pages: + return "\n\n".join( + page.markdown for page in ocr_response.pages if has_attributes(page, _OcrPage, "markdown") + ) return None diff --git a/litellm/videos/main.py b/litellm/videos/main.py index 66eb10e30db..f2681b03cb2 100644 --- a/litellm/videos/main.py +++ b/litellm/videos/main.py @@ -6,10 +6,13 @@ from functools import partial from typing import Final, Literal, overload from httpx._types import FileContent +from pydantic import TypeAdapter +from typing_extensions import TypedDict import litellm from litellm.constants import DEFAULT_VIDEO_ENDPOINT_MODEL from litellm.constants import request_timeout as DEFAULT_REQUEST_TIMEOUT +from litellm.litellm_core_utils.core_helpers import iter_items, keyword_arguments from litellm.litellm_core_utils.get_llm_provider_logic import get_llm_provider from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj from litellm.llms.base_llm.videos.transformation import BaseVideoConfig @@ -26,6 +29,16 @@ from litellm.types.videos.utils import decode_video_id_with_provider from litellm.utils import ProviderConfigManager, client from litellm.videos.utils import VideoGenerationRequestUtils +_DECODED_JSON: Final = TypeAdapter(object) + + +class _VideoObjectFields(TypedDict, total=False, extra_items=object): + pass + + +class _CharacterObjectFields(TypedDict, total=False, extra_items=object): + pass + def _unsupported_video_call( call_name: str, custom_llm_provider: str, model: str | None = None @@ -204,9 +217,11 @@ def video_generation( mock_response = kwargs.get("mock_response", None) if mock_response is not None: if isinstance(mock_response, str): - mock_response = json.loads(mock_response) + mock_response = _DECODED_JSON.validate_python(json.loads(mock_response)) - response: Final = VideoObject(**mock_response) + response: Final = VideoObject.model_validate( + _VideoObjectFields(**keyword_arguments(mock_response, VideoObject)) + ) return response # get llm provider logic @@ -593,9 +608,11 @@ def video_remix( mock_response = kwargs.get("mock_response", None) if mock_response is not None: if isinstance(mock_response, str): - mock_response = json.loads(mock_response) + mock_response = _DECODED_JSON.validate_python(json.loads(mock_response)) - response: Final = VideoObject(**mock_response) + response: Final = VideoObject.model_validate( + _VideoObjectFields(**keyword_arguments(mock_response, VideoObject)) + ) return response # Try to decode provider from video_id if not explicitly provided @@ -808,8 +825,11 @@ def video_list( mock_response = kwargs.get("mock_response", None) if mock_response is not None: if isinstance(mock_response, str): - mock_response = json.loads(mock_response) - return [VideoObject(**item) for item in mock_response] + mock_response = _DECODED_JSON.validate_python(json.loads(mock_response)) + return [ + VideoObject.model_validate(_VideoObjectFields(**keyword_arguments(item, VideoObject))) + for item in iter_items(mock_response) + ] # Ensure custom_llm_provider is not None - default to openai if not provided if custom_llm_provider is None: @@ -1026,9 +1046,11 @@ def video_status( mock_response = kwargs.get("mock_response", None) if mock_response is not None: if isinstance(mock_response, str): - mock_response = json.loads(mock_response) + mock_response = _DECODED_JSON.validate_python(json.loads(mock_response)) - response: Final = VideoObject(**mock_response) + response: Final = VideoObject.model_validate( + _VideoObjectFields(**keyword_arguments(mock_response, VideoObject)) + ) return response # Try to decode provider from video_id if not explicitly provided @@ -1173,8 +1195,10 @@ def video_create_character( mock_response = kwargs.get("mock_response", None) if mock_response is not None: if isinstance(mock_response, str): - mock_response = json.loads(mock_response) - return CharacterObject(**mock_response) + mock_response = _DECODED_JSON.validate_python(json.loads(mock_response)) + return CharacterObject.model_validate( + _CharacterObjectFields(**keyword_arguments(mock_response, CharacterObject)) + ) if custom_llm_provider is None: custom_llm_provider = "openai" @@ -1298,8 +1322,10 @@ def video_get_character( mock_response = kwargs.get("mock_response", None) if mock_response is not None: if isinstance(mock_response, str): - mock_response = json.loads(mock_response) - return CharacterObject(**mock_response) + mock_response = _DECODED_JSON.validate_python(json.loads(mock_response)) + return CharacterObject.model_validate( + _CharacterObjectFields(**keyword_arguments(mock_response, CharacterObject)) + ) if custom_llm_provider is None: custom_llm_provider = "openai" @@ -1430,8 +1456,8 @@ def video_edit( mock_response = kwargs.get("mock_response", None) if mock_response is not None: if isinstance(mock_response, str): - mock_response = json.loads(mock_response) - return VideoObject(**mock_response) + mock_response = _DECODED_JSON.validate_python(json.loads(mock_response)) + return VideoObject.model_validate(_VideoObjectFields(**keyword_arguments(mock_response, VideoObject))) if custom_llm_provider is None: decoded: Final = decode_video_id_with_provider(video_id) @@ -1564,8 +1590,8 @@ def video_extension( mock_response = kwargs.get("mock_response", None) if mock_response is not None: if isinstance(mock_response, str): - mock_response = json.loads(mock_response) - return VideoObject(**mock_response) + mock_response = _DECODED_JSON.validate_python(json.loads(mock_response)) + return VideoObject.model_validate(_VideoObjectFields(**keyword_arguments(mock_response, VideoObject))) if custom_llm_provider is None: decoded: Final = decode_video_id_with_provider(video_id)