scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-12 23:01:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 88

Fix mypy and code qa issues

Browse source
This commit is contained in:
Sameer Kankute 2026-03-20 23:18:58 +05:30
parent 0673c57fef
commit 2a69426e2f
3 changed files with 277 additions and 200 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -40,6 +40,7 @@ from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
from litellm.llms.custom_httpx.http_handler import get_async_httpx_client
from litellm.passthrough import BasePassthroughUtils
from litellm.proxy._types import (
CommonProxyErrors,
ConfigFieldInfo,
ConfigFieldUpdate,
LiteLLMRoutes,
@ -2155,6 +2156,102 @@ def _get_combined_pass_through_endpoints(
return pass_through_endpoints + config_pass_through_endpoints
async def _register_pass_through_endpoint(
endpoint: Union[Dict[str, Any], PassThroughGenericEndpoint],
app: FastAPI,
premium_user: bool,
visited_endpoints: set[str],
) -> None:
endpoint_data: Dict[str, Any]
if isinstance(endpoint, PassThroughGenericEndpoint):
endpoint_data = endpoint.model_dump()
else:
endpoint_data = endpoint
if endpoint_data.get("id") is None:
endpoint_data["id"] = str(uuid.uuid4())
endpoint_id = cast(str, endpoint_data["id"])
target = endpoint_data.get("target")
path = endpoint_data.get("path")
if path is None:
raise ValueError("Path is required for pass-through endpoint")
custom_headers = await set_env_variables_in_header(
custom_headers=endpoint_data.get("headers")
)
forward_headers = endpoint_data.get("forward_headers")
merge_query_params = endpoint_data.get("merge_query_params")
default_query_params = endpoint_data.get("default_query_params")
auth = endpoint_data.get("auth")
dependencies = None
if auth is not None and str(auth).lower() == "true":
if premium_user is not True:
raise ValueError(
"Error Setting Authentication on Pass Through Endpoint: {}".format(
CommonProxyErrors.not_premium_user.value
)
)
dependencies = [Depends(user_api_key_auth)]
if path not in LiteLLMRoutes.openai_routes.value:
LiteLLMRoutes.openai_routes.value.append(path)
if target is None:
return
guardrails = endpoint_data.get("guardrails")
methods = endpoint_data.get("methods")
cost_per_request = endpoint_data.get("cost_per_request")
verbose_proxy_logger.debug(
"Initializing pass through endpoint: %s (ID: %s)", path, endpoint_id
)
InitPassThroughEndpointHelpers.add_exact_path_route(
app=app,
path=path,
target=target,
custom_headers=custom_headers,
forward_headers=forward_headers,
merge_query_params=merge_query_params,
dependencies=dependencies,
cost_per_request=cost_per_request,
endpoint_id=endpoint_id,
guardrails=guardrails,
methods=methods,
default_query_params=default_query_params,
)
methods_for_key = methods if methods else ["GET", "POST", "PUT", "DELETE", "PATCH"]
methods_str = ",".join(sorted(methods_for_key))
visited_endpoints.add(f"{endpoint_id}:exact:{path}:{methods_str}")
if endpoint_data.get("include_subpath", False) is True:
if auth is not None and str(auth).lower() == "true":
wildcard_path = path.rstrip("/") + "/*"
if wildcard_path not in LiteLLMRoutes.openai_routes.value:
LiteLLMRoutes.openai_routes.value.append(wildcard_path)
InitPassThroughEndpointHelpers.add_subpath_route(
app=app,
path=path,
target=target,
custom_headers=custom_headers,
forward_headers=forward_headers,
merge_query_params=merge_query_params,
dependencies=dependencies,
cost_per_request=cost_per_request,
endpoint_id=endpoint_id,
guardrails=guardrails,
methods=methods,
default_query_params=default_query_params,
)
visited_endpoints.add(f"{endpoint_id}:subpath:{path}:{methods_str}")
verbose_proxy_logger.debug(
"Added new pass through endpoint: %s (ID: %s)", path, endpoint_id
)
async def initialize_pass_through_endpoints(
pass_through_endpoints: Union[List[Dict], List[PassThroughGenericEndpoint]],
):
@ -2171,10 +2268,7 @@ async def initialize_pass_through_endpoints(
Returns:
None
"""
from litellm._uuid import uuid
verbose_proxy_logger.debug("initializing pass through endpoints")
from litellm.proxy._types import CommonProxyErrors, LiteLLMRoutes
from litellm.proxy.proxy_server import (
app,
config_passthrough_endpoints,
@ -2201,105 +2295,14 @@ async def initialize_pass_through_endpoints(
InitPassThroughEndpointHelpers.get_all_registered_pass_through_routes()
)
visited_endpoints = set()
visited_endpoints: set[str] = set()
for endpoint in combined_pass_through_endpoints:
if isinstance(endpoint, PassThroughGenericEndpoint):
endpoint = endpoint.model_dump()
# Auto-generate ID for backwards compatibility if not present
if endpoint.get("id") is None:
endpoint["id"] = str(uuid.uuid4())
# Get the endpoint_id as a string (guaranteed to be set at this point)
endpoint_id: str = endpoint["id"]
_target = endpoint.get("target", None)
_path: Optional[str] = endpoint.get("path", None)
if _path is None:
raise ValueError("Path is required for pass-through endpoint")
_custom_headers = endpoint.get("headers", None)
_custom_headers = await set_env_variables_in_header(
custom_headers=_custom_headers
)
_forward_headers = endpoint.get("forward_headers", None)
_merge_query_params = endpoint.get("merge_query_params", None)
_default_query_params = endpoint.get("default_query_params", None)
_auth = endpoint.get("auth", None)
_dependencies = None
if _auth is not None and str(_auth).lower() == "true":
if premium_user is not True:
raise ValueError(
"Error Setting Authentication on Pass Through Endpoint: {}".format(
CommonProxyErrors.not_premium_user.value
)
)
_dependencies = [Depends(user_api_key_auth)]
if _path not in LiteLLMRoutes.openai_routes.value:
LiteLLMRoutes.openai_routes.value.append(_path)
if _target is None:
continue
# Get guardrails config if present
_guardrails = endpoint.get("guardrails", None)
# Get methods list if present (None means all methods for backward compatibility)
_methods = endpoint.get("methods", None)
# Add exact path route
verbose_proxy_logger.debug(
"Initializing pass through endpoint: %s (ID: %s)", _path, endpoint_id
)
InitPassThroughEndpointHelpers.add_exact_path_route(
await _register_pass_through_endpoint(
endpoint=endpoint,
app=app,
path=_path,
target=_target,
custom_headers=_custom_headers,
forward_headers=_forward_headers,
merge_query_params=_merge_query_params,
dependencies=_dependencies,
cost_per_request=endpoint.get("cost_per_request", None),
endpoint_id=endpoint_id,
guardrails=_guardrails,
methods=_methods,
default_query_params=_default_query_params,
)
# Generate route key with methods for tracking
methods_for_key = (
_methods if _methods else ["GET", "POST", "PUT", "DELETE", "PATCH"]
)
methods_str = ",".join(sorted(methods_for_key))
visited_endpoints.add(f"{endpoint_id}:exact:{_path}:{methods_str}")
# Add wildcard route for sub-paths
if endpoint.get("include_subpath", False) is True:
# Register wildcard path in openai_routes so non-admin users
# can access subpath routes when auth is enabled
if _auth is not None and str(_auth).lower() == "true":
_wildcard_path = _path.rstrip("/") + "/*"
if _wildcard_path not in LiteLLMRoutes.openai_routes.value:
LiteLLMRoutes.openai_routes.value.append(_wildcard_path)
InitPassThroughEndpointHelpers.add_subpath_route(
app=app,
path=_path,
target=_target,
custom_headers=_custom_headers,
forward_headers=_forward_headers,
merge_query_params=_merge_query_params,
dependencies=_dependencies,
cost_per_request=endpoint.get("cost_per_request", None),
endpoint_id=endpoint_id,
guardrails=_guardrails,
methods=_methods,
default_query_params=_default_query_params,
)
visited_endpoints.add(f"{endpoint_id}:subpath:{_path}:{methods_str}")
verbose_proxy_logger.debug(
"Added new pass through endpoint: %s (ID: %s)", _path, endpoint_id
premium_user=premium_user,
visited_endpoints=visited_endpoints,
)
# remove the ones that are not visited from the list

88
litellm/responses/file_search/emulated_handler.py
View file

@ -14,10 +14,10 @@ Flow:
import json
import time
import uuid
from typing import Any, Dict, Iterable, List, Optional, Tuple, cast
from typing import Any, Dict, Iterable, List, Optional, Tuple, Union, cast
from litellm._logging import verbose_logger
from litellm.types.llms.openai import ResponsesAPIResponse
from litellm.types.llms.openai import ResponseOutputItem, ResponsesAPIResponse
from litellm.types.vector_stores import VectorStoreSearchResult
# Keep ToolParam broad so we stay compatible with both dict and Pydantic forms
@ -30,6 +30,7 @@ FILE_SEARCH_FUNCTION_NAME = "litellm_file_search"
# Detection
# ---------------------------------------------------------------------------
def should_use_emulated_file_search(
tools: Optional[Iterable[ToolParam]],
provider_config: Any, # BaseResponsesAPIConfig
@ -37,9 +38,7 @@ def should_use_emulated_file_search(
"""Return True when there is a file_search tool and the provider can't handle it natively."""
if not tools:
return False
has_fs = any(
isinstance(t, dict) and t.get("type") == "file_search" for t in tools
)
has_fs = any(isinstance(t, dict) and t.get("type") == "file_search" for t in tools)
if not has_fs:
return False
return provider_config is None or not provider_config.supports_native_file_search()
@ -49,6 +48,7 @@ def should_use_emulated_file_search(
# Tool conversion
# ---------------------------------------------------------------------------
def _build_function_tool(vector_store_ids: List[str]) -> Dict[str, Any]:
"""
Create a Responses API function-tool definition that describes file search.
@ -104,7 +104,7 @@ def _replace_file_search_tools(
non_file_search: List[Dict[str, Any]] = []
vector_store_ids: List[str] = []
for tool in (tools or []):
for tool in tools or []:
if isinstance(tool, dict) and tool.get("type") == "file_search":
ids = tool.get("vector_store_ids") or []
vector_store_ids.extend(ids)
@ -123,6 +123,7 @@ def _replace_file_search_tools(
# Search execution
# ---------------------------------------------------------------------------
async def _run_vector_searches(
queries: List[str],
vector_store_ids: List[str],
@ -150,7 +151,11 @@ async def _run_vector_searches(
vector_store_id=vs_id,
query=query,
)
results_data = response.get("data") if isinstance(response, dict) else getattr(response, "data", None)
results_data = (
response.get("data")
if isinstance(response, dict)
else getattr(response, "data", None)
)
if results_data:
all_results.extend(results_data)
except Exception as exc:
@ -168,6 +173,7 @@ async def _run_vector_searches(
# Result formatting
# ---------------------------------------------------------------------------
def _get_field(result: Any, key: str, default: Any = None) -> Any:
"""Read a field from either a dict/TypedDict or an attribute-based object."""
if isinstance(result, dict):
@ -319,13 +325,27 @@ def _build_message_output(
def _extract_text_from_responses_output(response: ResponsesAPIResponse) -> str:
"""Pull the assistant's text from the provider's response."""
for item in response.output:
item_type = item.get("type") if isinstance(item, dict) else getattr(item, "type", None)
item_type = (
item.get("type") if isinstance(item, dict) else getattr(item, "type", None)
)
if item_type == "message":
content = item.get("content") if isinstance(item, dict) else getattr(item, "content", [])
for block in (content or []):
block_type = block.get("type") if isinstance(block, dict) else getattr(block, "type", None)
content = (
item.get("content")
if isinstance(item, dict)
else getattr(item, "content", [])
)
for block in content or []:
block_type = (
block.get("type")
if isinstance(block, dict)
else getattr(block, "type", None)
)
if block_type == "output_text":
raw = block.get("text") if isinstance(block, dict) else getattr(block, "text", "")
raw = (
block.get("text")
if isinstance(block, dict)
else getattr(block, "text", "")
)
return str(raw) if raw is not None else ""
return ""
@ -345,13 +365,16 @@ def _synthesize_responses_api_response(
synthesized _hidden_params so that billing callbacks see the total cost of
both provider calls that the emulated flow makes.
"""
synthesized_output: List[Dict[str, Any]] = [file_search_call_output, message_output]
synthesized = ResponsesAPIResponse(
id=getattr(original_response, "id", f"resp_{uuid.uuid4().hex}"),
object="response",
created_at=getattr(original_response, "created_at", int(time.time())),
status="completed",
model=getattr(original_response, "model", ""),
output=[file_search_call_output, message_output],
output=cast(
List[Union[ResponseOutputItem, Dict[str, Any]]], synthesized_output
),
usage=getattr(original_response, "usage", None),
error=None,
)
@ -359,9 +382,15 @@ def _synthesize_responses_api_response(
hidden = dict(getattr(original_response, "_hidden_params") or {})
if first_response is not None and hasattr(first_response, "_hidden_params"):
first_hidden = getattr(first_response, "_hidden_params") or {}
first_cost = first_hidden.get("response_cost") if isinstance(first_hidden, dict) else getattr(first_hidden, "response_cost", None)
first_cost = (
first_hidden.get("response_cost")
if isinstance(first_hidden, dict)
else getattr(first_hidden, "response_cost", None)
)
if first_cost is not None:
current_cost = hidden.get("response_cost") if isinstance(hidden, dict) else 0
current_cost = (
hidden.get("response_cost") if isinstance(hidden, dict) else 0
)
hidden["response_cost"] = (current_cost or 0) + first_cost
synthesized._hidden_params = hidden
return synthesized
@ -371,8 +400,12 @@ def _synthesize_responses_api_response(
# Main entry point
# ---------------------------------------------------------------------------
async def _call_aresponses(input, model, tools, **kwargs): # pragma: no cover – thin wrapper for patching in tests
async def _call_aresponses(
input, model, tools, **kwargs
): # pragma: no cover – thin wrapper for patching in tests
from litellm.responses.main import aresponses
return await aresponses(input=input, model=model, tools=tools, **kwargs)
@ -458,10 +491,17 @@ async def aresponses_with_emulated_file_search(
for tool_call in file_search_calls:
if isinstance(tool_call, dict):
call_id = tool_call.get("call_id") or tool_call.get("id") or file_search_call_id
call_id = str(
tool_call.get("call_id") or tool_call.get("id") or file_search_call_id
)
raw_args = tool_call.get("arguments") or "{}"
else:
call_id = getattr(tool_call, "call_id", None) or getattr(tool_call, "id", file_search_call_id)
raw_call_id = (
getattr(tool_call, "call_id", None)
or getattr(tool_call, "id", None)
or file_search_call_id
)
call_id = str(raw_call_id)
raw_args = getattr(tool_call, "arguments", "{}") or "{}"
try:
@ -500,7 +540,11 @@ async def aresponses_with_emulated_file_search(
# Including all output items (text blocks, reasoning, non-file-search calls) ensures providers
# like Anthropic that emit text before the tool call have complete conversation context.
# Serialize Pydantic model instances to plain dicts so the transformation layer can call .get().
original_input_items = list(input) if isinstance(input, (list, tuple)) else [{"role": "user", "content": str(input)}]
original_input_items = (
list(input)
if isinstance(input, (list, tuple))
else [{"role": "user", "content": str(input)}]
)
first_response_output_items: List[Any] = []
for _item in first_response.output:
if isinstance(_item, dict):
@ -510,11 +554,7 @@ async def aresponses_with_emulated_file_search(
else:
first_response_output_items.append(_item)
follow_up_input = (
original_input_items
+ first_response_output_items
+ tool_results
)
follow_up_input = original_input_items + first_response_output_items + tool_results
# 6. Follow-up call — provider writes the final answer given search results.
# Also an internal sub-call; billing is suppressed so the outer call fires once.

188
litellm/responses/main.py
View file

@ -77,9 +77,7 @@ def _has_file_search_tool(tools: Optional[Any]) -> bool:
"""Return True if any tool in the list has type 'file_search'."""
if not tools:
return False
return any(
isinstance(t, dict) and t.get("type") == "file_search" for t in tools
)
return any(isinstance(t, dict) and t.get("type") == "file_search" for t in tools)
def mock_responses_api_response(
@ -486,7 +484,9 @@ async def aresponses(
prompt_variables = cast(Optional[dict], kwargs.get("prompt_variables", None))
original_model = model
if isinstance(litellm_logging_obj, LiteLLMLoggingObj) and litellm_logging_obj.should_run_prompt_management_hooks(
if isinstance(
litellm_logging_obj, LiteLLMLoggingObj
) and litellm_logging_obj.should_run_prompt_management_hooks(
prompt_id=prompt_id, non_default_params=kwargs
):
if isinstance(input, str):
@ -514,9 +514,7 @@ async def aresponses(
)
input = cast(Union[str, ResponseInputParam], merged_input)
if model != original_model:
_, custom_llm_provider, _, _ = litellm.get_llm_provider(
model=model
)
_, custom_llm_provider, _, _ = litellm.get_llm_provider(model=model)
kwargs.pop("prompt_id", None)
kwargs["_async_prompt_merged_params"] = merged_optional_params
@ -588,6 +586,88 @@ async def aresponses(
)
def _apply_prompt_management_to_responses_call(
input: Union[str, ResponseInputParam],
model: str,
custom_llm_provider: Optional[str],
litellm_logging_obj: Optional[LiteLLMLoggingObj],
kwargs: Dict[str, Any],
local_vars: Dict[str, Any],
) -> tuple[Union[str, ResponseInputParam], str, Optional[str]]:
async_merged = kwargs.pop("_async_prompt_merged_params", None)
if async_merged is not None:
for key, value in async_merged.items():
local_vars[key] = value
return input, model, custom_llm_provider
prompt_id = cast(Optional[str], kwargs.get("prompt_id", None))
prompt_variables = cast(Optional[dict], kwargs.get("prompt_variables", None))
original_model = model
if isinstance(input, str):
client_input: List[AllMessageValues] = [{"role": "user", "content": input}]
else:
client_input = [
item # type: ignore[misc]
for item in input
if isinstance(item, dict) and "role" in item
]
if isinstance(
litellm_logging_obj, LiteLLMLoggingObj
) and litellm_logging_obj.should_run_prompt_management_hooks(
prompt_id=prompt_id, non_default_params=kwargs
):
(
model,
merged_input,
merged_optional_params,
) = litellm_logging_obj.get_chat_completion_prompt(
model=model,
messages=client_input,
non_default_params=kwargs,
prompt_id=prompt_id,
prompt_variables=prompt_variables,
prompt_label=kwargs.get("prompt_label", None),
prompt_version=kwargs.get("prompt_version", None),
)
input = cast(Union[str, ResponseInputParam], merged_input)
local_vars["input"] = input
local_vars["model"] = model
if model != original_model:
_, custom_llm_provider, _, _ = litellm.get_llm_provider(model=model)
local_vars["custom_llm_provider"] = custom_llm_provider
for key, value in merged_optional_params.items():
local_vars[key] = value
return input, model, custom_llm_provider
def _resolve_model_provider_for_responses(
model: str,
custom_llm_provider: Optional[str],
litellm_params: GenericLiteLLMParams,
local_vars: Dict[str, Any],
) -> tuple[str, Optional[str]]:
(
model,
custom_llm_provider,
dynamic_api_key,
dynamic_api_base,
) = litellm.get_llm_provider(
model=model,
custom_llm_provider=custom_llm_provider,
api_base=litellm_params.api_base,
api_key=litellm_params.api_key,
)
local_vars["custom_llm_provider"] = custom_llm_provider
if dynamic_api_key is not None:
litellm_params.api_key = dynamic_api_key
if dynamic_api_base is not None:
litellm_params.api_base = dynamic_api_base
return model, custom_llm_provider
@client
def responses(
input: Union[str, ResponseInputParam],
@ -659,80 +739,27 @@ def responses(
mock_response=litellm_params.mock_response
)
(
model,
custom_llm_provider,
dynamic_api_key,
dynamic_api_base,
) = litellm.get_llm_provider(
model, custom_llm_provider = _resolve_model_provider_for_responses(
model=model,
custom_llm_provider=custom_llm_provider,
api_base=litellm_params.api_base,
api_key=litellm_params.api_key,
litellm_params=litellm_params,
local_vars=local_vars,
)
# Update local_vars with detected provider (fixes #19782)
local_vars["custom_llm_provider"] = custom_llm_provider
# Use dynamic credentials from get_llm_provider (e.g., when use_litellm_proxy=True)
if dynamic_api_key is not None:
litellm_params.api_key = dynamic_api_key
if dynamic_api_base is not None:
litellm_params.api_base = dynamic_api_base
#########################################################
# PROMPT MANAGEMENT
# If aresponses() already ran the async hook, it pops prompt_id and
# passes the result via _async_prompt_merged_params — apply those
# directly and skip the sync hook to avoid double-merging.
#########################################################
_async_merged = kwargs.pop("_async_prompt_merged_params", None)
if _async_merged is not None:
for k, v in _async_merged.items():
local_vars[k] = v
else:
prompt_id = cast(Optional[str], kwargs.get("prompt_id", None))
prompt_variables = cast(
Optional[dict], kwargs.get("prompt_variables", None)
)
original_model = model
if isinstance(litellm_logging_obj, LiteLLMLoggingObj) and litellm_logging_obj.should_run_prompt_management_hooks(
prompt_id=prompt_id, non_default_params=kwargs
):
if isinstance(input, str):
client_input: List[AllMessageValues] = [
{"role": "user", "content": input}
]
else:
client_input = [
item # type: ignore[misc]
for item in input
if isinstance(item, dict) and "role" in item
]
(
model,
merged_input,
merged_optional_params,
) = litellm_logging_obj.get_chat_completion_prompt(
model=model,
messages=client_input,
non_default_params=kwargs,
prompt_id=prompt_id,
prompt_variables=prompt_variables,
prompt_label=kwargs.get("prompt_label", None),
prompt_version=kwargs.get("prompt_version", None),
)
input = cast(Union[str, ResponseInputParam], merged_input)
local_vars["input"] = input
local_vars["model"] = model
if model != original_model:
_, custom_llm_provider, _, _ = litellm.get_llm_provider(
model=model
)
local_vars["custom_llm_provider"] = custom_llm_provider
for k, v in merged_optional_params.items():
local_vars[k] = v
input, model, custom_llm_provider = _apply_prompt_management_to_responses_call(
input=input,
model=model,
custom_llm_provider=custom_llm_provider,
litellm_logging_obj=litellm_logging_obj,
kwargs=kwargs,
local_vars=local_vars,
)
#########################################################
# Update input and tools with provider-specific file IDs if managed files are used
@ -803,12 +830,16 @@ def responses(
return run_async_function(aresponses_api_with_mcp, **mcp_call_kwargs)
# get provider config
responses_api_provider_config: Optional[
BaseResponsesAPIConfig
] = ProviderConfigManager.get_provider_responses_api_config(
model=model,
provider=custom_llm_provider,
)
responses_api_provider_config: Optional[BaseResponsesAPIConfig]
if custom_llm_provider is None:
responses_api_provider_config = None
else:
responses_api_provider_config = (
ProviderConfigManager.get_provider_responses_api_config(
model=model,
provider=custom_llm_provider,
)
)
local_vars.update(kwargs)
# Map reasoning_effort (from litellm_params/proxy config) to reasoning when not set
@ -919,6 +950,9 @@ def responses(
)
# Call the handler with _is_async flag instead of directly calling the async handler
if custom_llm_provider is None:
raise ValueError("custom_llm_provider is required but passed as None")
response = base_llm_http_handler.response_api_handler(
model=model,
input=input,