diff --git a/litellm/files/main.py b/litellm/files/main.py index 144b929132..7bc2c13672 100644 --- a/litellm/files/main.py +++ b/litellm/files/main.py @@ -18,6 +18,7 @@ from litellm import get_secret_str from litellm.litellm_core_utils.get_llm_provider_logic import get_llm_provider from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj from litellm.llms.azure.files.handler import AzureOpenAIFilesAPI +from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler from litellm.llms.custom_httpx.llm_http_handler import BaseLLMHTTPHandler from litellm.llms.openai.openai import FileDeleted, FileObject, OpenAIFilesAPI from litellm.llms.vertex_ai.files.handler import VertexAIFilesHandler diff --git a/litellm/llms/openrouter/chat/transformation.py b/litellm/llms/openrouter/chat/transformation.py index a47689a30b..d5861c79b2 100644 --- a/litellm/llms/openrouter/chat/transformation.py +++ b/litellm/llms/openrouter/chat/transformation.py @@ -7,7 +7,7 @@ Docs: https://openrouter.ai/docs/parameters """ from enum import Enum -from typing import Any, AsyncIterator, Iterator, List, Optional, Tuple, Union +from typing import Any, AsyncIterator, Iterator, List, Optional, Tuple, Union, cast import httpx @@ -88,29 +88,31 @@ class OpenrouterConfig(OpenAIGPTConfig): Move cache_control from message level to content blocks. OpenRouter requires cache_control to be inside content blocks, not at message level. - When cache_control is at message level, it's added to ALL content blocks - to cache the entire message content. + To avoid exceeding Anthropic's limit of 4 cache breakpoints, cache_control is only + added to the LAST content block in each message. """ - transformed_messages = [] + transformed_messages: List[AllMessageValues] = [] for message in messages: - message_copy = dict(message) - cache_control = message_copy.pop("cache_control", None) + message_dict = dict(message) + cache_control = message_dict.pop("cache_control", None) if cache_control is not None: - content = message_copy.get("content") + content = message_dict.get("content") if isinstance(content, list): - # Content is already a list, add cache_control to all blocks + # Content is already a list, add cache_control only to the last block if len(content) > 0: content_copy = [] - for block in content: - block_copy = dict(block) - block_copy["cache_control"] = cache_control - content_copy.append(block_copy) - message_copy["content"] = content_copy + for i, block in enumerate(content): + block_dict = dict(block) + # Only add cache_control to the last content block + if i == len(content) - 1: + block_dict["cache_control"] = cache_control + content_copy.append(block_dict) + message_dict["content"] = content_copy else: # Content is a string, convert to structured format - message_copy["content"] = [ + message_dict["content"] = [ { "type": "text", "text": content, @@ -118,7 +120,8 @@ class OpenrouterConfig(OpenAIGPTConfig): } ] - transformed_messages.append(message_copy) + # Cast back to AllMessageValues after modification + transformed_messages.append(cast(AllMessageValues, message_dict)) return transformed_messages diff --git a/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py b/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py index ffdcfe8679..f8baddd1d3 100644 --- a/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py +++ b/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py @@ -1019,6 +1019,46 @@ class MCPServerManager: verbose_logger.error(f"Guardrail blocked MCP tool call pre call: {str(e)}") raise e + def _create_during_hook_task( + self, + name: str, + arguments: Dict[str, Any], + server_name_from_prefix: Optional[str], + user_api_key_auth: Optional[UserAPIKeyAuth], + proxy_logging_obj: ProxyLogging, + start_time: datetime.datetime, + ): + """Create and return a during hook task for MCP tool calls.""" + from litellm.types.llms.base import HiddenParams + from litellm.types.mcp import MCPDuringCallRequestObject + + request_obj = MCPDuringCallRequestObject( + tool_name=name, + arguments=arguments, + server_name=server_name_from_prefix, + start_time=start_time.timestamp() if start_time else None, + hidden_params=HiddenParams(), + ) + + during_hook_kwargs = { + "name": name, + "arguments": arguments, + "server_name": server_name_from_prefix, + "user_api_key_auth": user_api_key_auth, + } + + synthetic_llm_data = proxy_logging_obj._convert_mcp_to_llm_format( + request_obj, during_hook_kwargs + ) + + return asyncio.create_task( + proxy_logging_obj.during_call_hook( + user_api_key_dict=user_api_key_auth, + data=synthetic_llm_data, + call_type="mcp_call", # type: ignore + ) + ) + async def call_tool( self, name: str, @@ -1085,35 +1125,13 @@ class MCPServerManager: # Prepare tasks for during hooks tasks = [] if proxy_logging_obj: - # Create synthetic LLM data for during hook processing - from litellm.types.llms.base import HiddenParams - from litellm.types.mcp import MCPDuringCallRequestObject - - request_obj = MCPDuringCallRequestObject( - tool_name=name, + during_hook_task = self._create_during_hook_task( + name=name, arguments=arguments, - server_name=server_name_from_prefix, - start_time=start_time.timestamp() if start_time else None, - hidden_params=HiddenParams(), - ) - - during_hook_kwargs = { - "name": name, - "arguments": arguments, - "server_name": server_name_from_prefix, - "user_api_key_auth": user_api_key_auth, - } - - synthetic_llm_data = proxy_logging_obj._convert_mcp_to_llm_format( - request_obj, during_hook_kwargs - ) - - during_hook_task = asyncio.create_task( - proxy_logging_obj.during_call_hook( - user_api_key_dict=user_api_key_auth, - data=synthetic_llm_data, - call_type="mcp_call", # type: ignore - ) + server_name_from_prefix=server_name_from_prefix, + user_api_key_auth=user_api_key_auth, + proxy_logging_obj=proxy_logging_obj, + start_time=start_time, ) tasks.append(during_hook_task) diff --git a/litellm/proxy/guardrails/guardrail_hooks/enkryptai/__init__.py b/litellm/proxy/guardrails/guardrail_hooks/enkryptai/__init__.py index 9289693c2c..ab62679a60 100644 --- a/litellm/proxy/guardrails/guardrail_hooks/enkryptai/__init__.py +++ b/litellm/proxy/guardrails/guardrail_hooks/enkryptai/__init__.py @@ -7,8 +7,6 @@ from typing import TYPE_CHECKING from litellm.types.guardrails import SupportedGuardrailIntegrations -from .enkryptai import EnkryptAIGuardrails - if TYPE_CHECKING: from litellm.types.guardrails import Guardrail, LitellmParams diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index 50963be1ba..0104372d22 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -257,7 +257,9 @@ from litellm.proxy.management_endpoints.customer_endpoints import ( from litellm.proxy.management_endpoints.internal_user_endpoints import ( router as internal_user_router, ) -from litellm.proxy.management_endpoints.internal_user_endpoints import user_update +from litellm.proxy.management_endpoints.internal_user_endpoints import ( + user_update, +) from litellm.proxy.management_endpoints.key_management_endpoints import ( delete_verification_tokens, duration_in_seconds, @@ -304,7 +306,9 @@ from litellm.proxy.middleware.prometheus_auth_middleware import PrometheusAuthMi from litellm.proxy.openai_files_endpoints.files_endpoints import ( router as openai_files_router, ) -from litellm.proxy.openai_files_endpoints.files_endpoints import set_files_config +from litellm.proxy.openai_files_endpoints.files_endpoints import ( + set_files_config, +) from litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints import ( passthrough_endpoint_router, ) @@ -561,6 +565,31 @@ async def proxy_shutdown_event(): cleanup_router_config_variables() +async def _initialize_shared_aiohttp_session(): + """Initialize shared aiohttp session for connection reuse.""" + try: + from aiohttp import ClientSession, TCPConnector + + # Create connector with connection pooling settings optimized for long-lived connections + connector = TCPConnector( + limit=AIOHTTP_CONNECTOR_LIMIT, + keepalive_timeout=AIOHTTP_KEEPALIVE_TIMEOUT, + ttl_dns_cache=AIOHTTP_TTL_DNS_CACHE, + enable_cleanup_closed=True, + ) + + session = ClientSession(connector=connector) + verbose_proxy_logger.info( + f"SESSION REUSE: Created shared aiohttp session for connection pooling (ID: {id(session)})" + ) + return session + except Exception as e: + verbose_proxy_logger.warning( + f"Failed to create shared aiohttp session: {e}. Continuing without session reuse." + ) + return None + + @asynccontextmanager async def proxy_startup_event(app: FastAPI): global prisma_client, master_key, use_background_health_checks, llm_router, llm_model_list, general_settings, proxy_budget_rescheduler_min_time, proxy_budget_rescheduler_max_time, litellm_proxy_admin_name, db_writer_client, store_model_in_db, premium_user, _license_check, proxy_batch_polling_interval, shared_aiohttp_session @@ -679,25 +708,7 @@ async def proxy_startup_event(app: FastAPI): ProxyStartupEvent._init_dd_tracer() ## Initialize shared aiohttp session for connection reuse - try: - from aiohttp import ClientSession, TCPConnector - - # Create connector with connection pooling settings optimized for long-lived connections - connector = TCPConnector( - limit=AIOHTTP_CONNECTOR_LIMIT, - keepalive_timeout=AIOHTTP_KEEPALIVE_TIMEOUT, - ttl_dns_cache=AIOHTTP_TTL_DNS_CACHE, - enable_cleanup_closed=True, - ) - - shared_aiohttp_session = ClientSession(connector=connector) - verbose_proxy_logger.info( - f"SESSION REUSE: Created shared aiohttp session for connection pooling (ID: {id(shared_aiohttp_session)})" - ) - except Exception as e: - verbose_proxy_logger.warning( - f"Failed to create shared aiohttp session: {e}. Continuing without session reuse." - ) + shared_aiohttp_session = await _initialize_shared_aiohttp_session() # End of startup event yield diff --git a/litellm/types/utils.py b/litellm/types/utils.py index e241390fa5..7c914706c2 100644 --- a/litellm/types/utils.py +++ b/litellm/types/utils.py @@ -2254,7 +2254,6 @@ class CustomPricingLiteLLMParams(BaseModel): input_cost_per_token_cache_hit: Optional[float] = None input_cost_per_token_above_128k_tokens: Optional[float] = None input_cost_per_token_above_200k_tokens: Optional[float] = None - input_cost_per_character_above_128k_tokens: Optional[float] = None input_cost_per_query: Optional[float] = None input_cost_per_image: Optional[float] = None input_cost_per_image_above_128k_tokens: Optional[float] = None diff --git a/tests/test_litellm/llms/openrouter/chat/test_openrouter_chat_transformation.py b/tests/test_litellm/llms/openrouter/chat/test_openrouter_chat_transformation.py index ff81ba84c0..660ced53e5 100644 --- a/tests/test_litellm/llms/openrouter/chat/test_openrouter_chat_transformation.py +++ b/tests/test_litellm/llms/openrouter/chat/test_openrouter_chat_transformation.py @@ -187,7 +187,8 @@ def test_openrouter_transform_request_with_cache_control(): def test_openrouter_transform_request_with_cache_control_list_content(): """ - Test transform_request moves cache_control to all content blocks when content is already a list. + Test transform_request moves cache_control only to the last content block when content is already a list. + This prevents exceeding Anthropic's limit of 4 cache breakpoints. Input: { @@ -205,8 +206,7 @@ def test_openrouter_transform_request_with_cache_control_list_content(): "content": [ { "type": "text", - "text": "You are a historian...", - "cache_control": {"type": "ephemeral"} + "text": "You are a historian..." }, { "type": "text", @@ -258,7 +258,8 @@ def test_openrouter_transform_request_with_cache_control_list_content(): assert system_message["role"] == "system" assert isinstance(system_message["content"], list) assert len(system_message["content"]) == 2 - assert system_message["content"][0]["cache_control"] == {"type": "ephemeral"} + # Only the last content block should have cache_control + assert "cache_control" not in system_message["content"][0] assert system_message["content"][1]["cache_control"] == {"type": "ephemeral"} assert "cache_control" not in system_message @@ -316,4 +317,51 @@ def test_openrouter_transform_request_with_cache_control_gemini(): assert isinstance(user_message["content"], list) assert user_message["content"][0]["type"] == "text" assert user_message["content"][0]["cache_control"] == {"type": "ephemeral"} + + +def test_openrouter_transform_request_multiple_cache_controls(): + """ + Test that cache_control is only added to the last content block per message. + This prevents exceeding Anthropic's limit of 4 cache breakpoints. + + When a message has 5 content blocks with cache_control at message level, + only the 5th block should have cache_control, not all 5 blocks. + """ + import json + config = OpenrouterConfig() + + messages = [ + { + "role": "system", + "content": [ + {"type": "text", "text": "Block 1"}, + {"type": "text", "text": "Block 2"}, + {"type": "text", "text": "Block 3"}, + {"type": "text", "text": "Block 4"}, + {"type": "text", "text": "Block 5"} + ], + "cache_control": {"type": "ephemeral"} + } + ] + + transformed_request = config.transform_request( + model="openrouter/anthropic/claude-3-5-sonnet-20240620", + messages=messages, + optional_params={}, + litellm_params={}, + headers={}, + ) + + print("\n=== Transformed Request (Multiple Blocks) ===") + print(json.dumps(transformed_request, indent=4, default=str)) + + system_message = transformed_request["messages"][0] + assert len(system_message["content"]) == 5 + + # Only the last block should have cache_control + for i in range(4): + assert "cache_control" not in system_message["content"][i], f"Block {i} should not have cache_control" + + assert system_message["content"][4]["cache_control"] == {"type": "ephemeral"} + assert "cache_control" not in system_message \ No newline at end of file