mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-15 23:31:29 +00:00
e967bc8c4f
* fix(e2e): reference client.proxy in mid-conversation native providers test
EndpointsClient exposes the shared ProxyClient as .proxy and has never had a
.gateway attribute, so these two calls raised AttributeError at runtime and
failed the tests/e2e basedpyright zero-error gate for any PR touching e2e
files. Introduced in 23b5b7d199.
* test(e2e): cover 12 non-core LLM coverage registry cells
Raises Non-Core LLMs registry coverage from 24/50 to 36/50 (overall 51.9%
to 54.8%). Four cells were already asserted by existing tests and only
gain their covers marker (openai embeddings, openai image generation,
openai TTS, cohere rerank); one is dual-marked onto the existing
spend-tracking embeddings test rather than duplicated.
New tests: bedrock and vertex embeddings, streaming TTS (asserts chunked
transfer encoding so a buffered body cannot pass), audio transcriptions
via the realtime suite's wav fixture, moderations flag/pass pair, and
files list/retrieve in the batches suite.
Harness: e2e_http.upload generalized to any form model with a
file_content_type override (batches path unchanged), new stream_binary
primitive + BinaryStream for binary chunked responses, transcribe and
moderations client methods, file retrieve/list client methods.
* fix(e2e): close streamed TTS response on error paths and surface the error body
With stream=True a non-2xx response returned with the body unread, keeping
the socket checked out until garbage collection; the sibling
_streaming_outcome already consumes resp.text on error. The response now
closes on every path and BinaryStream carries a bounded error_body so a
failed stream call is triageable.
* test(e2e): assert streamed TTS response carries no content-length
429 lines
12 KiB
Python
429 lines
12 KiB
Python
"""Transport: the typed request primitives clients use, behind a Protocol.
|
|
|
|
`Transport` is what each client depends on (composition + DI); `HttpTransport` is
|
|
the concrete frozen-slots dataclass that fulfils it via the e2e_http wrapper. No
|
|
client touches requests.* or builds raw dicts; they pass pydantic models here.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Protocol
|
|
|
|
from pydantic import BaseModel
|
|
|
|
import e2e_http
|
|
from e2e_http import (
|
|
URL,
|
|
AuthHeaders,
|
|
BinaryStream,
|
|
ProbeResult,
|
|
Result,
|
|
StreamingResponse,
|
|
)
|
|
|
|
|
|
class Transport(Protocol):
|
|
def post[R: BaseModel](
|
|
self, path: str, *, headers: BaseModel, json: BaseModel, response_type: type[R]
|
|
) -> Result[R]: ...
|
|
|
|
def stream(
|
|
self, path: str, *, headers: BaseModel, json: BaseModel
|
|
) -> StreamingResponse: ...
|
|
|
|
def stream_binary(
|
|
self,
|
|
path: str,
|
|
*,
|
|
headers: BaseModel,
|
|
json: BaseModel,
|
|
chunk_size: int = 8192,
|
|
) -> BinaryStream: ...
|
|
|
|
def send(
|
|
self,
|
|
path: str,
|
|
*,
|
|
headers: BaseModel,
|
|
json: BaseModel,
|
|
params: BaseModel | None = None,
|
|
stream: bool = False,
|
|
) -> StreamingResponse: ...
|
|
|
|
def get[R: BaseModel](
|
|
self,
|
|
path: str,
|
|
*,
|
|
headers: BaseModel,
|
|
params: BaseModel,
|
|
response_type: type[R],
|
|
) -> Result[R]: ...
|
|
|
|
def delete[R: BaseModel](
|
|
self,
|
|
path: str,
|
|
*,
|
|
headers: BaseModel,
|
|
json: BaseModel,
|
|
response_type: type[R],
|
|
params: BaseModel | None = None,
|
|
) -> Result[R]: ...
|
|
|
|
def patch[R: BaseModel](
|
|
self, path: str, *, headers: BaseModel, json: BaseModel, response_type: type[R]
|
|
) -> Result[R]: ...
|
|
|
|
def put[R: BaseModel](
|
|
self, path: str, *, headers: BaseModel, json: BaseModel, response_type: type[R]
|
|
) -> Result[R]: ...
|
|
|
|
def probe(self, path: str, *, params: BaseModel) -> ProbeResult: ...
|
|
|
|
def upload[R: BaseModel](
|
|
self,
|
|
path: str,
|
|
*,
|
|
headers: BaseModel,
|
|
form: BaseModel,
|
|
filename: str,
|
|
content: bytes,
|
|
file_content_type: str = "application/jsonl",
|
|
params: BaseModel | None = None,
|
|
response_type: type[R],
|
|
) -> Result[R]: ...
|
|
|
|
def download(self, path: str, *, headers: BaseModel) -> StreamingResponse: ...
|
|
|
|
def bearer(self, key: str) -> AuthHeaders: ...
|
|
|
|
@property
|
|
def master(self) -> AuthHeaders: ...
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class HttpTransport:
|
|
base_url: str
|
|
master_key: str
|
|
request_timeout: float = 60.0
|
|
|
|
def _url(self, path: str) -> URL:
|
|
return URL(f"{self.base_url.rstrip('/')}{path}")
|
|
|
|
def bearer(self, key: str) -> AuthHeaders:
|
|
return AuthHeaders(authorization=f"Bearer {key}")
|
|
|
|
@property
|
|
def master(self) -> AuthHeaders:
|
|
return self.bearer(self.master_key)
|
|
|
|
def post[R: BaseModel](
|
|
self, path: str, *, headers: BaseModel, json: BaseModel, response_type: type[R]
|
|
) -> Result[R]:
|
|
return e2e_http.post(
|
|
self._url(path),
|
|
headers=headers,
|
|
json=json,
|
|
response_type=response_type,
|
|
timeout=self.request_timeout,
|
|
)
|
|
|
|
def get[R: BaseModel](
|
|
self,
|
|
path: str,
|
|
*,
|
|
headers: BaseModel,
|
|
params: BaseModel,
|
|
response_type: type[R],
|
|
) -> Result[R]:
|
|
return e2e_http.get(
|
|
self._url(path),
|
|
headers=headers,
|
|
params=params,
|
|
response_type=response_type,
|
|
timeout=self.request_timeout,
|
|
)
|
|
|
|
def delete[R: BaseModel](
|
|
self,
|
|
path: str,
|
|
*,
|
|
headers: BaseModel,
|
|
json: BaseModel,
|
|
response_type: type[R],
|
|
params: BaseModel | None = None,
|
|
) -> Result[R]:
|
|
return e2e_http.delete(
|
|
self._url(path),
|
|
headers=headers,
|
|
json=json,
|
|
params=params,
|
|
response_type=response_type,
|
|
timeout=self.request_timeout,
|
|
)
|
|
|
|
def patch[R: BaseModel](
|
|
self, path: str, *, headers: BaseModel, json: BaseModel, response_type: type[R]
|
|
) -> Result[R]:
|
|
return e2e_http.patch(
|
|
self._url(path),
|
|
headers=headers,
|
|
json=json,
|
|
response_type=response_type,
|
|
timeout=self.request_timeout,
|
|
)
|
|
|
|
def put[R: BaseModel](
|
|
self, path: str, *, headers: BaseModel, json: BaseModel, response_type: type[R]
|
|
) -> Result[R]:
|
|
return e2e_http.put(
|
|
self._url(path),
|
|
headers=headers,
|
|
json=json,
|
|
response_type=response_type,
|
|
timeout=self.request_timeout,
|
|
)
|
|
|
|
def stream(
|
|
self, path: str, *, headers: BaseModel, json: BaseModel
|
|
) -> StreamingResponse:
|
|
return e2e_http.stream(
|
|
self._url(path), headers=headers, json=json, timeout=self.request_timeout
|
|
)
|
|
|
|
def stream_binary(
|
|
self,
|
|
path: str,
|
|
*,
|
|
headers: BaseModel,
|
|
json: BaseModel,
|
|
chunk_size: int = 8192,
|
|
) -> BinaryStream:
|
|
return e2e_http.stream_binary(
|
|
self._url(path),
|
|
headers=headers,
|
|
json=json,
|
|
chunk_size=chunk_size,
|
|
timeout=self.request_timeout,
|
|
)
|
|
|
|
def send(
|
|
self,
|
|
path: str,
|
|
*,
|
|
headers: BaseModel,
|
|
json: BaseModel,
|
|
params: BaseModel | None = None,
|
|
stream: bool = False,
|
|
) -> StreamingResponse:
|
|
return e2e_http.send(
|
|
self._url(path),
|
|
headers=headers,
|
|
json=json,
|
|
params=params,
|
|
stream=stream,
|
|
timeout=self.request_timeout,
|
|
)
|
|
|
|
def probe(self, path: str, *, params: BaseModel) -> ProbeResult:
|
|
return e2e_http.probe(
|
|
self._url(path),
|
|
headers=self.master,
|
|
params=params,
|
|
timeout=self.request_timeout,
|
|
)
|
|
|
|
def upload[R: BaseModel](
|
|
self,
|
|
path: str,
|
|
*,
|
|
headers: BaseModel,
|
|
form: BaseModel,
|
|
filename: str,
|
|
content: bytes,
|
|
file_content_type: str = "application/jsonl",
|
|
params: BaseModel | None = None,
|
|
response_type: type[R],
|
|
) -> Result[R]:
|
|
return e2e_http.upload(
|
|
self._url(path),
|
|
headers=headers,
|
|
form=form,
|
|
filename=filename,
|
|
content=content,
|
|
file_content_type=file_content_type,
|
|
params=params,
|
|
response_type=response_type,
|
|
timeout=self.request_timeout,
|
|
)
|
|
|
|
def download(self, path: str, *, headers: BaseModel) -> StreamingResponse:
|
|
return e2e_http.download(
|
|
self._url(path), headers=headers, timeout=self.request_timeout
|
|
)
|
|
|
|
|
|
# Top-level management/admin route groups. In a split deployment these are served
|
|
# by the control plane (a different service from the LLM data plane). LLM routes
|
|
# (/chat, /embeddings, and native passthrough like /gemini, /anthropic) are NOT
|
|
# here and fall through to the data plane. Matched as path prefixes.
|
|
CONTROL_PLANE_PREFIXES: tuple[str, ...] = (
|
|
"/key",
|
|
"/user",
|
|
"/team",
|
|
"/organization",
|
|
"/customer",
|
|
"/end_user",
|
|
"/tag",
|
|
"/budget",
|
|
"/model/",
|
|
"/access_group",
|
|
"/spend",
|
|
"/global",
|
|
"/config",
|
|
"/guardrails",
|
|
"/openapi.json",
|
|
)
|
|
|
|
|
|
def is_control_plane_path(path: str) -> bool:
|
|
"""True if `path` is a management/admin route (served by the control plane in a
|
|
split deployment), false for LLM data-plane routes."""
|
|
return path.startswith(CONTROL_PLANE_PREFIXES)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class SplitTransport:
|
|
"""A Transport that dispatches each call by path to one of two backends: the
|
|
management/admin control plane or the LLM data plane.
|
|
|
|
Litellm can run as a split control-plane/data-plane deployment where the two
|
|
surfaces live on different services. Clients here stay plane-agnostic — they
|
|
keep calling ``transport.post("/budget/new", ...)`` or
|
|
``transport.send("/chat/completions", ...)`` — and routing happens in one place
|
|
by path (see ``CONTROL_PLANE_PREFIXES``). When ``control`` and ``data`` share a
|
|
base URL (the monolithic default), routing is a no-op. ``bearer``/``master``
|
|
are plane-agnostic (same master key both planes), so they come from ``data``.
|
|
"""
|
|
|
|
data: HttpTransport
|
|
control: HttpTransport
|
|
|
|
def _route(self, path: str) -> HttpTransport:
|
|
return self.control if is_control_plane_path(path) else self.data
|
|
|
|
def bearer(self, key: str) -> AuthHeaders:
|
|
return self.data.bearer(key)
|
|
|
|
@property
|
|
def master(self) -> AuthHeaders:
|
|
return self.data.master
|
|
|
|
def post[R: BaseModel](
|
|
self, path: str, *, headers: BaseModel, json: BaseModel, response_type: type[R]
|
|
) -> Result[R]:
|
|
return self._route(path).post(
|
|
path, headers=headers, json=json, response_type=response_type
|
|
)
|
|
|
|
def get[R: BaseModel](
|
|
self,
|
|
path: str,
|
|
*,
|
|
headers: BaseModel,
|
|
params: BaseModel,
|
|
response_type: type[R],
|
|
) -> Result[R]:
|
|
return self._route(path).get(
|
|
path, headers=headers, params=params, response_type=response_type
|
|
)
|
|
|
|
def delete[R: BaseModel](
|
|
self,
|
|
path: str,
|
|
*,
|
|
headers: BaseModel,
|
|
json: BaseModel,
|
|
response_type: type[R],
|
|
params: BaseModel | None = None,
|
|
) -> Result[R]:
|
|
return self._route(path).delete(
|
|
path,
|
|
headers=headers,
|
|
json=json,
|
|
response_type=response_type,
|
|
params=params,
|
|
)
|
|
|
|
def patch[R: BaseModel](
|
|
self, path: str, *, headers: BaseModel, json: BaseModel, response_type: type[R]
|
|
) -> Result[R]:
|
|
return self._route(path).patch(
|
|
path, headers=headers, json=json, response_type=response_type
|
|
)
|
|
|
|
def put[R: BaseModel](
|
|
self, path: str, *, headers: BaseModel, json: BaseModel, response_type: type[R]
|
|
) -> Result[R]:
|
|
return self._route(path).put(
|
|
path, headers=headers, json=json, response_type=response_type
|
|
)
|
|
|
|
def stream(
|
|
self, path: str, *, headers: BaseModel, json: BaseModel
|
|
) -> StreamingResponse:
|
|
return self._route(path).stream(path, headers=headers, json=json)
|
|
|
|
def stream_binary(
|
|
self,
|
|
path: str,
|
|
*,
|
|
headers: BaseModel,
|
|
json: BaseModel,
|
|
chunk_size: int = 8192,
|
|
) -> BinaryStream:
|
|
return self._route(path).stream_binary(
|
|
path, headers=headers, json=json, chunk_size=chunk_size
|
|
)
|
|
|
|
def send(
|
|
self,
|
|
path: str,
|
|
*,
|
|
headers: BaseModel,
|
|
json: BaseModel,
|
|
params: BaseModel | None = None,
|
|
stream: bool = False,
|
|
) -> StreamingResponse:
|
|
return self._route(path).send(
|
|
path, headers=headers, json=json, params=params, stream=stream
|
|
)
|
|
|
|
def probe(self, path: str, *, params: BaseModel) -> ProbeResult:
|
|
return self._route(path).probe(path, params=params)
|
|
|
|
def upload[R: BaseModel](
|
|
self,
|
|
path: str,
|
|
*,
|
|
headers: BaseModel,
|
|
form: BaseModel,
|
|
filename: str,
|
|
content: bytes,
|
|
file_content_type: str = "application/jsonl",
|
|
params: BaseModel | None = None,
|
|
response_type: type[R],
|
|
) -> Result[R]:
|
|
return self._route(path).upload(
|
|
path,
|
|
headers=headers,
|
|
form=form,
|
|
filename=filename,
|
|
content=content,
|
|
file_content_type=file_content_type,
|
|
params=params,
|
|
response_type=response_type,
|
|
)
|
|
|
|
def download(self, path: str, *, headers: BaseModel) -> StreamingResponse:
|
|
return self._route(path).download(path, headers=headers)
|