From 424363fb1d7037c8a82a0ae87f7ebbb59d823689 Mon Sep 17 00:00:00 2001 From: Ishaan Jaff Date: Wed, 17 Dec 2025 02:20:23 +0400 Subject: [PATCH] [Feat] Add New Google Interactions API on AI Gateway (#18081) * init BaseInteractionsAPIConfig * updated types for interactions API * fix SDK level interactions API * InteractionsAPIRequestUtils * init base * init interactions API * init Interactions API Types * init interactins API * GoogleAIStudioInteractionsConfig * init http handler * remove file no longer needed * test OPENAPI_SPEC_URL * TestGoogleInteractionsCreate * add google interctions routes * init interactions api * init interactions api * new routes * fix config * working streaming interactons api --- litellm/interactions/http_handler.py | 64 ++-- litellm/interactions/streaming_iterator.py | 266 +++++++++++++++ .../gemini/interactions/transformation.py | 2 +- litellm/proxy/_types.py | 7 + litellm/proxy/common_request_processing.py | 8 + litellm/proxy/google_endpoints/endpoints.py | 322 +++++++++++++++++- litellm/proxy/proxy_config.yaml | 19 +- litellm/proxy/route_llm_request.py | 16 + litellm/router.py | 78 ++++- 9 files changed, 715 insertions(+), 67 deletions(-) create mode 100644 litellm/interactions/streaming_iterator.py diff --git a/litellm/interactions/http_handler.py b/litellm/interactions/http_handler.py index 76aae6d5d4..5555b1d7e3 100644 --- a/litellm/interactions/http_handler.py +++ b/litellm/interactions/http_handler.py @@ -20,6 +20,10 @@ import httpx import litellm from litellm._logging import verbose_logger from litellm.constants import request_timeout +from litellm.interactions.streaming_iterator import ( + InteractionsAPIStreamingIterator, + SyncInteractionsAPIStreamingIterator, +) from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj from litellm.llms.base_llm.interactions.transformation import BaseInteractionsAPIConfig from litellm.llms.custom_httpx.http_handler import ( @@ -283,63 +287,37 @@ class InteractionsHTTPHandler: model: Optional[str], logging_obj: LiteLLMLoggingObj, interactions_api_config: BaseInteractionsAPIConfig, - ) -> Iterator[InteractionsAPIStreamingResponse]: + ) -> SyncInteractionsAPIStreamingIterator: """Create a synchronous streaming iterator. Google AI's streaming format uses SSE (Server-Sent Events). + Returns a proper streaming iterator that yields chunks as they arrive. """ - # Read the full response and parse as JSON array - full_content = response.read().decode("utf-8") - try: - chunks = json.loads(full_content) - if isinstance(chunks, list): - for parsed_chunk in chunks: - yield interactions_api_config.transform_streaming_response( - model=model, - parsed_chunk=parsed_chunk, - logging_obj=logging_obj, - ) - else: - # Single object response - yield interactions_api_config.transform_streaming_response( - model=model, - parsed_chunk=chunks, - logging_obj=logging_obj, - ) - except json.JSONDecodeError as e: - verbose_logger.debug(f"Failed to parse streaming response: {full_content[:500]}..., error: {e}") + return SyncInteractionsAPIStreamingIterator( + response=response, + model=model, + interactions_api_config=interactions_api_config, + logging_obj=logging_obj, + ) - async def _create_async_streaming_iterator( + def _create_async_streaming_iterator( self, response: httpx.Response, model: Optional[str], logging_obj: LiteLLMLoggingObj, interactions_api_config: BaseInteractionsAPIConfig, - ) -> AsyncIterator[InteractionsAPIStreamingResponse]: + ) -> InteractionsAPIStreamingIterator: """Create an asynchronous streaming iterator. Google AI's streaming format uses SSE (Server-Sent Events). + Returns a proper streaming iterator that yields chunks as they arrive. """ - # Read the full response and parse as JSON array - full_content = (await response.aread()).decode("utf-8") - try: - chunks = json.loads(full_content) - if isinstance(chunks, list): - for parsed_chunk in chunks: - yield interactions_api_config.transform_streaming_response( - model=model, - parsed_chunk=parsed_chunk, - logging_obj=logging_obj, - ) - else: - # Single object response - yield interactions_api_config.transform_streaming_response( - model=model, - parsed_chunk=chunks, - logging_obj=logging_obj, - ) - except json.JSONDecodeError as e: - verbose_logger.debug(f"Failed to parse async streaming response: {full_content[:500]}..., error: {e}") + return InteractionsAPIStreamingIterator( + response=response, + model=model, + interactions_api_config=interactions_api_config, + logging_obj=logging_obj, + ) # ========================================================= # GET INTERACTION diff --git a/litellm/interactions/streaming_iterator.py b/litellm/interactions/streaming_iterator.py new file mode 100644 index 0000000000..ad18477663 --- /dev/null +++ b/litellm/interactions/streaming_iterator.py @@ -0,0 +1,266 @@ +""" +Streaming iterators for the Interactions API. + +This module provides streaming iterators that properly stream SSE responses +from the Google Interactions API, similar to the responses API streaming iterator. +""" + +import asyncio +import json +from datetime import datetime +from typing import Any, Dict, Iterator, Optional + +import httpx + +import litellm +from litellm._logging import verbose_logger +from litellm.constants import STREAM_SSE_DONE_STRING +from litellm.litellm_core_utils.asyncify import run_async_function +from litellm.litellm_core_utils.core_helpers import process_response_headers +from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj +from litellm.litellm_core_utils.llm_response_utils.get_api_base import get_api_base +from litellm.litellm_core_utils.thread_pool_executor import executor +from litellm.llms.base_llm.interactions.transformation import BaseInteractionsAPIConfig +from litellm.types.interactions import ( + InteractionsAPIResponse, + InteractionsAPIStreamingResponse, +) +from litellm.utils import CustomStreamWrapper + + +class BaseInteractionsAPIStreamingIterator: + """ + Base class for streaming iterators that process responses from the Interactions API. + + This class contains shared logic for both synchronous and asynchronous iterators. + """ + + def __init__( + self, + response: httpx.Response, + model: Optional[str], + interactions_api_config: BaseInteractionsAPIConfig, + logging_obj: LiteLLMLoggingObj, + litellm_metadata: Optional[Dict[str, Any]] = None, + custom_llm_provider: Optional[str] = None, + ): + self.response = response + self.model = model + self.logging_obj = logging_obj + self.finished = False + self.interactions_api_config = interactions_api_config + self.completed_response: Optional[InteractionsAPIStreamingResponse] = None + self.start_time = datetime.now() + + # set request kwargs + self.litellm_metadata = litellm_metadata + self.custom_llm_provider = custom_llm_provider + + # set hidden params for response headers + _api_base = get_api_base( + model=model or "", + optional_params=self.logging_obj.model_call_details.get( + "litellm_params", {} + ), + ) + _model_info: Dict = litellm_metadata.get("model_info", {}) if litellm_metadata else {} + self._hidden_params = { + "model_id": _model_info.get("id", None), + "api_base": _api_base, + } + self._hidden_params["additional_headers"] = process_response_headers( + self.response.headers or {} + ) + + def _process_chunk(self, chunk: str) -> Optional[InteractionsAPIStreamingResponse]: + """Process a single chunk of data from the stream.""" + if not chunk: + return None + + # Handle SSE format (data: {...}) + stripped_chunk = CustomStreamWrapper._strip_sse_data_from_chunk(chunk) + if stripped_chunk is None: + return None + + # Handle "[DONE]" marker + if stripped_chunk == STREAM_SSE_DONE_STRING: + self.finished = True + return None + + try: + # Parse the JSON chunk + parsed_chunk = json.loads(stripped_chunk) + + # Format as InteractionsAPIStreamingResponse + if isinstance(parsed_chunk, dict): + streaming_response = self.interactions_api_config.transform_streaming_response( + model=self.model, + parsed_chunk=parsed_chunk, + logging_obj=self.logging_obj, + ) + + # Store the completed response (check for status=completed) + if ( + streaming_response + and getattr(streaming_response, "status", None) == "completed" + ): + self.completed_response = streaming_response + self._handle_logging_completed_response() + + return streaming_response + + return None + except json.JSONDecodeError: + # If we can't parse the chunk, continue + verbose_logger.debug(f"Failed to parse streaming chunk: {stripped_chunk[:200]}...") + return None + + def _handle_logging_completed_response(self): + """Base implementation - should be overridden by subclasses.""" + pass + + +class InteractionsAPIStreamingIterator(BaseInteractionsAPIStreamingIterator): + """ + Async iterator for processing streaming responses from the Interactions API. + """ + + def __init__( + self, + response: httpx.Response, + model: Optional[str], + interactions_api_config: BaseInteractionsAPIConfig, + logging_obj: LiteLLMLoggingObj, + litellm_metadata: Optional[Dict[str, Any]] = None, + custom_llm_provider: Optional[str] = None, + ): + super().__init__( + response=response, + model=model, + interactions_api_config=interactions_api_config, + logging_obj=logging_obj, + litellm_metadata=litellm_metadata, + custom_llm_provider=custom_llm_provider, + ) + self.stream_iterator = response.aiter_lines() + + def __aiter__(self): + return self + + async def __anext__(self) -> InteractionsAPIStreamingResponse: + try: + while True: + # Get the next chunk from the stream + try: + chunk = await self.stream_iterator.__anext__() + except StopAsyncIteration: + self.finished = True + raise StopAsyncIteration + + result = self._process_chunk(chunk) + + if self.finished: + raise StopAsyncIteration + elif result is not None: + return result + # If result is None, continue the loop to get the next chunk + + except httpx.HTTPError as e: + # Handle HTTP errors + self.finished = True + raise e + + def _handle_logging_completed_response(self): + """Handle logging for completed responses in async context.""" + import copy + logging_response = copy.deepcopy(self.completed_response) + + asyncio.create_task( + self.logging_obj.async_success_handler( + result=logging_response, + start_time=self.start_time, + end_time=datetime.now(), + cache_hit=None, + ) + ) + + executor.submit( + self.logging_obj.success_handler, + result=logging_response, + cache_hit=None, + start_time=self.start_time, + end_time=datetime.now(), + ) + + +class SyncInteractionsAPIStreamingIterator(BaseInteractionsAPIStreamingIterator): + """ + Synchronous iterator for processing streaming responses from the Interactions API. + """ + + def __init__( + self, + response: httpx.Response, + model: Optional[str], + interactions_api_config: BaseInteractionsAPIConfig, + logging_obj: LiteLLMLoggingObj, + litellm_metadata: Optional[Dict[str, Any]] = None, + custom_llm_provider: Optional[str] = None, + ): + super().__init__( + response=response, + model=model, + interactions_api_config=interactions_api_config, + logging_obj=logging_obj, + litellm_metadata=litellm_metadata, + custom_llm_provider=custom_llm_provider, + ) + self.stream_iterator = response.iter_lines() + + def __iter__(self): + return self + + def __next__(self) -> InteractionsAPIStreamingResponse: + try: + while True: + # Get the next chunk from the stream + try: + chunk = next(self.stream_iterator) + except StopIteration: + self.finished = True + raise StopIteration + + result = self._process_chunk(chunk) + + if self.finished: + raise StopIteration + elif result is not None: + return result + # If result is None, continue the loop to get the next chunk + + except httpx.HTTPError as e: + # Handle HTTP errors + self.finished = True + raise e + + def _handle_logging_completed_response(self): + """Handle logging for completed responses in sync context.""" + import copy + logging_response = copy.deepcopy(self.completed_response) + + run_async_function( + async_function=self.logging_obj.async_success_handler, + result=logging_response, + start_time=self.start_time, + end_time=datetime.now(), + cache_hit=None, + ) + + executor.submit( + self.logging_obj.success_handler, + result=logging_response, + cache_hit=None, + start_time=self.start_time, + end_time=datetime.now(), + ) + diff --git a/litellm/llms/gemini/interactions/transformation.py b/litellm/llms/gemini/interactions/transformation.py index d9ff468fd7..d21775eb23 100644 --- a/litellm/llms/gemini/interactions/transformation.py +++ b/litellm/llms/gemini/interactions/transformation.py @@ -123,7 +123,7 @@ class GoogleAIStudioInteractionsConfig(BaseInteractionsAPIConfig): # Pass through optional params directly (they match the spec) optional_keys = [ - "tools", "system_instruction", "generation_config", "store", + "tools", "system_instruction", "generation_config", "stream", "store", "background", "response_modalities", "response_format", "response_mime_type", "previous_interaction_id", ] diff --git a/litellm/proxy/_types.py b/litellm/proxy/_types.py index fcc4097e45..feafe35c80 100644 --- a/litellm/proxy/_types.py +++ b/litellm/proxy/_types.py @@ -418,6 +418,13 @@ class LiteLLMRoutes(enum.Enum): "/models/{model_name}:countTokens", "/models/{model_name}:generateContent", "/models/{model_name}:streamGenerateContent", + # Google Interactions API + "/interactions", + "/v1beta/interactions", + "/interactions/{interaction_id}", + "/v1beta/interactions/{interaction_id}", + "/interactions/{interaction_id}/cancel", + "/v1beta/interactions/{interaction_id}/cancel", ] apply_guardrail_routes = [ diff --git a/litellm/proxy/common_request_processing.py b/litellm/proxy/common_request_processing.py index 3f04ce3933..8637bc88c5 100644 --- a/litellm/proxy/common_request_processing.py +++ b/litellm/proxy/common_request_processing.py @@ -351,6 +351,10 @@ class ProxyBaseLLMRequestProcessing: "aget_skill", "adelete_skill", "anthropic_messages", + "acreate_interaction", + "aget_interaction", + "adelete_interaction", + "acancel_interaction", ], version: Optional[str] = None, user_model: Optional[str] = None, @@ -476,6 +480,10 @@ class ProxyBaseLLMRequestProcessing: "aget_skill", "adelete_skill", "anthropic_messages", + "acreate_interaction", + "aget_interaction", + "adelete_interaction", + "acancel_interaction", ], proxy_logging_obj: ProxyLogging, general_settings: dict, diff --git a/litellm/proxy/google_endpoints/endpoints.py b/litellm/proxy/google_endpoints/endpoints.py index 72620259b1..cc0719235a 100644 --- a/litellm/proxy/google_endpoints/endpoints.py +++ b/litellm/proxy/google_endpoints/endpoints.py @@ -1,9 +1,11 @@ -from fastapi import APIRouter, Depends, Request, Response, HTTPException -from fastapi.responses import StreamingResponse +from typing import Optional + +from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response +from fastapi.responses import ORJSONResponse, StreamingResponse from litellm.proxy._types import * from litellm.proxy.auth.user_api_key_auth import UserAPIKeyAuth, user_api_key_auth - +from litellm.proxy.common_request_processing import ProxyBaseLLMRequestProcessing from litellm.proxy.common_utils.http_parsing_utils import _read_request_body from litellm.types.llms.vertex_ai import TokenCountDetailsResponse @@ -25,8 +27,13 @@ async def google_generate_content( fastapi_response: Response, user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), ): - from litellm.proxy.proxy_server import llm_router, general_settings, proxy_config, version from litellm.proxy.litellm_pre_call_utils import add_litellm_data_to_request + from litellm.proxy.proxy_server import ( + general_settings, + llm_router, + proxy_config, + version, + ) data = await _read_request_body(request=request) if "model" not in data: @@ -63,8 +70,13 @@ async def google_stream_generate_content( fastapi_response: Response, user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), ): - from litellm.proxy.proxy_server import llm_router, general_settings, proxy_config, version from litellm.proxy.litellm_pre_call_utils import add_litellm_data_to_request + from litellm.proxy.proxy_server import ( + general_settings, + llm_router, + proxy_config, + version, + ) data = await _read_request_body(request=request) @@ -89,8 +101,8 @@ async def google_stream_generate_content( response = await llm_router.agenerate_content_stream(**data) # Check if response is an async iterator (streaming response) - if hasattr(response, "__aiter__"): - return StreamingResponse(response, media_type="text/event-stream") + if response is not None and hasattr(response, "__aiter__"): + return StreamingResponse(content=response, media_type="text/event-stream") return response @@ -167,3 +179,299 @@ async def google_count_tokens(request: Request, model_name: str): totalTokens=0, promptTokensDetails=[], ) + + +# ============================================================ +# Google Interactions API Endpoints +# Per OpenAPI spec: https://ai.google.dev/static/api/interactions.openapi.json +# ============================================================ + + +@router.post( + "/v1beta/interactions", + dependencies=[Depends(user_api_key_auth)], + response_class=ORJSONResponse, + tags=["interactions"], +) +@router.post( + "/interactions", + dependencies=[Depends(user_api_key_auth)], + response_class=ORJSONResponse, + tags=["interactions"], +) +async def create_interaction( + request: Request, + fastapi_response: Response, + user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), +): + """ + Create a new interaction using Google's Interactions API. + + Per OpenAPI spec: POST /{api_version}/interactions + + Supports both model interactions and agent interactions: + - Model: Provide `model` parameter (e.g., "gemini-2.5-flash") + - Agent: Provide `agent` parameter (e.g., "deep-research-pro-preview-12-2025") + + Example: + ```bash + curl -X POST "http://localhost:4000/v1beta/interactions" \ + -H "Authorization: Bearer sk-1234" \ + -H "Content-Type: application/json" \ + -d '{ + "model": "gemini/gemini-2.5-flash", + "input": "Hello, how are you?" + }' + ``` + """ + from litellm.proxy.proxy_server import ( + general_settings, + llm_router, + proxy_config, + proxy_logging_obj, + select_data_generator, + user_api_base, + user_max_tokens, + user_model, + user_request_timeout, + user_temperature, + version, + ) + + data = await _read_request_body(request=request) + + # Default to gemini provider for interactions + if "custom_llm_provider" not in data: + data["custom_llm_provider"] = "gemini" + + processor = ProxyBaseLLMRequestProcessing(data=data) + try: + return await processor.base_process_llm_request( + request=request, + fastapi_response=fastapi_response, + user_api_key_dict=user_api_key_dict, + route_type="acreate_interaction", + proxy_logging_obj=proxy_logging_obj, + llm_router=llm_router, + general_settings=general_settings, + proxy_config=proxy_config, + select_data_generator=select_data_generator, + model=data.get("model"), + user_model=user_model, + user_temperature=user_temperature, + user_request_timeout=user_request_timeout, + user_max_tokens=user_max_tokens, + user_api_base=user_api_base, + version=version, + ) + except Exception as e: + raise await processor._handle_llm_api_exception( + e=e, + user_api_key_dict=user_api_key_dict, + proxy_logging_obj=proxy_logging_obj, + version=version, + ) + + +@router.get( + "/v1beta/interactions/{interaction_id}", + dependencies=[Depends(user_api_key_auth)], + response_class=ORJSONResponse, + tags=["interactions"], +) +@router.get( + "/interactions/{interaction_id}", + dependencies=[Depends(user_api_key_auth)], + response_class=ORJSONResponse, + tags=["interactions"], +) +async def get_interaction( + request: Request, + interaction_id: str, + fastapi_response: Response, + user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), +): + """ + Get an interaction by ID. + + Per OpenAPI spec: GET /{api_version}/interactions/{interaction_id} + """ + from litellm.proxy.proxy_server import ( + general_settings, + llm_router, + proxy_config, + proxy_logging_obj, + select_data_generator, + user_api_base, + user_max_tokens, + user_model, + user_request_timeout, + user_temperature, + version, + ) + + data = {"interaction_id": interaction_id, "custom_llm_provider": "gemini"} + + processor = ProxyBaseLLMRequestProcessing(data=data) + try: + return await processor.base_process_llm_request( + request=request, + fastapi_response=fastapi_response, + user_api_key_dict=user_api_key_dict, + route_type="aget_interaction", + proxy_logging_obj=proxy_logging_obj, + llm_router=llm_router, + general_settings=general_settings, + proxy_config=proxy_config, + select_data_generator=select_data_generator, + model=None, + user_model=user_model, + user_temperature=user_temperature, + user_request_timeout=user_request_timeout, + user_max_tokens=user_max_tokens, + user_api_base=user_api_base, + version=version, + ) + except Exception as e: + raise await processor._handle_llm_api_exception( + e=e, + user_api_key_dict=user_api_key_dict, + proxy_logging_obj=proxy_logging_obj, + version=version, + ) + + +@router.delete( + "/v1beta/interactions/{interaction_id}", + dependencies=[Depends(user_api_key_auth)], + response_class=ORJSONResponse, + tags=["interactions"], +) +@router.delete( + "/interactions/{interaction_id}", + dependencies=[Depends(user_api_key_auth)], + response_class=ORJSONResponse, + tags=["interactions"], +) +async def delete_interaction( + request: Request, + interaction_id: str, + fastapi_response: Response, + user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), +): + """ + Delete an interaction by ID. + + Per OpenAPI spec: DELETE /{api_version}/interactions/{interaction_id} + """ + from litellm.proxy.proxy_server import ( + general_settings, + llm_router, + proxy_config, + proxy_logging_obj, + select_data_generator, + user_api_base, + user_max_tokens, + user_model, + user_request_timeout, + user_temperature, + version, + ) + + data = {"interaction_id": interaction_id, "custom_llm_provider": "gemini"} + + processor = ProxyBaseLLMRequestProcessing(data=data) + try: + return await processor.base_process_llm_request( + request=request, + fastapi_response=fastapi_response, + user_api_key_dict=user_api_key_dict, + route_type="adelete_interaction", + proxy_logging_obj=proxy_logging_obj, + llm_router=llm_router, + general_settings=general_settings, + proxy_config=proxy_config, + select_data_generator=select_data_generator, + model=None, + user_model=user_model, + user_temperature=user_temperature, + user_request_timeout=user_request_timeout, + user_max_tokens=user_max_tokens, + user_api_base=user_api_base, + version=version, + ) + except Exception as e: + raise await processor._handle_llm_api_exception( + e=e, + user_api_key_dict=user_api_key_dict, + proxy_logging_obj=proxy_logging_obj, + version=version, + ) + + +@router.post( + "/v1beta/interactions/{interaction_id}/cancel", + dependencies=[Depends(user_api_key_auth)], + response_class=ORJSONResponse, + tags=["interactions"], +) +@router.post( + "/interactions/{interaction_id}/cancel", + dependencies=[Depends(user_api_key_auth)], + response_class=ORJSONResponse, + tags=["interactions"], +) +async def cancel_interaction( + request: Request, + interaction_id: str, + fastapi_response: Response, + user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), +): + """ + Cancel an interaction by ID. + + Per OpenAPI spec: POST /{api_version}/interactions/{interaction_id}:cancel + """ + from litellm.proxy.proxy_server import ( + general_settings, + llm_router, + proxy_config, + proxy_logging_obj, + select_data_generator, + user_api_base, + user_max_tokens, + user_model, + user_request_timeout, + user_temperature, + version, + ) + + data = {"interaction_id": interaction_id, "custom_llm_provider": "gemini"} + + processor = ProxyBaseLLMRequestProcessing(data=data) + try: + return await processor.base_process_llm_request( + request=request, + fastapi_response=fastapi_response, + user_api_key_dict=user_api_key_dict, + route_type="acancel_interaction", + proxy_logging_obj=proxy_logging_obj, + llm_router=llm_router, + general_settings=general_settings, + proxy_config=proxy_config, + select_data_generator=select_data_generator, + model=None, + user_model=user_model, + user_temperature=user_temperature, + user_request_timeout=user_request_timeout, + user_max_tokens=user_max_tokens, + user_api_base=user_api_base, + version=version, + ) + except Exception as e: + raise await processor._handle_llm_api_exception( + e=e, + user_api_key_dict=user_api_key_dict, + proxy_logging_obj=proxy_logging_obj, + version=version, + ) diff --git a/litellm/proxy/proxy_config.yaml b/litellm/proxy/proxy_config.yaml index 47b8f2e945..af74853d82 100644 --- a/litellm/proxy/proxy_config.yaml +++ b/litellm/proxy/proxy_config.yaml @@ -1,20 +1,9 @@ model_list: - - model_name: openai/gpt-4o-mini + - model_name: gemini/* litellm_params: - model: openai/gpt-4o-mini - tpm: 1000 + model: gemini/* + - # LangGraph models - - model_name: langgraph/* - litellm_params: - model: langgraph/* litellm_settings: - callbacks: ["dynamic_rate_limiter_v3"] - priority_reservation: - "prod": 0.9 # 90% reserved for production - "dev": 0.1 # 10% reserved for development - - - - + callbacks: ["langfuse"] diff --git a/litellm/proxy/route_llm_request.py b/litellm/proxy/route_llm_request.py index 6b86d722b2..fd00cfc1c0 100644 --- a/litellm/proxy/route_llm_request.py +++ b/litellm/proxy/route_llm_request.py @@ -46,6 +46,11 @@ ROUTE_ENDPOINT_MAPPING = { "aget_skill": "/skills/{skill_id}", "adelete_skill": "/skills/{skill_id}", "aingest": "/rag/ingest", + # Google Interactions API routes + "acreate_interaction": "/interactions", + "aget_interaction": "/interactions/{interaction_id}", + "adelete_interaction": "/interactions/{interaction_id}", + "acancel_interaction": "/interactions/{interaction_id}/cancel", } @@ -147,6 +152,10 @@ async def route_request( "adelete_skill", "aingest", "anthropic_messages", + "acreate_interaction", + "aget_interaction", + "adelete_interaction", + "acancel_interaction", ], ): """ @@ -199,6 +208,13 @@ async def route_request( "aretrieve_container_file_content", ]: return getattr(llm_router, f"{route_type}")(**data) + # Interactions API: get/delete/cancel don't need model routing + if route_type in [ + "aget_interaction", + "adelete_interaction", + "acancel_interaction", + ]: + return getattr(llm_router, f"{route_type}")(**data) if route_type in [ "avideo_list", "avideo_status", diff --git a/litellm/router.py b/litellm/router.py index 5e6027671b..abb26456be 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -1065,8 +1065,44 @@ class Router: litellm.adelete_skill, call_type="adelete_skill" ) + def _initialize_interactions_endpoints(self): + """Initialize Google Interactions API endpoints.""" + from litellm.interactions import acancel as acancel_interaction + from litellm.interactions import acreate as acreate_interaction + from litellm.interactions import adelete as adelete_interaction + from litellm.interactions import aget as aget_interaction + from litellm.interactions import cancel as cancel_interaction + from litellm.interactions import create as create_interaction + from litellm.interactions import delete as delete_interaction + from litellm.interactions import get as get_interaction + + self.acreate_interaction = self.factory_function( + acreate_interaction, call_type="acreate_interaction" + ) + self.create_interaction = self.factory_function( + create_interaction, call_type="create_interaction" + ) + self.aget_interaction = self.factory_function( + aget_interaction, call_type="aget_interaction" + ) + self.get_interaction = self.factory_function( + get_interaction, call_type="get_interaction" + ) + self.adelete_interaction = self.factory_function( + adelete_interaction, call_type="adelete_interaction" + ) + self.delete_interaction = self.factory_function( + delete_interaction, call_type="delete_interaction" + ) + self.acancel_interaction = self.factory_function( + acancel_interaction, call_type="acancel_interaction" + ) + self.cancel_interaction = self.factory_function( + cancel_interaction, call_type="cancel_interaction" + ) + def _initialize_specialized_endpoints(self): - """Helper to initialize specialized router endpoints (vector store, OCR, search, video, container, skills).""" + """Helper to initialize specialized router endpoints (vector store, OCR, search, video, container, skills, interactions).""" self._initialize_vector_store_endpoints() self._initialize_vector_store_file_endpoints() self._initialize_google_genai_endpoints() @@ -1074,6 +1110,7 @@ class Router: self._initialize_video_endpoints() self._initialize_container_endpoints() self._initialize_skills_endpoints() + self._initialize_interactions_endpoints() def initialize_router_endpoints(self): self._initialize_core_endpoints() @@ -3858,6 +3895,14 @@ class Router: "alist_skills", "aget_skill", "adelete_skill", + "acreate_interaction", + "create_interaction", + "aget_interaction", + "get_interaction", + "adelete_interaction", + "delete_interaction", + "acancel_interaction", + "cancel_interaction", ] = "assistants", ): """ @@ -3979,6 +4024,8 @@ class Router: "alist_skills", "aget_skill", "adelete_skill", + "acreate_interaction", + "create_interaction", ): return await self._ageneric_api_call_with_fallbacks( original_function=original_function, @@ -4031,6 +4078,16 @@ class Router: client=client, **kwargs, ) + elif call_type in ( + "aget_interaction", + "adelete_interaction", + "acancel_interaction", + ): + return await self._init_interactions_api_endpoints( + original_function=original_function, + custom_llm_provider=custom_llm_provider, + **kwargs, + ) return async_wrapper @@ -4085,6 +4142,25 @@ class Router: **kwargs, ) + async def _init_interactions_api_endpoints( + self, + original_function: Callable, + custom_llm_provider: Optional[str] = None, + **kwargs, + ): + """ + Initialize the Interactions API endpoints on the router. + + GET, DELETE, CANCEL Interactions API Requests don't need model-based routing, + so we call the original function directly with the custom_llm_provider. + """ + if custom_llm_provider and "custom_llm_provider" not in kwargs: + kwargs["custom_llm_provider"] = custom_llm_provider + # Default to gemini for interactions API + if "custom_llm_provider" not in kwargs: + kwargs["custom_llm_provider"] = "gemini" + return await original_function(**kwargs) + async def _pass_through_assistants_endpoint_factory( self, original_function: Callable,