This commit is contained in:
Shi Peipei 2026-11-06 18:24:00 +00:00 • committed by GitHub
commit 9dd065d7c6
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
22 changed files with 1542 additions and 50 deletions

View file

@ -363,6 +363,7 @@ jobs:
tests/unit/llms/pg_vector
tests/unit/llms/pi
tests/unit/llms/publicai
tests/unit/llms/qca
tests/unit/llms/ragflow
tests/unit/llms/recraft
tests/unit/llms/reducto

View file

@ -0,0 +1,56 @@
# QCA managed interaction transport
## Status
Accepted.
## Context
QCA Cloud Agents exposes a session and event API rather than a single model inference request. A client creates a
session, appends a user message, then follows the session event stream until the agent returns to an idle or terminal
state. The Interactions API expects one logical request with a stable interaction ID, normalized steps, and an optional
stream.
The generic Interactions provider transformation supports one upstream HTTP request. Making it orchestrate several QCA
requests would mix transport state with request and response transformation. Routing QCA through the Responses API
bridge would also lose session continuation and event resume semantics.
## Decision
QCA uses a dedicated Interactions handler. It owns session creation, user event submission, SSE resume with
`Last-Event-ID`, terminal state detection, and normalization into LiteLLM Interactions responses.
An interaction ID contains the QCA session ID and accepted input event ID. Continuation reuses the session and submits a
new user event. The input event ID is also the SSE resume cursor so events emitted between event acceptance and stream
connection are not missed.
QCA credit fields remain provider usage extras under the normalized response usage object. LiteLLM does not convert
credits into a currency cost because QCA does not publish a stable conversion owned by LiteLLM.
The first implementation accepts async text interactions only. Unsupported tools, files, background execution, and
lifecycle operations fail before any upstream request.
## Alternatives considered
Extending the generic transformation was rejected because its one-request contract cannot represent session creation,
event submission, and streaming without provider-specific branches throughout the shared handler.
Mapping QCA to chat completions or Responses was rejected because neither surface preserves the accepted event cursor or
the managed session lifecycle.
## Consequences
The provider has a small dedicated transport module while the public Interactions entry point keeps one provider
dispatch branch. Router deployments can expose QCA through an ordinary model alias with server-owned credentials and an
environment ID.
The interaction ID is an opaque continuation token for clients. QCA lifecycle endpoints remain unsupported until their
semantics can be mapped without pretending a session delete or cancel is an interaction operation.
## References
QCA documents the session creation contract at <https://docs.qoder.com/cloud-agents/api/sessions/create> and event
submission at <https://docs.qoder.com/cloud-agents/api/sessions/events/send>.
The event stream and resume cursor are documented at <https://docs.qoder.com/cloud-agents/api/sessions/events/stream>
and <https://docs.qoder.com/cloud-agents/sse-initial-connection-behavior-change>.

View file

@ -0,0 +1,32 @@
# Qoder Cloud Agents budget enforcement
## Status
Accepted
## Context
Qoder Cloud Agents reports cumulative provider credits for a session. LiteLLM monetary budgets require a cost in a
configured currency for each request. Qoder does not publish a stable currency conversion, and charging the cumulative
session credits on every continuation would bill earlier turns more than once
## Decision
The proxy rejects Qoder Cloud Agents requests from callers with an enforced key, user, team, organization, end-user, or
model monetary budget. The rejection happens before creating or continuing a provider session. Callers without monetary
budgets can use Qoder Cloud Agents, and the response keeps the cumulative Qoder credits as provider usage fields
## Alternatives Considered
Converting credits with a fixed rate was rejected because LiteLLM does not own a stable Qoder credit price
Charging cumulative credits on every interaction was rejected because session continuations would repeatedly charge
earlier turns
Tracking session deltas in proxy memory was rejected because it would not be reliable across workers or restarts
## Consequences
Monetary budgets cannot silently admit unpriced Qoder usage. Operators that need budgeted access must wait for a stable,
deployment-owned conversion and durable per-session delta accounting. Existing unbudgeted integrations continue to
receive the provider credit fields without a guessed monetary cost

View file

