From adf41853ee71db9edcae5dbfe91608fc1d4a4aa4 Mon Sep 17 00:00:00 2001 From: cursor-agent Date: Wed, 20 May 2026 15:57:20 +0000 Subject: [PATCH] feat(tests): vendor example_openai_endpoint as a local mock server MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds a vendored copy of BerriAI/example_openai_endpoint under tests/mock_endpoints/example_openai_endpoint/ together with infrastructure to run it in-process during a test session, so CI no longer depends on the public Railway deployment (which has had multiple outages, e.g. the May 19 2026 GCP-account incident that took down 8 CI jobs). Pieces added: - tests/mock_endpoints/example_openai_endpoint/ vendored upstream source (main.py, batch_and_files_api.py, requirements.txt, Dockerfile), intended to be re-vendored as-is from upstream when needed (see README). - tests/mock_endpoints/__init__.py helper module exposing MOCK_OPENAI_BASE_URL (reads LITELLM_MOCK_OPENAI_BASE_URL with the Railway URL as fallback) and start_mock_server() — boots the server as a subprocess on a free port, polls /chat/completions until it responds, returns an RAII-style handle for teardown. - tests/mock_endpoints/conftest.py session-scoped pytest fixture mock_openai_endpoint_server that any suite can opt into via pytest_plugins. - tests/mock_endpoints/start_mock_server.sh bash launcher (foreground or --background) for shell-driven workflows. - tests/mock_endpoints/README.md local + CI usage docs and instructions for re-vendoring from upstream. - tests/test_litellm/mock_endpoints/ end-to-end smoke tests that boot the mock and exercise /chat/completions, /v1/embeddings, and the fine-tuning list endpoint. Wiring: - New mock-server dependency-group in pyproject.toml (just adds slowapi; the rest of the upstream deps already come in via dev/proxy-dev). - _test-unit-base.yml now syncs --group mock-server so any reusable unit test job has the deps available. - test-unit-misc.yml runs the new smoke tests as part of its path set. - Makefile gains 'make mock-server' (and an install-mock-server helper) for running the server locally with the project venv. The Railway URL is still the runtime fallback for any test that has not been migrated, so this PR is non-breaking; follow-up PRs can migrate hard-coded URLs to use MOCK_OPENAI_BASE_URL / the new fixture. Co-authored-by: Mateo Wang --- .github/workflows/_test-unit-base.yml | 2 +- .github/workflows/test-unit-misc.yml | 1 + Makefile | 15 +- pyproject.toml | 7 + tests/mock_endpoints/README.md | 113 + tests/mock_endpoints/__init__.py | 155 + tests/mock_endpoints/conftest.py | 54 + .../example_openai_endpoint/Dockerfile | 20 + .../batch_and_files_api.py | 310 ++ .../example_openai_endpoint/main.py | 2564 +++++++++++++++++ .../example_openai_endpoint/requirements.txt | 8 + tests/mock_endpoints/start_mock_server.sh | 61 + tests/test_litellm/mock_endpoints/__init__.py | 0 .../test_mock_openai_endpoint_server.py | 73 + uv.lock | 30 + 15 files changed, 3410 insertions(+), 3 deletions(-) create mode 100644 tests/mock_endpoints/README.md create mode 100644 tests/mock_endpoints/__init__.py create mode 100644 tests/mock_endpoints/conftest.py create mode 100644 tests/mock_endpoints/example_openai_endpoint/Dockerfile create mode 100644 tests/mock_endpoints/example_openai_endpoint/batch_and_files_api.py create mode 100644 tests/mock_endpoints/example_openai_endpoint/main.py create mode 100644 tests/mock_endpoints/example_openai_endpoint/requirements.txt create mode 100755 tests/mock_endpoints/start_mock_server.sh create mode 100644 tests/test_litellm/mock_endpoints/__init__.py create mode 100644 tests/test_litellm/mock_endpoints/test_mock_openai_endpoint_server.py diff --git a/.github/workflows/_test-unit-base.yml b/.github/workflows/_test-unit-base.yml index 7e91341ac77..d539c151084 100644 --- a/.github/workflows/_test-unit-base.yml +++ b/.github/workflows/_test-unit-base.yml @@ -68,7 +68,7 @@ jobs: - name: Install dependencies run: | - uv sync --frozen --group ci --group proxy-dev --extra google --extra proxy --extra semantic-router + uv sync --frozen --group ci --group proxy-dev --group mock-server --extra google --extra proxy --extra semantic-router - name: Generate Prisma client env: diff --git a/.github/workflows/test-unit-misc.yml b/.github/workflows/test-unit-misc.yml index 9add77ff424..1ed9aee4bcb 100644 --- a/.github/workflows/test-unit-misc.yml +++ b/.github/workflows/test-unit-misc.yml @@ -30,6 +30,7 @@ jobs: tests/test_litellm/experimental_mcp_client tests/test_litellm/images tests/test_litellm/interactions + tests/test_litellm/mock_endpoints tests/test_litellm/passthrough tests/test_litellm/vector_stores tests/test_litellm/test_*.py diff --git a/Makefile b/Makefile index 5dbd308a3e2..d8cec4a8aef 100644 --- a/Makefile +++ b/Makefile @@ -5,8 +5,8 @@ test-unit-integrations test-unit-core-utils test-unit-other test-unit-root \ test-proxy-unit-a test-proxy-unit-b test-integration test-unit-helm \ info lint lint-dev format \ - install-dev install-proxy-dev install-test-deps \ - install-helm-unittest check-circular-imports check-import-safety + install-dev install-proxy-dev install-test-deps install-mock-server \ + mock-server install-helm-unittest check-circular-imports check-import-safety # Default target help: @@ -17,6 +17,7 @@ help: @echo " make install-proxy-dev-ci - Install proxy dev dependencies (CI-compatible)" @echo " make install-test-deps - Install the full local test environment" @echo " make install-helm-unittest - Install helm unittest plugin" + @echo " make mock-server - Run the vendored mock OpenAI endpoint locally (PORT=8090)" @echo " make format - Apply Black code formatting" @echo " make format-check - Check Black code formatting (matches CI)" @echo " make lint - Run all linting (Ruff, MyPy, Black check, circular imports, import safety)" @@ -68,6 +69,16 @@ install-test-deps: install-proxy-dev install-helm-unittest: helm plugin install https://github.com/helm-unittest/helm-unittest --version v0.4.4 || echo "ignore error if plugin exists" +# Mock OpenAI/Anthropic/Vertex endpoint server (vendored copy of +# BerriAI/example_openai_endpoint). Use this to run tests offline / without +# relying on the Railway-hosted deployment. +install-mock-server: + $(UV) sync --group mock-server --group proxy-dev --extra proxy + +mock-server: install-mock-server + @echo "Starting mock OpenAI endpoint on http://0.0.0.0:$${PORT:-8090} (Ctrl+C to stop)" + $(UV_RUN) python tests/mock_endpoints/example_openai_endpoint/main.py + # Formatting format: install-dev cd litellm && $(UV_RUN) black . && cd .. diff --git a/pyproject.toml b/pyproject.toml index 70681c4ed6c..d1e588ae547 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -216,6 +216,13 @@ healthcheck = [ "httpx==0.28.1", "pyyaml==6.0.3", ] +mock-server = [ + # Extra deps required by the vendored mock OpenAI endpoint under + # tests/mock_endpoints/example_openai_endpoint/. Most of its deps (fastapi, + # uvicorn, pydantic, python-multipart, python-dotenv, websockets) are + # already pulled in by the proxy / dev groups; slowapi is the one extra. + "slowapi==0.1.9", +] [build-system] requires = ["uv_build==0.11.8"] diff --git a/tests/mock_endpoints/README.md b/tests/mock_endpoints/README.md new file mode 100644 index 00000000000..970be815ea7 --- /dev/null +++ b/tests/mock_endpoints/README.md @@ -0,0 +1,113 @@ +# Mock LLM Endpoints (for local dev + CI) + +This directory contains small, self-contained mock servers used by tests to +avoid hitting real provider APIs (and to avoid relying on third-party +hosting services like Railway). + +## `example_openai_endpoint/` + +A vendored copy of [BerriAI/example_openai_endpoint](https://github.com/BerriAI/example_openai_endpoint), +a FastAPI app that implements stub OpenAI- / Anthropic- / Vertex- / Bedrock- +compatible endpoints (chat completions, embeddings, audio, batches, etc.). +Many tests across the repo reference its public deployment at +`https://exampleopenaiendpoint-production.up.railway.app/`. + +The Railway deployment has had multiple outages +(see ) that break CI. The goal of +vendoring it here is to make the same stub server runnable in the same +container as CI (and locally on a developer machine), so tests do not depend +on any external service being up. + +### Running locally + +The easiest way to start the server (installs `slowapi` + friends into the +project venv first, then runs the server on `:8090`): + +```bash +make mock-server # foreground +PORT=18090 make mock-server # alt port +``` + +If you don't want to use the Makefile, you can run the server directly with +any Python interpreter that has the deps installed: + +```bash +python -m pip install -r tests/mock_endpoints/example_openai_endpoint/requirements.txt +python tests/mock_endpoints/example_openai_endpoint/main.py +``` + +Or use the bash helper (handy for CI / shell scripts — it starts the server +in the background, polls until it's ready, and writes logs to `/tmp`): + +```bash +./tests/mock_endpoints/start_mock_server.sh # foreground +./tests/mock_endpoints/start_mock_server.sh --background # background +``` + +### Using it from a pytest suite + +The recommended pattern is the session-scoped fixture defined in +[`tests/mock_endpoints/conftest.py`](./conftest.py). Any test file (or a +suite-level `conftest.py`) can opt in like so: + +```python +import pytest + +pytest_plugins = ("tests.mock_endpoints.conftest",) + + +def test_something(mock_openai_endpoint_server): + base_url = mock_openai_endpoint_server # e.g. "http://127.0.0.1:53892" + ... +``` + +The fixture: + +- Picks a free port automatically (so suites can run in parallel without + colliding). +- Boots the vendored server as a subprocess. +- Waits until `/chat/completions` returns 200 before yielding the URL. +- Exposes the URL as `LITELLM_MOCK_OPENAI_BASE_URL` in the environment so + any code that calls `tests.mock_endpoints.MOCK_OPENAI_BASE_URL` picks it + up automatically. +- Tears the subprocess down at session end. + +See [`tests/test_litellm/mock_endpoints/test_mock_openai_endpoint_server.py`](../test_litellm/mock_endpoints/test_mock_openai_endpoint_server.py) +for a working end-to-end example. + +### Quick sanity check + +```bash +curl -s http://127.0.0.1:8090/chat/completions \ + -H 'Authorization: Bearer sk-test' \ + -H 'Content-Type: application/json' \ + -d '{"model":"gpt-4o-mini","messages":[{"role":"user","content":"hi"}]}' +``` + +### Pointing tests at the local server + +Tests that currently hard-code +`https://exampleopenaiendpoint-production.up.railway.app/` can be pointed at +the local mock by changing the URL to `http://127.0.0.1:8090/`. New tests +should prefer this local mock over the Railway deployment. + +### Keeping the vendored copy in sync + +The upstream source of truth is still +. To pull the latest +version into this repo: + +```bash +curl -fsSL -o tests/mock_endpoints/example_openai_endpoint/main.py \ + https://raw.githubusercontent.com/BerriAI/example_openai_endpoint/main/main.py +curl -fsSL -o tests/mock_endpoints/example_openai_endpoint/batch_and_files_api.py \ + https://raw.githubusercontent.com/BerriAI/example_openai_endpoint/main/batch_and_files_api.py +curl -fsSL -o tests/mock_endpoints/example_openai_endpoint/requirements.txt \ + https://raw.githubusercontent.com/BerriAI/example_openai_endpoint/main/requirements.txt +curl -fsSL -o tests/mock_endpoints/example_openai_endpoint/Dockerfile \ + https://raw.githubusercontent.com/BerriAI/example_openai_endpoint/main/Dockerfile +``` + +These files should be copied **as-is** from upstream so the two repos stay +in sync; do not edit them in place. If an endpoint needs to change, change +it upstream first, then re-vendor. diff --git a/tests/mock_endpoints/__init__.py b/tests/mock_endpoints/__init__.py new file mode 100644 index 00000000000..e10bb13587b --- /dev/null +++ b/tests/mock_endpoints/__init__.py @@ -0,0 +1,155 @@ +"""Helpers for running the vendored mock OpenAI / Anthropic / Vertex endpoint. + +The mock server itself lives in ``tests/mock_endpoints/example_openai_endpoint/`` +and is a vendored copy of https://github.com/BerriAI/example_openai_endpoint. + +Many tests in this repo currently point at the Railway-hosted version of that +server (``https://exampleopenaiendpoint-production.up.railway.app``). Railway +outages take those tests down with it. This module exposes: + +* :data:`MOCK_OPENAI_BASE_URL` — the URL tests should hit. It reads + ``LITELLM_MOCK_OPENAI_BASE_URL`` from the environment, falling back to the + public Railway URL for backwards-compatibility with tests that have not been + migrated yet. +* :func:`start_mock_server` — spawn the vendored server as a subprocess on a + free port. Used by the pytest fixture in ``conftest.py`` but also callable + directly from scripts. + +The pytest fixture ``mock_openai_endpoint_server`` (see ``conftest.py``) is +the recommended way for a test suite to opt in: it starts the server once per +session, exposes the URL, and tears the process down on exit. +""" + +from __future__ import annotations + +import os +import socket +import subprocess +import sys +import time +from pathlib import Path +from typing import Optional + +import urllib.error +import urllib.request + +DEFAULT_REMOTE_URL = "https://exampleopenaiendpoint-production.up.railway.app" + +_SERVER_DIR = Path(__file__).resolve().parent / "example_openai_endpoint" +_SERVER_MAIN = _SERVER_DIR / "main.py" + + +def _resolve_base_url() -> str: + url = os.environ.get("LITELLM_MOCK_OPENAI_BASE_URL") + if url: + return url.rstrip("/") + return DEFAULT_REMOTE_URL + + +MOCK_OPENAI_BASE_URL = _resolve_base_url() + + +def _pick_free_port() -> int: + """Bind to port 0 and immediately release, returning the kernel-assigned port.""" + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: + sock.bind(("127.0.0.1", 0)) + return sock.getsockname()[1] + + +def _wait_for_ready(url: str, timeout: float = 30.0) -> None: + """Poll the mock until ``/chat/completions`` returns a 2xx response.""" + deadline = time.monotonic() + timeout + last_err: Optional[BaseException] = None + payload = b'{"model":"gpt-4o-mini","messages":[{"role":"user","content":"ping"}]}' + while time.monotonic() < deadline: + try: + req = urllib.request.Request( + f"{url}/chat/completions", + data=payload, + headers={ + "Authorization": "Bearer sk-test", + "Content-Type": "application/json", + }, + method="POST", + ) + with urllib.request.urlopen(req, timeout=2.0) as resp: + if 200 <= resp.status < 300: + return + except (urllib.error.URLError, ConnectionError, OSError) as err: + last_err = err + time.sleep(0.2) + raise RuntimeError( + f"Mock server at {url} did not become ready within {timeout}s " + f"(last error: {last_err!r})" + ) + + +class MockServerHandle: + """Tiny RAII-style handle for a running mock server subprocess.""" + + def __init__(self, process: subprocess.Popen, base_url: str) -> None: + self.process = process + self.base_url = base_url + + def stop(self) -> None: + if self.process.poll() is not None: + return + self.process.terminate() + try: + self.process.wait(timeout=10) + except subprocess.TimeoutExpired: + self.process.kill() + self.process.wait(timeout=5) + + +def start_mock_server( + port: Optional[int] = None, + *, + wait: bool = True, + timeout: float = 30.0, + log_file: Optional[Path] = None, + python_executable: Optional[str] = None, +) -> MockServerHandle: + """Start the vendored mock server as a child process. + + Args: + port: Port to bind. ``None`` (default) picks a free port automatically. + wait: If ``True``, block until ``/chat/completions`` returns a 2xx response. + timeout: Maximum seconds to wait for readiness. + log_file: Optional file to capture stdout/stderr. ``None`` inherits the parent's streams. + python_executable: Python interpreter to launch the server with. Defaults to ``sys.executable``. + """ + if port is None: + port = _pick_free_port() + + env = {**os.environ, "PORT": str(port)} + stdout = open(log_file, "w") if log_file is not None else None + stderr = subprocess.STDOUT if stdout is not None else None + + process = subprocess.Popen( + [python_executable or sys.executable, str(_SERVER_MAIN)], + env=env, + stdout=stdout, + stderr=stderr, + cwd=str(_SERVER_DIR), + ) + base_url = f"http://127.0.0.1:{port}" + handle = MockServerHandle(process=process, base_url=base_url) + + if wait: + try: + _wait_for_ready(base_url, timeout=timeout) + except Exception: + handle.stop() + if stdout is not None: + stdout.close() + raise + return handle + + +__all__ = [ + "DEFAULT_REMOTE_URL", + "MOCK_OPENAI_BASE_URL", + "MockServerHandle", + "start_mock_server", +] diff --git a/tests/mock_endpoints/conftest.py b/tests/mock_endpoints/conftest.py new file mode 100644 index 00000000000..be7b2dbb744 --- /dev/null +++ b/tests/mock_endpoints/conftest.py @@ -0,0 +1,54 @@ +"""Pytest fixtures for the vendored mock OpenAI endpoint. + +Any test suite can opt into a per-session local mock server like so:: + + pytest_plugins = ("tests.mock_endpoints.conftest",) + + def test_something(mock_openai_endpoint_server): + url = mock_openai_endpoint_server # e.g. "http://127.0.0.1:53892" + ... + +This avoids hitting the Railway-hosted deployment of +``BerriAI/example_openai_endpoint``, which has historically been a source of +flaky CI when Railway has incidents. +""" + +from __future__ import annotations + +import os +from pathlib import Path +from typing import Iterator + +import pytest + +from . import start_mock_server + + +@pytest.fixture(scope="session") +def mock_openai_endpoint_server( + tmp_path_factory: pytest.TempPathFactory, +) -> Iterator[str]: + """Session-scoped fixture that boots the vendored mock server. + + Yields the base URL (e.g. ``http://127.0.0.1:53892``). The server is + killed at session teardown. Server logs are written to + ``/mock_openai_endpoint.log`` for debugging. + """ + log_dir: Path = tmp_path_factory.mktemp("mock_openai_endpoint") + log_file = log_dir / "mock_openai_endpoint.log" + + handle = start_mock_server(log_file=log_file) + + # Expose the URL to any code that reads the env var (this is what the + # ``MOCK_OPENAI_BASE_URL`` helper checks). Restore the previous value on + # teardown so we don't leak it across test sessions. + previous = os.environ.get("LITELLM_MOCK_OPENAI_BASE_URL") + os.environ["LITELLM_MOCK_OPENAI_BASE_URL"] = handle.base_url + try: + yield handle.base_url + finally: + if previous is None: + os.environ.pop("LITELLM_MOCK_OPENAI_BASE_URL", None) + else: + os.environ["LITELLM_MOCK_OPENAI_BASE_URL"] = previous + handle.stop() diff --git a/tests/mock_endpoints/example_openai_endpoint/Dockerfile b/tests/mock_endpoints/example_openai_endpoint/Dockerfile new file mode 100644 index 00000000000..f5787f0da38 --- /dev/null +++ b/tests/mock_endpoints/example_openai_endpoint/Dockerfile @@ -0,0 +1,20 @@ +# Use the official Python image as the base image +FROM python:3.9-slim + +# Set the working directory in the container +WORKDIR /app + +# Copy the Python requirements file +COPY requirements.txt . + +# Install the Python dependencies +RUN pip install --no-cache-dir -r requirements.txt + +# Copy the application code +COPY . . + +# Expose the port the app will run on +EXPOSE 8090 + +# Start the application +CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8090"] \ No newline at end of file diff --git a/tests/mock_endpoints/example_openai_endpoint/batch_and_files_api.py b/tests/mock_endpoints/example_openai_endpoint/batch_and_files_api.py new file mode 100644 index 00000000000..ec4d74029cd --- /dev/null +++ b/tests/mock_endpoints/example_openai_endpoint/batch_and_files_api.py @@ -0,0 +1,310 @@ +from fastapi import APIRouter, File, Form, HTTPException, UploadFile +from pydantic import BaseModel +from typing import Dict, Literal, Optional +import uuid +import hashlib +from datetime import datetime + +router = APIRouter() + + +# ============= Models ============= + +class FileObject(BaseModel): + id: str + object: Literal["file"] = "file" + bytes: int + created_at: int + filename: str + purpose: str + status: str + + +class BatchRequestCounts(BaseModel): + total: int + completed: int + failed: int + + +class BatchObject(BaseModel): + id: str + object: Literal["batch"] = "batch" + endpoint: str + errors: Optional[Dict] = None + input_file_id: str + completion_window: str + status: Literal["validating", "failed", "in_progress", "finalizing", "completed", "expired", "cancelling", "cancelled"] + output_file_id: Optional[str] = None + error_file_id: Optional[str] = None + created_at: int + in_progress_at: Optional[int] = None + expires_at: Optional[int] = None + finalizing_at: Optional[int] = None + completed_at: Optional[int] = None + failed_at: Optional[int] = None + expired_at: Optional[int] = None + cancelling_at: Optional[int] = None + cancelled_at: Optional[int] = None + request_counts: Optional[BatchRequestCounts] = None + metadata: Optional[Dict[str, str]] = None + + +class CreateBatchRequest(BaseModel): + input_file_id: str + endpoint: Literal["/v1/chat/completions", "/v1/embeddings", "/v1/completions"] + completion_window: Literal["24h"] + metadata: Optional[Dict[str, str]] = None + status: str + + +# ============= Files Endpoints ============= + +@router.post("/files", response_model=FileObject) +async def create_file( + file: UploadFile = File(...), + purpose: str = Form(...) +): + """ + Upload a file that can be used for batch processing. + + Compatible with: https://platform.openai.com/docs/api-reference/files/create + """ + content = await file.read() + + # Generate consistent file ID based on filename and content + content_hash = hashlib.md5(f"{file.filename}{len(content)}".encode()).hexdigest()[:8] + file_id = f"file-{content_hash}" + + return FileObject( + id=file_id, + bytes=len(content), + created_at=int(datetime.now().timestamp()), + filename=file.filename or "uploaded_file", + purpose=purpose, + status="completed" + ) + + +@router.get("/files/{file_id}", response_model=FileObject) +async def retrieve_file(file_id: str): + """ + Returns information about a specific file. + + Compatible with: https://platform.openai.com/docs/api-reference/files/retrieve + """ + # Return stubbed file information + return FileObject( + id=file_id, + bytes=1024, # Stubbed file size + created_at=1698768000, # Stubbed timestamp + filename="example_file.jsonl", + purpose="batch", + status="completed" + ) + + +@router.get("/files/{file_id}/content") +async def retrieve_file_content(file_id: str): + """ + Returns the contents of the specified file. + + Compatible with: https://platform.openai.com/docs/api-reference/files/retrieve-contents + """ + # Return stubbed file content + stubbed_content = '{"custom_id": "request-1", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "gpt-3.5-turbo", "messages": [{"role": "user", "content": "Hello world"}]}}\n{"custom_id": "request-2", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "gpt-3.5-turbo", "messages": [{"role": "user", "content": "How are you?"}]}}' + return stubbed_content + + +@router.delete("/files/{file_id}") +async def delete_file(file_id: str): + """ + Delete a file. + + Compatible with: https://platform.openai.com/docs/api-reference/files/delete + """ + # Return stubbed deletion response + return {"id": file_id, "object": "file", "deleted": True} + + +# ============= Batches Endpoints ============= + +@router.post("/batches", response_model=BatchObject) +async def create_batch(request: CreateBatchRequest): + """ + Creates and executes a batch from an uploaded file of requests. + + Compatible with: https://platform.openai.com/docs/api-reference/batch/create + """ + # Generate consistent batch ID based on input file ID + batch_hash = hashlib.md5(request.input_file_id.encode()).hexdigest()[:8] + batch_id = f"batch_{batch_hash}" + created_at = int(datetime.now().timestamp()) + + # Return stubbed batch object + return BatchObject( + id=batch_id, + object="batch", + endpoint=request.endpoint, + errors=None, + input_file_id=request.input_file_id, + completion_window=request.completion_window, + status="completed", + output_file_id=f"file-output-{batch_hash}", + error_file_id=None, + created_at=created_at, + in_progress_at=created_at + 10, + expires_at=created_at + 86400, # 24 hours + finalizing_at=created_at + 300, + completed_at=created_at + 600, + failed_at=None, + expired_at=None, + cancelling_at=None, + cancelled_at=None, + request_counts=BatchRequestCounts(total=2, completed=2, failed=0), + metadata=request.metadata or {} + ) + + +@router.get("/batches/{batch_id}", response_model=BatchObject) +async def retrieve_batch(batch_id: str): + """ + Retrieves a batch. + + Compatible with: https://platform.openai.com/docs/api-reference/batch/retrieve + """ + # Extract hash from batch_id for consistent output file ID + batch_hash = batch_id.split("_")[-1] if "_" in batch_id else "stubbed" + created_at = 1698768000 # Stubbed timestamp + + # Return stubbed batch object + return BatchObject( + id=batch_id, + object="batch", + endpoint="/v1/chat/completions", + errors=None, + input_file_id=f"file-{batch_hash}", + completion_window="24h", + status="completed", + output_file_id=f"file-output-{batch_hash}", + error_file_id=None, + created_at=created_at, + in_progress_at=created_at + 10, + expires_at=created_at + 86400, + finalizing_at=created_at + 300, + completed_at=created_at + 600, + failed_at=None, + expired_at=None, + cancelling_at=None, + cancelled_at=None, + request_counts=BatchRequestCounts(total=2, completed=2, failed=0), + metadata={} + ) + + +@router.post("/batches/{batch_id}/cancel", response_model=BatchObject) +async def cancel_batch(batch_id: str): + """ + Cancels an in-progress batch. + + Compatible with: https://platform.openai.com/docs/api-reference/batch/cancel + """ + # Extract hash from batch_id for consistent output file ID + batch_hash = batch_id.split("_")[-1] if "_" in batch_id else "stubbed" + created_at = 1698768000 # Stubbed timestamp + cancelled_at = int(datetime.now().timestamp()) + + # Return stubbed cancelled batch object + return BatchObject( + id=batch_id, + object="batch", + endpoint="/v1/chat/completions", + errors=None, + input_file_id=f"file-{batch_hash}", + completion_window="24h", + status="cancelled", + output_file_id=None, + error_file_id=None, + created_at=created_at, + in_progress_at=created_at + 10, + expires_at=created_at + 86400, + finalizing_at=None, + completed_at=None, + failed_at=None, + expired_at=None, + cancelling_at=cancelled_at - 5, + cancelled_at=cancelled_at, + request_counts=BatchRequestCounts(total=2, completed=0, failed=0), + metadata={} + ) + + +@router.get("/batches") +async def list_batches(limit: int = 20, after: Optional[str] = None): + """ + List your organization's batches. + + Compatible with: https://platform.openai.com/docs/api-reference/batch/list + """ + # Return stubbed list of batches + stubbed_batches = [ + BatchObject( + id="batch_example1", + object="batch", + endpoint="/v1/chat/completions", + errors=None, + input_file_id="file-example1", + completion_window="24h", + status="completed", + output_file_id="file-output-example1", + error_file_id=None, + created_at=1698768000, + in_progress_at=1698768010, + expires_at=1698854400, + finalizing_at=1698768300, + completed_at=1698768600, + failed_at=None, + expired_at=None, + cancelling_at=None, + cancelled_at=None, + request_counts=BatchRequestCounts(total=5, completed=5, failed=0), + metadata={} + ), + BatchObject( + id="batch_example2", + object="batch", + endpoint="/v1/embeddings", + errors=None, + input_file_id="file-example2", + completion_window="24h", + status="in_progress", + output_file_id=None, + error_file_id=None, + created_at=1698767000, + in_progress_at=1698767010, + expires_at=1698853400, + finalizing_at=None, + completed_at=None, + failed_at=None, + expired_at=None, + cancelling_at=None, + cancelled_at=None, + request_counts=BatchRequestCounts(total=3, completed=1, failed=0), + metadata={"project": "test"} + ) + ] + + # Apply pagination logic to stubbed data + if after: + try: + start_idx = next(i for i, b in enumerate(stubbed_batches) if b.id == after) + 1 + stubbed_batches = stubbed_batches[start_idx:] + except StopIteration: + pass + + stubbed_batches = stubbed_batches[:limit] + + return { + "object": "list", + "data": stubbed_batches, + "has_more": False # Stubbed - no more data + } diff --git a/tests/mock_endpoints/example_openai_endpoint/main.py b/tests/mock_endpoints/example_openai_endpoint/main.py new file mode 100644 index 00000000000..b09ab0f3ae2 --- /dev/null +++ b/tests/mock_endpoints/example_openai_endpoint/main.py @@ -0,0 +1,2564 @@ +from fastapi import FastAPI, Request, status, HTTPException, Depends, Header, WebSocket, WebSocketDisconnect, UploadFile, Form +from fastapi.responses import StreamingResponse, Response, PlainTextResponse, JSONResponse +from fastapi.security import OAuth2PasswordBearer +from fastapi.middleware.cors import CORSMiddleware +import asyncio +import json +import uuid +import asyncio +import os +import time +import random +import warnings +import logging +from dotenv import load_dotenv +from slowapi import Limiter +from collections import deque +from datetime import datetime, timedelta +from typing import List, Dict, Any, Callable, Optional +from pydantic import BaseModel + +from batch_and_files_api import router as batch_files_router + +from slowapi import Limiter, _rate_limit_exceeded_handler + + +def _normalize_model_for_response(model: str) -> str: + """Normalize deployment model IDs (e.g. gpt-4o-mini-data-zone) to client-facing names (gpt-4o-mini) + so LiteLLM does not log 'response model mismatch' when backend returns a deployment variant.""" + if not model or not isinstance(model, str): + return model or "" + # Strip known deployment suffixes so response matches what the client requested + for suffix in ("-data-zone", "-eu", "-us", "-preview"): + if model.endswith(suffix): + return model[: -len(suffix)] + return model +from slowapi.util import get_remote_address +from slowapi.errors import RateLimitExceeded + + +# Preloaded audio in three sizes: small, medium, large +# Structure: _PRELOADED_AUDIO[format][size] = bytes +_PRELOADED_AUDIO: Dict[str, Dict[str, bytes]] = {} + +# Size definitions in bytes +AUDIO_SIZES = { + "small": 20_000, # ~20 KB - for short inputs (< 100 chars) + "medium": 100_000, # ~100 KB - for medium inputs (100-1000 chars) + "large": 400_000, # ~400 KB - for large inputs (> 1000 chars) +} + + +def generate_minimal_audio(format: str = "mp3") -> bytes: + """Generate minimal valid audio data for different formats.""" + if format == "mp3": + # Minimal valid MP3 frame (silent audio, ~1 second) + # This is a minimal MP3 header + frame + mp3_header = bytes([ + 0xFF, 0xFB, 0x90, 0x00, # MP3 sync word + header + ]) + # Add some minimal frame data using a non-zero pattern + frame_data = bytes([0x55] * 100) # 0x55 pattern instead of all zeros + return mp3_header + frame_data + elif format == "opus": + # Minimal Opus header (OggS) + opus_header = b"OggS\x00\x02\x00\x00\x00\x00\x00\x00\x00\x00" + # Use a simple non-zero pattern for payload + return opus_header + bytes([0x33] * 50) + elif format == "aac": + # Minimal AAC header + aac_header = bytes([0xFF, 0xF1]) # ADTS sync word + # Non-zero pattern payload + return aac_header + bytes([0x77] * 50) + elif format == "flac": + # Minimal FLAC header (fLaC) + flac_header = b"fLaC" + # Non-zero pattern payload + return flac_header + bytes([0x99] * 50) + elif format == "pcm": + # Minimal PCM WAV header + # WAV header structure + wav_header = ( + b"RIFF" + # ChunkID + (36).to_bytes(4, byteorder="little") + # ChunkSize + b"WAVE" + # Format + b"fmt " + # Subchunk1ID + (16).to_bytes(4, byteorder="little") + # Subchunk1Size + (1).to_bytes(2, byteorder="little") + # AudioFormat (PCM) + (1).to_bytes(2, byteorder="little") + # NumChannels + (16000).to_bytes(4, byteorder="little") + # SampleRate + (32000).to_bytes(4, byteorder="little") + # ByteRate + (2).to_bytes(2, byteorder="little") + # BlockAlign + (16).to_bytes(2, byteorder="little") + # BitsPerSample + b"data" + # Subchunk2ID + (0).to_bytes(4, byteorder="little") # Subchunk2Size + ) + return wav_header + else: + # Default to MP3 + return generate_minimal_audio("mp3") + + +def generate_audio_by_size(format: str, target_size: int, speed: float = 1.0) -> bytes: + """ + Generate audio data of approximately the target size. + + Args: + format: Audio format (mp3, opus, aac, flac, pcm) + target_size: Target size in bytes + speed: Speech speed (affects duration, so faster = smaller for same text) + + Returns: + bytes: Audio data approximately matching target_size + """ + # Adjust target size based on speed (faster speed = shorter duration = smaller file) + # Speed affects duration linearly, so we divide by speed + adjusted_size = int(target_size / speed) + + # Get the minimal header/base for this format + base = generate_minimal_audio(format) + base_size = len(base) + + # If target is smaller than base, return base (minimum valid audio) + if adjusted_size <= base_size: + return base + + # Calculate how much payload we need + payload_size = adjusted_size - base_size + + # Generate payload data with a pattern that looks like audio + # Use a repeating pattern that varies to avoid compression artifacts + pattern = [0x55, 0xAA, 0x33, 0xCC, 0x66, 0x99, 0x11, 0xEE] + payload = bytes([pattern[i % len(pattern)] for i in range(payload_size)]) + + return base + payload + + +def _load_preloaded_audio() -> None: + """ + Preload three sizes (small, medium, large) of audio for each format. + This is done once at startup to avoid per-request generation overhead. + """ + global _PRELOADED_AUDIO + + # Formats we support for the speech endpoint + formats = ["mp3", "opus", "aac", "flac", "pcm"] + + for fmt in formats: + _PRELOADED_AUDIO[fmt] = {} + for size_name, target_size in AUDIO_SIZES.items(): + # Generate audio of the target size + audio_data = generate_audio_by_size(fmt, target_size, speed=1.0) + _PRELOADED_AUDIO[fmt][size_name] = audio_data + + +# Initialize preloaded audio at import time +_load_preloaded_audio() + + +def get_histogram_sleep_time() -> float: + """ + Returns a sleep time based on a histogram distribution: + - Most requests (~70%): ~10 seconds (8-12s range) + - Some requests (~25%): ~60 seconds (55-65s range) + - Very few (~5%): 60+ seconds (60-300s range) + + This simulates realistic degraded provider behavior where most requests + are slow but not terrible, some are very slow, and a few are extremely slow. + """ + rand = random.random() + + if rand < 0.70: # 70% of requests: ~10 seconds + # Normal distribution around 10 seconds, std dev of 2 + sleep_time = max(1.0, random.gauss(10, 2)) + return sleep_time + elif rand < 0.95: # 25% of requests: ~60 seconds + # Normal distribution around 60 seconds, std dev of 5 + sleep_time = max(30.0, random.gauss(60, 5)) + return sleep_time + else: # 5% of requests: 60+ seconds (up to 5 minutes) + # Uniform distribution between 60 and 300 seconds + sleep_time = random.uniform(60, 300) + return sleep_time + + +def get_request_url(request: Request): + return str(request.url) + + +def format_detailed_error( + error: Exception, + request: Request, + context: str = "", + include_body: bool = False +) -> dict: + """Format a detailed error response with request context""" + error_detail = { + "error": { + "message": str(error), + "type": type(error).__name__, + "context": context, + "request": { + "path": request.url.path, + "method": request.method, + "query": str(request.url.query) if request.url.query else None, + } + } + } + + # Clearer message for KeyError (e.g. 'content' or 'candidates') to aid debugging format mismatches + if isinstance(error, KeyError): + key = str(error).strip("'\"") + error_detail["error"]["message"] = ( + f"Missing required key '{key}' in response. " + "Claude/Anthropic models expect a top-level 'content' array; Gemini models expect 'candidates'. " + f"This may indicate a response-format mismatch. Original: {error}." + ) + + # Add sanitized headers (mask sensitive values) + headers_dict = dict(request.headers) + sanitized_headers = {} + for key, value in headers_dict.items(): + if key.lower() in ["authorization", "api-key", "x-api-key", "cookie"]: + # Show only first 20 chars of sensitive headers + sanitized_headers[key] = f"{value[:20]}..." if len(value) > 20 else value + else: + sanitized_headers[key] = value + error_detail["error"]["request"]["headers"] = sanitized_headers + + # Optionally include request body preview + if include_body: + try: + # Try to get body without consuming it (if already read) + # This might not work if body was already consumed, but worth trying + pass # Body reading is async, so we'll do it in the caller if needed + except: + pass + + return error_detail + + +async def get_error_detail_with_body( + error: Exception, + request: Request, + context: str = "" +) -> dict: + """Get detailed error response including request body if possible""" + error_detail = format_detailed_error(error, request, context, include_body=False) + + # Try to get request body + try: + body = await request.body() + if body: + body_str = body.decode('utf-8', errors='ignore') + # Limit body size in error response (first 500 chars) + error_detail["error"]["request"]["body_preview"] = body_str[:500] + if len(body_str) > 500: + error_detail["error"]["request"]["body_preview"] += "... (truncated)" + except Exception: + error_detail["error"]["request"]["body_preview"] = "Unable to read request body" + + return error_detail + + +def _validate_response_format(response: dict, model: Optional[str], is_gemini: bool) -> Optional[dict]: + """ + Validate that the response matches the expected format for the model. + Returns an error dict for HTTP 500 if invalid, None if ok. + Logs clearly on the server when a format mismatch or structural bug is detected. + """ + model_lower = (model or "").lower() + has_candidates = "candidates" in response + has_content = "content" in response and isinstance(response.get("content"), list) + + # Misclassification: Claude model but we built Gemini format (causes KeyError: 'content' in LiteLLM) + if "claude" in model_lower and has_candidates and not has_content: + log.error( + "Format misclassification: Claude model '%s' received Gemini format (candidates, no top-level content). " + "Response keys: %s. This will cause KeyError: 'content' in LiteLLM Anthropic transformation.", + model, list(response.keys()) + ) + return { + "error": { + "message": ( + f"Response format mismatch: Model '{model}' is a Claude/Anthropic model but the server produced " + "a Gemini-format response (with 'candidates' instead of 'content'). " + "The LiteLLM/Anthropic client expects a top-level 'content' array. " + "This is a format-detection bug on the server. Please report." + ), + "code": "FORMAT_MISMATCH_CLAUDE_GOT_GEMINI", + } + } + # Misclassification: Gemini model but we built Anthropic format + if "gemini" in model_lower and has_content and not has_candidates: + log.error( + "Format misclassification: Gemini model '%s' received Anthropic format (content, no candidates). " + "Response keys: %s.", + model, list(response.keys()) + ) + return { + "error": { + "message": ( + f"Response format mismatch: Model '{model}' is a Gemini model but the server produced " + "an Anthropic-format response (with 'content' instead of 'candidates'). " + "The Vertex/Gemini client expects 'candidates'. This is a format-detection bug on the server. Please report." + ), + "code": "FORMAT_MISMATCH_GEMINI_GOT_ANTHROPIC", + } + } + + # Structural validation: required top-level key must exist + if is_gemini: + if not has_candidates: + log.error("Invalid Gemini response: missing 'candidates'. Model: %s, response keys: %s", model, list(response.keys())) + return { + "error": { + "message": f"Response format error: Gemini format requires a top-level 'candidates' array. Model: {model}. This is a server-side bug.", + "code": "INVALID_GEMINI_RESPONSE", + } + } + else: + if not has_content: + log.error("Invalid Anthropic response: missing or invalid 'content'. Model: %s, response keys: %s", model, list(response.keys())) + return { + "error": { + "message": f"Response format error: Anthropic/Claude format requires a top-level 'content' array. Model: {model}. This is a server-side bug.", + "code": "INVALID_ANTHROPIC_RESPONSE", + } + } + return None + + +limiter = Limiter(key_func=get_request_url) +load_dotenv() + +# Suppress python-multipart's "Skipping data after last boundary" warning +# This warning appears even in patched versions (0.0.18+) where the DoS vulnerability is fixed. +# The warning is harmless but clutters logs. We suppress it via logging configuration. +# Note: Upgraded to python-multipart 0.0.20 which has better handling, but warning may still appear. +logging.getLogger("multipart").setLevel(logging.ERROR) # Only show errors, not warnings + +log = logging.getLogger(__name__) + +app = FastAPI() +app.state.limiter = limiter +app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) + + +@app.exception_handler(Exception) +async def global_exception_handler(request: Request, exc: Exception): + """Return a clear, non-leaking error to the client; log the full traceback on the server.""" + if isinstance(exc, HTTPException): + raise exc + log.exception("Unhandled exception: %s", exc) + return JSONResponse( + status_code=500, + content={ + "error": { + "message": "Internal server error. The request could not be processed. Check server logs for details.", + "code": "INTERNAL_ERROR", + } + }, + ) + + +app.add_middleware( + CORSMiddleware, + allow_origins=["*"], + allow_credentials=True, + allow_methods=["*"], + allow_headers=["*"], +) + +# Include the batch and files router with /v1 prefix +app.include_router(batch_files_router, prefix="/v1") + + +def data_generator(model=None): + response_id = uuid.uuid4().hex + sentence = "Hello this is a test response from a fixed OpenAI endpoint." + words = sentence.split(" ") + _model = model if isinstance(model, str) else "gpt-3.5-turbo-0125" + for word in words: + word = word + " " + chunk = { + "id": f"chatcmpl-{response_id}", + "object": "chat.completion.chunk", + "created": 1677652288, + "model": _model, + "choices": [{"index": 0, "delta": {"content": word}}], + } + try: + yield f"data: {json.dumps(chunk.dict())}\n\n" + except: + yield f"data: {json.dumps(chunk)}\n\n" + + +# for completion +@app.post("/chat/completions") +@app.post("/v1/chat/completions") +@app.post("/openai/deployments/{model:path}/chat/completions") # azure compatible endpoint +async def completion(request: Request, authorization: str = Header(None)): + # Accept both Authorization: Bearer and x-goog-api-key (optional for OpenAI-compatible endpoints) + # This allows Gemini models routed through /chat/completions to work + has_valid_auth(request, authorization) + + _time_to_sleep = os.getenv("TIME_TO_SLEEP", None) + if _time_to_sleep is not None: + print("sleeping for " + _time_to_sleep) + await asyncio.sleep(float(_time_to_sleep)) + + data = await request.json() + data = data if isinstance(data, dict) else {} + + if data.get("model") == "429": + raise HTTPException(status_code=status.HTTP_429_TOO_MANY_REQUESTS, detail="Too many requests") + + if data.get("model") == "random_sleep": + # sleep for a random time between 1 and 10 seconds + sleep_time = random.randint(1, 10) + print("sleeping for " + str(sleep_time) + " seconds") + await asyncio.sleep(sleep_time) + + # Degraded provider simulation: sleep with histogram distribution to simulate hanging requests + # This causes event loop blocking, file descriptor accumulation, and tests LiteLLM's + # behavior under degraded provider conditions + if data.get("model") in ["degraded", "slow_provider", "blocked"]: + # Histogram distribution: most ~10s, some ~60s, few 60s+ + sleep_time = get_histogram_sleep_time() + print(f"[DEGRADED MODE] Sleeping for {sleep_time:.1f} seconds ({sleep_time/60:.1f} minutes) - histogram distribution") + await asyncio.sleep(sleep_time) + # For /v1/chat/completions, LiteLLM's Vertex AI Anthropic expects a body with "content". + # data_generator() yields OpenAI format (choices/delta); returning it causes KeyError: 'content'. + # So when path is /v1/chat/completions, do NOT stream—fall through to return Anthropic JSON. + _path_for_claude = request.url.path + if data.get("stream") == True and "/v1/chat/completions" not in _path_for_claude: + _stream_model = ( + data.get("litellm_model_id") or data.get("model") or data.get("model_name") + or data.get("modelId") or data.get("model_id") + or (getattr(request, "path_params", None) or {}).get("model") + or "gpt-3.5-turbo-0125" + ) + _stream_model = _normalize_model_for_response( + _stream_model if isinstance(_stream_model, str) else "gpt-3.5-turbo-0125" + ) + return StreamingResponse( + content=data_generator(model=_stream_model), + media_type="text/event-stream", + ) + # Model: body (model, model_name, modelId, model_id, litellm_model_id), path (Azure), query, + # generationConfig, metadata, or headers (X-Model, X-LiteLLM-Model). LiteLLM may put model + # in different places when calling /v1/chat/completions for Vertex AI Anthropic. + _path_params = getattr(request, "path_params", None) or {} + _model = ( + data.get("model") + or data.get("model_name") + or data.get("modelId") + or data.get("model_id") + or data.get("litellm_model_id") + or _path_params.get("model") + or (getattr(request, "query_params", None) or {}).get("model") + or (data.get("generationConfig") or {}).get("model") + or (data.get("metadata") or {}).get("model") + or (data.get("metadata") or {}).get("model_name") + or request.headers.get("X-Model") + or request.headers.get("x-model") + or request.headers.get("X-LiteLLM-Model") + or request.headers.get("x-litellm-model") + or "" + ) + # Coerce to str so .lower() is safe; non-string (e.g. dict) treated as no model. + _model = _model if isinstance(_model, str) else "" + # Vertex-style body ("contents") without "model" or "messages": LiteLLM may omit model; + # assume Anthropic so we return "content" and avoid KeyError in anthropic transformation. + if not _model and "contents" in data and "messages" not in data: + _model = "claude" + # LiteLLM routes Vertex AI Anthropic (Claude) to /v1/chat/completions and may omit model + # in the body. If we still have no model and path is exactly /v1/chat/completions, + # assume Claude so we return Anthropic format (content) and avoid KeyError. + if not _model and "/v1/chat/completions" in _path_for_claude: + _model = "claude" + _model = _model or "" + _model_lower = _model.lower() + + # When LiteLLM routes Claude (Vertex AI Anthropic) to /v1/chat/completions, it + # expects Anthropic Messages API format (top-level "content" array). OpenAI format + # (choices/usage) causes KeyError: 'content' in litellm anthropic transformation. + # Also: path /v1/chat/completions is used by LiteLLM for Vertex AI Anthropic; the + # model in the body may be omitted or not contain "claude", so treat that path as Claude. + is_claude = (_model and "claude" in _model_lower) or "/v1/chat/completions" in _path_for_claude + if is_claude: + _disp = _normalize_model_for_response(_model or "claude") + resp = { + "id": f"msg_{uuid.uuid4().hex}", + "type": "message", + "role": "assistant", + "content": [ + {"type": "text", "text": "\n\nHello there, how may I assist you today?"} + ], + "model": _disp, + "stop_reason": "end_turn", + "stop_sequence": None, + "usage": {"input_tokens": 10, "output_tokens": 20, "total_tokens": 30}, + } + print(f"[fmt] completion path={request.url.path} body_keys={list(data.keys())} model={_disp} branch=claude_anthropic response_keys={list(resp.keys())}") + return resp + + if _model == "gpt-5": + _model = "gpt-12" + elif not _model: + _model = "gpt-3.5-turbo-0301" # fallback when no model in request + else: + _model = _normalize_model_for_response(_model) # e.g. gpt-4o-mini-data-zone -> gpt-4o-mini + response_id = uuid.uuid4().hex + response = { + "id": f"chatcmpl-{response_id}", + "object": "chat.completion", + "created": 1677652288, + "model": _model, + "system_fingerprint": "fp_44709d6fcb", + "choices": [ + { + "index": 0, + "message": { + "role": "assistant", + "content": "\n\nHello there, how may I assist you today?", + }, + "logprobs": None, + "finish_reason": "stop", + } + ], + "usage": {"prompt_tokens": 9, "completion_tokens": 12, "total_tokens": 21}, + } + print(f"[fmt] completion path={request.url.path} body_keys={list(data.keys())} model={_model} branch=openai response_keys={list(response.keys())}") + return response + + +# for completion +@app.post("/completions") +@app.post("/v1/completions") +async def text_completion(request: Request): + data = await request.json() + _model = ( + data.get("litellm_model_id") or data.get("model") or data.get("model_name") + or data.get("modelId") or data.get("model_id") or "unknown" + ) + _model = _normalize_model_for_response(_model if isinstance(_model, str) else "unknown") + + if data.get("stream") == True: + return StreamingResponse( + content=data_generator(model=_model), + media_type="text/event-stream", + ) + else: + response_id = uuid.uuid4().hex + response = { + "id": "cmpl-9B2ycsf0odECdLmrVzm2y8Q12csjW", + "choices": [ + { + "finish_reason": "length", + "index": 0, + "logprobs": None, + "text": "\n\nA test request, how intriguing\nAn invitation for knowledge bringing\nWith words" + } + ], + "created": 1712420078, + "model": _model, + "object": "text_completion", + "system_fingerprint": None, + "usage": { + "completion_tokens": 16, + "prompt_tokens": 10, + "total_tokens": 26 + } + } + + return response + + + + +# for completion +@app.post("/invocations") +@app.post("/invocations/") +async def invocation(request: Request): + _time_to_sleep = os.getenv("TIME_TO_SLEEP", None) + if _time_to_sleep is not None: + print("sleeping for " + _time_to_sleep) + await asyncio.sleep(float(_time_to_sleep)) + data = await request.json() + if data.get("model") == "429": + raise HTTPException(status_code=status.HTTP_429_TOO_MANY_REQUESTS, detail="Too many requests") + else: + response_id = uuid.uuid4().hex + return { + "generated_text": "This is a mock response from SageMaker.", + "id": "cmpl-mockid", + "object": "text_completion", + "created": 1629800000, + "model": "sagemaker/jumpstart-dft-hf-textgeneration1-mp-20240815-185614", + "choices": [ + { + "text": "This is a mock response from SageMaker.", + "index": 0, + "logprobs": None, + "finish_reason": "length", + } + ], + "usage": {"prompt_tokens": 1, "completion_tokens": 8, "total_tokens": 9}, + } + +@app.post("/embeddings") +@app.post("/v1/embeddings") +@app.post("/openai/deployments/{model:path}/embeddings") # azure compatible endpoint +async def embeddings(request: Request, authorization: str = Header(None)): + # Accept both Authorization: Bearer and x-goog-api-key (optional for OpenAI-compatible endpoints) + # This allows Gemini models routed through /embeddings to work + has_valid_auth(request, authorization) + + _small_embedding = [ + -0.006929283495992422, + -0.005336422007530928, + -4.547132266452536e-05, + -0.024047505110502243, + ] + + big_embedding = _small_embedding * 100 + return { + "object": "list", + "data": [ + { + "object": "embedding", + "index": 0, + "embedding": big_embedding + } + ], + "model": "text-embedding-3-small", + "usage": { + "prompt_tokens": 5, + "total_tokens": 5 + } + } + + +@app.post("/audio/speech") +@app.post("/v1/audio/speech") +async def audio_speech(request: Request): + """OpenAI Audio Speech endpoint - Text to Speech""" + _time_to_sleep = os.getenv("TIME_TO_SLEEP", None) + if _time_to_sleep is not None: + print("sleeping for " + _time_to_sleep) + await asyncio.sleep(float(_time_to_sleep)) + + data = await request.json() + + # Extract parameters + model = data.get("model", "tts-1") + input_text = data.get("input", "") + voice = data.get("voice", "alloy") + response_format = data.get("response_format", "mp3") + speed = data.get("speed", 1.0) + + # Validate required parameters + if not input_text: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="Missing required parameter: input" + ) + + # Validate voice + valid_voices = ["alloy", "echo", "fable", "onyx", "nova", "shimmer"] + if voice not in valid_voices: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"Invalid voice. Must be one of: {', '.join(valid_voices)}" + ) + + # Validate response_format + valid_formats = ["mp3", "opus", "aac", "flac", "pcm"] + if response_format not in valid_formats: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"Invalid response_format. Must be one of: {', '.join(valid_formats)}" + ) + + # Validate speed + if not (0.25 <= speed <= 4.0): + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="speed must be between 0.25 and 4.0" + ) + + # Select audio size based on input text length + # This allows testing different response sizes without dynamic generation overhead + text_length = len(input_text) + + if text_length < 100: + size_key = "small" + elif text_length < 1000: + size_key = "medium" + else: + size_key = "large" + + # Get preloaded audio data for the selected size + audio_data = _PRELOADED_AUDIO.get(response_format, {}).get(size_key) + if audio_data is None: + # Fallback to on-demand minimal generation if something is missing + audio_data = generate_minimal_audio(response_format) + + # Adjust for speed parameter (faster speed = smaller file) + # For simplicity, we'll just use the preloaded data as-is + # In a real implementation, speed would affect the audio duration + # Optional: Log the size for testing/debugging (can be enabled via env var) + if os.getenv("LOG_AUDIO_SIZES", "").lower() == "true": + print(f"[Audio Speech] Input length: {text_length} chars, Format: {response_format}, " + f"Size: {size_key}, Actual size: {len(audio_data)} bytes, Speed: {speed}") + + # Set appropriate content type + content_types = { + "mp3": "audio/mpeg", + "opus": "audio/ogg", + "aac": "audio/aac", + "flac": "audio/flac", + "pcm": "audio/wav" + } + content_type = content_types.get(response_format, "audio/mpeg") + + # Return binary audio response with preloaded size based on input text length + return Response( + content=audio_data, + media_type=content_type, + headers={ + "Content-Disposition": f'attachment; filename="speech.{response_format}"' + }, + ) + + +# Pre-computed responses for maximum speed +_TRANSCRIPTION_TEXT = "This is a mock transcription of the audio file. The audio has been processed and transcribed to text." +_TRANSCRIPTION_JSON = {"text": _TRANSCRIPTION_TEXT} +_TRANSCRIPTION_VERBOSE_JSON = { + "text": _TRANSCRIPTION_TEXT, + "language": "en", + "duration": 5.0, + "segments": [{"id": 0, "start": 0.0, "end": 5.0, "text": _TRANSCRIPTION_TEXT}] +} +_TRANSCRIPTION_SRT = f"1\n00:00:00,000 --> 00:00:05,000\n{_TRANSCRIPTION_TEXT}\n\n" +_TRANSCRIPTION_VTT = f"WEBVTT\n\n00:00:00.000 --> 00:00:05.000\n{_TRANSCRIPTION_TEXT}\n" + +@app.post("/audio/transcriptions") +@app.post("/v1/audio/transcriptions") +async def audio_transcriptions( + file: UploadFile, + model: str = Form(...), + language: Optional[str] = Form(None), + prompt: Optional[str] = Form(None), + temperature: Optional[float] = Form(0.0), + response_format: Optional[str] = Form("json") +): + """OpenAI Audio Transcriptions endpoint - Speech to Text""" + _time_to_sleep = os.getenv("TIME_TO_SLEEP", None) + if _time_to_sleep is not None: + print("sleeping for " + _time_to_sleep) + await asyncio.sleep(float(_time_to_sleep)) + + # Minimal validation - only check critical errors + if model != "whisper-1": + raise HTTPException(status_code=400, detail="Invalid model. Must be one of: whisper-1") + + # Fast path: return pre-computed responses based on format + # Use simple dict returns (FastAPI auto-encodes to JSON) for maximum speed + if response_format == "text": + return PlainTextResponse(content=_TRANSCRIPTION_TEXT) + elif response_format == "srt": + return PlainTextResponse(content=_TRANSCRIPTION_SRT, media_type="text/plain") + elif response_format == "vtt": + return PlainTextResponse(content=_TRANSCRIPTION_VTT, media_type="text/vtt") + elif response_format == "verbose_json": + # Only modify language if different from default + if language and language != "en": + return {"text": _TRANSCRIPTION_TEXT, "language": language, "duration": 5.0, "segments": _TRANSCRIPTION_VERBOSE_JSON["segments"]} + return _TRANSCRIPTION_VERBOSE_JSON + else: # json or default + return _TRANSCRIPTION_JSON + + + + +@app.post("/triton/embeddings") +async def embeddings(request: Request): + try: + input_data = await request.json() + assert "inputs" in input_data + + inputs = input_data["inputs"] + element_one = inputs[0] + + assert "name" in element_one, "Missing name in inputs" + assert "shape" in element_one, "Missing shape in inputs" + assert "datatype" in element_one, "Missing datatype in inputs" + assert "data" in element_one, "Missing data in inputs" + + + except (ValueError, KeyError) as e: + return HTTPException(status_code=400, detail=str(e)) + + output_data = { + "model_name": "triton-embeddings", + "model_version": "1", + "parameters": { + "sequence_id": 0, + "sequence_start": False, + "sequence_end": False + }, + "outputs": [ + { + "name": "embedding_output", + "datatype": "FP32", + "shape": [2, 2], + "data": [0.1, 0.2] # Replace with actual output data + } + ] + } + + return output_data + + +@app.post("/openai/fine_tuning/jobs") # azure compatible endpoint +async def fine_tuning(request: Request): + _time_to_sleep = os.getenv("TIME_TO_SLEEP", None) + + print("inside fine tuning /jobs endpoint") + if _time_to_sleep is not None: + print("sleeping for " + _time_to_sleep) + await asyncio.sleep(float(_time_to_sleep)) + + data = await request.json() + + if data.get("model") == "429": + raise HTTPException(status_code=status.HTTP_429_TOO_MANY_REQUESTS, detail="Too many requests") + + print("got request=" + json.dumps(data)) + + return { + "object": "fine_tuning.job", + "id": "ftjob-abc123", + "model": "davinci-002", + "created_at": 1692661014, + "finished_at": 1692661190, + "fine_tuned_model": "ft:davinci-002:my-org:custom_suffix:7q8mpxmy", + "organization_id": "org-123", + "result_files": [ + "file-abc123" + ], + "status": "succeeded", + "validation_file": None, + "training_file": "file-abc123", + "hyperparameters": { + "n_epochs": 4, + "batch_size": 1, + "learning_rate_multiplier": 1.0 + }, + "trained_tokens": 5768, + "integrations": [], + "seed": 0, + "estimated_finish": 0 + } + + +@app.get("/openai/fine_tuning/jobs") # azure compatible endpoint +async def list_fine_tuning(request: Request): + _time_to_sleep = os.getenv("TIME_TO_SLEEP", None) + + return { + "object": "list", + "data": [ + { + "object": "fine_tuning.job.event", + "id": "ft-event-TjX0lMfOniCZX64t9PUQT5hn", + "created_at": 1689813489, + "level": "warn", + "message": "Fine tuning process stopping due to job cancellation", + "data": None, + "type": "message" + }, + ], "has_more": True + } + + + +@app.post("/openai/fine_tuning/jobs/{fine_tuning_job_id:path}/cancel") # azure compatible endpoint +async def cancel_fine_tuning(request: Request): + _time_to_sleep = os.getenv("TIME_TO_SLEEP", None) + + return { + "object": "fine_tuning.job", + "id": "ftjob-abc123", + "model": "gpt-4o-mini-2024-07-18", + "created_at": 1721764800, + "fine_tuned_model": None, + "organization_id": "org-123", + "result_files": [], + "hyperparameters": { + "n_epochs": "auto" + }, + "status": "cancelled", + "validation_file": "file-abc123", + "training_file": "file-abc123" + } + + + + + +@app.post("/openai/files") # azure compatible endpoint +async def openai_files(request: Request): + _time_to_sleep = os.getenv("TIME_TO_SLEEP", None) + + print("inside fine tuning /jobs endpoint") + if _time_to_sleep is not None: + print("sleeping for " + _time_to_sleep) + await asyncio.sleep(float(_time_to_sleep)) + + + return { + "id": "file-abc123", + "object": "file", + "bytes": 120000, + "created_at": 1677610602, + "filename": "mydata.jsonl", + "purpose": "fine-tune", + } + + +### FAKE BEDROCK ENDPOINT ### + +@app.post("/model/{modelId}/converse") +async def fake_bedrock_endpoint(request: Request): + return {"metrics":{"latencyMs":393},"output":{"message":{"content":[{"text":"Good morning to you too! I am not Claude, however. Claude is a large language model trained by Google, while I am Gemini, a multi-modal AI model, developed by Google as well. Is there anything I can help you with today?"}],"role":"assistant"}},"stopReason":"end_turn","usage":{"inputTokens":37,"outputTokens":8,"totalTokens":45}} + +@app.post("/model/{modelId}/invoke") +async def fake_bedrock_invoke(request: Request, modelId: str): + """Bedrock invoke endpoint for text generation and embeddings""" + data = await request.json() + modelId_lower = modelId.lower() + + print(f"[bedrock_invoke] modelId={modelId}, modelId_lower={modelId_lower}") + + # Check for Mistral models FIRST (before embedding check) + # Mistral model IDs: mistral.mistral-7b-instruct-v0:2, mistral.mixtral-8x7b-instruct-v0:1 + if "mistral" in modelId_lower: + print(f"[bedrock_invoke] Detected Mistral model, returning outputs format") + # Mistral models expect "outputs" array with "text" and "stop_reason" + # Reference: litellm/llms/bedrock/chat/invoke_handler.py line 657-661 + return { + "outputs": [ + { + "text": "This is a mock response from Bedrock Mistral model.", + "stop_reason": "stop" + } + ] + } + # Check if this is an embedding model + elif "embed" in modelId_lower or "titan-embed" in modelId_lower: + print(f"[bedrock_invoke] Detected embedding model") + # Return embedding format + input_text = data.get("inputText", data.get("text", "")) + if isinstance(input_text, list): + input_text = " ".join(str(t) for t in input_text) + + # Generate fake embedding (768 dimensions for most models, 1024 for Cohere v3) + embedding_dim = 1024 if "cohere.embed" in modelId_lower else 768 + embedding = [random.uniform(-0.15, 0.15) for _ in range(embedding_dim)] + + # Cohere models expect "embeddings" (plural) in an array + # Model IDs: cohere.embed-english-v3, cohere.embed-multilingual-v3 + if "cohere.embed" in modelId_lower: + print(f"[bedrock_invoke] Detected Cohere embedding model, returning embeddings array") + return { + "embeddings": [embedding], + "id": f"embed_{uuid.uuid4().hex}", + "response_type": "embeddings_floats" + } + else: + # Titan and other embedding models expect "embedding" (singular) + print(f"[bedrock_invoke] Detected Titan/other embedding model, returning embedding format") + return { + "embedding": embedding, + "inputTextTokenCount": len(input_text.split()) if input_text else 0 + } + else: + # Return text generation format for other models (Amazon Titan text, etc.) + print(f"[bedrock_invoke] Default to Amazon Titan text format with results") + return { + "results": [ + { + "outputText": "This is a mock response from Bedrock.", + "completionReason": "FINISH", + "tokenCount": { + "inputTokens": 10, + "outputTokens": 8, + "totalTokens": 18 + } + } + ] + } + +### FAKE VERTEX ENDPOINT ### + +def validate_google_auth(request: Request, authorization: str = None, required: bool = True): + """Validate authentication for Google endpoints - accepts both Bearer token and x-goog-api-key""" + # Check Authorization header (Vertex AI format) + if authorization and authorization.startswith("Bearer "): + return True + + # Check x-goog-api-key header (Gemini API format) + api_key = request.headers.get("x-goog-api-key") or request.headers.get("X-Goog-Api-Key") + if api_key: + return True + + # No valid auth found + if required: + raise HTTPException(status_code=401, detail="Invalid or missing Authorization header. Use 'Authorization: Bearer ' or 'x-goog-api-key: '") + return False + +def has_valid_auth(request: Request, authorization: str = None) -> bool: + """Check if request has valid authentication (either format) - returns True/False without raising""" + try: + return validate_google_auth(request, authorization, required=False) + except HTTPException: + return False + +@app.post("/generateContent") +@app.post("/v1/projects/adroit-crow-413218/locations/us-central1/publishers/google/models/gemini-1.0-pro-vision-001:generateContent") +@app.post("/v1/projects/pathrise-convert-1606954137718/locations/us-central1/publishers/google/models/gemini-1.0-pro-vision-001:generateContent") +@app.post("/v1beta/models/gemini-1.5-flash:generateContent") +async def generate_content(request: Request, authorization: str = Header(None)): + validate_google_auth(request, authorization) + + data = await request.json() + + request_path = request.url.path + + # Extract model name from path + model = None + if "/models/" in request_path: + try: + model_part = request_path.split("/models/")[1].split(":")[0] + if model_part: + model = model_part + except: + pass + + # Also check request body for model information (fallback). Prefer litellm_model_id (client-facing). + if isinstance(data, dict): + model = ( + data.get("litellm_model_id") or model + or data.get("model") or data.get("model_name") or data.get("modelId") or data.get("model_id") + ) + + # Determine if this is a Gemini model: check path OR model name + # Claude models MUST get Anthropic format (with 'content'). Gemini gets Gemini format (with 'candidates', 'usageMetadata'). + # Check path (only /models/gemini or /models/claude to avoid false positives from project IDs) + request_path_lower = request_path.lower() + path_has_gemini = "/models/gemini" in request_path_lower + path_has_claude = "/models/claude" in request_path_lower + + # Check model name from path or body + model_lower = (model or "").lower() + model_has_gemini = "gemini" in model_lower if model else False + model_has_claude = "claude" in model_lower if model else False + + # Claude models MUST get Anthropic format. Gemini MUST get Gemini format (usageMetadata, candidates). + # When model/path give no signal: default to Gemini (Vertex generateContent is mostly Gemini; returning + # Anthropic for Gemini causes "usageMetadata not found" in LiteLLM). + is_gemini = (path_has_gemini or model_has_gemini or (not path_has_claude and not model_has_claude)) and not (path_has_claude or model_has_claude) + + print(f"[fmt] generate_content path={request_path} model={model} is_gemini={is_gemini} path_g={path_has_gemini} path_c={path_has_claude} model_g={model_has_gemini} model_c={model_has_claude}") + + # CRITICAL: Gemini models MUST get Gemini format, Anthropic models MUST get Anthropic format + if is_gemini: + # Return Gemini format - DO NOT return Anthropic format for Gemini models + # Skip to Gemini format section below + pass + else: + # Return Anthropic format for Anthropic/Claude models only (we only reach here when we detected Claude) + # Anthropic Messages API format - MUST include 'content' array and 'usage' (not usageMetadata) + response = { + "id": f"msg_{uuid.uuid4().hex}", + "type": "message", + "role": "assistant", + "content": [ # REQUIRED: This field must be present and be an array + { + "type": "text", + "text": "Hello! This is a mock response from the Vertex AI Anthropic endpoint. I'm processing your request." + } + ], + "model": _normalize_model_for_response(model or "unknown"), + "stop_reason": "end_turn", + "stop_sequence": None, + "usage": { + "input_tokens": 10, + "output_tokens": 20, + "total_tokens": 30 + } + } + # Ensure content field exists + if 'content' not in response or not isinstance(response.get('content'), list): + response['content'] = [{"type": "text", "text": "Hello! This is a mock response from the Vertex AI Anthropic endpoint. I'm processing your request."}] + + err = _validate_response_format(response, model, False) + if err: + return JSONResponse(status_code=500, content=err) + print(f"[fmt] generate_content branch=anthropic response_keys={list(response.keys())}") + return response + + # Return Vertex AI Gemini format (only reached if is_gemini is True) + # IMPORTANT: Gemini format requires "usageMetadata" (not "usage") + response = { + "candidates": [ + { + "content": { + "role": "model", + "parts": [ + { + "text": f"Hello! This is a mock response from Vertex AI Gemini endpoint. Model: {model or 'gemini'}" + } + ] + }, + "finishReason": "STOP", + "safetyRatings": [ + { + "category": "HARM_CATEGORY_HATE_SPEECH", + "probability": "NEGLIGIBLE", + "probabilityScore": 0.037353516, + "severity": "HARM_SEVERITY_NEGLIGIBLE", + "severityScore": 0.03515625 + }, + { + "category": "HARM_CATEGORY_DANGEROUS_CONTENT", + "probability": "NEGLIGIBLE", + "probabilityScore": 0.017944336, + "severity": "HARM_SEVERITY_NEGLIGIBLE", + "severityScore": 0.020019531 + }, + { + "category": "HARM_CATEGORY_HARASSMENT", + "probability": "NEGLIGIBLE", + "probabilityScore": 0.06738281, + "severity": "HARM_SEVERITY_NEGLIGIBLE", + "severityScore": 0.03173828 + }, + { + "category": "HARM_CATEGORY_SEXUALLY_EXPLICIT", + "probability": "NEGLIGIBLE", + "probabilityScore": 0.11279297, + "severity": "HARM_SEVERITY_NEGLIGIBLE", + "severityScore": 0.057373047 + } + ], + "avgLogprobs": -0.30250951355578853 + } + ], + "usageMetadata": { # REQUIRED: Gemini format uses "usageMetadata" (not "usage") + "promptTokenCount": 5, + "candidatesTokenCount": 51, + "totalTokenCount": 56 + } + } + + err = _validate_response_format(response, model, True) + if err: + return JSONResponse(status_code=500, content=err) + print(f"[fmt] generate_content branch=gemini response_keys={list(response.keys())}") + return response + + +import random + +request_counter = 0 + +@app.post("/generateContent") +@app.post("/v1/projects/bad-adroit-crow-413218/locations/us-central1/publishers/google/models/gemini-1.0-pro-vision-001:generateContent") +@limiter.limit("10000/minute") +async def generate_content_bad(request: Request, authorization: str = Header(None)): + global request_counter + request_counter += 1 + + validate_google_auth(request, authorization) + + # Raise an error for every 200th request + if request_counter % 200 == 0: + raise HTTPException(status_code=500, detail="Internal Server Error: Simulated error for every 200th request") + + # Introduce a 0.5% chance of error for other requests + if random.random() < 0.005: + raise HTTPException(status_code=500, detail="Internal Server Error: Random error (0.5% chance)") + + data = await request.json() + + # You can process the input data here if needed + # For now, we'll just return the hardcoded response + + response = { + "candidates": [ + { + "content": { + "role": "model", + "parts": [ + { + "text": "Good morning to you too! I am not Claude, however. Claude is a large language model trained by Google, while I am Gemini, a multi-modal AI model, developed by Google as well. Is there anything I can help you with today?" + } + ] + }, + "finishReason": "STOP", + "safetyRatings": [ + { + "category": "HARM_CATEGORY_HATE_SPEECH", + "probability": "NEGLIGIBLE", + "probabilityScore": 0.037353516, + "severity": "HARM_SEVERITY_NEGLIGIBLE", + "severityScore": 0.03515625 + }, + { + "category": "HARM_CATEGORY_DANGEROUS_CONTENT", + "probability": "NEGLIGIBLE", + "probabilityScore": 0.017944336, + "severity": "HARM_SEVERITY_NEGLIGIBLE", + "severityScore": 0.020019531 + }, + { + "category": "HARM_CATEGORY_HARASSMENT", + "probability": "NEGLIGIBLE", + "probabilityScore": 0.06738281, + "severity": "HARM_SEVERITY_NEGLIGIBLE", + "severityScore": 0.03173828 + }, + { + "category": "HARM_CATEGORY_SEXUALLY_EXPLICIT", + "probability": "NEGLIGIBLE", + "probabilityScore": 0.11279297, + "severity": "HARM_SEVERITY_NEGLIGIBLE", + "severityScore": 0.057373047 + } + ], + "avgLogprobs": -0.30250951355578853 + } + ], + "usageMetadata": { + "promptTokenCount": 5, + "candidatesTokenCount": 51, + "totalTokenCount": 56 + } + } + + return response + + +@app.post("/models/{model}:batchEmbedContents") +async def gemini_batch_embed_contents(request: Request, model: str, authorization: str = Header(None)): + """Gemini batch embedding endpoint""" + validate_google_auth(request, authorization) + + data = await request.json() + + # Extract requests from the body + requests_list = data.get("requests", []) + if not requests_list: + raise HTTPException(status_code=400, detail="Missing 'requests' field in body") + + # Generate embeddings for each request + embeddings = [] + for req in requests_list: + # Extract text from the request + text = "" + if "text" in req: + text = req["text"] + elif "content" in req: + content = req["content"] + if isinstance(content, list) and len(content) > 0: + if "parts" in content[0]: + parts = content[0]["parts"] + if isinstance(parts, list) and len(parts) > 0: + text = parts[0].get("text", "") + + # Generate fake embedding (768 dimensions) + embedding = [random.uniform(-0.15, 0.15) for _ in range(768)] + + embeddings.append({ + "values": embedding + }) + + return { + "embeddings": embeddings + } + + +@app.post("/predict") +@app.post("/rawPredict") +@app.post("/v1/projects/adroit-crow-413218/locations/us-central1/publishers/google/models/textembedding-gecko@001:predict") +@app.post("/v1/projects/pathrise-convert-1606954137718/locations/us-central1/publishers/google/models/textembedding-gecko@001:predict") +async def predict(request: Request, authorization: str = Header(None)): + validate_google_auth(request, authorization) + + data = await request.json() + + # Extract model from path, body, or headers to determine if this is a Claude model + request_path = request.url.path + model_from_path = None + if "/models/" in request_path: + try: + model_part = request_path.split("/models/")[1].split(":")[0] + if model_part: + model_from_path = model_part + except: + pass + + # Also check request body for model information (prefer litellm_model_id = client-facing name) + model_from_body = None + if isinstance(data, dict): + model_from_body = ( + data.get("litellm_model_id") or data.get("model") or data.get("model_name") + or data.get("modelId") or data.get("model_id") + ) + + # Check headers for model information (LiteLLM may pass model in headers) + model_from_headers = ( + request.headers.get("X-Model") or + request.headers.get("x-model") or + request.headers.get("X-LiteLLM-Model") or + request.headers.get("x-litellm-model") + ) + + # Determine final model (prefer body = client-facing name over path = deployment ID) + final_model = model_from_body or model_from_path or model_from_headers or "" + model_lower = (final_model or "").lower() + + # Check request body structure: Claude requests use "contents" (Anthropic format) or "messages" (chat format), + # while embedding requests use "instances" (Vertex AI embedding format) + has_contents = isinstance(data, dict) and "contents" in data + has_messages = isinstance(data, dict) and "messages" in data + has_instances = isinstance(data, dict) and "instances" in data + # If it has messages or contents (but not instances), it's a Claude request + body_suggests_claude = (has_contents or has_messages) and not has_instances + + # Check if this is a Claude model + path_has_claude = "/models/claude" in request_path.lower() + model_has_claude = "claude" in model_lower if final_model else False + + # For :rawPredict or :predict endpoints, if we can't determine the model and the request + # doesn't have "instances" (embedding format), assume it's Claude since LiteLLM uses + # these endpoints for Claude models via Vertex AI + is_raw_predict_endpoint = ":rawPredict" in request_path or request_path.endswith("rawPredict") or ":predict" in request_path or request_path.endswith("predict") + no_instances = not has_instances + fallback_to_claude = is_raw_predict_endpoint and no_instances and not final_model + + # If body has "contents" but no "instances", it's likely a Claude request + # Or if it's a rawPredict/predict endpoint without instances and no model detected, assume Claude + is_claude = path_has_claude or model_has_claude or body_suggests_claude or fallback_to_claude + + # Log for debugging + print(f"[predict] path={request_path} model={final_model} path_has_claude={path_has_claude} model_has_claude={model_has_claude} has_contents={has_contents} has_messages={has_messages} has_instances={has_instances} body_suggests_claude={body_suggests_claude} fallback_to_claude={fallback_to_claude} is_claude={is_claude}") + + # If this is a Claude model, return Anthropic format (not embedding format) + if is_claude: + response = { + "id": f"msg_{uuid.uuid4().hex}", + "type": "message", + "role": "assistant", + "content": [ + { + "type": "text", + "text": f"Hello! This is a mock response from Vertex AI Anthropic endpoint via predict/rawPredict. Model: {final_model or 'claude'}" + } + ], + "model": _normalize_model_for_response(final_model or "claude"), + "stop_reason": "end_turn", + "stop_sequence": None, + "usage": { + "input_tokens": 10, + "output_tokens": 20, + "total_tokens": 30 + } + } + # Ensure content field exists + if 'content' not in response or not isinstance(response.get('content'), list): + response['content'] = [{"type": "text", "text": f"Hello! This is a mock response from Vertex AI Anthropic endpoint via predict/rawPredict. Model: {final_model or 'claude'}"}] + + err = _validate_response_format(response, final_model or "claude", False) + if err: + return JSONResponse(status_code=500, content=err) + print(f"[fmt] predict/rawPredict path={request_path} model={final_model} branch=anthropic response_keys={list(response.keys())}") + return response + + # Otherwise, return embedding format (for embedding models) + # Process the input data + instances = data.get('instances', []) + num_instances = len(instances) + + # Generate fake embeddings + predictions = [] + for _ in range(num_instances): + embedding = [random.uniform(-0.15, 0.15) for _ in range(768)] # 768-dimensional embedding + predictions.append({ + "embeddings": { + "values": embedding, + "statistics": { + "truncated": False, + "token_count": random.randint(4, 10) + } + } + }) + + # Calculate billable character count + billable_character_count = sum(len(instance.get('content', '')) for instance in instances) + + response = { + "predictions": predictions, + "metadata": { + "billableCharacterCount": billable_character_count + } + } + + return response + + +# Add catch-all routes for Vertex AI endpoints +# These must come AFTER the specific routes but handle any project/location/model combination + +@app.post("/v1/projects/{project}/locations/{location}/publishers/google/models/{model}:generateContent") +async def vertex_generate_content_catchall(request: Request, project: str, location: str, model: str, authorization: str = Header(None)): + """Catch-all endpoint for Vertex AI generateContent - accepts any project/location/model""" + validate_google_auth(request, authorization) + + data = await request.json() + + # Also check request body for model information (prefer litellm_model_id = client-facing name) + model_from_body = None + if isinstance(data, dict): + model_from_body = ( + data.get("litellm_model_id") or data.get("model") or data.get("model_name") + or data.get("modelId") or data.get("model_id") + ) + + # Use model from body (client-facing) or path + final_model = model_from_body or model or "" + + # Model detection: only /models/gemini or /models/claude in path to avoid false positives (e.g. "gemini" in project ID) + request_path_lower = request.url.path.lower() + path_has_gemini = "/models/gemini" in request_path_lower + path_has_claude = "/models/claude" in request_path_lower + + model_lower = (final_model or "").lower() + model_has_gemini = "gemini" in model_lower if final_model else False + model_has_claude = "claude" in model_lower if final_model else False + + # Claude models MUST get Anthropic format (content, usage). Gemini MUST get Gemini format (candidates, usageMetadata). + # When model/path give no signal: default to Gemini (Vertex generateContent is mostly Gemini). + is_gemini = (path_has_gemini or model_has_gemini or (not path_has_claude and not model_has_claude)) and not (path_has_claude or model_has_claude) + + print(f"[fmt] vertex_catchall path={request.url.path} model={final_model} is_gemini={is_gemini} path_g={path_has_gemini} path_c={path_has_claude} model_g={model_has_gemini} model_c={model_has_claude}") + + # CRITICAL: Gemini models MUST get Gemini format, Anthropic models MUST get Anthropic format + if is_gemini: + # Return Gemini format - DO NOT return Anthropic format for Gemini models + # Skip to Gemini format section below + pass + else: + # Return Anthropic format for Anthropic/Claude models only (we only reach here when we detected Claude) + response = { + "id": f"msg_{uuid.uuid4().hex}", + "type": "message", + "role": "assistant", + "content": [ + { + "type": "text", + "text": f"Hello! This is a mock response from Vertex AI Anthropic endpoint. Model: {final_model or model}" + } + ], + "model": _normalize_model_for_response(final_model or model or "unknown"), + "stop_reason": "end_turn", + "stop_sequence": None, + "usage": { + "input_tokens": 10, + "output_tokens": 20, + "total_tokens": 30 + } + } + # Safety check: Ensure content field exists and is a list + if 'content' not in response or not isinstance(response.get('content'), list): + response['content'] = [{"type": "text", "text": f"Hello! This is a mock response from Vertex AI Anthropic endpoint. Model: {final_model or model}"}] + + err = _validate_response_format(response, final_model or model, False) + if err: + return JSONResponse(status_code=500, content=err) + print(f"[fmt] vertex_catchall branch=anthropic response_keys={list(response.keys())}") + return response + + # Return Vertex AI Gemini format (only reached if is_gemini is True) + response = { + "candidates": [ + { + "content": { + "role": "model", + "parts": [ + { + "text": "Hello! This is a mock response from Vertex AI. Model: " + model + } + ] + }, + "finishReason": "STOP", + "safetyRatings": [ + { + "category": "HARM_CATEGORY_HATE_SPEECH", + "probability": "NEGLIGIBLE", + "probabilityScore": 0.037353516, + "severity": "HARM_SEVERITY_NEGLIGIBLE", + "severityScore": 0.03515625 + }, + { + "category": "HARM_CATEGORY_DANGEROUS_CONTENT", + "probability": "NEGLIGIBLE", + "probabilityScore": 0.017944336, + "severity": "HARM_SEVERITY_NEGLIGIBLE", + "severityScore": 0.020019531 + }, + { + "category": "HARM_CATEGORY_HARASSMENT", + "probability": "NEGLIGIBLE", + "probabilityScore": 0.06738281, + "severity": "HARM_SEVERITY_NEGLIGIBLE", + "severityScore": 0.03173828 + }, + { + "category": "HARM_CATEGORY_SEXUALLY_EXPLICIT", + "probability": "NEGLIGIBLE", + "probabilityScore": 0.11279297, + "severity": "HARM_SEVERITY_NEGLIGIBLE", + "severityScore": 0.057373047 + } + ], + "avgLogprobs": -0.30250951355578853 + } + ], + "usageMetadata": { + "promptTokenCount": 5, + "candidatesTokenCount": 51, + "totalTokenCount": 56 + } + } + err = _validate_response_format(response, final_model or model, True) + if err: + return JSONResponse(status_code=500, content=err) + print(f"[fmt] vertex_catchall branch=gemini response_keys={list(response.keys())}") + return response + + +@app.post("/v1/projects/{project}/locations/{location}/publishers/google/models/{model}:predict") +async def vertex_predict_catchall(request: Request, project: str, location: str, model: str, authorization: str = Header(None)): + """Catch-all endpoint for Vertex AI predict (embeddings) - accepts any project/location/model""" + validate_google_auth(request, authorization) + + data = await request.json() + + # Check headers for model information (LiteLLM may pass model in headers) + model_from_headers = ( + request.headers.get("X-Model") or + request.headers.get("x-model") or + request.headers.get("X-LiteLLM-Model") or + request.headers.get("x-litellm-model") + ) + + # Use model from body (prefer litellm_model_id), headers, or path + model_from_body = None + if isinstance(data, dict): + model_from_body = ( + data.get("litellm_model_id") or data.get("model") or data.get("model_name") + or data.get("modelId") or data.get("model_id") + ) + final_model = model_from_body or model or model_from_headers or "" + + # Check request body structure: Claude requests use "contents" (Anthropic format) or "messages" (chat format), + # while embedding requests use "instances" (Vertex AI embedding format) + has_contents = isinstance(data, dict) and "contents" in data + has_messages = isinstance(data, dict) and "messages" in data + has_instances = isinstance(data, dict) and "instances" in data + # If it has messages or contents (but not instances), it's a Claude request + body_suggests_claude = (has_contents or has_messages) and not has_instances + + # Check if this is a Claude model + model_lower = (final_model or "").lower() + model_has_claude = "claude" in model_lower if final_model else False + request_path = request.url.path + path_has_claude = "/models/claude" in request_path.lower() + + # For :predict endpoints, if we can't determine the model and the request + # doesn't have "instances" (embedding format), assume it's Claude since LiteLLM uses + # these endpoints for Claude models via Vertex AI + is_predict_endpoint = ":predict" in request_path or request_path.endswith("predict") + no_instances = not has_instances + fallback_to_claude = is_predict_endpoint and no_instances and not final_model + + is_claude = path_has_claude or model_has_claude or body_suggests_claude or fallback_to_claude + + # Log for debugging + print(f"[vertex_predict_catchall] path={request_path} model={final_model or model} path_has_claude={path_has_claude} model_has_claude={model_has_claude} has_contents={has_contents} has_messages={has_messages} has_instances={has_instances} body_suggests_claude={body_suggests_claude} fallback_to_claude={fallback_to_claude} is_claude={is_claude}") + + # If this is a Claude model, return Anthropic format (not embedding format) + if is_claude: + response = { + "id": f"msg_{uuid.uuid4().hex}", + "type": "message", + "role": "assistant", + "content": [ + { + "type": "text", + "text": f"Hello! This is a mock response from Vertex AI Anthropic endpoint via predict. Model: {final_model or model}" + } + ], + "model": _normalize_model_for_response(final_model or model or "claude"), + "stop_reason": "end_turn", + "stop_sequence": None, + "usage": { + "input_tokens": 10, + "output_tokens": 20, + "total_tokens": 30 + } + } + # Ensure content field exists + if 'content' not in response or not isinstance(response.get('content'), list): + response['content'] = [{"type": "text", "text": f"Hello! This is a mock response from Vertex AI Anthropic endpoint via predict. Model: {final_model or model}"}] + + err = _validate_response_format(response, final_model or model, False) + if err: + return JSONResponse(status_code=500, content=err) + print(f"[fmt] vertex_predict_catchall path={request_path} model={final_model or model} branch=anthropic response_keys={list(response.keys())}") + return response + + # Otherwise, return embedding format (for embedding models) + instances = data.get('instances', []) + num_instances = len(instances) + + predictions = [] + for _ in range(num_instances): + embedding = [random.uniform(-0.15, 0.15) for _ in range(768)] + predictions.append({ + "embeddings": { + "values": embedding, + "statistics": { + "truncated": False, + "token_count": random.randint(4, 10) + } + } + }) + + billable_character_count = sum(len(instance.get('content', '')) for instance in instances) + + return { + "predictions": predictions, + "metadata": { + "billableCharacterCount": billable_character_count + } + } + + +@app.post("/v1/projects/{project}/locations/{location}/publishers/google/models/{model}:rawPredict") +async def vertex_raw_predict_catchall(request: Request, project: str, location: str, model: str, authorization: str = Header(None)): + """Catch-all endpoint for Vertex AI rawPredict - accepts any project/location/model""" + validate_google_auth(request, authorization) + # rawPredict is similar to predict, so delegate to the same handler + return await vertex_predict_catchall(request, project, location, model, authorization) + + +@app.post("/runs") +@app.post("/runs/batch") +async def runs(request: Request): + start_time = time.perf_counter() + + # Simulate some minimal processing + data = await request.json() + + # Create a simple response + response = { + "id": str(uuid.uuid4()), + "status": "completed", + "created_at": int(time.time()), + "request": data + } + + # Ensure the response takes at least 0.05 ms + elapsed_time = (time.perf_counter() - start_time) * 1000 # Convert to milliseconds + if elapsed_time < 0.05: + time.sleep((0.05 - elapsed_time) / 1000) # Convert back to seconds for sleep + + return response + + +@app.post("/traces") +async def traces(request: Request): + try: + start_time = time.perf_counter() + + # Attempt to parse the request body + try: + data = await request.json() + except json.JSONDecodeError: + # If JSON parsing fails, try to read the raw body + body = await request.body() + return HTTPException(status_code=400, detail=f"Invalid JSON: {body.decode('utf-8', errors='ignore')}") + except UnicodeDecodeError: + # If decoding fails, return an error about invalid encoding + return HTTPException(status_code=400, detail="Request body is not valid UTF-8 encoded") + + # Rest of the function remains the same + response = { + "id": str(uuid.uuid4()), + "status": "completed", + "created_at": int(time.time()), + "trace_data": { + "events": [ + { + "timestamp": int(time.time()), + "type": "start", + "details": "Trace started" + }, + { + "timestamp": int(time.time()) + 1, + "type": "end", + "details": "Trace completed" + } + ], + } + } + + # Ensure the response takes at least 0.05 ms + elapsed_time = (time.perf_counter() - start_time) * 1000 # Convert to milliseconds + if elapsed_time < 0.05: + time.sleep((0.05 - elapsed_time) / 1000) # Convert back to seconds for sleep + + return response + except Exception as e: + import traceback + traceback.print_exc() + return HTTPException(status_code=500, detail=str(e)) + +import gzip +import io + +@app.post("/api/v2/logs") +async def logs(request: Request): + await asyncio.sleep(60) # Wait for 1 second + return {"status": "done"} + start_time = time.perf_counter() + + # Check if the content is gzipped + content_encoding = request.headers.get("Content-Encoding", "").lower() + + # Read the raw body + body = await request.body() + + # Decompress if gzipped + if content_encoding == "gzip": + try: + body = gzip.decompress(body) + except gzip.BadGzipFile: + return HTTPException(status_code=400, detail="Invalid gzip data") + + # Attempt to parse the request body + try: + data = json.loads(body) + except json.JSONDecodeError: + return HTTPException(status_code=400, detail=f"Invalid JSON: {body.decode('utf-8', errors='ignore')}") + except UnicodeDecodeError: + return HTTPException(status_code=400, detail="Request body is not valid UTF-8 encoded") + + # Create a log response + response = { + "id": str(uuid.uuid4()), + "timestamp": int(time.time()), + "level": "info", + "message": "Log entry received", + "data": data + } + + # Ensure the response takes at least 0.05 ms + elapsed_time = (time.perf_counter() - start_time) * 1000 # Convert to milliseconds + if elapsed_time < 0.05: + time.sleep((0.05 - elapsed_time) / 1000) # Convert back to seconds for sleep + + return Response( + content=json.dumps(response), + status_code=202, + ) +slack_requests = deque(maxlen=10) +slack_requests = deque(maxlen=10) + +class SlackRequest(BaseModel): + timestamp: datetime + data: Dict[str, Any] + +@app.post("/slack") +async def slack_endpoint(request: Request): + current_time = datetime.now() + request_data = await request.json() + + # Add the current request to the deque + slack_requests.append(SlackRequest(timestamp=current_time, data=request_data)) + + # Remove requests older than 10 minutes + slack_requests_list = list(slack_requests) + slack_requests_list = [req for req in slack_requests_list if current_time - req.timestamp <= timedelta(minutes=10)] + slack_requests.clear() + slack_requests.extend(slack_requests_list) + + return {"message": "Request received and stored"} + +@app.get("/slack/history", response_model=List[SlackRequest]) +async def get_slack_history(): + return list(slack_requests) + + + + + +def data_generator_anthropic(model=None): + response_id = uuid.uuid4().hex + sentence = "Hello this is a test response from a fixed OpenAI endpoint." + words = sentence.split(" ") + _model = model if isinstance(model, str) else "claude-3-opus-20240229" + for word in words: + word = word + " " + chunk = { + "id": f"chatcmpl-{response_id}", + "object": "chat.completion.chunk", + "created": 1677652288, + "model": _model, + "choices": [{"index": 0, "delta": {"content": word}}], + } + try: + yield f"data: {json.dumps(chunk.dict())}\n\n" + except: + yield f"data: {json.dumps(chunk)}\n\n" + + + +# for completion +@app.post("/v1/messages") +async def completion_anthropic(request: Request): + data = await request.json() + _model = ( + data.get("litellm_model_id") # Prefer client-facing model name + or data.get("model") or data.get("model_name") or data.get("modelId") or data.get("model_id") + or request.headers.get("X-LiteLLM-Model") or request.headers.get("x-litellm-model") + or "claude-3-opus-20240229" + ) + _model = _model if isinstance(_model, str) else "claude-3-opus-20240229" + _model = _normalize_model_for_response(_model) + + if data.get("stream") == True: + return StreamingResponse( + content=data_generator_anthropic(model=_model), + media_type="text/event-stream", + ) + else: + response = { + "id": "msg_01G7MsdWPT2JZMUuc1UXRavn", + "type": "message", + "role": "assistant", + "content": [ + { + "type": "text", + "text": "I'm sorry, but the string of characters \"123450000s0 p kk\" doesn't appear to have any clear meaning or context. It seems to be a random combination of numbers and letters. If you could provide more information or clarify what you're trying to communicate, I'll do my best to assist you." + } + ], + "model": _model, + "stop_reason": "end_turn", + "stop_sequence": None, + "usage": { + "input_tokens": 17, + "output_tokens": 71, + "total_tokens": 88 + } + } + + return response + +@app.post("/load_test/api/public/ingestion") +async def mock_ingestion(request: Request): + time.sleep(0.5) + return {"status": "done"} + + +seen_langfuse_request_ids = set() + +@app.post("/api/public/ingestion") +async def ingestion(request: Request): + try: + global seen_langfuse_request_ids + data = await request.json() + + # Extract request IDs from the batch + for item in data.get('batch', []): + if item.get('type') == 'generation-create': + full_request_id = item.get('body', {}).get('id') + if full_request_id and '_' in full_request_id: + # Split on underscore and take the second part (the chatcmpl ID) + clean_request_id = full_request_id.split('_')[1] + seen_langfuse_request_ids.add(clean_request_id) + + print(f"Stored request IDs (total: {len(seen_langfuse_request_ids)}): {seen_langfuse_request_ids}") + await asyncio.sleep(1) # Original delay + return {"status": "done", "stored_ids_count": len(seen_langfuse_request_ids)} + + except Exception as e: + print(f"Error processing ingestion request: {str(e)}") + raise HTTPException(status_code=500, detail=f"Error processing request: {str(e)}") + +@app.get("/langfuse/trace/{request_id}") +async def has_request_id(request_id: str): + return { + "exists": request_id in seen_langfuse_request_ids, + "request_id": request_id + } + + +# for responses +def responses_data_generator(input_text="", model=None): + """Generator for streaming Responses API chunks""" + # Ensure input_text is always a string + if isinstance(input_text, list): + input_text = " ".join(str(msg) for msg in input_text) + elif isinstance(input_text, dict): + input_text = str(input_text) + else: + input_text = str(input_text) if input_text else "" + + _model = _normalize_model_for_response(model) if isinstance(model, str) else "gpt-4.1" + response_id = uuid.uuid4().hex + item_id = f"msg_{uuid.uuid4().hex}" + sentence = f"Hello! I received your input: '{input_text}'. This is a mock response from the Responses API." + words = sentence.split(" ") + current_time = int(time.time()) + + # 1. Send response.created event + response_created = { + 'type': 'response.created', + 'response': { + 'id': f'resp_{response_id}', + 'object': 'response', + 'created_at': current_time, + 'model': _model, + 'status': 'in_progress', + 'output': [] + } + } + yield f"data: {json.dumps(response_created)}\n\n" + + # 2. Send output_item.added event + output_item_added = { + 'type': 'response.output_item.added', + 'output_index': 0, + 'item': { + 'type': 'message', + 'id': item_id, + 'role': 'assistant', + 'status': 'in_progress', + 'content': [] + } + } + yield f"data: {json.dumps(output_item_added)}\n\n" + + # 3. Send text deltas + for word in words: + text_delta = { + 'type': 'response.output_text.delta', + 'item_id': item_id, + 'output_index': 0, + 'content_index': 0, + 'delta': word + ' ' + } + yield f"data: {json.dumps(text_delta)}\n\n" + + # 4. Send output_text.done event + output_text_done = { + 'type': 'response.output_text.done', + 'item_id': item_id, + 'output_index': 0, + 'content_index': 0, + 'text': sentence + } + yield f"data: {json.dumps(output_text_done)}\n\n" + + # 5. Send response.completed event + response_completed = { + 'type': 'response.completed', + 'response': { + 'id': f'resp_{response_id}', + 'object': 'response', + 'created_at': current_time, + 'model': _model, + 'status': 'completed', + 'output': [{ + 'type': 'message', + 'id': item_id, + 'role': 'assistant', + 'status': 'completed', + 'content': [{'type': 'output_text', 'text': sentence}] + }], + 'usage': { + 'input_tokens': len(input_text.split()) if input_text else 0, + 'output_tokens': len(sentence.split()), + 'total_tokens': (len(input_text.split()) if input_text else 0) + len(sentence.split()) + } + } + } + yield f"data: {json.dumps(response_completed)}\n\n" + + # 6. Send [DONE] + yield "data: [DONE]\n\n" + + +async def _send_realtime_text_response( + websocket: WebSocket, + text: str, + model: str, + next_event_id: Callable[[], str], + conversation_id: str, +) -> None: + """Emit realtime events that align with OpenAI's websocket schema.""" + + current_time = int(time.time()) + response_id = f"resp_{uuid.uuid4().hex}" + item_id = f"msg_{uuid.uuid4().hex}" + + await websocket.send_json( + { + "event_id": next_event_id(), + "type": "response.created", + "response": { + "id": response_id, + "object": "realtime.response", + "created_at": current_time, + "status": "in_progress", + "model": model, + "conversation_id": conversation_id, + "output": [], + }, + } + ) + + await websocket.send_json( + { + "event_id": next_event_id(), + "type": "response.output_item.added", + "response_id": response_id, + "output_index": 0, + "item": { + "id": item_id, + "object": "realtime.item", + "type": "message", + "role": "assistant", + "status": "in_progress", + "content": [], + }, + } + ) + + await websocket.send_json( + { + "event_id": next_event_id(), + "type": "response.content_part.added", + "response_id": response_id, + "output_index": 0, + "item_id": item_id, + "content_index": 0, + "part": { + "type": "text", + "text": "", + }, + } + ) + + await websocket.send_json( + { + "event_id": next_event_id(), + "type": "response.text.delta", + "response_id": response_id, + "output_index": 0, + "item_id": item_id, + "content_index": 0, + "delta": text, + } + ) + + await websocket.send_json( + { + "event_id": next_event_id(), + "type": "response.text.done", + "response_id": response_id, + "output_index": 0, + "item_id": item_id, + "content_index": 0, + "text": text, + } + ) + + await websocket.send_json( + { + "event_id": next_event_id(), + "type": "response.content_part.done", + "response_id": response_id, + "output_index": 0, + "item_id": item_id, + "content_index": 0, + "part": { + "type": "text", + "text": text, + }, + } + ) + + await websocket.send_json( + { + "event_id": next_event_id(), + "type": "response.output_item.done", + "response_id": response_id, + "output_index": 0, + "item": { + "id": item_id, + "object": "realtime.item", + "type": "message", + "role": "assistant", + "status": "completed", + "content": [ + { + "type": "text", + "text": text, + } + ], + }, + } + ) + + await websocket.send_json( + { + "event_id": next_event_id(), + "type": "response.done", + "response": { + "id": response_id, + "object": "realtime.response", + "status": "completed", + "model": model, + "conversation_id": conversation_id, + "output": [ + { + "object": "realtime.item", + "type": "message", + "id": item_id, + "role": "assistant", + "status": "completed", + "content": [ + { + "type": "text", + "text": text, + } + ], + } + ], + "usage": { + "input_tokens": 0, + "output_tokens": len(text.split()), + "total_tokens": len(text.split()), + }, + }, + } + ) + + +@app.post("/responses") +@app.post("/v1/responses") +async def create_response(request: Request): + """OpenAI Responses API endpoint - Create a response""" + _time_to_sleep = os.getenv("TIME_TO_SLEEP", None) + if _time_to_sleep is not None: + print("sleeping for " + _time_to_sleep) + await asyncio.sleep(float(_time_to_sleep)) + + data = await request.json() + + # Handle error cases + if data.get("model") == "429": + raise HTTPException(status_code=status.HTTP_429_TOO_MANY_REQUESTS, detail="Too many requests") + + if data.get("model") == "random_sleep": + sleep_time = random.randint(1, 10) + print("sleeping for " + str(sleep_time) + " seconds") + await asyncio.sleep(sleep_time) + + # Degraded provider simulation for responses endpoint + if data.get("model") in ["degraded", "slow_provider", "blocked"]: + sleep_time = get_histogram_sleep_time() + print(f"[DEGRADED MODE] Sleeping for {sleep_time:.1f} seconds ({sleep_time/60:.1f} minutes) - histogram distribution") + await asyncio.sleep(sleep_time) + + # Non-streaming response setup + response_id = uuid.uuid4().hex + model = ( + data.get("litellm_model_id") or data.get("model") or data.get("model_name") + or data.get("modelId") or data.get("model_id") or "gpt-4.1" + ) + model = _normalize_model_for_response(model if isinstance(model, str) else "gpt-4.1") + input_data = data.get("input", "") + + # Handle input: can be string, list, or dict (Azure format) + if isinstance(input_data, list): + # Convert list of messages to string representation for token counting + input_text = " ".join(str(msg) for msg in input_data) + elif isinstance(input_data, dict): + # Handle dict format (e.g., {"messages": [...]}) + input_text = str(input_data) + else: + # Assume it's a string + input_text = str(input_data) if input_data else "" + + if data.get("stream") == True: + return StreamingResponse( + content=responses_data_generator(input_text=input_text, model=model), + media_type="text/event-stream", + ) + tools = data.get("tools", []) + reasoning = data.get("reasoning", {}) + background = data.get("background", False) + + # Generate output text based on input (use original input_data for display) + input_display = input_data if isinstance(input_data, str) else str(input_data) + output_text = f"Hello! I received your input: '{input_display}'. This is a mock response from the Responses API." + + # Handle background mode + if background: + # In background mode, return a job-like response + return { + "id": f"resp_{response_id}", + "object": "response", + "created_at": int(time.time()), + "model": model, + "status": "processing", + "background": True, + "output": [], # Empty output for processing state + "tools": tools, + "reasoning": reasoning + } + + # Regular response + response = { + "id": f"resp_{response_id}", + "object": "response", + "created_at": int(time.time()), + "model": model, + "output": [ + { + "type": "message", + "id": f"msg_{uuid.uuid4().hex}", + "status": "completed", + "role": "assistant", + "content": [ + { + "type": "output_text", + "text": output_text + } + ] + } + ], + "tools": tools if tools else [], + "reasoning": reasoning if reasoning else {}, + "usage": { + "input_tokens": len(input_text.split()) if input_text else 0, + "output_tokens": len(output_text.split()), + "total_tokens": len(input_text.split()) + len(output_text.split()) if input_text else len(output_text.split()) + } + } + + return response + + +@app.post("/openai/responses") +@app.post("/openai/v1/responses") +async def azure_responses_api(request: Request): + """Azure Responses API endpoint - delegates to the same handler as /responses""" + try: + # Reuse the exact same logic as your existing /responses endpoint + return await create_response(request) + except Exception as e: + # Return detailed error response to client + error_detail = await get_error_detail_with_body(e, request, "azure_responses_api") + raise HTTPException( + status_code=500, + detail=error_detail + ) + + +@app.get("/v1/responses/{response_id}") +async def get_response(response_id: str): + """OpenAI Responses API endpoint - Get a response by ID""" + # Return a mock response with the given ID + return { + "id": response_id, + "object": "response", + "created_at": int(time.time()), + "model": "gpt-4.1", + "output": [ + { + "type": "message", + "id": f"msg_{uuid.uuid4().hex}", + "status": "completed", + "role": "assistant", + "content": [ + { + "type": "output_text", + "text": "This is a mock response retrieved by ID." + } + ] + } + ], + "tools": [], + "reasoning": {}, + "status": "completed", + "usage": { + "input_tokens": 5, + "output_tokens": 10, + "total_tokens": 15 + } + } + + +@app.websocket("/v1/realtime") +async def realtime_endpoint(websocket: WebSocket): + """Simplified realtime endpoint compatible with LiteLLM proxy tests.""" + + model = websocket.query_params.get("model", "gpt-4o-realtime-preview-2024-10-01") + session_id = f"sess_{uuid.uuid4().hex}" + conversation_id = f"conv_{uuid.uuid4().hex}" + + event_counter = 0 + + def next_event_id() -> str: + nonlocal event_counter + event_counter += 1 + return f"evt_{session_id}_{event_counter}" + + await websocket.accept() + + await websocket.send_json( + { + "event_id": next_event_id(), + "type": "session.created", + "session": { + "id": session_id, + "model": model, + "created_at": int(time.time()), + "modalities": ["text"], + }, + } + ) + + try: + while True: + try: + incoming = await websocket.receive() + except RuntimeError as e: + # Connection was closed by client + if "already completed" in str(e) or "websocket.close" in str(e): + break + raise + + if incoming.get("type") == "websocket.close": + break + + message_text = incoming.get("text") + if message_text is None: + # Skip non-text messages (binary, ping/pong, etc.) + continue + + try: + payload = json.loads(message_text) + except json.JSONDecodeError: + await websocket.send_json( + { + "type": "error", + "error": { + "type": "invalid_request_error", + "message": "Payload must be valid JSON", + }, + } + ) + continue + + event_type = payload.get("type") + + if event_type == "session.update": + await websocket.send_json( + { + "event_id": next_event_id(), + "type": "session.updated", + "session": { + "id": session_id, + "model": model, + "modalities": payload.get("session", {}).get("modalities", ["text"]), + }, + } + ) + elif event_type == "response.create": + response_payload = payload.get("response", {}) + instructions = response_payload.get("instructions", "") + prefix = "[fake-realtime] " + output_text = ( + f"{prefix}{instructions}" + if instructions + else f"{prefix}Hello! This is a realtime response from the fake endpoint." + ) + await _send_realtime_text_response( + websocket, + output_text, + model, + next_event_id, + conversation_id, + ) + else: + await websocket.send_json( + { + "type": "error", + "error": { + "type": "unsupported_event", + "message": f"Unsupported realtime event: {event_type}", + }, + } + ) + + except WebSocketDisconnect: + pass + except Exception as exc: + try: + await websocket.close(code=1011, reason=str(exc)) + except RuntimeError: + pass + finally: + try: + await websocket.close() + except RuntimeError: + pass + + +# Catch-all route for Vertex AI endpoints with colons (e.g., :generateContent, :predict) +# This must come at the END after all other routes +# Only matches paths that look like Vertex AI endpoints (contains /v1/projects/ or ends with colon methods) +@app.post("/{path:path}") +async def catch_all_vertex_with_colons(request: Request, path: str): + """Catch-all for Vertex AI endpoints with colons (e.g., :generateContent, :predict)""" + # Extract the actual path from the request URL + request_path = request.url.path + + # Only handle paths that look like Vertex AI endpoints + # Check for Vertex AI patterns: /v1/projects/, paths with colons, or common Vertex methods + is_vertex_path = ( + "/v1/projects/" in request_path or + ":generateContent" in request_path or + ":predict" in request_path or + ":rawPredict" in request_path or + request_path.endswith("generateContent") or + request_path.endswith("predict") or + request_path.endswith("rawPredict") or + request_path.endswith(":generateContent") or + request_path.endswith(":predict") or + request_path.endswith(":rawPredict") or + path.endswith("generateContent") or + path.endswith("predict") or + path.endswith("rawPredict") or + path.endswith(":generateContent") or + path.endswith(":predict") or + path.endswith(":rawPredict") + ) + + if not is_vertex_path: + # Not a Vertex AI path, return 404 with details + error_detail = format_detailed_error( + Exception("Path does not match Vertex AI endpoint patterns"), + request, + "catch_all_vertex_path_check" + ) + error_detail["error"]["message"] = f"No handler found for path: {request_path}. Supported Vertex AI patterns: /v1/projects/.../:generateContent, /v1/projects/.../:predict, /:generateContent, /:predict, /:rawPredict" + raise HTTPException(status_code=404, detail=error_detail) + + # Check if this is a generateContent endpoint (multiple patterns) + is_generate_content = ( + ":generateContent" in request_path or + request_path.endswith("generateContent") or + request_path.endswith(":generateContent") or + ":generateContent" in path or + path.endswith("generateContent") or + path.endswith(":generateContent") + ) + + if is_generate_content: + authorization = request.headers.get("authorization") or request.headers.get("Authorization") + try: + return await generate_content(request, authorization) + except Exception as e: + # Return detailed error response to client + error_detail = await get_error_detail_with_body(e, request, "vertex_generate_content") + raise HTTPException(status_code=500, detail=error_detail) + + # Check if this is a predict endpoint (multiple patterns) + is_predict = ( + ":predict" in request_path or + request_path.endswith("predict") or + request_path.endswith(":predict") or + ":predict" in path or + path.endswith("predict") or + path.endswith(":predict") + ) + + if is_predict: + authorization = request.headers.get("authorization") or request.headers.get("Authorization") + try: + return await predict(request, authorization) + except Exception as e: + # Return detailed error response to client + error_detail = await get_error_detail_with_body(e, request, "vertex_predict") + raise HTTPException(status_code=500, detail=error_detail) + + # Check if this is a rawPredict endpoint (multiple patterns) + # rawPredict is similar to predict, so we'll use the same handler + is_raw_predict = ( + ":rawPredict" in request_path or + request_path.endswith("rawPredict") or + request_path.endswith(":rawPredict") or + ":rawPredict" in path or + path.endswith("rawPredict") or + path.endswith(":rawPredict") + ) + + if is_raw_predict: + authorization = request.headers.get("authorization") or request.headers.get("Authorization") + try: + return await predict(request, authorization) + except Exception as e: + # Return detailed error response to client + error_detail = await get_error_detail_with_body(e, request, "vertex_raw_predict") + raise HTTPException(status_code=500, detail=error_detail) + + # If it's a Vertex path but doesn't match our handlers, return 404 with details + error_detail = format_detailed_error( + Exception(f"Vertex AI path recognized but no handler available"), + request, + "catch_all_vertex_no_handler" + ) + error_detail["error"]["message"] = f"Vertex AI path '{request_path}' recognized but no handler matched. Supported handlers: generateContent, predict, rawPredict" + raise HTTPException(status_code=404, detail=error_detail) + + +if __name__ == "__main__": + import os + import uvicorn + + port = int(os.getenv("PORT", "8090")) + uvicorn.run(app, host="0.0.0.0", port=port, log_level="info") \ No newline at end of file diff --git a/tests/mock_endpoints/example_openai_endpoint/requirements.txt b/tests/mock_endpoints/example_openai_endpoint/requirements.txt new file mode 100644 index 00000000000..32d48733bed --- /dev/null +++ b/tests/mock_endpoints/example_openai_endpoint/requirements.txt @@ -0,0 +1,8 @@ +fastapi +uvicorn[standard] +slowapi +uuid +python-multipart>=0.0.20 +pydantic +python-dotenv>=0.2.0 # for env +websockets \ No newline at end of file diff --git a/tests/mock_endpoints/start_mock_server.sh b/tests/mock_endpoints/start_mock_server.sh new file mode 100755 index 00000000000..6f330cfc315 --- /dev/null +++ b/tests/mock_endpoints/start_mock_server.sh @@ -0,0 +1,61 @@ +#!/usr/bin/env bash +# +# Start the vendored example_openai_endpoint mock server. +# +# Usage: +# ./tests/mock_endpoints/start_mock_server.sh # foreground +# ./tests/mock_endpoints/start_mock_server.sh --background # background, prints PID +# +# Environment: +# PORT - port to bind (default 8090) +# MOCK_SERVER_LOG_FILE - log file when running in background (default /tmp/mock_openai_endpoint.log) +# MOCK_SERVER_PID_FILE - PID file when running in background (default /tmp/mock_openai_endpoint.pid) +# MOCK_SERVER_TIMEOUT - seconds to wait for /chat/completions to respond in --background (default 30) +# +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +APP_DIR="${SCRIPT_DIR}/example_openai_endpoint" + +PORT="${PORT:-8090}" +LOG_FILE="${MOCK_SERVER_LOG_FILE:-/tmp/mock_openai_endpoint.log}" +PID_FILE="${MOCK_SERVER_PID_FILE:-/tmp/mock_openai_endpoint.pid}" +TIMEOUT="${MOCK_SERVER_TIMEOUT:-30}" + +# Prefer the project's venv if it has the deps; otherwise fall back to system python3. +# Callers can override with PYTHON_BIN=/path/to/python. +REPO_ROOT="$(cd "${SCRIPT_DIR}/../.." && pwd)" +if [[ -z "${PYTHON_BIN:-}" ]]; then + if [[ -x "${REPO_ROOT}/.venv/bin/python" ]]; then + PYTHON_BIN="${REPO_ROOT}/.venv/bin/python" + else + PYTHON_BIN="python3" + fi +fi + +if [[ "${1:-}" == "--background" ]]; then + PORT="${PORT}" nohup "${PYTHON_BIN}" "${APP_DIR}/main.py" >"${LOG_FILE}" 2>&1 & + PID=$! + echo "${PID}" >"${PID_FILE}" + echo "Started mock server: pid=${PID} port=${PORT} log=${LOG_FILE}" + + # Wait until the server responds (or until we exhaust the timeout). + for _ in $(seq 1 "${TIMEOUT}"); do + if curl -fsS -o /dev/null \ + -X POST "http://127.0.0.1:${PORT}/chat/completions" \ + -H 'Authorization: Bearer sk-test' \ + -H 'Content-Type: application/json' \ + -d '{"model":"gpt-4o-mini","messages":[{"role":"user","content":"hi"}]}'; then + echo "Mock server is ready on http://127.0.0.1:${PORT}" + exit 0 + fi + sleep 1 + done + + echo "Mock server failed to become ready within ${TIMEOUT}s. Logs:" >&2 + tail -n 100 "${LOG_FILE}" >&2 || true + kill "${PID}" 2>/dev/null || true + exit 1 +fi + +exec env PORT="${PORT}" "${PYTHON_BIN}" "${APP_DIR}/main.py" diff --git a/tests/test_litellm/mock_endpoints/__init__.py b/tests/test_litellm/mock_endpoints/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/tests/test_litellm/mock_endpoints/test_mock_openai_endpoint_server.py b/tests/test_litellm/mock_endpoints/test_mock_openai_endpoint_server.py new file mode 100644 index 00000000000..0695cfe526f --- /dev/null +++ b/tests/test_litellm/mock_endpoints/test_mock_openai_endpoint_server.py @@ -0,0 +1,73 @@ +"""End-to-end smoke tests for the vendored mock OpenAI endpoint. + +These tests boot the mock server in-process (via the session-scoped fixture +in ``tests/mock_endpoints/conftest.py``) and verify that the key endpoints +that production tests rely on actually respond correctly. + +If this test passes in CI we know the local mock works as a drop-in +replacement for the Railway deployment for chat-completions, embeddings, and +fine-tuning routes — which is the whole point of vendoring it. +""" + +from __future__ import annotations + +import json +import urllib.request + +import pytest + +pytest_plugins = ("tests.mock_endpoints.conftest",) + + +def _post_json(url: str, payload: dict) -> dict: + req = urllib.request.Request( + url, + data=json.dumps(payload).encode("utf-8"), + headers={ + "Authorization": "Bearer sk-test", + "Content-Type": "application/json", + }, + method="POST", + ) + with urllib.request.urlopen(req, timeout=10) as resp: + return json.loads(resp.read().decode("utf-8")) + + +def _get_json(url: str) -> dict: + req = urllib.request.Request(url, method="GET") + with urllib.request.urlopen(req, timeout=10) as resp: + return json.loads(resp.read().decode("utf-8")) + + +@pytest.mark.usefixtures("mock_openai_endpoint_server") +def test_should_serve_openai_chat_completion(mock_openai_endpoint_server: str) -> None: + body = _post_json( + f"{mock_openai_endpoint_server}/chat/completions", + { + "model": "gpt-4o-mini", + "messages": [{"role": "user", "content": "hi"}], + }, + ) + + assert body["object"] == "chat.completion" + assert body["model"] == "gpt-4o-mini" + assert body["choices"][0]["message"]["role"] == "assistant" + assert body["choices"][0]["message"]["content"] + + +def test_should_serve_embeddings(mock_openai_endpoint_server: str) -> None: + body = _post_json( + f"{mock_openai_endpoint_server}/v1/embeddings", + {"model": "text-embedding-3-small", "input": "hello world"}, + ) + assert body["object"] == "list" + assert body["model"] == "text-embedding-3-small" + assert len(body["data"][0]["embedding"]) > 0 + + +def test_should_serve_fine_tuning_jobs_list( + mock_openai_endpoint_server: str, +) -> None: + body = _get_json(f"{mock_openai_endpoint_server}/openai/fine_tuning/jobs") + assert body["object"] == "list" + assert isinstance(body["data"], list) diff --git a/uv.lock b/uv.lock index cafb6664958..43e59b14c79 100644 --- a/uv.lock +++ b/uv.lock @@ -3187,6 +3187,20 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/3b/e4/b2f4ca7965ca373b491cdb4bc25cdb30c1649ca81a8782056a83850292a9/librt-0.11.0-cp313-cp313-win_arm64.whl", hash = "sha256:96f044bb325fd9cf1a723015638c219e9143f0dfbc0ca54c565df2b7fc748b44", size = 103448, upload-time = "2026-05-10T18:16:29.066Z" }, ] +[[package]] +name = "limits" +version = "5.8.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "deprecated" }, + { name = "packaging" }, + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/71/69/826a5d1f45426c68d8f6539f8d275c0e4fcaa57f0c017ec3100986558a41/limits-5.8.0.tar.gz", hash = "sha256:c9e0d74aed837e8f6f50d1fcebcf5fd8130957287206bc3799adaee5092655da", size = 226104, upload-time = "2026-02-05T07:17:35.859Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/b9/98/cb5ca20618d205a09d5bec7591fbc4130369c7e6308d9a676a28ff3ab22c/limits-5.8.0-py3-none-any.whl", hash = "sha256:ae1b008a43eb43073c3c579398bd4eb4c795de60952532dc24720ab45e1ac6b8", size = 60954, upload-time = "2026-02-05T07:17:34.425Z" }, +] + [[package]] name = "litellm" version = "1.87.0" @@ -3361,6 +3375,9 @@ healthcheck = [ { name = "httpx" }, { name = "pyyaml" }, ] +mock-server = [ + { name = "slowapi" }, +] proxy-dev = [ { name = "a2a-sdk" }, { name = "azure-identity" }, @@ -3521,6 +3538,7 @@ healthcheck = [ { name = "httpx", specifier = "==0.28.1" }, { name = "pyyaml", specifier = "==6.0.3" }, ] +mock-server = [{ name = "slowapi", specifier = "==0.1.9" }] proxy-dev = [ { name = "a2a-sdk", specifier = "==0.3.24" }, { name = "azure-identity", specifier = "==1.25.2" }, @@ -6925,6 +6943,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/e7/0e/3ae19fa941522cd98e119762e7181d371c8dba0b2d72bfaf9522692e329c/skops-0.14.0-py3-none-any.whl", hash = "sha256:60a5db78a9db46ccee2139a0ba13ab5afb1c96f4749b382e75a371291bbe3e36", size = 132198, upload-time = "2026-04-20T18:23:54.018Z" }, ] +[[package]] +name = "slowapi" +version = "0.1.9" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "limits" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/a0/99/adfc7f94ca024736f061257d39118e1542bade7a52e86415a4c4ae92d8ff/slowapi-0.1.9.tar.gz", hash = "sha256:639192d0f1ca01b1c6d95bf6c71d794c3a9ee189855337b4821f7f457dddad77", size = 14028, upload-time = "2024-02-05T12:11:52.13Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/2b/bb/f71c4b7d7e7eb3fc1e8c0458a8979b912f40b58002b9fbf37729b8cb464b/slowapi-0.1.9-py3-none-any.whl", hash = "sha256:cfad116cfb84ad9d763ee155c1e5c5cbf00b0d47399a769b227865f5df576e36", size = 14670, upload-time = "2024-02-05T12:11:50.898Z" }, +] + [[package]] name = "smmap" version = "5.0.3"