tiennm99/litellm
1
0
Fork 0
mirror of https://github.com/tiennm99/litellm.git synced 2026-07-13 01:04:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

[Feat] Add New Google Interactions API on AI Gateway (#18081)

Browse Source 
* 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
This commit is contained in:
Ishaan Jaff
2025-12-17 02:20:23 +04:00
committed by GitHub
parent c727c8216f
commit 424363fb1d
9 changed files with 715 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files
litellm/interactions/http_handler.py
+21 -43
View File
@@ -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
litellm/interactions/streaming_iterator.py
+266
View File
@@ -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(),
)
litellm/llms/gemini/interactions/transformation.py
+1 -1
View File
@@ -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",
]
litellm/proxy/_types.py
+7
View File
@@ -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 = [
litellm/proxy/common_request_processing.py
+8
View File
@@ -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,
litellm/proxy/google_endpoints/endpoints.py
+315 -7
View File
@@ -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,
)
litellm/proxy/proxy_config.yaml
+4 -15
View File
@@ -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"]
litellm/proxy/route_llm_request.py
+16
View File
@@ -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",
litellm/router.py
+77 -1
View File
@@ -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,