@ -310,6 +310,47 @@ def create(
local_vars.update(kwargs)
optional_params: Final = InteractionsAPIRequestUtils.get_requested_interactions_api_optional_params(local_vars)
if custom_llm_provider == "qca":
from litellm.llms.qca.interactions import create_qca_interaction
if not _is_async:
raise litellm.UnsupportedParamsError(
message="QCA interactions require litellm.interactions.acreate",
llm_provider="qca",
model=model,
)
litellm_logging_obj.update_from_kwargs( # pyright: ignore[reportUnknownMemberType] # old API untyped
kwargs=kwargs,
model=model,
optional_params=dict(optional_params),
litellm_params={"litellm_call_id": litellm_call_id},
custom_llm_provider=custom_llm_provider,
)
return create_qca_interaction(
agent=agent or model,
input=input,
environment=environment,
previous_interaction_id=previous_interaction_id,
stream=stream,
api_key=litellm_params.api_key if isinstance(litellm_params.api_key, str) else None,
api_base=litellm_params.api_base,
timeout=timeout,
logging_obj=litellm_logging_obj,
unsupported_params={
"tools": tools,
"system_instruction": system_instruction,
"generation_config": generation_config,
"store": store,
"background": background,
"response_modalities": response_modalities,
"response_format": response_format,
"response_mime_type": response_mime_type,
"agent_config": agent_config,
"extra_headers": extra_headers,
"extra_body": extra_body,
},
)
# Check if this is a bridge provider (litellm_responses) - similar to responses API
# Either provider is explicitly "litellm_responses" or no config found (bridge to responses)
if custom_llm_provider == "litellm_responses" or interactions_api_config is None:

View file

@ -0,0 +1 @@

View file

@ -0,0 +1,3 @@
from litellm.llms.qca.interactions.handler import create_qca_interaction
__all__ = ["create_qca_interaction"]

View file

@ -0,0 +1,466 @@
from collections.abc import AsyncIterator, Mapping
from datetime import datetime, timezone
from typing import Final
from urllib.parse import urlsplit
import httpx
from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError
from litellm.exceptions import APIError, AuthenticationError, BadRequestError, RateLimitError
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.llms.custom_httpx.sse_lines import aiter_sse_lines
from litellm.types.interactions import Error, InteractionsAPIResponse, InteractionsAPIStreamingResponse
QCA_DEFAULT_API_BASE: Final = "https://api.qoder.com/api/v1/cloud"
_JSON_OBJECT: Final = TypeAdapter(dict[str, object])
_JSON_OBJECTS: Final = TypeAdapter(list[dict[str, object]])
class QCAEventContent(BaseModel):
model_config = ConfigDict(frozen=True, extra="ignore")
type: str
text: str | None = None
class QCAStopReason(BaseModel):
model_config = ConfigDict(frozen=True, extra="ignore")
type: str
class QCAEvent(BaseModel):
model_config = ConfigDict(frozen=True, extra="ignore")
id: str
type: str
content: tuple[QCAEventContent, ...] = ()
stop_reason: QCAStopReason | None = None
model_credits: float | None = None
sandbox_runtime_credits: float | None = None
total_credits: float | None = None
class QCASessionAgent(BaseModel):
model_config = ConfigDict(frozen=True, extra="ignore")
id: str
class QCASession(BaseModel):
model_config = ConfigDict(frozen=True, extra="ignore")
id: str
agent: str | QCASessionAgent | None = None
environment_id: str | None = None
class QCAEventEnvelope(BaseModel):
model_config = ConfigDict(frozen=True, extra="ignore")
data: QCAEvent | None = None
event: QCAEvent | None = None
class QCAEventsPage(BaseModel):
model_config = ConfigDict(frozen=True, extra="ignore")
data: tuple[QCAEvent, ...]
class QCAValidationError(ValueError):
pass
def _require_identifier(value: str | None, prefix: str, name: str) -> str:
if value is None or not value.startswith(prefix):
raise QCAValidationError(f"QCA {name} must start with {prefix!r}")
return value
def _api_base(value: str | None) -> str:
base: Final = (value or QCA_DEFAULT_API_BASE).rstrip("/")
parsed: Final = urlsplit(base)
invalid: Final = (
parsed.scheme not in {"http", "https"}
or not parsed.netloc
or parsed.username is not None
or bool(parsed.query)
or bool(parsed.fragment)
)
if invalid:
raise QCAValidationError("QCA api_base must be an absolute HTTP URL without credentials, query, or fragment")
return base
def _text_input(value: object) -> str:
if not isinstance(value, str) or not value:
raise QCAValidationError("QCA interactions currently require a non-empty text input")
return value
def _previous_session_id(previous_interaction_id: str | None) -> str | None:
if previous_interaction_id is None:
return None
parts: Final = previous_interaction_id.split(".")
if len(parts) != 3 or parts[0] != "qca1":
raise QCAValidationError("QCA previous_interaction_id is invalid")
_require_identifier(parts[2], "evt_", "event id")
return _require_identifier(parts[1], "sess_", "session id")
def _event_from_payload(payload: str) -> QCAEvent:
decoded: Final = _JSON_OBJECT.validate_json(payload)
envelope: Final = QCAEventEnvelope.model_validate(decoded)
return envelope.data or envelope.event or QCAEvent.model_validate(decoded)
def _accepted_event(payload: object) -> QCAEvent:
page: Final = QCAEventsPage.model_validate(payload)
if not page.data:
raise ValueError("QCA accepted the user message without returning its event")
return page.data[0]
def _provider_error(error: httpx.HTTPStatusError, model: str) -> Exception:
response: Final = error.response
message: Final = response.text or str(error)
match response.status_code:
case 401 | 403:
return AuthenticationError(message=message, llm_provider="qca", model=model, response=response)
case 429:
return RateLimitError(message=message, llm_provider="qca", model=model, response=response)
case status_code:
return APIError(
status_code=status_code,
message=message,
llm_provider="qca",
model=model,
request=response.request,
response=response,
)
def _interaction_id(session_id: str, input_event_id: str) -> str:
return f"qca1.{session_id}.{input_event_id}"
def _usage(event: QCAEvent) -> Mapping[str, object] | None:
if event.total_credits is None and event.model_credits is None and event.sandbox_runtime_credits is None:
return None
return {
"qca_session_credits": {
"model_credits": event.model_credits,
"sandbox_runtime_credits": event.sandbox_runtime_credits,
"total_credits": event.total_credits,
},
"qca_usage_scope": "session_cumulative",
"qca_usage_is_final": event.type == "session.status_idle",
}
def _text_step(event: QCAEvent) -> Mapping[str, object] | None:
text: Final = "".join(content.text or "" for content in event.content if content.type == "text")
if event.type != "agent.message" or not text:
return None
return {"type": "model_output", "content": [{"type": "text", "text": text}]}
def _text_delta(step: Mapping[str, object]) -> Mapping[str, object]:
try:
content: Final = _JSON_OBJECTS.validate_python(step.get("content"))
except ValidationError:
return {}
return content[0] if content else {}
def _step_from_delta(delta: Mapping[str, object]) -> Mapping[str, object]:
return {"type": "model_output", "content": [dict(delta)]}
def _terminal_status(event: QCAEvent) -> str | None:
if event.type in {"session.error", "session.deleted", "session.status_terminated"}:
return "failed"
if event.type != "session.status_idle":
return None
if event.stop_reason is not None and event.stop_reason.type == "requires_action":
return "requires_action"
return "completed"
def _normalized_event(event: QCAEvent) -> tuple[Mapping[str, object] | None, str | None]:
return _text_step(event), _terminal_status(event)
async def _stream_events(response: httpx.Response) -> AsyncIterator[QCAEvent]:
data_lines: tuple[str, ...] = () # rebind-ok: immutable SSE record accumulation
async for line in aiter_sse_lines(response.aiter_text()):
if line == "":
if data_lines:
yield _event_from_payload("\n".join(data_lines))
data_lines = ()
continue
if line.startswith("data:"):
data_lines = (*data_lines, line[5:].lstrip())
if data_lines:
yield _event_from_payload("\n".join(data_lines))
async def _normalized_stream(
response: httpx.Response, interaction_id: str, agent: str
) -> AsyncIterator[InteractionsAPIStreamingResponse]:
created: Final = datetime.now(timezone.utc).isoformat()
try:
yield InteractionsAPIStreamingResponse(
event_type="interaction.created",
interaction_id=interaction_id,
interaction={
"id": interaction_id,
"agent": agent,
"status": "in_progress",
"created": created,
"updated": created,
"steps": [],
},
)
async for event in _stream_events(response):
step, status = _normalized_event(event)
if step is not None:
yield InteractionsAPIStreamingResponse(
event_type="step.delta",
interaction_id=interaction_id,
index=0,
delta=dict(_text_delta(step)),
)
if status is not None:
if status == "failed":
yield InteractionsAPIStreamingResponse(
event_type="error",
interaction_id=interaction_id,
error=Error(
code="qca_session_failed",
message=f"Qoder Cloud Agents session ended with {event.type}",
),
)
return
yield InteractionsAPIStreamingResponse(
event_type=f"interaction.{status}",
interaction_id=interaction_id,
interaction={
"id": interaction_id,
"agent": agent,
"status": status,
"created": created,
"updated": datetime.now(timezone.utc).isoformat(),
"usage": _usage(event),
},
)
return
raise httpx.RemoteProtocolError("QCA event stream ended before the session reached a terminal state")
finally:
await response.aclose()
async def _closing_stream(
stream: AsyncIterator[InteractionsAPIStreamingResponse], client: httpx.AsyncClient
) -> AsyncIterator[InteractionsAPIStreamingResponse]:
try:
async for chunk in stream:
yield chunk
finally:
await client.aclose()
def _response_from_terminal(
terminal: InteractionsAPIStreamingResponse,
interaction_id: str,
agent: str,
steps: tuple[Mapping[str, object], ...],
) -> InteractionsAPIResponse:
if terminal.interaction is None:
raise httpx.RemoteProtocolError("QCA terminal event is missing its interaction")
status: Final = terminal.interaction.get("status")
usage_value: Final = terminal.interaction.get("usage")
created: Final = terminal.interaction.get("created")
updated: Final = terminal.interaction.get("updated")
usage: Final = _JSON_OBJECT.validate_python(usage_value) if isinstance(usage_value, dict) else None
return InteractionsAPIResponse(
id=interaction_id,
agent=agent,
status=status if isinstance(status, str) else None,
created=created if isinstance(created, str) else None,
updated=updated if isinstance(updated, str) else None,
steps=[dict(step) for step in steps],
usage=usage,
)
def _terminal_for_logging(
terminal: InteractionsAPIStreamingResponse,
interaction_id: str,
agent: str,
steps: tuple[Mapping[str, object], ...],
) -> InteractionsAPIStreamingResponse:
response: Final = _response_from_terminal(terminal, interaction_id, agent, steps)
return terminal.model_copy(update={"interaction": response.model_dump()})
async def _observed_stream(
stream: AsyncIterator[InteractionsAPIStreamingResponse],
interaction_id: str,
agent: str,
logging_obj: LiteLLMLoggingObj,
start_time: datetime,
) -> AsyncIterator[InteractionsAPIStreamingResponse]:
steps: tuple[Mapping[str, object], ...] = () # rebind-ok: immutable response accumulation
async for chunk in stream:
if chunk.event_type == "step.delta" and chunk.delta is not None:
steps = (*steps, _step_from_delta(chunk.delta))
if chunk.event_type in {"interaction.completed", "interaction.requires_action"}:
await logging_obj.dispatch_success_handlers( # pyright: ignore[reportUnknownMemberType] # legacy callback API is untyped
result=_terminal_for_logging(chunk, interaction_id, agent, steps),
start_time=start_time,
end_time=datetime.now(timezone.utc),
cache_hit=None,
prefer_async_handlers=True,
)
yield chunk
async def _collect_response(
stream: AsyncIterator[InteractionsAPIStreamingResponse], interaction_id: str, agent: str
) -> InteractionsAPIResponse:
steps: tuple[Mapping[str, object], ...] = () # rebind-ok: immutable response accumulation
terminal: InteractionsAPIStreamingResponse | None = None # rebind-ok: captures the terminal chunk
async for chunk in stream:
if chunk.event_type == "step.delta" and chunk.delta is not None:
steps = (*steps, _step_from_delta(chunk.delta))
if chunk.event_type == "error":
raise APIError(
status_code=502,
message=(
chunk.error.message
if chunk.error is not None and chunk.error.message is not None
else "Qoder Cloud Agents session failed"
),
llm_provider="qca",
model=agent,
)
if chunk.event_type in {"interaction.completed", "interaction.requires_action"}:
terminal = chunk
if terminal is None:
raise httpx.RemoteProtocolError("QCA event stream did not produce a terminal interaction")
return _response_from_terminal(terminal, interaction_id, agent, steps)
async def _session_id(
client: httpx.AsyncClient,
base: str,
headers: Mapping[str, str],
agent_id: str,
environment_id: str,
previous_session_id: str | None,
) -> str:
if previous_session_id is not None:
existing_response: Final = await client.get(f"{base}/sessions/{previous_session_id}", headers=headers)
existing_response.raise_for_status()
session: Final = QCASession.model_validate(existing_response.json())
session_agent: Final = session.agent.id if isinstance(session.agent, QCASessionAgent) else session.agent
if session.id != previous_session_id or session_agent != agent_id or session.environment_id != environment_id:
raise QCAValidationError("QCA previous_interaction_id does not match the configured agent and environment")
return previous_session_id
create_response: Final = await client.post(
f"{base}/sessions",
headers=headers,
json={"agent": agent_id, "environment_id": environment_id},
)
create_response.raise_for_status()
return _require_identifier(QCASession.model_validate(create_response.json()).id, "sess_", "session id")
async def create_qca_interaction(
*,
agent: str | None,
input: object,
environment: object,
previous_interaction_id: str | None,
stream: bool | None,
api_key: str | None,
api_base: str | None,
timeout: float | httpx.Timeout | None,
unsupported_params: Mapping[str, object],
logging_obj: LiteLLMLoggingObj,
client: httpx.AsyncClient | None = None,
) -> InteractionsAPIResponse | AsyncIterator[InteractionsAPIStreamingResponse]:
try:
unsupported: Final = tuple(
name
for name, value in unsupported_params.items()
if value not in (None, (), {}) and (name == "store" or value is not False)
)
if unsupported:
raise QCAValidationError(f"QCA interactions do not support {', '.join(sorted(unsupported))}")
agent_id: Final = _require_identifier(agent, "agent_", "agent id")
environment_id: Final = _require_identifier(
environment if isinstance(environment, str) else None,
"env_",
"environment id",
)
token: Final = api_key
if not token:
raise QCAValidationError("QCA api_key is required")
base: Final = _api_base(api_base)
text: Final = _text_input(input)
previous_session_id: Final = _previous_session_id(previous_interaction_id)
except QCAValidationError as error:
raise BadRequestError(message=str(error), model=agent or "", llm_provider="qca") from error
owns_client: Final = client is None
http_client: Final = client or httpx.AsyncClient(timeout=timeout)
start_time: Final = datetime.now(timezone.utc)
try:
headers: Final = {"Authorization": f"Bearer {token}", "Accept": "application/json"}
session_id: Final = await _session_id(
http_client,
base,
headers,
agent_id,
environment_id,
previous_session_id,
)
event_response: Final = await http_client.post(
f"{base}/sessions/{session_id}/events",
headers=headers,
json={"events": [{"type": "user.message", "content": [{"type": "text", "text": text}]}]},
)
event_response.raise_for_status()
input_event: Final = _accepted_event(_JSON_OBJECT.validate_json(event_response.content))
input_event_id: Final = _require_identifier(input_event.id, "evt_", "event id")
interaction_id: Final = _interaction_id(session_id, input_event_id)
stream_request: Final = http_client.build_request(
"GET",
f"{base}/sessions/{session_id}/events/stream",
headers={**headers, "Accept": "text/event-stream", "Last-Event-ID": input_event_id},
)
stream_response: Final = await http_client.send(stream_request, stream=True)
stream_response.raise_for_status()
normalized: Final = _normalized_stream(stream_response, interaction_id, agent_id)
if stream is True:
observed: Final = _observed_stream(normalized, interaction_id, agent_id, logging_obj, start_time)
return _closing_stream(observed, http_client) if owns_client else observed
result: Final = await _collect_response(normalized, interaction_id, agent_id)
except httpx.HTTPStatusError as error:
if owns_client:
await http_client.aclose()
raise _provider_error(error, agent_id) from error
except QCAValidationError as error:
if owns_client:
await http_client.aclose()
raise BadRequestError(message=str(error), model=agent_id, llm_provider="qca") from error
except BaseException:
if owns_client:
await http_client.aclose()
raise
else:
if owns_client:
await http_client.aclose()
return result

View file

@ -446,10 +446,6 @@ async def create_interaction(
data: Final = 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: Final = ProxyBaseLLMRequestProcessing(data=data)
try:
return await processor.base_process_llm_request(

View file

@ -57,6 +57,24 @@ _DECODED_JSON: Final = TypeAdapter(object)
_RequestData: TypeAlias = dict[str, object]
def _raise_for_unmapped_interactions_guardrail(
call_type: CallTypesLiteral | None,
mappings: Mapping[CallTypes, type["BaseTranslation"]],
) -> None:
if call_type is None:
return
try:
resolved_call_type: Final = CallTypes(call_type)
except ValueError:
return
if resolved_call_type != CallTypes.acreate_interaction or resolved_call_type in mappings:
return
raise HTTPException(
status_code=400,
detail="Interactions requests cannot enforce configured apply_guardrail policies without a translation handler",
)
@runtime_checkable
class RequestAttachmentScanner(Protocol):
async def async_scan_request_attachments(
@ -295,6 +313,8 @@ class UnifiedLLMGuardrails(CustomLogger):
mappings: Final = load_guardrail_translation_mappings()
_raise_for_unmapped_interactions_guardrail(call_type, mappings)
try:
if CallTypes(call_type) not in mappings:
return data
@ -346,6 +366,7 @@ class UnifiedLLMGuardrails(CustomLogger):
return data
mappings: Final = load_guardrail_translation_mappings()
_raise_for_unmapped_interactions_guardrail(call_type, mappings)
if call_type is not None and CallTypes(call_type) not in mappings:
return data
@ -422,6 +443,8 @@ class UnifiedLLMGuardrails(CustomLogger):
mappings: Final = load_guardrail_translation_mappings()
_raise_for_unmapped_interactions_guardrail(call_type, mappings)
if CallTypes(call_type) not in mappings:
verbose_proxy_logger.warning(
"Guardrail '%s' selected for route '%s' but call type '%s' has no guardrail translation handler; "

View file

@ -7,7 +7,7 @@ instead of writing immediately on each request.
from collections.abc import AsyncGenerator, Callable, Mapping
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, TypeVar
from typing import TYPE_CHECKING, Any, Final, TypeAlias, TypeVar
from fastapi import HTTPException
from pydantic import TypeAdapter, ValidationError
@ -22,6 +22,12 @@ from litellm.proxy.common_utils.encrypt_decrypt_utils import (
encrypt_value_helper,
)
from litellm.responses.utils import ResponsesAPIRequestUtils
from litellm.types.interactions import (
CancelInteractionResult,
DeleteInteractionResult,
InteractionsAPIResponse,
InteractionsAPIStreamingResponse,
)
from litellm.types.llms.openai import (
BaseLiteLLMOpenAIResponseObject,
ResponsesAPIResponse,
@ -41,7 +47,7 @@ if TYPE_CHECKING:
_RESPONSES_API_PROVIDER_PREFIX: Final = "/openai"
_RESPONSES_API_ROUTES: Final = frozenset({"/v1/responses", "/responses"})
_RESPONSES_API_ROUTES: Final = frozenset({"/v1/responses", "/responses", "/v1beta/interactions", "/interactions"})
_ADDRESSED_RESPONSE_ID_FIELD_BY_CALL_TYPE: Final[Mapping[str, str]] = MappingProxyType(
{
"aresponses": "previous_response_id",
@ -50,6 +56,10 @@ _ADDRESSED_RESPONSE_ID_FIELD_BY_CALL_TYPE: Final[Mapping[str, str]] = MappingPro
"adelete_responses": "response_id",
"acancel_responses": "response_id",
"alist_input_items": "response_id",
"acreate_interaction": "previous_interaction_id",
"aget_interaction": "interaction_id",
"adelete_interaction": "interaction_id",
"acancel_interaction": "interaction_id",
}
)
@ -79,7 +89,14 @@ def _proxy_signing_key() -> str | None:
_RESPONSE_PAYLOAD_ADAPTER: Final = TypeAdapter(Mapping[str, object])
_REQUEST_DATA_ADAPTER: Final = TypeAdapter(dict[str, object])
_ResponseObjectT = TypeVar("_ResponseObjectT", bound=BaseLiteLLMOpenAIResponseObject)
_AddressedResultT = TypeVar("_AddressedResultT", bound=ResponsesAPIResponse | DeleteResponseResult)
_AddressedResult: TypeAlias = (
ResponsesAPIResponse
| InteractionsAPIResponse
| DeleteResponseResult
| DeleteInteractionResult
| CancelInteractionResult
)
_AddressedResultT: Final = TypeVar("_AddressedResultT", bound=_AddressedResult)
def _response_payload(response_obj: object) -> Mapping[str, object] | None:
@ -93,8 +110,28 @@ def _rewrite_advertised_id(
event: _ResponseObjectT,
rewrite: Callable[[str], str],
) -> _ResponseObjectT:
if isinstance(event, InteractionsAPIStreamingResponse):
interaction_id: Final = event.interaction_id
rewritten_interaction_id: Final = rewrite(interaction_id) if isinstance(interaction_id, str) else None
interaction: Final = _response_payload(event.interaction)
nested_id: Final = None if interaction is None else interaction.get("id")
rewritten_interaction: Final = (
{**interaction, "id": rewrite(nested_id)}
if interaction is not None and isinstance(nested_id, str)
else None
)
if rewritten_interaction_id is not None or rewritten_interaction is not None:
return event.model_copy(
update={
**({"interaction_id": rewritten_interaction_id} if rewritten_interaction_id is not None else {}),
**({"interaction": rewritten_interaction} if rewritten_interaction is not None else {}),
}
)
event_id: Final = getattr(event, "id", None)
if isinstance(event_id, str) and event_id.startswith("resp_"):
if isinstance(event_id, str) and (
event_id.startswith("resp_") or isinstance(event, InteractionsAPIResponse | InteractionsAPIStreamingResponse)
):
setattr(event, "id", rewrite(event_id))
return event
@ -151,7 +188,7 @@ def _held_id_naming(response_id: str | None, held_ids: Mapping[str, str]) -> str
)
def _id_sent_back(response: ResponsesAPIResponse | DeleteResponseResult, held_ids: Mapping[str, str]) -> str | None:
def _id_sent_back(response: _AddressedResult, held_ids: Mapping[str, str]) -> str | None:
held_id: Final = _held_id_naming(response.id, held_ids)
if held_id is not None:
return held_id
@ -455,7 +492,14 @@ class ResponsesIDSecurity(CustomLogger):
if general_settings.get("disable_responses_id_security", False):
return response
if isinstance(response, ResponsesAPIResponse | DeleteResponseResult):
if isinstance(
response,
ResponsesAPIResponse
| InteractionsAPIResponse
| DeleteResponseResult
| DeleteInteractionResult
| CancelInteractionResult,
):
return self._advertised_response(response, user_api_key_dict, data)
return response

View file

@ -11,7 +11,7 @@ from uuid import uuid4
import httpx
from fastapi import HTTPException, status
from pydantic import BeforeValidator, TypeAdapter, ValidationError
from pydantic import BaseModel, BeforeValidator, ConfigDict, TypeAdapter, ValidationError
import litellm
from litellm.litellm_core_utils.core_helpers import normalize_drop_params
@ -61,6 +61,50 @@ _PER_REQUEST_ROUTER_SETTINGS: Final = (
_ROUTER_ONLY_SETTINGS: Final = frozenset(
("fallbacks", "context_window_fallbacks", "content_policy_fallbacks", "model_group_retry_policy")
)
_ROUTER_DEPLOYMENTS: Final = TypeAdapter(list[dict[str, object]])
class _QCAMonetaryBudgetContext(BaseModel):
model_config = ConfigDict(frozen=True, from_attributes=True)
max_budget: float | None = None
user_max_budget: float | None = None
team_max_budget: float | None = None
organization_max_budget: float | None = None
end_user_max_budget: float | None = None
budget_id: str | None = None
budget_reservation: Mapping[str, object] | None = None
model_max_budget: Mapping[str, object] | None = None
user_model_max_budget: Mapping[str, object] | None = None
team_model_max_budget: Mapping[str, object] | None = None
team_member_model_max_budget: Mapping[str, object] | None = None
end_user_model_max_budget: Mapping[str, object] | None = None
def is_limited(self) -> bool:
values: Final = (
self.max_budget,
self.user_max_budget,
self.team_max_budget,
self.organization_max_budget,
self.end_user_max_budget,
self.budget_id,
self.budget_reservation,
self.model_max_budget,
self.user_model_max_budget,
self.team_model_max_budget,
self.team_member_model_max_budget,
self.end_user_model_max_budget,
)
return any(value is not None and value != {} for value in values)
def _provider_model(deployment: Mapping[str, object]) -> str | None:
unvalidated_litellm_params: Final = deployment.get("litellm_params")
if not isinstance(unvalidated_litellm_params, Mapping):
return None
litellm_params: Final = JSON_OBJECT_ADAPTER.validate_python(unvalidated_litellm_params)
model: Final = litellm_params.get("model")
return model if isinstance(model, str) else None
class UserConfigRouters:
@ -144,6 +188,53 @@ def _raise_if_model_fully_blocked(llm_router: LitellmRouter, model_name: object,
)
def _raise_if_qca_deployment_selectors_are_overridden(
llm_router: LitellmRouter,
model_name: object,
request_data: Mapping[str, object],
user_api_key_dict: UserAPIKeyAuth | None = None,
) -> None:
if not isinstance(model_name, str):
return
deployment_by_id: Final = llm_router.registry().deployment(model_name)
request_deployments: Final = _ROUTER_DEPLOYMENTS.validate_python(
llm_router.registry().deployments(
model_name,
source="for_request",
request=request_data,
call_type="acreate_interaction",
)
)
configured_models: Final = (
(deployment_by_id.litellm_params.model,)
if deployment_by_id is not None
else tuple(model for deployment in request_deployments if (model := _provider_model(deployment)) is not None)
)
if not model_name.startswith("qca/") and not any(model.startswith("qca/") for model in configured_models):
return
budget_context: Final = (
None if user_api_key_dict is None else _QCAMonetaryBudgetContext.model_validate(user_api_key_dict)
)
if budget_context is not None and budget_context.is_limited():
raise ProxyException(
message=(
"Qoder Cloud Agents cannot enforce monetary budgets because Qoder credits do not have a configured "
"currency conversion"
),
type="invalid_request_error",
param="model",
code=status.HTTP_400_BAD_REQUEST,
)
overridden: Final = tuple(name for name in ("agent", "environment") if request_data.get(name) is not None)
if overridden:
raise ProxyException(
message=f"Qoder Cloud Agents deployment selectors cannot be overridden: {', '.join(overridden)}",
type="auth_error",
param=overridden[0],
code=status.HTTP_403_FORBIDDEN,
)
ROUTE_ENDPOINT_MAPPING: Final = {
"acompletion": "/chat/completions",
"atext_completion": "/completions",
@ -458,13 +549,14 @@ def endpoint_unsupported_by_model(
if not isinstance(model, str) or llm_router is None:
return False
metadata: Final = data.get("metadata")
request_kwargs: Final = {
**data,
"metadata": {
**(metadata if isinstance(metadata, Mapping) else {}),
validated_metadata: Final = JSON_OBJECT_ADAPTER.validate_python(metadata) if isinstance(metadata, Mapping) else {}
request_metadata: Final = MappingProxyType[str, object](
{
**validated_metadata,
"user_api_key_team_id": team_id,
},
}
}
)
request_kwargs: Final = MappingProxyType[str, object]({**data, "metadata": request_metadata})
registry: Final = llm_router.registry()
return bool(registry.deployments(model, source="routable", request=request_kwargs)) and not registry.deployments(
model, source="routable", request=request_kwargs, call_type=route_type
@ -823,6 +915,15 @@ async def _route_request_single_attempt( # noqa: ANN202 # returns unawaited pr
router_model_names: Final = llm_router.registry().names(kind="group_set") if llm_router is not None else []
is_proxy_admin_without_team: Final = team_id is None and is_proxy_admin_request(data)
if route_type == "acreate_interaction" and llm_router is not None:
interaction_request: Final = JSON_OBJECT_ADAPTER.validate_python(data)
_raise_if_qca_deployment_selectors_are_overridden(
llm_router,
interaction_request.get("model"),
interaction_request,
user_api_key_dict,
)
# Preprocess Google GenAI generate content requests
if route_type in ["agenerate_content", "agenerate_content_stream"]:
# Map generationConfig to config parameter for Google GenAI compatibility

View file

@ -24,6 +24,7 @@ from litellm.types.interactions.generated import (
DeleteInteractionResult,
DocumentContent,
DynamicAgentConfig,
Error,
ErrorEvent,
FileSearch,
FileSearchResultContent,
@ -92,6 +93,7 @@ __all__ = [
"DeleteInteractionResult",
"DocumentContent",
"DynamicAgentConfig",
"Error",
"ErrorEvent",
"FileSearch",
"FileSearchResultContent",

View file

@ -1144,6 +1144,7 @@ class InteractionsAPIStreamingResponse(BaseLiteLLMOpenAIResponseObject):
steps: list[dict[str, Any]] | None = None
usage: dict[str, Any] | None = None
delta: dict[str, Any] | None = None
error: Error | None = None
# New schema streaming fields
index: int | None = None
step: dict[str, Any] | None = None

View file

@ -666,6 +666,12 @@ class CallTypes(str, Enum):
#########################################################
create_interaction = "create_interaction"
acreate_interaction = "acreate_interaction"
get_interaction = "get_interaction"
aget_interaction = "aget_interaction"
delete_interaction = "delete_interaction"
adelete_interaction = "adelete_interaction"
cancel_interaction = "cancel_interaction"
acancel_interaction = "acancel_interaction"
#########################################################
# Container Call Types
@ -808,6 +814,10 @@ CallTypesLiteral = Literal[
"list_mcp_tools",
"asend_message",
"send_message",
"acreate_interaction",
"aget_interaction",
"adelete_interaction",
"acancel_interaction",
"aresponses",
"responses",
"compact_responses",
@ -4678,6 +4688,7 @@ class LlmProviders(str, Enum):
CODESTRAL = "codestral"
TEXT_COMPLETION_CODESTRAL = "text-completion-codestral"
DASHSCOPE = "dashscope"
QCA = "qca"
QWENCLOUD = "qwencloud"
QWEN_AI_PLATFORM = "qwen_ai_platform"
MODELSCOPE = "modelscope"

View file

@ -12,6 +12,7 @@ from respx import MockRouter
import litellm
from litellm import interactions
from litellm.integrations.custom_logger import CustomLogger
from litellm.types.interactions import InteractionsAPIResponse
_JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue])
@ -862,6 +863,43 @@ def test_create_streaming_hands_completed_interaction_to_success_callback(
assert recorder.logged == (_GEMINI_COMPLETED_INTERACTION,)
@pytest.mark.asyncio
async def test_acreate_qca_runs_managed_session_flow(respx_mock: MockRouter) -> None:
base: Final = "https://qca.test/api/v1/cloud"
respx_mock.post(f"{base}/sessions").mock(return_value=httpx.Response(200, json={"id": "sess_public"}))
respx_mock.post(f"{base}/sessions/sess_public/events").mock(
return_value=httpx.Response(202, json={"data": [{"id": "evt_input", "type": "user.message"}]})
)
respx_mock.get(f"{base}/sessions/sess_public/events/stream").mock(
return_value=httpx.Response(
200,
text=(
'data: {"id":"evt_answer","type":"agent.message",'
'"content":[{"type":"text","text":"ready"}]}\n\n'
'data: {"id":"evt_idle","type":"session.status_idle",'
'"stop_reason":{"type":"end_turn"}}\n\n'
),
headers={"content-type": "text/event-stream"},
)
)
response: Final = await interactions.acreate(
model="qca/agent_public",
input="run",
environment="env_public",
api_key="qca-secret",
api_base=base,
)
assert isinstance(response, InteractionsAPIResponse)
assert response.id == "qca1.sess_public.evt_input"
assert response.status == "completed"
assert response.steps == [
{
"type": "model_output",
"content": [{"type": "text", "text": "ready"}],
}
]
_GEMINI_URL: Final = "https://generativelanguage.googleapis.com/v1beta/interactions"
_LONG_DECODE_REQUEST: Final = {
"model": "gemini/gemini-4-argon",

View file

@ -0,0 +1,325 @@
import asyncio
import json
from datetime import datetime
from typing import Final
from unittest.mock import MagicMock
import httpx
import pytest
from litellm.exceptions import BadRequestError, RateLimitError
from litellm.integrations.custom_logger import CustomLogger
from litellm.litellm_core_utils import litellm_logging
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.llms.qca.interactions.handler import create_qca_interaction
from litellm.types.interactions import InteractionsAPIResponse, InteractionSseEvent
def _sse_event(payload: dict[str, object]) -> bytes:
return f"data: {json.dumps(payload)}\n\n".encode()
class _SuccessRecorder(CustomLogger):
def __init__(self) -> None:
self.responses: list[object] = []
async def async_log_success_event(
self, kwargs: dict[str, object], response_obj: object, start_time: datetime, end_time: datetime
) -> None:
self.responses.append(response_obj)
def _stream_logging(recorder: _SuccessRecorder, include_sync_callback: bool = False) -> LiteLLMLoggingObj:
logging_obj: Final = LiteLLMLoggingObj(
model="qca/agent_test",
messages=[],
stream=True,
call_type="acreate_interaction",
start_time=datetime.now(),
litellm_call_id="qca-test-call",
function_id="qca-test-function",
dynamic_async_success_callbacks=[recorder],
)
logging_obj.update_environment_variables(
model="qca/agent_test", user="", optional_params={}, litellm_params={"api_base": "https://qca.test"}
)
if include_sync_callback:
logging_obj.dynamic_success_callbacks = ["s3"]
return logging_obj
@pytest.mark.asyncio
async def test_create_qca_interaction_maps_session_events_to_completed_response() -> None:
requests: list[httpx.Request] = []
def route(request: httpx.Request) -> httpx.Response:
requests.append(request)
if request.url.path.endswith("/sessions"):
return httpx.Response(200, json={"id": "sess_test"})
if request.url.path.endswith("/events"):
return httpx.Response(202, json={"data": [{"id": "evt_input", "type": "user.message"}]})
body: Final = b"".join(
(
_sse_event(
{
"id": "evt_answer",
"type": "agent.message",
"content": [{"type": "text", "text": "done"}],
}
),
_sse_event(
{
"id": "evt_idle",
"type": "session.status_idle",
"stop_reason": {"type": "end_turn"},
"total_credits": 3.5,
}
),
)
)
return httpx.Response(200, content=body, headers={"content-type": "text/event-stream"})
async with httpx.AsyncClient(transport=httpx.MockTransport(route)) as client:
logging_obj: Final = MagicMock()
response: Final = await create_qca_interaction(
agent="agent_test",
input="finish the task",
environment="env_test",
previous_interaction_id=None,
stream=False,
api_key="secret",
api_base="https://qca.test/api/v1/cloud",
timeout=30,
unsupported_params={},
logging_obj=logging_obj,
client=client,
)
assert isinstance(response, InteractionsAPIResponse)
assert response.id == "qca1.sess_test.evt_input"
assert response.status == "completed"
assert response.created is not None
assert response.updated is not None
assert datetime.fromisoformat(response.updated) >= datetime.fromisoformat(response.created)
assert response.steps == [
{
"type": "model_output",
"content": [{"type": "text", "text": "done"}],
}
]
assert response.usage == {
"qca_session_credits": {
"model_credits": None,
"sandbox_runtime_credits": None,
"total_credits": 3.5,
},
"qca_usage_scope": "session_cumulative",
"qca_usage_is_final": True,
}
assert requests[-1].headers["last-event-id"] == "evt_input"
@pytest.mark.asyncio
async def test_create_qca_interaction_continues_session_and_streams_normalized_events(
monkeypatch: pytest.MonkeyPatch,
) -> None:
requests: list[httpx.Request] = []
sync_logger: Final = MagicMock()
monkeypatch.setattr(litellm_logging, "s3Logger", sync_logger)
def route(request: httpx.Request) -> httpx.Response:
requests.append(request)
if request.method == "GET" and request.url.path.endswith("/sessions/sess_existing"):
return httpx.Response(
200,
json={"id": "sess_existing", "agent": {"id": "agent_test"}, "environment_id": "env_test"},
)
if request.url.path.endswith("/events"):
return httpx.Response(202, json={"data": [{"id": "evt_next", "type": "user.message"}]})
body: Final = b"".join(
(
_sse_event(
{
"id": "evt_answer",
"type": "agent.message",
"content": [{"type": "text", "text": "continued"}],
}
),
_sse_event(
{
"id": "evt_action",
"type": "session.status_idle",
"stop_reason": {"type": "end_turn"},
}
),
)
)
return httpx.Response(200, content=body, headers={"content-type": "text/event-stream"})
async with httpx.AsyncClient(transport=httpx.MockTransport(route)) as client:
recorder: Final = _SuccessRecorder()
logging_obj: Final = _stream_logging(recorder, include_sync_callback=True)
result: Final = await create_qca_interaction(
agent="agent_test",
input="continue",
environment="env_test",
previous_interaction_id="qca1.sess_existing.evt_old",
stream=True,
api_key="secret",
api_base="https://qca.test/api/v1/cloud",
timeout=30,
unsupported_params={},
logging_obj=logging_obj,
client=client,
)
assert not isinstance(result, InteractionsAPIResponse)
chunks: Final = [chunk async for chunk in result]
assert [chunk.event_type for chunk in chunks] == [
"interaction.created",
"step.delta",
"interaction.completed",
]
assert chunks[1].delta == {"type": "text", "text": "continued"}
assert chunks[-1].interaction_id == "qca1.sess_existing.evt_next"
assert len(recorder.responses) == 1
logged: Final = recorder.responses[0]
assert isinstance(logged, InteractionsAPIResponse)
assert logged.steps == [{"type": "model_output", "content": [{"type": "text", "text": "continued"}]}]
for _ in range(200):
if sync_logger.log_event.call_count == 1:
break
await asyncio.sleep(0.01)
sync_logger.log_event.assert_called_once()
assert all(not request.url.path.endswith("/sessions") for request in requests)
@pytest.mark.asyncio
async def test_create_qca_interaction_stream_emits_schema_valid_error_event() -> None:
def route(request: httpx.Request) -> httpx.Response:
if request.url.path.endswith("/sessions"):
return httpx.Response(200, json={"id": "sess_test"})
if request.url.path.endswith("/events"):
return httpx.Response(202, json={"data": [{"id": "evt_input", "type": "user.message"}]})
return httpx.Response(
200,
content=_sse_event({"id": "evt_error", "type": "session.error"}),
headers={"content-type": "text/event-stream"},
)
async with httpx.AsyncClient(transport=httpx.MockTransport(route)) as client:
result: Final = await create_qca_interaction(
agent="agent_test",
input="fail",
environment="env_test",
previous_interaction_id=None,
stream=True,
api_key="secret",
api_base="https://qca.test/api/v1/cloud",
timeout=30,
unsupported_params={},
logging_obj=_stream_logging(_SuccessRecorder()),
client=client,
)
assert not isinstance(result, InteractionsAPIResponse)
chunks: Final = [chunk async for chunk in result]
assert chunks[-1].event_type == "error"
validated: Final = InteractionSseEvent.model_validate(chunks[-1].model_dump(exclude_none=True))
assert validated.root.event_type == "error"
@pytest.mark.asyncio
async def test_create_qca_interaction_rejects_unsupported_features_before_network_call() -> None:
async with httpx.AsyncClient(transport=httpx.MockTransport(lambda _: httpx.Response(500))) as client:
logging_obj: Final = MagicMock()
with pytest.raises(BadRequestError, match="background") as error:
await create_qca_interaction(
agent="agent_test",
input="hello",
environment="env_test",
previous_interaction_id=None,
stream=False,
api_key="secret",
api_base=None,
timeout=30,
unsupported_params={"background": True},
logging_obj=logging_obj,
client=client,
)
assert error.value.status_code == 400
@pytest.mark.asyncio
async def test_create_qca_interaction_rejects_store_false_before_network_call() -> None:
async with httpx.AsyncClient(transport=httpx.MockTransport(lambda _: httpx.Response(500))) as client:
with pytest.raises(BadRequestError, match="store") as error:
await create_qca_interaction(
agent="agent_test",
input="hello",
environment="env_test",
previous_interaction_id=None,
stream=False,
api_key="secret",
api_base=None,
timeout=30,
unsupported_params={"store": False},
logging_obj=MagicMock(),
client=client,
)
assert error.value.status_code == 400
@pytest.mark.asyncio
async def test_create_qca_interaction_preserves_rate_limit_status() -> None:
def route(request: httpx.Request) -> httpx.Response:
return httpx.Response(429, text="rate limited", request=request)
async with httpx.AsyncClient(transport=httpx.MockTransport(route)) as client:
with pytest.raises(RateLimitError) as error:
await create_qca_interaction(
agent="agent_test",
input="hello",
environment="env_test",
previous_interaction_id=None,
stream=False,
api_key="secret",
api_base=None,
timeout=30,
unsupported_params={},
logging_obj=MagicMock(),
client=client,
)
assert error.value.status_code == 429
@pytest.mark.asyncio
async def test_create_qca_interaction_rejects_continuation_for_another_agent() -> None:
requests: list[httpx.Request] = []
def route(request: httpx.Request) -> httpx.Response:
requests.append(request)
return httpx.Response(
200,
json={"id": "sess_existing", "agent": {"id": "agent_other"}, "environment_id": "env_test"},
)
async with httpx.AsyncClient(transport=httpx.MockTransport(route)) as client:
logging_obj: Final = MagicMock()
with pytest.raises(BadRequestError, match="configured agent and environment") as error:
await create_qca_interaction(
agent="agent_test",
input="continue",
environment="env_test",
previous_interaction_id="qca1.sess_existing.evt_old",
stream=False,
api_key="secret",
api_base=None,
timeout=30,
unsupported_params={},
logging_obj=logging_obj,
client=client,
)
assert error.value.status_code == 400
assert [request.method for request in requests] == ["GET"]

View file

@ -12,6 +12,7 @@ from typing import TYPE_CHECKING, Final, Literal
import anyio
import pytest
from fastapi import HTTPException
import litellm
from litellm.caching import DualCache
@ -338,6 +339,27 @@ class TestUnifiedLLMGuardrails:
assert "tool" in roles
class TestAsyncPreCallHook:
@pytest.mark.asyncio
async def test_interactions_rejects_apply_guardrail_without_translation(self) -> None:
guardrail: Final = RecordingGuardrail()
data: Final = {
"guardrail_to_apply": guardrail,
"model": "qca-agent-alias",
"input": "sensitive input",
}
with pytest.raises(HTTPException) as error:
await UnifiedLLMGuardrails().async_pre_call_hook(
user_api_key_dict=UserAPIKeyAuth(api_key="test-key"),
cache=DualCache(),
data=data,
call_type=CallTypes.acreate_interaction.value,
)
assert error.value.status_code == 400
assert "cannot enforce configured apply_guardrail policies" in error.value.detail
assert guardrail.apply_calls == []
@pytest.mark.asyncio
async def test_uses_mcp_event_type(self):
"""pre_call hook should swap to GuardrailEventHooks.pre_mcp_call for MCP calls."""

View file

@ -1,46 +1,200 @@
import asyncio
import contextlib
import gc
import json
import threading
import weakref
from collections.abc import Awaitable, Callable, Coroutine, Iterator, Mapping
from datetime import datetime
from types import SimpleNamespace
from typing import Final
from unittest.mock import MagicMock
from uuid import uuid4
import httpx
import json
import litellm
import pytest
import redis
import respx
from collections.abc import Awaitable, Callable, Coroutine, Iterator, Mapping
from datetime import datetime
from types import SimpleNamespace
from uuid import uuid4
from typing import Final
from unittest.mock import MagicMock
from litellm.router_api import RouterSettings
from tests.unit.proxy.router_entry_point_stubs import route_entry_points_to_stubs
from fakeredis import TcpFakeServer
from fastapi import HTTPException
from litellm.caching.caching import DualCache
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.proxy._types import UserAPIKeyAuth
from litellm.proxy._types import ProxyException, UserAPIKeyAuth
from litellm.proxy.route_llm_request import (
JSON_OBJECT_ADAPTER,
ProxyModelNotFoundError,
UserConfigRouters,
_raise_if_qca_deployment_selectors_are_overridden,
endpoint_unsupported_by_model,
route_request,
user_config_routers,
)
from litellm.proxy.utils import ProxyLogging
from litellm.router_api import RouterSettings
from litellm.router_strategy.adaptive_router.config import (
ADAPTIVE_ROUTER_CHOSEN_MODEL_KEY,
ADAPTIVE_ROUTER_RESPONSE_HEADER,
)
from litellm.types.router import RouterErrors, RouterGeneralSettings, RouterRateLimitError
from tests.unit.proxy.router_entry_point_stubs import route_entry_points_to_stubs
@pytest.mark.parametrize("selector", ("agent", "environment"))
def test_qca_alias_rejects_request_selector_overrides(selector: str) -> None:
router: Final = litellm.Router(
model_list=[
{
"model_name": "qoder-agent",
"litellm_params": {
"model": "qca/agent_configured",
"environment": "env_configured",
"api_key": "unit-provider-key",
},
}
]
)
with pytest.raises(ProxyException) as error:
_raise_if_qca_deployment_selectors_are_overridden(
router,
"qoder-agent",
{"model": "qoder-agent", selector: f"{selector}_attacker"},
)
assert error.value.code == "403"
@pytest.mark.parametrize("selector", ("agent", "environment"))
def test_qca_model_group_alias_rejects_request_selector_overrides(selector: str) -> None:
router: Final = litellm.Router(
model_list=[
{
"model_name": "qoder-agent",
"litellm_params": {
"model": "qca/agent_configured",
"environment": "env_configured",
"api_key": "unit-provider-key",
},
}
],
model_group_alias={"public-agent": "qoder-agent"},
)
with pytest.raises(ProxyException) as error:
_raise_if_qca_deployment_selectors_are_overridden(
router,
"public-agent",
{"model": "public-agent", selector: f"{selector}_attacker"},
)
assert error.value.code == "403"
@pytest.mark.parametrize("selector", ("agent", "environment"))
def test_qca_deployment_id_rejects_request_selector_overrides(selector: str) -> None:
deployment_id: Final = "qoder-deployment-id"
router: Final = litellm.Router(
model_list=[
{
"model_name": "qoder-agent",
"litellm_params": {
"model": "qca/agent_configured",
"environment": "env_configured",
"api_key": "unit-provider-key",
},
"model_info": {"id": deployment_id},
}
]
)
with pytest.raises(ProxyException) as error:
_raise_if_qca_deployment_selectors_are_overridden(
router,
deployment_id,
{"model": deployment_id, selector: f"{selector}_attacker"},
)
assert error.value.code == "403"
@pytest.mark.parametrize("selector", ("agent", "environment"))
def test_qca_team_public_name_rejects_request_selector_overrides(selector: str) -> None:
team_id: Final = "qoder-team"
public_name: Final = "team-qoder-agent"
router: Final = litellm.Router(
model_list=[
{
"model_name": "private-qoder-agent",
"litellm_params": {
"model": "qca/agent_configured",
"environment": "env_configured",
"api_key": "unit-provider-key",
},
"model_info": {"team_id": team_id, "team_public_model_name": public_name},
}
]
)
request: Final = {
"model": public_name,
selector: f"{selector}_attacker",
"metadata": {"user_api_key_team_id": team_id},
}
with pytest.raises(ProxyException) as error:
_raise_if_qca_deployment_selectors_are_overridden(
router,
public_name,
request,
UserAPIKeyAuth(team_id=team_id),
)
assert error.value.code == "403"
@pytest.mark.parametrize(
"budget_field,budget_value",
(
("max_budget", 10.0),
("user_max_budget", 10.0),
("team_max_budget", 10.0),
("organization_max_budget", 10.0),
("end_user_max_budget", 10.0),
("budget_id", "budget-config-id"),
("budget_reservation", {"entries": [{"counter_key": "spend:team:team-id"}]}),
("model_max_budget", {"qoder-agent": {"max_budget": 10.0}}),
("user_model_max_budget", {"qoder-agent": {"max_budget": 10.0}}),
("team_model_max_budget", {"qoder-agent": {"max_budget": 10.0}}),
("team_member_model_max_budget", {"qoder-agent": {"max_budget": 10.0}}),
("end_user_model_max_budget", {"qoder-agent": {"max_budget": 10.0}}),
),
)
def test_qca_rejects_callers_with_unenforceable_monetary_budgets(budget_field: str, budget_value: object) -> None:
router: Final = litellm.Router(
model_list=[
{
"model_name": "qoder-agent",
"litellm_params": {
"model": "qca/agent_configured",
"environment": "env_configured",
"api_key": "unit-provider-key",
},
}
]
)
caller: Final = UserAPIKeyAuth.model_validate({budget_field: budget_value})
with pytest.raises(ProxyException) as error:
_raise_if_qca_deployment_selectors_are_overridden(
router,
"qoder-agent",
{"model": "qoder-agent"},
caller,
)
assert error.value.code == "400"
assert "cannot enforce monetary budgets" in error.value.message
@pytest.mark.parametrize(
@ -1399,7 +1553,9 @@ async def test_route_request_applies_override_fallbacks_when_request_carries_cre
@pytest.mark.parametrize(
"router_only_setting",
[
pytest.param({"model_group_retry_policy": {"gpt-4": {"TimeoutErrorRetries": 1}}}, id="model-group-retry-policy"),
pytest.param(
{"model_group_retry_policy": {"gpt-4": {"TimeoutErrorRetries": 1}}}, id="model-group-retry-policy"
),
pytest.param({"context_window_fallbacks": [{"gpt-4": ["gpt-3.5-turbo"]}]}, id="context-window-fallbacks"),
pytest.param({"content_policy_fallbacks": [{"gpt-4": ["gpt-3.5-turbo"]}]}, id="content-policy-fallbacks"),
],
@ -2636,7 +2792,9 @@ async def test_a_user_config_adaptive_request_answers_with_the_model_it_picked()
},
{
**_mock_deployment("smart"),
"model_info": {"adaptive_router_preferences": {"quality_tier": 3, "strengths": ["code_generation"]}},
"model_info": {
"adaptive_router_preferences": {"quality_tier": 3, "strengths": ["code_generation"]}
},
},
]
},

View file

@ -449,9 +449,7 @@ def test_finalize_prunes_stale_adaptive_router_hooks_from_callbacks():
# Snapshot any pre-existing AdaptiveRouterPostCallHook entries so we can
# restore them — other tests may have registered hooks we shouldn't drop.
pre_hooks = [
cb for cb in litellm.callbacks if isinstance(cb, AdaptiveRouterPostCallHook)
]
pre_hooks = [cb for cb in litellm.callbacks if isinstance(cb, AdaptiveRouterPostCallHook)]
for cb in pre_hooks:
litellm.callbacks.remove(cb)
@ -459,14 +457,9 @@ def test_finalize_prunes_stale_adaptive_router_hooks_from_callbacks():
Router(model_list=model_list)
Router(model_list=model_list) # simulate hot-reload
adaptive_hooks = [
cb
for cb in litellm.callbacks
if isinstance(cb, AdaptiveRouterPostCallHook)
]
adaptive_hooks = [cb for cb in litellm.callbacks if isinstance(cb, AdaptiveRouterPostCallHook)]
assert len(adaptive_hooks) == 1, (
f"expected exactly one AdaptiveRouterPostCallHook after hot-reload, "
f"got {len(adaptive_hooks)}"
f"expected exactly one AdaptiveRouterPostCallHook after hot-reload, got {len(adaptive_hooks)}"
)
finally:
# Best-effort cleanup: remove whatever this test added, then restore.
@ -475,5 +468,3 @@ def test_finalize_prunes_stale_adaptive_router_hooks_from_callbacks():
litellm.callbacks.remove(cb)
for cb in pre_hooks:
litellm.callbacks.append(cb)

View file

@ -23,6 +23,11 @@ from litellm.proxy.hooks.responses_id_security import (
)
from litellm.proxy.spend_tracking.spend_tracking_utils import get_logging_payload
from litellm.responses.utils import ResponsesAPIRequestUtils
from litellm.types.interactions import (
CancelInteractionResult,
InteractionsAPIResponse,
InteractionsAPIStreamingResponse,
)
from litellm.types.llms.openai import (
GenericEvent,
ResponseAPIUsage,
@ -803,6 +808,172 @@ def _comparison_options(data: object, nested: bool) -> Mapping[str, str]:
return TypeAdapter(dict[str, str]).validate_python(container["prompt_cache_options"])
@pytest.mark.asyncio
async def test_interaction_continuation_id_is_bound_to_its_user_and_team(salt_key_env: str) -> None:
hook: Final = _hook(signing_key=salt_key_env)
owner: Final = _auth()
issued: Final = InteractionsAPIResponse.model_validate(
await hook.async_post_call_success_hook( # pyright: ignore[reportArgumentType] # interactions share this proxy hook
data={},
user_api_key_dict=owner,
response=InteractionsAPIResponse(id="qca1.sess_owner.evt_input", status="completed"),
)
)
assert issued.id is not None
assert issued.id != "qca1.sess_owner.evt_input"
authorized: Final = TypeAdapter(dict[str, object]).validate_python(
await hook.async_pre_call_hook(
user_api_key_dict=owner,
cache=DualCache(),
data={"previous_interaction_id": issued.id},
call_type="acreate_interaction",
)
)
assert authorized["previous_interaction_id"] == "qca1.sess_owner.evt_input"
with pytest.raises(HTTPException) as refusal:
await hook.async_pre_call_hook(
user_api_key_dict=_auth(user_id="other-user"),
cache=DualCache(),
data={"previous_interaction_id": issued.id},
call_type="acreate_interaction",
)
assert refusal.value.status_code == 403
@pytest.mark.asyncio
async def test_streamed_interaction_ids_are_bound_to_the_caller(salt_key_env: str) -> None:
hook: Final = _hook(signing_key=salt_key_env)
caller: Final = _auth().model_copy(update={"request_route": "/interactions"})
async def chunks():
yield InteractionsAPIStreamingResponse(
event_type="interaction.created", id="qca1.sess_owner.evt_input", status="in_progress"
)
protected: Final = tuple(
[
chunk
async for chunk in hook.async_post_call_streaming_iterator_hook(
user_api_key_dict=caller,
response=chunks(),
request_data={},
)
]
)
protected_id: Final = protected[0].id
assert protected_id is not None
assert hook._decrypt_response_id(protected_id) == ( # pyright: ignore[reportPrivateUsage] # verifies ownership payload
"qca1.sess_owner.evt_input",
"owner-user",
"owner-team",
)
@pytest.mark.asyncio
@pytest.mark.parametrize("call_type", ("aget_interaction", "adelete_interaction", "acancel_interaction"))
async def test_interaction_lifecycle_ids_are_authorized_before_provider_calls(
salt_key_env: str, call_type: CallTypesLiteral
) -> None:
hook: Final = _hook(signing_key=salt_key_env)
owner: Final = _auth()
issued_response: Final = hook._encrypt_response_id( # pyright: ignore[reportPrivateUsage] # mints ids the way create does
InteractionsAPIResponse(id="interaction-provider-id", status="completed"),
owner,
)
issued: Final = InteractionsAPIResponse.model_validate(issued_response).id
assert issued is not None
authorized: Final = TypeAdapter(dict[str, object]).validate_python(
await hook.async_pre_call_hook(
user_api_key_dict=owner,
cache=DualCache(),
data={"interaction_id": issued},
call_type=call_type,
)
)
assert authorized["interaction_id"] == "interaction-provider-id"
with pytest.raises(HTTPException) as refusal:
await hook.async_pre_call_hook(
user_api_key_dict=_auth(user_id="other-user"),
cache=DualCache(),
data={"interaction_id": issued},
call_type=call_type,
)
assert refusal.value.status_code == 403
@pytest.mark.asyncio
async def test_cancel_interaction_returns_the_managed_id_held_by_the_caller(salt_key_env: str) -> None:
hook: Final = _hook(signing_key=salt_key_env)
owner: Final = _auth()
provider_id: Final = "interaction-provider-id"
issued_response: Final = hook._encrypt_response_id( # pyright: ignore[reportPrivateUsage] # mints ids the way create does
InteractionsAPIResponse(id=provider_id, status="in_progress"),
owner,
)
issued: Final = InteractionsAPIResponse.model_validate(issued_response).id
assert issued is not None
authorized: Final = TypeAdapter(dict[str, object]).validate_python(
await hook.async_pre_call_hook(
user_api_key_dict=owner,
cache=DualCache(),
data={"interaction_id": issued},
call_type="acancel_interaction",
)
)
result: Final = await hook.async_post_call_success_hook( # pyright: ignore[reportArgumentType] # interactions share this proxy hook
data=authorized,
user_api_key_dict=owner,
response=CancelInteractionResult(id=provider_id, status="cancelled"),
)
assert CancelInteractionResult.model_validate(result).id == issued
@pytest.mark.asyncio
@pytest.mark.parametrize("id_location", ("interaction_id", "interaction"))
async def test_streamed_interaction_schema_ids_are_bound_to_the_caller(salt_key_env: str, id_location: str) -> None:
hook: Final = _hook(signing_key=salt_key_env)
caller: Final = _auth().model_copy(update={"request_route": "/interactions"})
provider_id: Final = "qca1.sess_owner.evt_input"
async def chunks():
yield InteractionsAPIStreamingResponse(
event_type="interaction.created",
**(
{"interaction_id": provider_id}
if id_location == "interaction_id"
else {"interaction": {"id": provider_id}}
),
)
protected: Final = tuple(
[
chunk
async for chunk in hook.async_post_call_streaming_iterator_hook(
user_api_key_dict=caller,
response=chunks(),
request_data={},
)
]
)
protected_id: Final = (
protected[0].interaction_id
if id_location == "interaction_id"
else TypeAdapter(dict[str, str]).validate_python(protected[0].interaction)["id"]
)
assert protected_id != provider_id
assert hook._decrypt_response_id(protected_id) == ( # pyright: ignore[reportPrivateUsage] # verifies ownership payload
provider_id,
"owner-user",
"owner-team",
)
@pytest.mark.asyncio
@pytest.mark.parametrize("nested", (False, True))
@pytest.mark.parametrize("stranger", ("user", "team"))

View file

@ -16,7 +16,13 @@ from litellm.exceptions import BadRequestError, EndpointNotSupportedByModelError
from litellm.integrations.custom_logger import CustomLogger
from litellm.types.router import RouterRateLimitError
from tests.unit.test_router.core.fallbacks_helpers import MESSAGES, MakeCore, complete, complete_sync, headers
from tests.unit.test_router.core.scripted_host import CORE_BUILDERS, ScriptedHost, completion_response, deployment, server_error
from tests.unit.test_router.core.scripted_host import (
CORE_BUILDERS,
ScriptedHost,
completion_response,
deployment,
server_error,
)
GROUP_LIMITS: Final = {"rpm": 5, "tpm": 1000}
TEAM_OWNED: Final = {"team_id": "T1"}
@ -131,7 +137,9 @@ async def test_a_guardrail_result_carries_the_router_headers(make_core: MakeCore
# S2 and Y4 are ruled to differ: the Rust core filters every pick, so these run on it only.
def test_the_rust_core_filters_a_team_owned_deployment_on_a_sync_pick(host: ScriptedHost) -> None:
core: Final = CORE_BUILDERS["rust"](host, model_list=[deployment("own", "o1", model_info=TEAM_OWNED)], num_retries=0)
core: Final = CORE_BUILDERS["rust"](
host, model_list=[deployment("own", "o1", model_info=TEAM_OWNED)], num_retries=0
)
with pytest.raises(RouterRateLimitError):
_ = complete_sync(core, "own")
assert host.deployments_called("completion") == []
@ -153,7 +161,9 @@ async def test_an_async_pick_runs_the_team_filter(make_core: MakeCore) -> None:
async def test_the_rust_core_refuses_a_team_owned_deployment_named_by_its_id(
host: ScriptedHost, request_fields: Mapping[str, object]
) -> None:
core: Final = CORE_BUILDERS["rust"](host, model_list=[deployment("own", "o1", model_info=TEAM_OWNED)], num_retries=0)
core: Final = CORE_BUILDERS["rust"](
host, model_list=[deployment("own", "o1", model_info=TEAM_OWNED)], num_retries=0
)
with pytest.raises(RouterRateLimitError):
_ = await complete(core, "o1", **request_fields)
assert host.deployments_called("acompletion") == []
@ -202,4 +212,3 @@ def test_a_mocked_model_response_warns_once_per_dump_python_makes(make_core: Mak
_ = make_core(model_list=[deployment("g", "d1", mock_response=BLOCKED_RESPONSE)], num_retries=0)
serializer: Final = [w for w in caught if "Pydantic serializer warnings" in str(w.message)]
assert len(serializer) == PYTHON_DUMPS_OF_A_DEPLOYMENT

View file

@ -29442,7 +29442,7 @@ export interface components {
* CallTypes
* @enum {string}
*/
CallTypes: "embedding" | "aembedding" | "completion" | "acompletion" | "atext_completion" | "text_completion" | "image_generation" | "aimage_generation" | "image_edit" | "aimage_edit" | "moderation" | "amoderation" | "atranscription" | "transcription" | "aspeech" | "speech" | "rerank" | "arerank" | "search" | "asearch" | "decisions" | "adecisions" | "systemone" | "asystemone" | "extraction" | "aextraction" | "_arealtime" | "_aresponses_websocket" | "acreate_realtime_client_secret" | "arealtime_calls" | "acreate_realtime_transcription_session" | "acreate_realtime_translation_client_secret" | "arealtime_translation_calls" | "create_batch" | "acreate_batch" | "aretrieve_batch" | "retrieve_batch" | "acancel_batch" | "cancel_batch" | "pass_through_endpoint" | "anthropic_messages" | "aanthropic_messages" | "get_assistants" | "aget_assistants" | "create_assistants" | "acreate_assistants" | "delete_assistant" | "adelete_assistant" | "acreate_thread" | "create_thread" | "aget_thread" | "get_thread" | "a_add_message" | "add_message" | "aget_messages" | "get_messages" | "arun_thread" | "run_thread" | "arun_thread_stream" | "run_thread_stream" | "afile_retrieve" | "file_retrieve" | "afile_delete" | "file_delete" | "afile_list" | "file_list" | "acreate_file" | "create_file" | "afile_content" | "file_content" | "create_fine_tuning_job" | "acreate_fine_tuning_job" | "create_video" | "acreate_video" | "video_generation" | "avideo_generation" | "avideo_retrieve" | "video_retrieve" | "avideo_content" | "video_content" | "video_remix" | "avideo_remix" | "video_list" | "avideo_list" | "video_retrieve_job" | "avideo_retrieve_job" | "video_delete" | "avideo_delete" | "video_create_character" | "avideo_create_character" | "video_get_character" | "avideo_get_character" | "video_edit" | "avideo_edit" | "video_extension" | "avideo_extension" | "vector_store_file_create" | "avector_store_file_create" | "vector_store_file_list" | "avector_store_file_list" | "vector_store_file_retrieve" | "avector_store_file_retrieve" | "vector_store_file_content" | "avector_store_file_content" | "vector_store_file_update" | "avector_store_file_update" | "vector_store_file_delete" | "avector_store_file_delete" | "vector_store_create" | "avector_store_create" | "vector_store_search" | "avector_store_search" | "ingest" | "aingest" | "query" | "aquery" | "create_interaction" | "acreate_interaction" | "create_container" | "acreate_container" | "list_containers" | "alist_containers" | "retrieve_container" | "aretrieve_container" | "delete_container" | "adelete_container" | "list_container_files" | "alist_container_files" | "upload_container_file" | "aupload_container_file" | "create_sandbox" | "acreate_sandbox" | "delete_sandbox" | "adelete_sandbox" | "run_code" | "arun_code" | "code_interpreter_tool" | "acode_interpreter_tool" | "acancel_fine_tuning_job" | "cancel_fine_tuning_job" | "alist_fine_tuning_jobs" | "list_fine_tuning_jobs" | "aretrieve_fine_tuning_job" | "retrieve_fine_tuning_job" | "responses" | "aresponses" | "get_responses" | "aget_responses" | "delete_responses" | "adelete_responses" | "cancel_responses" | "acancel_responses" | "compact_responses" | "acompact_responses" | "alist_input_items" | "llm_passthrough_route" | "allm_passthrough_route" | "generate_content" | "agenerate_content" | "generate_content_stream" | "agenerate_content_stream" | "ocr" | "aocr" | "call_mcp_tool" | "list_mcp_tools" | "asend_message" | "send_message" | "acreate_skill";
CallTypes: "embedding" | "aembedding" | "completion" | "acompletion" | "atext_completion" | "text_completion" | "image_generation" | "aimage_generation" | "image_edit" | "aimage_edit" | "moderation" | "amoderation" | "atranscription" | "transcription" | "aspeech" | "speech" | "rerank" | "arerank" | "search" | "asearch" | "decisions" | "adecisions" | "systemone" | "asystemone" | "extraction" | "aextraction" | "_arealtime" | "_aresponses_websocket" | "acreate_realtime_client_secret" | "arealtime_calls" | "acreate_realtime_transcription_session" | "acreate_realtime_translation_client_secret" | "arealtime_translation_calls" | "create_batch" | "acreate_batch" | "aretrieve_batch" | "retrieve_batch" | "acancel_batch" | "cancel_batch" | "pass_through_endpoint" | "anthropic_messages" | "aanthropic_messages" | "get_assistants" | "aget_assistants" | "create_assistants" | "acreate_assistants" | "delete_assistant" | "adelete_assistant" | "acreate_thread" | "create_thread" | "aget_thread" | "get_thread" | "a_add_message" | "add_message" | "aget_messages" | "get_messages" | "arun_thread" | "run_thread" | "arun_thread_stream" | "run_thread_stream" | "afile_retrieve" | "file_retrieve" | "afile_delete" | "file_delete" | "afile_list" | "file_list" | "acreate_file" | "create_file" | "afile_content" | "file_content" | "create_fine_tuning_job" | "acreate_fine_tuning_job" | "create_video" | "acreate_video" | "video_generation" | "avideo_generation" | "avideo_retrieve" | "video_retrieve" | "avideo_content" | "video_content" | "video_remix" | "avideo_remix" | "video_list" | "avideo_list" | "video_retrieve_job" | "avideo_retrieve_job" | "video_delete" | "avideo_delete" | "video_create_character" | "avideo_create_character" | "video_get_character" | "avideo_get_character" | "video_edit" | "avideo_edit" | "video_extension" | "avideo_extension" | "vector_store_file_create" | "avector_store_file_create" | "vector_store_file_list" | "avector_store_file_list" | "vector_store_file_retrieve" | "avector_store_file_retrieve" | "vector_store_file_content" | "avector_store_file_content" | "vector_store_file_update" | "avector_store_file_update" | "vector_store_file_delete" | "avector_store_file_delete" | "vector_store_create" | "avector_store_create" | "vector_store_search" | "avector_store_search" | "ingest" | "aingest" | "query" | "aquery" | "create_interaction" | "acreate_interaction" | "get_interaction" | "aget_interaction" | "delete_interaction" | "adelete_interaction" | "cancel_interaction" | "acancel_interaction" | "create_container" | "acreate_container" | "list_containers" | "alist_containers" | "retrieve_container" | "aretrieve_container" | "delete_container" | "adelete_container" | "list_container_files" | "alist_container_files" | "upload_container_file" | "aupload_container_file" | "create_sandbox" | "acreate_sandbox" | "delete_sandbox" | "adelete_sandbox" | "run_code" | "arun_code" | "code_interpreter_tool" | "acode_interpreter_tool" | "acancel_fine_tuning_job" | "cancel_fine_tuning_job" | "alist_fine_tuning_jobs" | "list_fine_tuning_jobs" | "aretrieve_fine_tuning_job" | "retrieve_fine_tuning_job" | "responses" | "aresponses" | "get_responses" | "aget_responses" | "delete_responses" | "adelete_responses" | "cancel_responses" | "acancel_responses" | "compact_responses" | "acompact_responses" | "alist_input_items" | "llm_passthrough_route" | "allm_passthrough_route" | "generate_content" | "agenerate_content" | "generate_content_stream" | "agenerate_content_stream" | "ocr" | "aocr" | "call_mcp_tool" | "list_mcp_tools" | "asend_message" | "send_message" | "acreate_skill";
/** CallbackDelete */
CallbackDelete: {
/** Callback Name */