scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-14 23:21:35 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 102

chore(typing): clear basedpyright Any errors in streaming/proxy/responses modules

Browse source 
Replace reportAny/reportExplicitAny seams with real types (TypedDicts,
Protocols, and concrete request/response types) in the eight production
files with the highest combined error counts: streaming_iterator.py,
streaming_handler.py, managed_id_rewriter.py, model_management_endpoints.py,
streaming_chunk_builder_utils.py, sampling_handler.py, transformation.py,
and pass_through_endpoints.py.

reportAny: 19435 -> 19133 (-302)
reportExplicitAny: 6518 -> 6386 (-132)
No other basedpyright rule regressed repo-wide; total errors 148372 -> 147787.
This commit is contained in:
mateo-berri 2026-08-04 01:27:02 +00:00
parent 491eda319c
commit 029311a895
No known key found for this signature in database
11 changed files with 440 additions and 297 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

16
basedpyright-code-budget.json
View file

@ -1,6 +1,6 @@
{
"reportAny": {
"limit": 29813
"limit": 28907
},
"reportArgumentType": {
"limit": 2645
@ -24,7 +24,7 @@
"limit": 42
},
"reportExplicitAny": {
"limit": 9473
"limit": 9077
},
"reportFunctionMemberAccess": {
"limit": 11
@ -57,7 +57,7 @@
"limit": 5855
},
"reportMissingTypeArgument": {
"limit": 15852
"limit": 15825
},
"reportMissingTypeStubs": {
"limit": 41
@ -90,7 +90,7 @@
"limit": 12
},
"reportReturnType": {
"limit": 219
"limit": 216
},
"reportTypedDictNotRequiredAccess": {
"limit": 27
@ -99,19 +99,19 @@
"limit": 0
},
"reportUnknownArgumentType": {
"limit": 45324
"limit": 45150
},
"reportUnknownLambdaType": {
"limit": 113
},
"reportUnknownMemberType": {
"limit": 40452
"limit": 40386
},
"reportUnknownParameterType": {
"limit": 20309
"limit": 20282
},
"reportUnknownVariableType": {
"limit": 31978
"limit": 31822
},
"reportUnnecessaryCast": {
"limit": 177

149
litellm/litellm_core_utils/streaming_chunk_builder_utils.py
View file

@ -1,6 +1,6 @@
import base64
import time
from typing import TYPE_CHECKING, Any, Union, cast
from typing import TYPE_CHECKING, Any, TypedDict, Union, cast
from litellm._logging import verbose_logger
from litellm.types.llms.openai import (
@ -25,6 +25,7 @@ from litellm.types.utils import (
from litellm.utils import print_verbose, token_counter
if TYPE_CHECKING:
from litellm.litellm_core_utils.litellm_logging import Logging
from litellm.types.litellm_core_utils.streaming_chunk_builder_utils import (
UsagePerChunk,
)
@ -33,35 +34,46 @@ if TYPE_CHECKING:
ChatCompletionThinkingBlock,
)
_JsonDict = dict[str, Any]
_StreamChunk = _JsonDict | ModelResponse | ModelResponseStream
class _ToolCallAccumulator(TypedDict):
id: str | None
name: str | None
type: str | None
arguments: list[str]
provider_specific_fields: _JsonDict | None
class _UsageChunkFields(TypedDict):
prompt_tokens: int
completion_tokens: int
cache_creation_input_tokens: int | None
cache_read_input_tokens: int | None
completion_tokens_details: CompletionTokensDetails | None
prompt_tokens_details: PromptTokensDetailsWrapper | None
cost: float | None
class ChunkProcessor:
def __init__(self, chunks: list, messages: list | None = None):
def __init__(self, chunks: list["_StreamChunk"], messages: list | None = None):
self.chunks = self._sort_chunks(chunks)
self.messages = messages
self.first_chunk = chunks[0]
def _sort_chunks(self, chunks: list) -> list:
def _sort_chunks(self, chunks: list["_StreamChunk"]) -> list["_StreamChunk"]:
if not chunks:
return []
first_chunk = chunks[0]
first_hidden_params: dict[str, Any] = {}
if isinstance(first_chunk, dict):
candidate = first_chunk.get("_hidden_params", {})
if isinstance(candidate, dict):
first_hidden_params = candidate
else:
candidate = getattr(first_chunk, "_hidden_params", {})
if isinstance(candidate, dict):
first_hidden_params = candidate
candidate = first_chunk.get("_hidden_params", {})
first_hidden_params: _JsonDict = candidate if isinstance(candidate, dict) else {}
if first_hidden_params.get("created_at"):
def _created_at(chunk: Any) -> int | float:
if isinstance(chunk, dict):
params = chunk.get("_hidden_params", {})
else:
params = getattr(chunk, "_hidden_params", {})
def _created_at(chunk: "_StreamChunk") -> int | float:
params = chunk.get("_hidden_params", {})
if isinstance(params, dict):
return cast(int | float, params.get("created_at", float("inf")))
return float("inf")
@ -70,25 +82,26 @@ class ChunkProcessor:
return chunks
def update_model_response_with_hidden_params(
self, model_response: ModelResponse, chunk: dict[str, Any] | None = None
self, model_response: ModelResponse, chunk: "_StreamChunk | None" = None
) -> ModelResponse:
if chunk is None:
return model_response
# set hidden params from chunk to model_response
if model_response is not None and hasattr(model_response, "_hidden_params"):
model_response._hidden_params = chunk.get("_hidden_params", {})
hidden_params = chunk.get("_hidden_params", {})
model_response._hidden_params = hidden_params if isinstance(hidden_params, dict) else {}
return model_response
@staticmethod
def apply_provider_assembled_streaming_metadata(
response: ModelResponse,
chunks: list[Any],
logging_obj: Any | None = None,
chunks: list["_StreamChunk"],
logging_obj: "Logging | None" = None,
) -> None:
if not chunks:
return
model = getattr(response, "model", None)
model = response.model
if not model:
return
@ -126,7 +139,7 @@ class ChunkProcessor:
)
@staticmethod
def _get_chunk_id(chunks: list[dict[str, Any]]) -> str:
def _get_chunk_id(chunks: list[_JsonDict]) -> str:
"""
Chunks:
[{"id": ""}, {"id": "1"}, {"id": "1"}]
@ -137,7 +150,7 @@ class ChunkProcessor:
return ""
@staticmethod
def _get_model_from_chunks(chunks: list[dict[str, Any]], first_chunk_model: str) -> str:
def _get_model_from_chunks(chunks: list[_JsonDict], first_chunk_model: str) -> str:
"""
Get the actual model from chunks, preferring a model that differs from the first chunk.
@ -153,7 +166,7 @@ class ChunkProcessor:
# Fall back to first chunk's model if no different model found
return first_chunk_model
def build_base_response(self, chunks: list[dict[str, Any]]) -> ModelResponse:
def build_base_response(self, chunks: list[_JsonDict]) -> ModelResponse:
chunk = self.first_chunk
id = ChunkProcessor._get_chunk_id(chunks)
object = chunk["object"]
@ -202,9 +215,9 @@ class ChunkProcessor:
response = self.update_model_response_with_hidden_params(model_response=response, chunk=chunk)
return response
def get_combined_tool_content(self, tool_call_chunks: list[dict[str, Any]]) -> list[ChatCompletionMessageToolCall]:
def get_combined_tool_content(self, tool_call_chunks: list[_JsonDict]) -> list[ChatCompletionMessageToolCall]:
tool_calls_list: list[ChatCompletionMessageToolCall] = []
tool_call_map: dict[int, dict[str, Any]] = {} # Map to store tool calls by index
tool_call_map: dict[int, _ToolCallAccumulator] = {} # Map to store tool calls by index
for chunk in tool_call_chunks:
choices = chunk["choices"]
@ -291,10 +304,12 @@ class ChunkProcessor:
if provider_fields:
# Merge provider_specific_fields if multiple chunks have them
if tool_call_map[index]["provider_specific_fields"] is None:
tool_call_map[index]["provider_specific_fields"] = {}
accumulated_provider_fields = tool_call_map[index]["provider_specific_fields"]
if accumulated_provider_fields is None:
accumulated_provider_fields = {}
tool_call_map[index]["provider_specific_fields"] = accumulated_provider_fields
if isinstance(provider_fields, dict):
tool_call_map[index]["provider_specific_fields"].update(provider_fields)
accumulated_provider_fields.update(provider_fields)
# Convert the map to a list of tool calls
for index in sorted(tool_call_map.keys()):
@ -308,23 +323,26 @@ class ChunkProcessor:
name=tool_call_data["name"],
)
# Prepare params for ChatCompletionMessageToolCall
tool_call_params = {
"id": tool_call_data["id"],
"function": function,
"type": tool_call_data["type"] or "function",
}
# Add provider_specific_fields if present (for thought signatures in Gemini 3)
if tool_call_data.get("provider_specific_fields"):
tool_call_params["provider_specific_fields"] = tool_call_data["provider_specific_fields"]
tool_call = ChatCompletionMessageToolCall(
id=tool_call_data["id"],
function=function,
type=tool_call_data["type"] or "function",
provider_specific_fields=tool_call_data["provider_specific_fields"],
)
else:
tool_call = ChatCompletionMessageToolCall(
id=tool_call_data["id"],
function=function,
type=tool_call_data["type"] or "function",
)
tool_call = ChatCompletionMessageToolCall(**tool_call_params)
tool_calls_list.append(tool_call)
return tool_calls_list
def get_combined_function_call_content(self, function_call_chunks: list[dict[str, Any]]) -> FunctionCall:
def get_combined_function_call_content(self, function_call_chunks: list[_JsonDict]) -> FunctionCall:
argument_list = []
delta = function_call_chunks[0]["choices"][0]["delta"]
function_call = delta.get("function_call", "")
@ -350,7 +368,7 @@ class ChunkProcessor:
)
def get_combined_content(
self, chunks: list[dict[str, Any]], delta_key: str = "content"
self, chunks: list[_JsonDict], delta_key: str = "content"
) -> ChatCompletionAssistantContentValue:
content_list: list[str] = []
for chunk in chunks:
@ -369,7 +387,7 @@ class ChunkProcessor:
return combined_content
def get_combined_thinking_content(
self, chunks: list[dict[str, Any]]
self, chunks: list[_JsonDict]
) -> list[Union["ChatCompletionThinkingBlock", "ChatCompletionRedactedThinkingBlock"]] | None:
from litellm.types.llms.openai import (
ChatCompletionRedactedThinkingBlock,
@ -426,10 +444,10 @@ class ChunkProcessor:
return thinking_blocks
return None
def get_combined_reasoning_content(self, chunks: list[dict[str, Any]]) -> ChatCompletionAssistantContentValue:
def get_combined_reasoning_content(self, chunks: list[_JsonDict]) -> ChatCompletionAssistantContentValue:
return self.get_combined_content(chunks, delta_key="reasoning_content")
def get_combined_audio_content(self, chunks: list[dict[str, Any]]) -> ChatCompletionAudioResponse:
def get_combined_audio_content(self, chunks: list[_JsonDict]) -> ChatCompletionAudioResponse:
base64_data_list: list[str] = []
transcript_list: list[str] = []
expires_at: int | None = None
@ -459,7 +477,7 @@ class ChunkProcessor:
id=id,
)
def _usage_chunk_calculation_helper(self, usage_chunk: Usage) -> dict:
def _usage_chunk_calculation_helper(self, usage_chunk: Usage) -> _UsageChunkFields:
prompt_tokens = 0
completion_tokens = 0
## anthropic prompt caching information ##
@ -517,8 +535,8 @@ class ChunkProcessor:
return reasoning_tokens
@staticmethod
def _extract_usage_chunk(chunk: dict[str, Any] | ModelResponse | ModelResponseStream) -> Usage | None:
usage_chunk: Usage | dict[str, Any] | None = None
def _extract_usage_chunk(chunk: _StreamChunk) -> Usage | None:
usage_chunk: Usage | _JsonDict | None = None
if hasattr(chunk, "usage") and chunk.usage is not None:
usage_chunk = chunk.usage
elif "usage" in chunk:
@ -534,7 +552,7 @@ class ChunkProcessor:
def _calculate_usage_per_chunk(
self,
chunks: list[dict[str, Any] | ModelResponse],
chunks: list[_StreamChunk],
) -> "UsagePerChunk":
from litellm.types.litellm_core_utils.streaming_chunk_builder_utils import (
UsagePerChunk,
@ -575,9 +593,9 @@ class ChunkProcessor:
if usage_chunk is not None:
usage_chunk_dict = self._usage_chunk_calculation_helper(usage_chunk)
if usage_chunk_dict["prompt_tokens"] is not None and usage_chunk_dict["prompt_tokens"] > 0:
if usage_chunk_dict["prompt_tokens"] > 0:
prompt_tokens = usage_chunk_dict["prompt_tokens"]
if usage_chunk_dict["completion_tokens"] is not None and usage_chunk_dict["completion_tokens"] > 0:
if usage_chunk_dict["completion_tokens"] > 0:
completion_tokens = usage_chunk_dict["completion_tokens"]
completion_usage_updates += 1
if usage_chunk_dict["cache_creation_input_tokens"] is not None and (
@ -601,24 +619,11 @@ class ChunkProcessor:
server_tool_use = usage_chunk.server_tool_use
else:
server_tool_use = ServerToolUse.model_validate(usage_chunk.server_tool_use)
if (
usage_chunk_dict["prompt_tokens_details"] is not None
and getattr(
usage_chunk_dict["prompt_tokens_details"],
"web_search_requests",
None,
)
is not None
):
web_search_requests = getattr(
usage_chunk_dict["prompt_tokens_details"],
"web_search_requests",
)
prompt_tokens_details = cast(
PromptTokensDetailsWrapper | None,
usage_chunk_dict["prompt_tokens_details"],
)
prompt_tokens_details = usage_chunk_dict["prompt_tokens_details"]
if prompt_tokens_details is not None:
candidate_web_search_requests = getattr(prompt_tokens_details, "web_search_requests", None)
if candidate_web_search_requests is not None:
web_search_requests = candidate_web_search_requests
cache_creation_token_details = self._capture_cache_creation_token_details(
prompt_tokens_details, cache_creation_token_details
@ -679,7 +684,7 @@ class ChunkProcessor:
@staticmethod
def _reset_anthropic_cursor_completion_tokens(
chunks: list[dict[str, Any] | ModelResponse],
chunks: list[_StreamChunk],
completion_tokens: int,
completion_usage_updates: int,
) -> int:
@ -718,7 +723,7 @@ class ChunkProcessor:
def calculate_usage(
self,
chunks: list[dict[str, Any] | ModelResponse],
chunks: list[_StreamChunk],
model: str,
completion_output: str,
messages: list | None = None,
@ -772,8 +777,8 @@ class ChunkProcessor:
setattr(returned_usage, "cache_read_input_tokens", cache_read_input_tokens) # for anthropic
if completion_tokens_details is not None:
if isinstance(completion_tokens_details, CompletionTokensDetails):
returned_usage.completion_tokens_details = CompletionTokensDetailsWrapper(
**completion_tokens_details.model_dump()
returned_usage.completion_tokens_details = CompletionTokensDetailsWrapper.model_validate(
completion_tokens_details.model_dump()
)
else:
returned_usage.completion_tokens_details = completion_tokens_details

41
litellm/litellm_core_utils/streaming_handler.py
View file

@ -11,13 +11,14 @@ from dataclasses import dataclass
from typing import (
Any,
NoReturn,
Union,
TypedDict,
cast,
)
import anyio
import httpx
from pydantic import BaseModel
from typing_extensions import NotRequired
import litellm
from litellm import verbose_logger
@ -98,10 +99,23 @@ class _ProviderChunkParsed:
@dataclass(frozen=True, slots=True)
class _ProviderChunkEarlyReturn:
value: Any
value: ModelResponseStream | None
_ProviderChunkResult = Union[_ProviderChunkParsed, _ProviderChunkEarlyReturn]
_ProviderChunkResult = _ProviderChunkParsed | _ProviderChunkEarlyReturn
class _AzureChunkDelta(TypedDict, total=False):
content: str
class _AzureChunkChoice(TypedDict):
delta: _AzureChunkDelta | None
finish_reason: NotRequired[str]
class _AzureChunkPayload(TypedDict):
choices: list[_AzureChunkChoice]
class CustomStreamWrapper:
@ -109,7 +123,7 @@ class CustomStreamWrapper:
self,
completion_stream,
model,
logging_obj: Any,
logging_obj: LiteLLMLoggingObject,
custom_llm_provider: str | None = None,
stream_options=None,
make_call: Callable | None = None,
@ -124,9 +138,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", {}))
)
_litellm_params_raw: dict[str, object] = self.logging_obj.model_call_details.get("litellm_params", {})
litellm_params: GenericLiteLLMParams = GenericLiteLLMParams.model_validate(dict(**_litellm_params_raw))
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 +164,7 @@ class CustomStreamWrapper:
_api_base = get_api_base(
model=model or "",
optional_params=self.logging_obj.model_call_details.get("litellm_params", {}),
optional_params=_litellm_params_raw,
)
self._hidden_params = {
@ -462,14 +475,16 @@ class CustomStreamWrapper:
"finish_reason": finish_reason,
}
elif chunk.startswith("data:"):
data_json = json.loads(chunk[5:]) # chunk.startswith("data:"):
data_json: _AzureChunkPayload = json.loads(chunk[5:]) # chunk.startswith("data:"):
try:
if len(data_json["choices"]) > 0:
delta = data_json["choices"][0]["delta"]
choice = data_json["choices"][0]
delta = choice["delta"]
text = "" if delta is None else delta.get("content", "")
if data_json["choices"][0].get("finish_reason", None):
_choice_finish_reason = choice.get("finish_reason", None)
if _choice_finish_reason:
is_finished = True
finish_reason = data_json["choices"][0]["finish_reason"]
finish_reason = _choice_finish_reason
print_verbose(f"text: {text}; is_finished: {is_finished}; finish_reason: {finish_reason}")
return {
"text": text,
@ -1656,7 +1671,7 @@ class CustomStreamWrapper:
else:
asyncio.run(self.logging_obj.async_success_handler(processed_chunk, None, None, cache_hit))
## SYNC LOGGING — only for sync SDK entrypoints; async proxy paths export via async_success_handler
litellm_params = self.logging_obj.model_call_details.get("litellm_params", {})
litellm_params: dict[str, object] = self.logging_obj.model_call_details.get("litellm_params", {})
if self.logging_obj._is_sync_litellm_request(litellm_params):
self.logging_obj.success_handler(processed_chunk, None, None, cache_hit)

73
litellm/proxy/_experimental/mcp_server/sampling_handler.py
View file

@ -271,7 +271,7 @@ def _select_model_by_priority(
def _convert_mcp_content_to_openai(
content: "SamplingMessageContentBlock | Sequence[SamplingMessageContentBlock]",
content: "SamplingMessageContentBlock | list[SamplingMessageContentBlock]",
) -> "str | dict[str, object] | list[dict[str, object]]":
"""
Convert MCP SamplingMessage content to OpenAI message content format.
@ -296,7 +296,7 @@ def _convert_mcp_content_to_openai(
def _convert_single_content(
content: Any,
content: "SamplingMessageContentBlock",
) -> "dict[str, object] | list[dict[str, object]]":
"""Convert a single MCP content item to OpenAI format.
@ -308,19 +308,14 @@ def _convert_single_content(
"""
import json
content_type = getattr(content, "type", None)
if content_type == "text":
if content.type == "text":
return {"type": "text", "text": content.text}
elif content_type == "image":
data = getattr(content, "data", "")
mime_type = getattr(content, "mimeType", "image/png")
elif content.type == "image":
return {
"type": "image_url",
"image_url": {"url": f"data:{mime_type};base64,{data}"},
"image_url": {"url": f"data:{content.mimeType};base64,{content.data}"},
}
elif content_type == "audio":
data = getattr(content, "data", "")
mime_type = getattr(content, "mimeType", "audio/wav")
elif content.type == "audio":
# Map MIME type to OpenAI audio format
format_map = {
"audio/wav": "wav",
@ -329,36 +324,33 @@ def _convert_single_content(
"audio/flac": "flac",
"audio/ogg": "ogg",
}
audio_format = format_map.get(mime_type, "wav")
audio_format = format_map.get(content.mimeType, "wav")
return {
"type": "input_audio",
"input_audio": {"data": data, "format": audio_format},
"input_audio": {"data": content.data, "format": audio_format},
}
elif content_type == "tool_use":
elif content.type == "tool_use":
# ToolUseContent → proper OpenAI function-call representation.
# The ``_marker_type`` key lets the message-level converter
# hoist this into the ``tool_calls`` array on the assistant
# message instead of embedding it inline as a content part.
return {
"_marker_type": "tool_use",
"id": getattr(content, "id", f"call_{id(content)}"),
"id": content.id,
"type": "function",
"function": {
"name": getattr(content, "name", ""),
"arguments": json.dumps(getattr(content, "input", {}), default=str),
"name": content.name,
"arguments": json.dumps(content.input, default=str),
},
}
elif content_type == "tool_result":
elif content.type == "tool_result":
# ToolResultContent → proper OpenAI tool-role message.
# Marked so the message-level converter can emit it as a
# separate ``{"role": "tool", ...}`` message.
tool_use_id = getattr(content, "toolUseId", "")
nested_content: Sequence[ContentBlock] = getattr(content, "content", [])
if isinstance(nested_content, list):
text_parts = [getattr(c, "text", str(c)) for c in nested_content if getattr(c, "type", None) == "text"]
result_text = "\n".join(text_parts) if text_parts else ""
else:
result_text = str(nested_content)
tool_use_id = content.toolUseId
nested_content: Sequence[ContentBlock] = content.content
text_parts = [getattr(c, "text", str(c)) for c in nested_content if getattr(c, "type", None) == "text"]
result_text = "\n".join(text_parts) if text_parts else ""
return {
"_marker_type": "tool_result",
"role": "tool",
@ -581,12 +573,28 @@ def _convert_mcp_tool_choice_to_openai(
return "auto"
class _SamplingToolCallFunction(Protocol):
@property
def name(self) -> str: ...
@property
def arguments(self) -> str: ...
class _SamplingToolCall(Protocol):
@property
def id(self) -> str: ...
@property
def function(self) -> _SamplingToolCallFunction: ...
class _SamplingResponseMessage(Protocol):
@property
def content(self) -> str | None: ...
@property
def tool_calls(self) -> Sequence[object] | None: ...
def tool_calls(self) -> Sequence[_SamplingToolCall] | None: ...
class _SamplingResponseChoice(Protocol):
@ -641,7 +649,7 @@ def _convert_openai_response_to_mcp_result(
stop_reason = "endTurn"
actual_model: str = getattr(response, "model", model_name) or model_name
# Check if response has tool calls
tool_calls = getattr(message, "tool_calls", None)
tool_calls = message.tool_calls
if tool_calls:
# Build ToolUseContent items
content_parts: list[SamplingMessageContentBlock] = []
@ -652,12 +660,11 @@ def _convert_openai_response_to_mcp_result(
for tc in tool_calls:
import json
tool_input = tc.function.arguments
if isinstance(tool_input, str):
try:
tool_input = json.loads(tool_input)
except (json.JSONDecodeError, TypeError):
tool_input = {"raw": tool_input}
tool_input: dict[str, object]
try:
tool_input = json.loads(tc.function.arguments)
except (json.JSONDecodeError, TypeError):
tool_input = {"raw": tc.function.arguments}
content_parts.append(
ToolUseContent(
type="tool_use",

72
litellm/proxy/management_endpoints/model_management_endpoints.py
View file

@ -14,7 +14,7 @@ import asyncio
import datetime
import json
from collections.abc import Mapping, Sequence
from typing import Any, Literal, cast
from typing import Literal, Protocol, cast
from fastapi import APIRouter, Depends, Header, HTTPException, Request, status
from pydantic import BaseModel, ConfigDict, Field
@ -322,7 +322,7 @@ async def patch_model(
update_data["updated_at"] = cast(str, get_utc_datetime())
# Perform partial update
updated_model = await ModelRepository(prisma_client).table.update(
updated_model: LiteLLM_ProxyModelTable = await ModelRepository(prisma_client).table.update(
where={"model_id": model_id},
data=update_data,
)
@ -425,7 +425,7 @@ async def _set_model_blocked_status(
param=None,
)
updated_model = await ModelRepository(prisma_client).table.update(
updated_model: LiteLLM_ProxyModelTable = await ModelRepository(prisma_client).table.update(
where={"model_id": data.model_id},
data={
"blocked": blocked,
@ -444,9 +444,7 @@ async def _set_model_blocked_status(
user_api_key_dict=user_api_key_dict,
table_name=LitellmTableNames.PROXY_MODEL_TABLE_NAME,
before_value=db_model.model_dump_json(exclude_none=True),
after_value=(
updated_model.model_dump_json(exclude_none=True) if isinstance(updated_model, BaseModel) else None
),
after_value=updated_model.model_dump_json(exclude_none=True),
litellm_changed_by=litellm_changed_by,
litellm_proxy_admin_name=litellm_proxy_admin_name,
)
@ -564,6 +562,7 @@ async def _add_model_to_db(
}
if model_params.model_info.id is not None:
_data["model_id"] = model_params.model_info.id
model_response: LiteLLM_ProxyModelTable
if should_create_model_in_db:
model_response = await ModelRepository(prisma_client).table.create(
data=_data # type: ignore
@ -756,9 +755,19 @@ async def _setup_new_team_model_assignment(
)
class _ProxyModelDeploymentRow(Protocol):
model_id: str
model_name: str
model_info: object
class _ProxyModelTableReader(Protocol):
async def find_many(self, where: Mapping[str, object]) -> Sequence[_ProxyModelDeploymentRow]: ...
async def _get_team_deployments(
team_id: str, prisma_client: PrismaClient, table: Any | None = None
) -> list[LiteLLM_ProxyModelTable]:
team_id: str, prisma_client: PrismaClient, table: _ProxyModelTableReader | None = None
) -> list[_ProxyModelDeploymentRow]:
"""
Fetch all deployments for a given team_id from the database.
@ -794,7 +803,7 @@ async def _get_team_deployments(
async def delete_team_models(
team_ids: list[str],
prisma_client: PrismaClient,
llm_router: Any | None,
llm_router: Router | None,
) -> list[str]:
"""
Delete every BYOK model owned by the given teams, from the DB and the router.
@ -941,7 +950,7 @@ async def _update_existing_team_model_assignment(
"""
def _get_team_public_model_name(
model_info: dict | str | None,
model_info: object,
) -> str | None:
parsed = model_info_as_mapping(model_info)
if parsed is None:
@ -1050,7 +1059,7 @@ class ModelManagementAuthChecks:
detail={"error": CommonProxyErrors.not_premium_user.value},
)
_existing_team_row = await TeamRepository(prisma_client).table.find_unique(
_existing_team_row: LiteLLM_TeamTable | None = await TeamRepository(prisma_client).table.find_unique(
where={"team_id": model_params.model_info.team_id}
)
@ -1079,7 +1088,7 @@ class ModelManagementAuthChecks:
) -> Literal[True]:
## Check team model auth
if model_params.model_info is not None and model_params.model_info.team_id is not None:
team_obj_row = await TeamRepository(prisma_client).table.find_unique(
team_obj_row: LiteLLM_TeamTable | None = await TeamRepository(prisma_client).table.find_unique(
where={"team_id": model_params.model_info.team_id}
)
if team_obj_row is None:
@ -1157,7 +1166,9 @@ async def delete_model(
},
)
model_in_db = await ModelRepository(prisma_client).table.find_unique(where={"model_id": model_info.id})
model_in_db: LiteLLM_ProxyModelTable | None = await ModelRepository(prisma_client).table.find_unique(
where={"model_id": model_info.id}
)
if model_in_db is None:
raise HTTPException(
status_code=400,
@ -1180,7 +1191,9 @@ async def delete_model(
- store keys separately
"""
# encrypt litellm params #
result = await ModelRepository(prisma_client).table.delete(where={"model_id": model_info.id})
result: LiteLLM_ProxyModelTable | None = await ModelRepository(prisma_client).table.delete(
where={"model_id": model_info.id}
)
if result is None:
raise HTTPException(
@ -1241,6 +1254,22 @@ async def delete_model(
)
class _ModelAliasTeamRef(Protocol):
team_id: str
class _ModelAliasRow(Protocol):
id: int
model_aliases: dict[str, str]
team: _ModelAliasTeamRef | None
class _ModelAliasTableClient(Protocol):
async def find_many(self, include: Mapping[str, object]) -> Sequence[_ModelAliasRow]: ...
async def update(self, where: Mapping[str, object], data: Mapping[str, object]) -> object: ...
async def delete_team_model_alias(
public_model_name: str,
prisma_client: PrismaClient,
@ -1253,7 +1282,8 @@ async def delete_team_model_alias(
Returns:
- List of team id + model alias pairs that were removed
"""
team_model_aliases = await ModelTableRepository(prisma_client).table.find_many(include={"team": True})
table_client: _ModelAliasTableClient = ModelTableRepository(prisma_client).table
team_model_aliases = await table_client.find_many(include={"team": True})
tasks = []
removed_model_aliases = []
for team_model_alias in team_model_aliases:
@ -1266,7 +1296,7 @@ async def delete_team_model_alias(
removed_model_aliases.append((team_model_alias.team.team_id, key))
del model_aliases[key]
tasks.append(
ModelTableRepository(prisma_client).table.update(
table_client.update(
where={"id": id},
data={"model_aliases": json.dumps(model_aliases)},
)
@ -1543,7 +1573,7 @@ async def update_model(
"litellm_params": json.dumps(merged_dictionary), # type: ignore
"updated_by": user_api_key_dict.user_id or LITELLM_PROXY_ADMIN_NAME,
}
model_response = await ModelRepository(prisma_client).table.update(
model_response: LiteLLM_ProxyModelTable = await ModelRepository(prisma_client).table.update(
where={"model_id": _model_id},
data=_data, # type: ignore
)
@ -1563,11 +1593,7 @@ async def update_model(
if isinstance(_existing_litellm_params, BaseModel)
else None
),
after_value=(
model_response.model_dump_json(exclude_none=True)
if isinstance(model_response, BaseModel)
else None
),
after_value=model_response.model_dump_json(exclude_none=True),
litellm_changed_by=user_api_key_dict.user_id,
litellm_proxy_admin_name=LITELLM_PROXY_ADMIN_NAME,
)
@ -1787,7 +1813,7 @@ def model_info_as_mapping(model_info: object) -> Mapping[str, object] | None:
if not isinstance(model_info, str):
return None
try:
parsed = json.loads(model_info)
parsed: object = json.loads(model_info)
except (TypeError, ValueError):
return None
return parsed if isinstance(parsed, Mapping) else None

175
litellm/proxy/pass_through_endpoints/managed_id_rewriter.py
View file

@ -32,7 +32,8 @@ from __future__ import annotations
import json
import re
from typing import Any
from collections.abc import Sequence
from typing import TYPE_CHECKING, Any, Literal, Protocol, TypeAlias, TypedDict, Union
from urllib.parse import quote, unquote
from fastapi import HTTPException
@ -42,7 +43,7 @@ from litellm.llms.base_llm.managed_resources.isolation import (
build_owner_filter,
can_access_resource,
)
from litellm.proxy._types import UserAPIKeyAuth
from litellm.proxy._types import LiteLLM_ManagedFileTable, UserAPIKeyAuth
from litellm.repositories.table_repositories import (
ManagedFileRepository,
ManagedObjectRepository,
@ -51,6 +52,40 @@ from litellm.types.llms.openai import OpenAIFileObject
from .managed_id_codec import ManagedIdPayload, decode, is_managed, new_managed_id
if TYPE_CHECKING:
from prisma.models import LiteLLM_ManagedFileTable as PrismaManagedFileRow
from prisma.models import LiteLLM_ManagedObjectTable as PrismaManagedObjectRow
from litellm.proxy.utils import PrismaClient
JSONValue: TypeAlias = Union[str, int, float, bool, None, "dict[str, JSONValue]", "list[JSONValue]"]
class _ManagedFilesHook(Protocol):
"""Structural shape this module needs from the enterprise managed-files hook."""
async def get_unified_file_id(
self, file_id: str, litellm_parent_otel_span: None = None
) -> LiteLLM_ManagedFileTable | None: ...
async def store_unified_file_id(
self,
file_id: str,
file_object: OpenAIFileObject | None,
litellm_parent_otel_span: None,
model_mappings: dict[str, str],
user_api_key_dict: UserAPIKeyAuth,
) -> None: ...
class _PassthroughListPage(TypedDict):
object: Literal["list"]
data: list[dict[str, JSONValue]]
first_id: str | None
last_id: str | None
has_more: bool
# ---------------------------------------------------------------------------
# Field map
# ---------------------------------------------------------------------------
@ -263,8 +298,8 @@ async def _resolve_one(
managed_id: str,
provider: str,
user_api_key_dict: UserAPIKeyAuth,
prisma_client: Any,
managed_files_hook: Any,
prisma_client: PrismaClient | None,
managed_files_hook: _ManagedFilesHook | None,
) -> str:
"""
Resolve a single value that may be a passthrough managed ID.
@ -322,7 +357,7 @@ async def _resolve_one(
)
if not found and prisma_client is not None:
try:
db_row = await ManagedFileRepository(prisma_client).table.find_first(
db_row: PrismaManagedFileRow | None = await ManagedFileRepository(prisma_client).table.find_first(
where={"unified_file_id": managed_id}
)
if db_row is not None:
@ -338,7 +373,7 @@ async def _resolve_one(
# Object table (batches, responses)
if prisma_client is not None:
try:
obj_row = await ManagedObjectRepository(prisma_client).table.find_first(
obj_row: PrismaManagedObjectRow | None = await ManagedObjectRepository(prisma_client).table.find_first(
where={"unified_object_id": managed_id}
)
if obj_row is not None:
@ -372,7 +407,7 @@ async def _guard_raw_provider_id(
raw_id: str,
provider: str,
user_api_key_dict: UserAPIKeyAuth,
prisma_client: Any,
prisma_client: PrismaClient | None,
budget: _RawIdGuardBudget | None = None,
) -> None:
"""Deny a raw provider ID that maps to a managed resource the caller does
@ -398,7 +433,7 @@ async def _guard_raw_provider_id(
# id and scope to the current provider in the application layer (same as
# _mint_or_reuse_file's dedup).
try:
candidates = await ManagedFileRepository(prisma_client).table.find_many(
candidates: list[PrismaManagedFileRow] = await ManagedFileRepository(prisma_client).table.find_many(
where={"flat_model_file_ids": {"has": raw_id}},
)
except Exception:
@ -419,7 +454,7 @@ async def _guard_raw_provider_id(
# Object rows store model_object_id as "passthrough:{provider}:{raw}", so
# the lookup is exact and already provider-scoped.
try:
existing = await ManagedObjectRepository(prisma_client).table.find_first(
existing: PrismaManagedObjectRow | None = await ManagedObjectRepository(prisma_client).table.find_first(
where={"model_object_id": f"passthrough:{provider}:{raw_id}"}
)
except Exception:
@ -434,7 +469,7 @@ async def _guard_raw_provider_id(
# ---------------------------------------------------------------------------
def _build_managed_file_object(snapshot: dict[str, Any] | None, managed_id: str) -> OpenAIFileObject | None:
def _build_managed_file_object(snapshot: dict[str, JSONValue] | None, managed_id: str) -> OpenAIFileObject | None:
"""Build an ``OpenAIFileObject`` (with the managed ID swapped in) from an
upstream file response so the DB-served list returns the same metadata as a
direct file GET. Returns ``None`` when no usable snapshot is available, in
@ -442,7 +477,7 @@ def _build_managed_file_object(snapshot: dict[str, Any] | None, managed_id: str)
if not snapshot:
return None
try:
return OpenAIFileObject(**{**snapshot, "id": managed_id})
return OpenAIFileObject.model_validate({**snapshot, "id": managed_id})
except Exception:
verbose_proxy_logger.debug(
"managed_id_rewriter: file object snapshot incomplete; storing file row without list metadata",
@ -455,9 +490,9 @@ async def _mint_or_reuse_file(
raw_id: str,
provider: str,
user_api_key_dict: UserAPIKeyAuth,
prisma_client: Any,
managed_files_hook: Any,
file_object_snapshot: dict[str, Any] | None = None,
prisma_client: PrismaClient | None,
managed_files_hook: _ManagedFilesHook | None,
file_object_snapshot: dict[str, JSONValue] | None = None,
is_create_route: bool = True,
) -> str:
"""Return an existing managed file ID or mint + store a new one."""
@ -479,7 +514,7 @@ async def _mint_or_reuse_file(
# reuse a stable row instead of minting duplicate rows on every call.
if prisma_client is not None:
try:
candidates = await ManagedFileRepository(prisma_client).table.find_many(
candidates: list[PrismaManagedFileRow] = await ManagedFileRepository(prisma_client).table.find_many(
where={"flat_model_file_ids": {"has": raw_id}},
order={"created_at": "asc"},
)
@ -551,9 +586,9 @@ async def _mint_or_reuse_object(
raw_id: str,
provider: str,
file_purpose: str,
body_snapshot: dict,
body_snapshot: dict[str, JSONValue],
user_api_key_dict: UserAPIKeyAuth,
prisma_client: Any,
prisma_client: PrismaClient | None,
is_create_route: bool,
) -> str:
"""Return an existing managed object ID (batch/response) or mint + store one."""
@ -569,7 +604,7 @@ async def _mint_or_reuse_object(
# f"{purpose}:{provider}:{raw_id}" for the same reason.
namespaced_model_object_id = f"passthrough:{provider}:{raw_id}"
async def _reuse_existing(existing: Any, refresh_snapshot: bool) -> str:
async def _reuse_existing(existing: PrismaManagedObjectRow, refresh_snapshot: bool) -> str:
"""Resolve an already-persisted namespaced row: enforce the access
check, optionally refresh the snapshot, and return its managed ID."""
if not can_access_resource(user_api_key_dict, existing.created_by, existing.team_id):
@ -618,7 +653,7 @@ async def _mint_or_reuse_object(
# Dedup: look up by the namespaced key — guaranteed unique per provider.
try:
existing = await ManagedObjectRepository(prisma_client).table.find_first(
existing: PrismaManagedObjectRow | None = await ManagedObjectRepository(prisma_client).table.find_first(
where={"model_object_id": namespaced_model_object_id}
)
except Exception:
@ -659,7 +694,7 @@ async def _mint_or_reuse_object(
# the winner's managed ID so both callers converge on one ID instead of
# the loser silently keeping the raw id.
try:
raced = await ManagedObjectRepository(prisma_client).table.find_first(
raced: PrismaManagedObjectRow | None = await ManagedObjectRepository(prisma_client).table.find_first(
where={"model_object_id": namespaced_model_object_id}
)
except Exception:
@ -681,11 +716,11 @@ async def rewrite_response_ids(
provider: str,
method: str,
route: str,
body: dict,
body: dict[str, JSONValue],
user_api_key_dict: UserAPIKeyAuth,
prisma_client: Any,
managed_files_hook: Any,
) -> dict:
prisma_client: PrismaClient | None,
managed_files_hook: _ManagedFilesHook | None,
) -> dict[str, JSONValue]:
"""
Mint managed IDs for raw provider values listed in
``BUILTIN_OUTPUT_ID_FIELD_MAP`` and swap them into *body*.
@ -795,7 +830,7 @@ def is_passthrough_list_route(provider: str, method: str, route: str) -> bool:
return (provider, canonical) in _LIST_ROUTE_TABLE
def _parse_file_object(file_object: Any) -> Any:
def _parse_file_object(file_object: JSONValue | None) -> JSONValue | None:
"""Prisma may return ``Json`` columns as either a parsed dict or the raw
JSON string (depending on driver / row source). Mirror the handling used
elsewhere (see ``openai_files_endpoints/common_utils.py``) so callers can
@ -803,13 +838,14 @@ def _parse_file_object(file_object: Any) -> Any:
"""
if isinstance(file_object, str):
try:
return json.loads(file_object)
parsed: JSONValue = json.loads(file_object)
except (TypeError, ValueError):
return None
return parsed
return file_object
def _empty_list_response() -> dict[str, Any]:
def _empty_list_response() -> _PassthroughListPage:
return {
"object": "list",
"data": [],
@ -819,7 +855,7 @@ def _empty_list_response() -> dict[str, Any]:
}
def _parse_list_limit(query_params: dict[str, Any] | None) -> tuple[int, int]:
def _parse_list_limit(query_params: dict[str, str] | None) -> tuple[int, int]:
params = query_params or {}
try:
raw_limit = int(params.get("limit", 20))
@ -830,17 +866,17 @@ def _parse_list_limit(query_params: dict[str, Any] | None) -> tuple[int, int]:
async def _build_list_where_with_cursor(
prisma_client: Any,
prisma_client: PrismaClient | None,
resource_kind: str,
provider: str,
owner_filter: dict[str, Any],
query_params: dict[str, Any] | None,
) -> tuple[dict[str, Any], str]:
owner_filter: dict[str, object],
query_params: dict[str, str] | None,
) -> tuple[dict[str, object], str]:
"""Return a Prisma ``where`` clause and fetch order for a list query."""
params = query_params or {}
after_id: str | None = params.get("after")
before_id: str | None = params.get("before")
where: dict[str, Any] = dict(owner_filter)
where: dict[str, object] = dict(owner_filter)
fetch_order = "desc"
cursor_id = after_id or before_id
@ -857,7 +893,9 @@ async def _build_list_where_with_cursor(
)
cursor_field = "unified_file_id" if resource_kind == "files" else "unified_object_id"
try:
cursor_row = await cursor_table.find_first(where={**owner_filter, cursor_field: cursor_id})
cursor_row: PrismaManagedFileRow | PrismaManagedObjectRow | None = await cursor_table.find_first(
where={**owner_filter, cursor_field: cursor_id}
)
if cursor_row is not None:
if after_id:
op = "lt"
@ -867,7 +905,7 @@ async def _build_list_where_with_cursor(
# created_at is not unique, so the boundary must also compare the
# unique id (the secondary sort key) to avoid skipping or repeating
# rows that share the cursor row's timestamp across a page boundary.
boundary = {
boundary: dict[str, object] = {
"OR": [
{"created_at": {op: cursor_row.created_at}},
{
@ -885,9 +923,9 @@ async def _build_list_where_with_cursor(
async def _fetch_list_rows(
prisma_client: Any,
prisma_client: PrismaClient | None,
resource_kind: str,
where: dict[str, Any],
where: dict[str, object],
fetch_order: str,
fetch_limit: int,
) -> list[Any] | None:
@ -912,10 +950,10 @@ async def _fetch_list_rows(
async def _fetch_provider_scoped_list_rows(
prisma_client: Any,
prisma_client: PrismaClient | None,
resource_kind: str,
provider: str,
where: dict[str, Any],
where: dict[str, object],
fetch_order: str,
raw_limit: int,
fetch_limit: int,
@ -951,8 +989,8 @@ async def _fetch_provider_scoped_list_rows(
return page, has_more
def _serialize_file_list_item(row: Any) -> dict[str, Any]:
item: dict[str, Any] = {
def _serialize_file_list_item(row: PrismaManagedFileRow) -> dict[str, JSONValue]:
item: dict[str, JSONValue] = {
"id": row.unified_file_id,
"object": "file",
"created_at": int(row.created_at.timestamp()) if row.created_at else None,
@ -964,8 +1002,8 @@ def _serialize_file_list_item(row: Any) -> dict[str, Any]:
return item
def _serialize_batch_list_item(row: Any) -> dict[str, Any]:
item: dict[str, Any] = {}
def _serialize_batch_list_item(row: PrismaManagedObjectRow) -> dict[str, JSONValue]:
item: dict[str, JSONValue] = {}
file_object = _parse_file_object(row.file_object)
if isinstance(file_object, dict):
item.update(file_object)
@ -974,7 +1012,10 @@ def _serialize_batch_list_item(row: Any) -> dict[str, Any]:
return item
def _list_boundary_ids(rows: list[Any], resource_kind: str) -> tuple[str | None, str | None]:
def _list_boundary_ids(
rows: Sequence[PrismaManagedFileRow] | Sequence[PrismaManagedObjectRow],
resource_kind: str,
) -> tuple[str | None, str | None]:
if not rows:
return None, None
id_attr = "unified_file_id" if resource_kind == "files" else "unified_object_id"
@ -985,9 +1026,9 @@ async def list_passthrough_ids_from_db(
provider: str,
route: str,
user_api_key_dict: UserAPIKeyAuth,
prisma_client: Any,
query_params: dict[str, Any] | None = None,
) -> dict[str, Any] | None:
prisma_client: PrismaClient | None,
query_params: dict[str, str] | None = None,
) -> _PassthroughListPage | None:
"""Query the DB for managed IDs the caller owns and return an OpenAI-style
paginated list response.
@ -1060,8 +1101,8 @@ async def rewrite_path_ids(
path: str,
provider: str,
user_api_key_dict: UserAPIKeyAuth,
prisma_client: Any,
managed_files_hook: Any,
prisma_client: PrismaClient | None,
managed_files_hook: _ManagedFilesHook | None,
) -> str:
"""
Walk URL path segments and resolve any passthrough managed IDs to raw
@ -1092,12 +1133,12 @@ async def rewrite_path_ids(
async def rewrite_query_ids(
params: dict[str, Any] | None,
params: dict[str, str] | None,
provider: str,
user_api_key_dict: UserAPIKeyAuth,
prisma_client: Any,
managed_files_hook: Any,
) -> dict[str, Any] | None:
prisma_client: PrismaClient | None,
managed_files_hook: _ManagedFilesHook | None,
) -> dict[str, str] | None:
"""
Walk query param values and resolve any passthrough managed IDs.
Returns *params* unchanged (same object) when nothing is resolved.
@ -1108,12 +1149,11 @@ async def rewrite_query_ids(
mutated = dict(params)
rewritten_keys: list[str] = []
for key, val in list(mutated.items()):
if isinstance(val, str):
if is_managed(val):
mutated[key] = await _resolve_one(val, provider, user_api_key_dict, prisma_client, managed_files_hook)
rewritten_keys.append(key)
else:
await _guard_raw_provider_id(val, provider, user_api_key_dict, prisma_client, budget)
if is_managed(val):
mutated[key] = await _resolve_one(val, provider, user_api_key_dict, prisma_client, managed_files_hook)
rewritten_keys.append(key)
else:
await _guard_raw_provider_id(val, provider, user_api_key_dict, prisma_client, budget)
if rewritten_keys:
verbose_proxy_logger.debug(
"managed_id_rewriter: query ids rewritten provider=%s keys=%s",
@ -1124,12 +1164,12 @@ async def rewrite_query_ids(
async def rewrite_body_ids(
body: dict[str, Any] | None,
body: dict[str, JSONValue] | None,
provider: str,
user_api_key_dict: UserAPIKeyAuth,
prisma_client: Any,
managed_files_hook: Any,
) -> dict[str, Any] | None:
prisma_client: PrismaClient | None,
managed_files_hook: _ManagedFilesHook | None,
) -> dict[str, JSONValue] | None:
"""
Recursively walk a request body dict/list and resolve any passthrough
managed IDs. Skips litellm internal keys (``litellm_*``).
@ -1140,15 +1180,15 @@ async def rewrite_body_ids(
budget = _RawIdGuardBudget()
async def _walk(node: Any, depth: int) -> Any:
async def _walk(node: JSONValue, depth: int) -> JSONValue:
if depth >= _MAX_BODY_REWRITE_DEPTH:
return node
if isinstance(node, dict):
result: dict[str, Any] = {}
result: dict[str, JSONValue] = {}
changed_inner = False
for k, v in node.items():
# Skip litellm internal injection keys (e.g. litellm_logging_obj)
if isinstance(k, str) and k.startswith("litellm_"):
if k.startswith("litellm_"):
result[k] = v
continue
new_v = await _walk(v, depth + 1)
@ -1168,7 +1208,8 @@ async def rewrite_body_ids(
return node
return node
rewritten = await _walk(body, 0)
walked = await _walk(body, 0)
rewritten = walked if isinstance(walked, dict) else body
if rewritten is not body:
verbose_proxy_logger.debug("managed_id_rewriter: body ids rewritten provider=%s", provider)
return rewritten

40
litellm/proxy/pass_through_endpoints/pass_through_endpoints.py
View file

@ -5,7 +5,7 @@ import json
import posixpath
import traceback
from base64 import b64encode
from collections.abc import AsyncGenerator, Mapping
from collections.abc import AsyncGenerator, Callable, Mapping
from datetime import datetime
from itertools import groupby
from typing import Any, cast
@ -23,7 +23,9 @@ from fastapi import (
WebSocket,
status,
)
from fastapi.params import Depends as ParamsDepends
from fastapi.responses import StreamingResponse
from starlette.datastructures import State
from starlette.datastructures import UploadFile as StarletteUploadFile
from starlette.websockets import WebSocketState
from websockets.asyncio.client import connect
@ -57,6 +59,7 @@ from litellm.proxy._types import (
LiteLLMRoutes,
PassThroughEndpointResponse,
PassThroughGenericEndpoint,
PassThroughGuardrailsConfig,
ProxyException,
UserAPIKeyAuth,
)
@ -848,7 +851,7 @@ async def pass_through_request(
# parsed dict (hooks mutate it, breaking the signature / Content-Length).
# Tolerate request objects without `state` (test fixtures) and only honor
# values httpx accepts for `content=`.
_request_state = getattr(request, "state", None)
_request_state: State | None = getattr(request, "state", None)
state_raw_body: str | bytes | None = (
getattr(_request_state, LITELLM_PASS_THROUGH_RAW_BODY_STATE_KEY, None)
if _request_state is not None
@ -1643,7 +1646,7 @@ async def _parse_request_data_by_content_type(
def create_pass_through_route(
endpoint,
target: str,
custom_headers: Mapping[str, Any] | None = None,
custom_headers: Mapping[str, str] | None = None,
_forward_headers: bool | None = False,
_merge_query_params: bool | None = False,
dependencies: list | None = None,
@ -1653,7 +1656,7 @@ def create_pass_through_route(
is_streaming_request: bool | None = False,
query_params: dict | None = None,
default_query_params: dict | None = None,
guardrails: dict[str, Any] | None = None,
guardrails: PassThroughGuardrailsConfig | None = None,
config_file_path: str | None = None,
timeout: float | None = None,
):
@ -2272,7 +2275,7 @@ async def websocket_passthrough_request(
def _is_streaming_response(response: httpx.Response) -> bool:
_content_type = response.headers.get("content-type")
_content_type: str | None = response.headers.get("content-type")
if _content_type is not None and "text/event-stream" in _content_type:
return True
return False
@ -2290,7 +2293,8 @@ def _should_buffer_passthrough_response(response: httpx.Response) -> bool:
"""
if response.status_code >= 400:
return True
media_type = response.headers.get("content-type", "").split(";")[0].strip().lower()
content_type: str = response.headers.get("content-type", "")
media_type = content_type.split(";")[0].strip().lower()
return media_type in ("", "application/json") or media_type.endswith("+json")
@ -2408,9 +2412,9 @@ class SafeRouteAdder:
def add_api_route_if_not_exists(
app: FastAPI,
path: str,
endpoint: Any,
endpoint: Callable[..., object],
methods: list[str],
dependencies: list | None = None,
dependencies: list[ParamsDepends] | None = None,
) -> bool:
"""
Add an API route to the app only if it doesn't already exist.
@ -2925,12 +2929,12 @@ def _get_pass_through_endpoints_from_config() -> list[PassThroughGenericEndpoint
if isinstance(endpoint, dict):
endpoint_dict = dict(endpoint)
endpoint_dict["is_from_config"] = True
returned_endpoints.append(PassThroughGenericEndpoint(**endpoint_dict))
returned_endpoints.append(PassThroughGenericEndpoint.model_validate(endpoint_dict))
elif isinstance(endpoint, PassThroughGenericEndpoint):
# Create a copy with is_from_config=True
endpoint_dict = endpoint.model_dump()
endpoint_dict["is_from_config"] = True
returned_endpoints.append(PassThroughGenericEndpoint(**endpoint_dict))
returned_endpoints.append(PassThroughGenericEndpoint.model_validate(endpoint_dict))
except ValidationError as e:
verbose_proxy_logger.warning(
"Skipping malformed pass-through endpoint from config: %s",
@ -2968,11 +2972,11 @@ async def _get_pass_through_endpoints_from_db(
if isinstance(endpoint, dict):
endpoint_dict = dict(endpoint)
endpoint_dict["is_from_config"] = False
returned_endpoints.append(PassThroughGenericEndpoint(**endpoint_dict))
returned_endpoints.append(PassThroughGenericEndpoint.model_validate(endpoint_dict))
elif isinstance(endpoint, PassThroughGenericEndpoint):
endpoint_dict = endpoint.model_dump()
endpoint_dict["is_from_config"] = False
returned_endpoints.append(PassThroughGenericEndpoint(**endpoint_dict))
returned_endpoints.append(PassThroughGenericEndpoint.model_validate(endpoint_dict))
else:
# Find specific endpoint by ID
found_endpoint = _find_endpoint_by_id(pass_through_endpoint_data, endpoint_id)
@ -2983,7 +2987,7 @@ async def _get_pass_through_endpoints_from_db(
else dict(found_endpoint)
)
endpoint_dict["is_from_config"] = False
returned_endpoints.append(PassThroughGenericEndpoint(**endpoint_dict))
returned_endpoints.append(PassThroughGenericEndpoint.model_validate(endpoint_dict))
return returned_endpoints
@ -3134,7 +3138,7 @@ async def update_pass_through_endpoints(
# Find the index for updating the list
endpoint_index = None
for idx, endpoint in enumerate(pass_through_endpoint_data):
_endpoint = PassThroughGenericEndpoint(**endpoint) if isinstance(endpoint, dict) else endpoint
_endpoint = PassThroughGenericEndpoint.model_validate(endpoint) if isinstance(endpoint, dict) else endpoint
if _endpoint.id == endpoint_id:
endpoint_index = idx
break
@ -3165,7 +3169,7 @@ async def update_pass_through_endpoints(
endpoint_dict.pop("is_from_config", None)
# Create updated endpoint object
updated_endpoint = PassThroughGenericEndpoint(**endpoint_dict)
updated_endpoint = PassThroughGenericEndpoint.model_validate(endpoint_dict)
# Update the list
pass_through_endpoint_data[endpoint_index] = endpoint_dict
@ -3271,7 +3275,7 @@ async def create_pass_through_endpoints(
await update_config_general_settings(data=updated_data, user_api_key_dict=user_api_key_dict)
# Return the created endpoint with the generated ID
created_endpoint = PassThroughGenericEndpoint(**data_dict)
created_endpoint = PassThroughGenericEndpoint.model_validate(data_dict)
# Register the new route
_custom_headers: dict | None = created_endpoint.headers or {}
@ -3363,7 +3367,7 @@ async def delete_pass_through_endpoints(
# Find the index for deleting from the list
endpoint_index = None
for idx, endpoint in enumerate(pass_through_endpoint_data):
_endpoint = PassThroughGenericEndpoint(**endpoint) if isinstance(endpoint, dict) else endpoint
_endpoint = PassThroughGenericEndpoint.model_validate(endpoint) if isinstance(endpoint, dict) else endpoint
if _endpoint.id == endpoint_id:
endpoint_index = idx
break
@ -3409,7 +3413,7 @@ def _find_endpoint_by_id(
for endpoint in endpoints_data:
_endpoint: PassThroughGenericEndpoint | None = None
if isinstance(endpoint, dict):
_endpoint = PassThroughGenericEndpoint(**endpoint)
_endpoint = PassThroughGenericEndpoint.model_validate(endpoint)
elif isinstance(endpoint, PassThroughGenericEndpoint):
_endpoint = endpoint

106
litellm/responses/litellm_completion_transformation/transformation.py
View file

@ -4,7 +4,7 @@ Handles transforming from Responses API -> LiteLLM completion (Chat Completion
import json
import re
from collections.abc import Sequence
from collections.abc import Mapping, Sequence
from typing import Any, Literal, cast
from openai.types.responses import ResponseFunctionToolCall
@ -116,8 +116,8 @@ class LiteLLMCompletionResponsesConfig:
@staticmethod
def _transform_tool_choice(
tool_choice: Any,
) -> str | dict[str, Any] | None:
tool_choice: str | Mapping[str, object] | None,
) -> str | Mapping[str, object] | None:
"""
Transform tool_choice from various formats to OpenAI Chat Completion format.
@ -145,7 +145,8 @@ class LiteLLMCompletionResponsesConfig:
tool_choice_type = tool_choice.get("type")
# If it has a function with name, it's standard OpenAI format - pass through
if tool_choice.get("function") and tool_choice.get("function", {}).get("name"):
tool_choice_function = tool_choice.get("function")
if isinstance(tool_choice_function, dict) and tool_choice_function.get("name"):
return tool_choice
# Handle Cursor IDE dict formats without function name
@ -189,7 +190,7 @@ class LiteLLMCompletionResponsesConfig:
responses_api_request: ResponsesAPIOptionalRequestParams,
custom_llm_provider: str | None = None,
stream: bool | None = None,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
**kwargs,
) -> dict:
"""
@ -446,7 +447,9 @@ class LiteLLMCompletionResponsesConfig:
if not chat_completion_messages:
continue
deduped_in_place: list[Any] = []
deduped_in_place: list[
AllMessageValues | GenericChatCompletionMessage | ChatCompletionResponseMessage
] = []
for m in chat_completion_messages:
role = ""
if isinstance(m, dict):
@ -456,7 +459,7 @@ class LiteLLMCompletionResponsesConfig:
# Drop assistant tool_calls wrappers if we already have this call_id
if role == "assistant":
tool_calls: Any = (
tool_calls: object = (
m.get("tool_calls") if isinstance(m, dict) else getattr(m, "tool_calls", None)
)
call_id = ""
@ -518,7 +521,7 @@ class LiteLLMCompletionResponsesConfig:
call_id = ""
if role == "assistant":
tool_calls: Any = None
tool_calls: object = None
if isinstance(tool_call_message, dict):
tool_calls = tool_call_message.get("tool_calls")
else:
@ -562,7 +565,16 @@ class LiteLLMCompletionResponsesConfig:
return False
@staticmethod
def _find_previous_assistant_idx(messages: list[Any], current_idx: int) -> int | None:
def _find_previous_assistant_idx(
messages: list[
AllMessageValues
| GenericChatCompletionMessage
| ChatCompletionResponseMessage
| ChatCompletionMessageToolCall
| Message
],
current_idx: int,
) -> int | None:
"""Find the index of the previous assistant message."""
for j in range(current_idx - 1, -1, -1):
if messages[j].get("role") == "assistant":
@ -570,7 +582,22 @@ class LiteLLMCompletionResponsesConfig:
return None
@staticmethod
def _recover_tool_call_id_from_assistant(assistant_message: Any, message: Any) -> str:
def _recover_tool_call_id_from_assistant(
assistant_message: (
AllMessageValues
| GenericChatCompletionMessage
| ChatCompletionResponseMessage
| ChatCompletionMessageToolCall
| Message
),
message: (
AllMessageValues
| GenericChatCompletionMessage
| ChatCompletionResponseMessage
| ChatCompletionMessageToolCall
| Message
),
) -> str:
"""Try to recover empty tool_call_id from assistant message's tool_calls."""
tool_calls_raw = (
assistant_message.get("tool_calls")
@ -588,7 +615,15 @@ class LiteLLMCompletionResponsesConfig:
return ""
@staticmethod
def _get_tool_calls_list(assistant_message: Any) -> list[Any]:
def _get_tool_calls_list(
assistant_message: (
AllMessageValues
| GenericChatCompletionMessage
| ChatCompletionResponseMessage
| ChatCompletionMessageToolCall
| Message
),
) -> Sequence[object]:
"""Extract tool_calls as a list from assistant message."""
tool_calls_raw = (
assistant_message.get("tool_calls")
@ -604,10 +639,10 @@ class LiteLLMCompletionResponsesConfig:
return []
@staticmethod
def _check_tool_call_exists(tool_calls: list[Any], tool_call_id: str) -> bool:
def _check_tool_call_exists(tool_calls: Sequence[object], tool_call_id: str) -> bool:
"""Check if a tool_call with the given ID exists in the list."""
for tool_call in tool_calls:
tool_call_id_to_check: str | None = None
tool_call_id_to_check: object = None
if isinstance(tool_call, dict):
tool_call_id_to_check = tool_call.get("id")
elif hasattr(tool_call, "id"):
@ -617,7 +652,7 @@ class LiteLLMCompletionResponsesConfig:
return False
@staticmethod
def _reconstruct_tool_call_from_tools(tool_call_id: str, tools: list[Any]) -> dict[str, Any] | None:
def _reconstruct_tool_call_from_tools(tool_call_id: str, tools: list[object]) -> dict[str, object] | None:
"""Reconstruct a minimal tool_call definition from tools list."""
for tool in tools:
if isinstance(tool, dict):
@ -635,7 +670,7 @@ class LiteLLMCompletionResponsesConfig:
return None
@staticmethod
def _get_mapping_or_attr_value(obj: Any, key: str, default: Any = None) -> Any:
def _get_mapping_or_attr_value(obj: object, key: str, default: object = None) -> object:
"""
Safely read a field from dict-like or attribute-based objects.
"""
@ -656,13 +691,13 @@ class LiteLLMCompletionResponsesConfig:
@staticmethod
def _create_tool_call_chunk(
tool_use_definition: dict[str, Any], tool_call_id: str, index: int
tool_use_definition: Mapping[str, object], tool_call_id: str, index: int
) -> ChatCompletionToolCallChunk:
"""Create a ChatCompletionToolCallChunk from tool_use_definition."""
function_raw = LiteLLMCompletionResponsesConfig._get_mapping_or_attr_value(tool_use_definition, "function")
function_name_raw = LiteLLMCompletionResponsesConfig._get_mapping_or_attr_value(function_raw, "name")
function_arguments_raw = LiteLLMCompletionResponsesConfig._get_mapping_or_attr_value(function_raw, "arguments")
function: dict[str, Any] = {
function: dict[str, object] = {
"name": function_name_raw or "",
"arguments": function_arguments_raw or "{}",
}
@ -681,15 +716,16 @@ class LiteLLMCompletionResponsesConfig:
)
@staticmethod
def _normalize_tool_use_definition(tool_use_definition: Any, tool_call_id: str) -> dict[str, Any] | None:
def _normalize_tool_use_definition(tool_use_definition: object, tool_call_id: str) -> dict[str, object] | None:
"""
Normalize cached tool_call definitions to a dict-like shape consumed by _create_tool_call_chunk.
"""
if not tool_use_definition:
return None
normalized_definition: dict[str, object]
if isinstance(tool_use_definition, dict):
normalized_definition: dict[str, Any] = dict(tool_use_definition)
normalized_definition = dict(tool_use_definition)
else:
tool_use_id_raw = LiteLLMCompletionResponsesConfig._get_mapping_or_attr_value(tool_use_definition, "id")
tool_use_type_raw = LiteLLMCompletionResponsesConfig._get_mapping_or_attr_value(tool_use_definition, "type")
@ -746,7 +782,7 @@ class LiteLLMCompletionResponsesConfig:
| ChatCompletionMessageToolCall
| Message
],
tools: list[Any] | None = None,
tools: list[object] | None = None,
) -> list[
AllMessageValues
| GenericChatCompletionMessage
@ -930,7 +966,7 @@ class LiteLLMCompletionResponsesConfig:
@staticmethod
def _transform_responses_api_tool_call_output_to_chat_completion_message(
tool_call_output: dict[str, Any],
tool_call_output: dict[str, object],
) -> list[AllMessageValues | GenericChatCompletionMessage | ChatCompletionResponseMessage]:
"""
ChatCompletionToolMessage is used to indicate the output from a tool call
@ -942,7 +978,7 @@ class LiteLLMCompletionResponsesConfig:
return []
def _normalize_function_call_output_to_tool_content(
output: Any,
output: object,
) -> Any:
"""
Normalize Responses API function_call_output.output into a shape that downstream
@ -965,7 +1001,7 @@ class LiteLLMCompletionResponsesConfig:
# Some adapters represent tool output as a list of "input_*" parts
if isinstance(output, list):
normalized_blocks: list[dict[str, Any]] = []
normalized_blocks: list[dict[str, object]] = []
text_acc: list[str] = []
for part in output:
if not isinstance(part, dict):
@ -1157,8 +1193,8 @@ class LiteLLMCompletionResponsesConfig:
@staticmethod
def _transform_responses_api_content_to_chat_completion_content(
content: Any,
) -> str | list[str | dict[str, Any]]:
content: object,
) -> str | list[str | dict[str, object]]:
"""
Transform a Responses API content into a Chat Completion content
@ -1172,7 +1208,7 @@ class LiteLLMCompletionResponsesConfig:
elif isinstance(content, str):
return content
elif isinstance(content, list):
content_list: list[str | dict[str, Any]] = []
content_list: list[str | dict[str, object]] = []
for item in content:
if isinstance(item, str):
content_list.append(item)
@ -1193,7 +1229,7 @@ class LiteLLMCompletionResponsesConfig:
text_value = item.get("text")
if text_value is None:
continue
content_block: dict[str, Any] = {
content_block: dict[str, object] = {
"type": LiteLLMCompletionResponsesConfig._get_chat_completion_request_content_type(
item.get("type") or "text"
),
@ -1283,7 +1319,7 @@ class LiteLLMCompletionResponsesConfig:
parameters = dict(typed_tool.get("parameters", {}) or {})
if not parameters or "type" not in parameters:
parameters["type"] = "object"
chat_completion_tool: dict[str, Any] = {
chat_completion_tool: dict[str, object] = {
"type": "function",
"function": {
"name": typed_tool.get("name") or "",
@ -1498,7 +1534,7 @@ class LiteLLMCompletionResponsesConfig:
def convert_response_function_tool_call_to_chat_completion_tool_call(
tool_call_item: Any,
index: int = 0,
) -> dict[str, Any]:
) -> dict[str, object]:
"""
Convert ResponseFunctionToolCall to ChatCompletionToolCallChunk format.
@ -1528,7 +1564,7 @@ class LiteLLMCompletionResponsesConfig:
)
)
function_dict: dict[str, Any] = {
function_dict: dict[str, object] = {
"name": tool_call_item.name,
"arguments": tool_call_item.arguments,
}
@ -1536,7 +1572,7 @@ class LiteLLMCompletionResponsesConfig:
if provider_specific_fields:
function_dict["provider_specific_fields"] = provider_specific_fields
tool_call_dict: dict[str, Any] = {
tool_call_dict: dict[str, object] = {
"id": LiteLLMCompletionResponsesConfig._tool_call_id_from_responses_item(
getattr(tool_call_item, "id", None),
getattr(tool_call_item, "call_id", None),
@ -1555,7 +1591,7 @@ class LiteLLMCompletionResponsesConfig:
def convert_apply_patch_tool_call_to_chat_completion_tool_call(
tool_call_item: Any,
index: int = 0,
) -> dict[str, Any]:
) -> dict[str, object]:
"""
Convert ResponseApplyPatchToolCall to ChatCompletionToolCallChunk format.
@ -1573,7 +1609,7 @@ class LiteLLMCompletionResponsesConfig:
import json
operation_dict = tool_call_item.operation.model_dump()
tool_call_dict: dict[str, Any] = {
tool_call_dict: dict[str, object] = {
"id": tool_call_item.call_id,
"function": {
"name": "apply_patch",
@ -2034,8 +2070,8 @@ class LiteLLMCompletionResponsesConfig:
@staticmethod
def _transform_text_format_to_response_format(
text_param: dict[str, Any] | Any,
) -> dict[str, Any] | None:
text_param: Mapping[str, object] | None,
) -> dict[str, object] | None:
"""
Transform Responses API text.format parameter to Chat Completion response_format parameter.

55
litellm/responses/streaming_iterator.py
View file

@ -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
import httpx
from openai._streaming import SSEDecoder
@ -34,6 +34,16 @@ 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 litellm.types.llms.openai import (
PART_UNION_TYPES,
ContentPartDoneEvent,
ResponseCreatedEvent,
ResponseInProgressEvent,
ResponsesAPIResponse,
ResponsesAPIStreamingResponse,
)
@lru_cache(maxsize=1)
def _get_openai_response_types():
@ -42,7 +52,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()
@ -451,7 +461,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):
@ -880,7 +890,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 +904,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 +918,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 +933,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 +951,13 @@ class CachedResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator):
)
self._completed_response_cache_hit = True
self._persist_completed_response_before_logging = False
self._events: list[Any] = []
self._events: list[ResponsesAPIStreamingResponse] = []
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 +971,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 +985,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]
@ -1000,8 +1010,8 @@ def _build_response_status_event(
"response.created",
"response.in_progress",
],
transformed: Any,
) -> Any:
transformed: ResponsesAPIResponse,
) -> ResponseCreatedEvent | ResponseInProgressEvent:
openai_types = _get_openai_response_types()
in_progress_response = transformed.model_copy(
deep=True,
@ -1018,10 +1028,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: PART_UNION_TYPES
if part_type == "output_text":
annotations = [
openai_types.BaseLiteLLMOpenAIResponseObject(**annotation)
@ -1057,7 +1067,7 @@ def _build_content_part_done_event(
def _add_text_like_part_events(
*,
events: list[Any],
events: list[ResponsesAPIStreamingResponse],
item_id: str,
output_index: int,
content_index: int,
@ -1123,13 +1133,13 @@ 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]:
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,10 +1148,9 @@ def _build_synthetic_response_events(
except Exception:
pass
events: list[Any] = [
_build_response_status_event(openai_types.ResponsesAPIStreamEvents.RESPONSE_CREATED, transformed),
_build_response_status_event(openai_types.ResponsesAPIStreamEvents.RESPONSE_IN_PROGRESS, transformed),
]
events: list[ResponsesAPIStreamingResponse] = []
events.append(_build_response_status_event(openai_types.ResponsesAPIStreamEvents.RESPONSE_CREATED, transformed))
events.append(_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 []):

4
ruff-strict-budget.json
View file

@ -24,7 +24,7 @@
"limit": 130
},
"ANN401": {
"limit": 1851
"limit": 1686
},
"ASYNC230": {
"limit": 14
@ -39,7 +39,7 @@
"limit": 505
},
"B009": {
"limit": 84
"limit": 81
},
"B010": {
"limit": 194

6
type-discipline-budget.json
View file

@ -1,9 +1,9 @@
{
"LIT001": {
"limit": 23350
"limit": 23308
},
"LIT002": {
"limit": 27256
"limit": 27238
},
"LIT003": {
"limit": 292
@ -15,7 +15,7 @@
"limit": 0
},
"LIT006": {
"limit": 1105
"limit": 1102
},
"LIT007": {
"limit": 0