From 1d012a92ef6fc5724acfc30301a7ac5394978f91 Mon Sep 17 00:00:00 2001 From: alpha nerd Date: Mon, 6 Jul 2026 10:46:18 +0200 Subject: [PATCH] feat: transparent anthropic api incl. native anthropic api backend --- README.md | 41 ++++ api/messages.py | 329 ++++++++++++++++++++++++++ api/openai.py | 16 ++ backends/normalize.py | 14 +- backends/probe.py | 63 +++-- config.py | 6 + config.yaml | 8 + requests/anthropic.py | 519 ++++++++++++++++++++++++++++++++++++++++++ router.py | 9 + routing.py | 6 +- test/test_messages.py | 440 +++++++++++++++++++++++++++++++++++ 11 files changed, 1431 insertions(+), 20 deletions(-) create mode 100644 api/messages.py create mode 100644 requests/anthropic.py create mode 100644 test/test_messages.py diff --git a/README.md b/README.md index 52f7b99..b2cf971 100644 --- a/README.md +++ b/README.md @@ -167,6 +167,47 @@ multi-worker/replica deployment polling works via the shared DB, but `cancel` on running task in the worker that started it (other workers just mark the stored row cancelled). A background task interrupted by a server restart is reconciled to `failed` on the next startup. +## Anthropic Messages API + +NOMYO Router also exposes the Anthropic **Messages API**: + +``` +POST /v1/messages # create a message (stream or non-stream) +POST /v1/messages/count_tokens # count input tokens for a request +``` + +It works transparently across **all** backends. For Ollama / llama-server / llama-swap the router +translates Messages ⇄ Chat Completions in both directions (request, response, and streaming typed +SSE events — `message_start` → `content_block_*` → `message_delta` → `message_stop`), so clients get +a consistent `/v1/messages` surface regardless of backend. The API is stateless — there is no store, +background mode, or conversation persistence. + +### Native Anthropic upstream + +Configure real Anthropic endpoints under the `anthropic_endpoints` config key (base URL **without** +a `/v1` suffix). Requests routed to a model advertised by such an endpoint are **forwarded verbatim** +over the Anthropic wire format — the router injects the endpoint's `api_keys` entry as the `x-api-key` +header and pins `anthropic-version`, passing through the client's `anthropic-beta`. Their advertised +models are treated as always-loaded, like external OpenAI endpoints. + +```yaml +anthropic_endpoints: + - https://api.anthropic.com +api_keys: + "https://api.anthropic.com": "${ANTHROPIC_API_KEY}" +``` + +### Thinking + +An inbound `thinking` block is mapped to the backend's `reasoning_effort` (budget → `low`/`medium`/ +`high`); a backend that streams `reasoning_content` is surfaced back as Anthropic `thinking` content +blocks / `thinking_delta` events. On native endpoints, `thinking` passes through untouched. + +### Caching + +Set `nomyo: {"cache": true}` on the request body to consult the router's semantic LLM cache; a hit is +reflected via `usage.cache_read_input_tokens` (input tokens served from cache rather than re-processed). + ## Semantic LLM Cache NOMYO Router includes an optional semantic cache that serves repeated or semantically similar LLM requests from cache — no endpoint round-trip, no token cost, response in <10 ms. diff --git a/api/messages.py b/api/messages.py new file mode 100644 index 0000000..90dd945 --- /dev/null +++ b/api/messages.py @@ -0,0 +1,329 @@ +"""Anthropic **Messages API** routes (``/v1/messages`` and ``/v1/messages/count_tokens``). + +The router speaks Chat Completions to its local backends, so this layer: + + * **native** (configured ``anthropic_endpoints``): forwards the Anthropic request + verbatim over httpx with ``x-api-key`` / ``anthropic-version`` headers and streams + the upstream SSE straight back. + * **translated** (Ollama / llama-server / llama-swap): converts the request to chat, + reuses the resilient ``create_chat_with_retries`` ladder, and re-emits the result as + Anthropic typed SSE events (``requests/anthropic.py``). + +The Messages API is stateless, so — unlike ``/v1/responses`` — there is no store, +background mode, or DB. An optional ``nomyo.cache`` extension field reflects hits back +through the router's semantic LLM cache (a hit is reported via ``usage.cache_read_input_tokens``). +""" +import codecs + +import httpx +import orjson +from fastapi import APIRouter, HTTPException, Request +from starlette.responses import JSONResponse, StreamingResponse + +from cache import get_llm_cache +from config import get_config +from context_window import _count_message_tokens +from fingerprint import _conversation_fingerprint +from state import app_state, token_queue, default_headers +from backends.normalize import is_anthropic_endpoint +from backends.probe import ANTHROPIC_VERSION +from backends.sessions import _make_openai_client +from routing import choose_endpoint, decrement_usage +from api.openai import create_chat_with_retries +from requests.anthropic import ( + ChatToMessagesStream, + anthropic_messages_to_chat, + anthropic_to_chat_send_params, + build_message_object, + chat_message_to_content_blocks, + finish_reason_to_stop_reason, + message_object_to_sse, + new_message_id, + usage_chat_to_anthropic, +) + +router = APIRouter() + +CACHE_ROUTE = "anthropic_messages" + + +# --------------------------------------------------------------------------- +# helpers +# --------------------------------------------------------------------------- +def _anthropic_http_client(endpoint: str) -> httpx.AsyncClient: + """Return the warmed httpx client for a native Anthropic endpoint. + + Startup pre-creates one per configured endpoint; fall back to an on-demand + client (cached in app_state) for tests that skip the lifespan startup. + """ + client = app_state["httpx_clients"].get(endpoint) + if client is None: + client = httpx.AsyncClient(timeout=httpx.Timeout(300.0, connect=15.0)) + app_state["httpx_clients"][endpoint] = client + return client + + +def _native_headers(request: Request, api_key: str) -> dict: + """Build outbound headers for a native Anthropic forward. + + Injects the router's stored key as ``x-api-key`` and pins ``anthropic-version``, + passing through the client's ``anthropic-beta`` / ``anthropic-version`` when present. + """ + headers = { + "content-type": "application/json", + "x-api-key": api_key, + "anthropic-version": request.headers.get("anthropic-version", ANTHROPIC_VERSION), + } + beta = request.headers.get("anthropic-beta") + if beta: + headers["anthropic-beta"] = beta + return headers + + +async def _track(endpoint, tracking_model, prompt_tok, comp_tok): + if prompt_tok or comp_tok: + await token_queue.put((endpoint, tracking_model, prompt_tok, comp_tok)) + + +def _serve_cache_hit(cached: bytes, message_id: str, stream: bool): + """Serve a stored message object as a cache hit (input tokens → cache_read).""" + obj = orjson.loads(cached) + obj["id"] = message_id + u = obj.get("usage") or {} + read = u.get("input_tokens", 0) or 0 + obj["usage"] = { + **u, + "input_tokens": 0, + "cache_read_input_tokens": read, + "cache_creation_input_tokens": 0, + } + if stream: + async def _served(): + yield message_object_to_sse(obj) + return StreamingResponse(_served(), media_type="text/event-stream") + return JSONResponse(content=obj) + + +# --------------------------------------------------------------------------- +# POST /v1/messages +# --------------------------------------------------------------------------- +@router.post("/v1/messages") +async def anthropic_messages_proxy(request: Request): + config = get_config() + raw_body = await request.body() + try: + payload = orjson.loads(raw_body.decode("utf-8")) + except orjson.JSONDecodeError as e: + raise HTTPException(status_code=400, detail=f"Invalid JSON: {e}") from e + + model = payload.get("model") + messages = payload.get("messages") + system = payload.get("system") + stream = bool(payload.get("stream")) + _cache_enabled = payload.get("nomyo", {}).get("cache", False) + + if not model: + raise HTTPException(status_code=400, detail="Missing required field 'model'") + if messages is None: + raise HTTPException(status_code=400, detail="Missing required field 'messages'") + if payload.get("max_tokens") is None: + raise HTTPException(status_code=400, detail="Missing required field 'max_tokens'") + + if ":latest" in model: + model = model.split(":latest")[0] + + chat_messages = anthropic_messages_to_chat(system, messages) + message_id = new_message_id() + + # Cache lookup (foreground) — before endpoint selection, keyed on the chat form. + _cache = get_llm_cache() + if _cache is not None and _cache_enabled: + cached = await _cache.get_chat(CACHE_ROUTE, model, chat_messages) + if cached is not None: + return _serve_cache_hit(cached, message_id, stream) + + async def _cache_store(obj): + if _cache is None or not _cache_enabled or not obj.get("content"): + return + try: + await _cache.set_chat(CACHE_ROUTE, model, chat_messages, orjson.dumps(obj)) + except Exception as _ce: + print(f"[cache] set_chat ({CACHE_ROUTE}) failed: {_ce}") + + # Endpoint selection reserves a slot — released exactly once per branch. + _affinity_key = _conversation_fingerprint(model, chat_messages, None) + endpoint, tracking_model = await choose_endpoint(model, affinity_key=_affinity_key) + try: + native = is_anthropic_endpoint(endpoint) + if not native: + oclient = _make_openai_client(endpoint, default_headers=default_headers, + api_key=config.api_keys.get(endpoint, "no-key")) + send_params = anthropic_to_chat_send_params(payload, chat_messages, model) + except BaseException: + await decrement_usage(endpoint, tracking_model) + raise + + # ---- native passthrough ----------------------------------------------- + if native: + return await _handle_native( + request, payload, endpoint, tracking_model, stream, + api_key=config.api_keys.get(endpoint, "no-key"), cache_store=_cache_store) + + # ---- translated streaming --------------------------------------------- + if stream: + try: + source = await create_chat_with_retries( + oclient, {**send_params, "stream": True, + "stream_options": {"include_usage": True}}, + endpoint, model, tracking_model) + except BaseException: + await decrement_usage(endpoint, tracking_model) + raise + translator = ChatToMessagesStream(message_id, model) + + async def _stream(): + try: + async for sse in translator.events(source): + yield sse + prompt = (translator.usage or {}).get("prompt_tokens", 0) + comp = (translator.usage or {}).get("completion_tokens", 0) + await _track(endpoint, tracking_model, prompt, comp) + obj = build_message_object( + message_id=message_id, model=model, + content_blocks=translator.content_blocks, + stop_reason=translator.stop_reason, + usage=usage_chat_to_anthropic(translator.usage)) + await _cache_store(obj) + finally: + await decrement_usage(endpoint, tracking_model) + + return StreamingResponse(_stream(), media_type="text/event-stream") + + # ---- translated non-streaming ----------------------------------------- + try: + result = await create_chat_with_retries( + oclient, {**send_params, "stream": False}, endpoint, model, tracking_model) + message = result.choices[0].message.model_dump() if result.choices else {} + usage = result.usage.model_dump() if result.usage is not None else None + content_blocks = chat_message_to_content_blocks(message) + finish_reason = getattr(result.choices[0], "finish_reason", None) if result.choices else None + has_tool_use = any(b.get("type") == "tool_use" for b in content_blocks) + stop_reason = finish_reason_to_stop_reason(finish_reason, has_tool_use=has_tool_use) + await _track(endpoint, tracking_model, + (usage or {}).get("prompt_tokens", 0), + (usage or {}).get("completion_tokens", 0)) + finally: + await decrement_usage(endpoint, tracking_model) + + obj = build_message_object( + message_id=message_id, model=model, content_blocks=content_blocks, + stop_reason=stop_reason, usage=usage_chat_to_anthropic(usage)) + await _cache_store(obj) + return JSONResponse(content=obj) + + +async def _handle_native(request, payload, endpoint, tracking_model, stream, + *, api_key, cache_store): + """Forward an Anthropic request verbatim to a native upstream.""" + client = _anthropic_http_client(endpoint) + headers = _native_headers(request, api_key) + forward = {k: v for k, v in payload.items() if k != "nomyo"} + url = f"{endpoint.rstrip('/')}/v1/messages" + + if not stream: + forward["stream"] = False + try: + resp = await client.post(url, headers=headers, json=forward) + except BaseException: + await decrement_usage(endpoint, tracking_model) + raise + try: + data = resp.json() + except Exception: + data = {"type": "error", "error": {"message": resp.text[:500]}} + if resp.status_code == 200 and isinstance(data, dict): + u = data.get("usage") or {} + await _track(endpoint, tracking_model, + u.get("input_tokens", 0) or 0, u.get("output_tokens", 0) or 0) + await cache_store(data) + await decrement_usage(endpoint, tracking_model) + return JSONResponse(content=data, status_code=resp.status_code) + + forward["stream"] = True + + async def _proxy_stream(): + decoder = codecs.getincrementaldecoder("utf-8")() + buffer = "" + input_tok = 0 + output_tok = 0 + try: + async with client.stream("POST", url, headers=headers, json=forward) as resp: + async for raw in resp.aiter_bytes(): + if not raw: + continue + yield raw + # Parse a light copy to capture usage for token tracking. + buffer += decoder.decode(raw) + while "\n" in buffer: + line, buffer = buffer.split("\n", 1) + line = line.strip() + if not line.startswith("data:"): + continue + payload_str = line[len("data:"):].strip() + if not payload_str or payload_str == "[DONE]": + continue + try: + evt = orjson.loads(payload_str) + except orjson.JSONDecodeError: + continue + if evt.get("type") == "message_start": + u = (evt.get("message") or {}).get("usage") or {} + input_tok = u.get("input_tokens", 0) or input_tok + elif evt.get("type") == "message_delta": + u = evt.get("usage") or {} + output_tok = u.get("output_tokens", 0) or output_tok + finally: + await _track(endpoint, tracking_model, input_tok, output_tok) + await decrement_usage(endpoint, tracking_model) + + return StreamingResponse(_proxy_stream(), media_type="text/event-stream") + + +# --------------------------------------------------------------------------- +# POST /v1/messages/count_tokens +# --------------------------------------------------------------------------- +@router.post("/v1/messages/count_tokens") +async def anthropic_count_tokens(request: Request): + config = get_config() + try: + payload = orjson.loads((await request.body()).decode("utf-8")) + except orjson.JSONDecodeError as e: + raise HTTPException(status_code=400, detail=f"Invalid JSON: {e}") from e + + model = payload.get("model") + messages = payload.get("messages") + if not model: + raise HTTPException(status_code=400, detail="Missing required field 'model'") + if messages is None: + raise HTTPException(status_code=400, detail="Missing required field 'messages'") + if ":latest" in model: + model = model.split(":latest")[0] + + chat_messages = anthropic_messages_to_chat(payload.get("system"), messages) + + # No slot reservation — this is a metadata call, not a completion. + endpoint, _tracking = await choose_endpoint(model, reserve=False) + + if is_anthropic_endpoint(endpoint): + client = _anthropic_http_client(endpoint) + headers = _native_headers(request, config.api_keys.get(endpoint, "no-key")) + forward = {k: v for k, v in payload.items() if k != "nomyo"} + try: + resp = await client.post( + f"{endpoint.rstrip('/')}/v1/messages/count_tokens", + headers=headers, json=forward) + return JSONResponse(content=resp.json(), status_code=resp.status_code) + except Exception as e: + raise HTTPException(status_code=502, detail=f"count_tokens upstream failed: {e}") from e + + return JSONResponse(content={"input_tokens": _count_message_tokens(chat_messages)}) diff --git a/api/openai.py b/api/openai.py index dc5432c..9b64fe9 100644 --- a/api/openai.py +++ b/api/openai.py @@ -665,10 +665,16 @@ async def openai_models_proxy(request: Request): fetch.endpoint_details(ep, "/models", "data", config.api_keys.get(ep), skip_error_cache=True, timeout=8) for ep in all_llama_endpoints ] + # 4. Query native Anthropic endpoints via /v1/models (auth headers picked by endpoint type) + anthropic_tasks = [ + fetch.endpoint_details(ep, "/v1/models", "data", config.api_keys.get(ep), skip_error_cache=True, timeout=8) + for ep in config.anthropic_endpoints + ] ollama_models = await asyncio.gather(*ollama_tasks) if ollama_tasks else [] ext_openai_models = await asyncio.gather(*ext_openai_tasks) if ext_openai_tasks else [] llama_models = await asyncio.gather(*llama_tasks) if llama_tasks else [] + anthropic_models = await asyncio.gather(*anthropic_tasks) if anthropic_tasks else [] models = {'data': []} @@ -702,6 +708,16 @@ async def openai_models_proxy(request: Request): model['name'] = model['id'] models['data'].append(model) + # Add native Anthropic models (if any) + if anthropic_models: + for modellist in anthropic_models: + for model in modellist: + if not "id" in model.keys(): + model['id'] = model.get('name', model.get('id', '')) + else: + model['name'] = model['id'] + models['data'].append(model) + # 2. Return a JSONResponse with a deduplicated list of unique models for inference return JSONResponse( content={"data": dedupe_on_keys(models['data'], ['name'])}, diff --git a/backends/normalize.py b/backends/normalize.py index 41fc199..eef57cc 100644 --- a/backends/normalize.py +++ b/backends/normalize.py @@ -70,6 +70,16 @@ def llama_endpoints(cfg) -> list: return list(dict.fromkeys([*cfg.llama_server_endpoints, *cfg.llama_swap_endpoints])) +def is_anthropic_endpoint(endpoint: str) -> bool: + """True if the endpoint is a configured native Anthropic Messages-API backend. + + These speak the Anthropic wire format (``x-api-key`` / ``anthropic-version`` + headers, ``/v1/messages``), not OpenAI Chat Completions, so requests routed to + them are forwarded verbatim rather than translated. + """ + return endpoint in get_config().anthropic_endpoints + + def is_ext_openai_endpoint(endpoint: str) -> bool: """ Determine if an endpoint is an external OpenAI-compatible endpoint (not Ollama, llama-server or llama-swap). @@ -117,8 +127,8 @@ def get_tracking_model(endpoint: str, model: str) -> str: This ensures consistent model naming across all routes for usage tracking. """ - # External OpenAI endpoints are not shown in PS, keep as-is - if is_ext_openai_endpoint(endpoint): + # External OpenAI / native Anthropic endpoints are not shown in PS, keep as-is + if is_ext_openai_endpoint(endpoint) or is_anthropic_endpoint(endpoint): return model # llama-server / llama-swap endpoints use normalized names in PS diff --git a/backends/probe.py b/backends/probe.py index 2e53f01..0d10da8 100644 --- a/backends/probe.py +++ b/backends/probe.py @@ -46,7 +46,32 @@ from backends.health import ( _format_connection_issue, _is_llama_model_loaded, ) -from backends.normalize import is_ext_openai_endpoint, is_openai_compatible, is_llama_server, is_llama_swap +from backends.normalize import ( + is_ext_openai_endpoint, + is_openai_compatible, + is_llama_server, + is_llama_swap, + is_anthropic_endpoint, +) + +# Anthropic Messages API version pinned on every native probe/proxy request. +ANTHROPIC_VERSION = "2023-06-01" + + +def _auth_headers(endpoint: str, api_key: Optional[str]) -> dict: + """Build outbound auth headers for a probe, honoring the backend's scheme. + + Native Anthropic endpoints authenticate with ``x-api-key`` + a pinned + ``anthropic-version`` header; everything else uses OpenAI-style Bearer auth. + """ + headers = {"Referer": default_headers.get("HTTP-Referer", "https://nomyo.ai")} + if is_anthropic_endpoint(endpoint): + if api_key is not None: + headers["x-api-key"] = api_key + headers["anthropic-version"] = ANTHROPIC_VERSION + elif api_key is not None: + headers["Authorization"] = "Bearer " + api_key + return headers class fetch: @@ -56,12 +81,13 @@ class fetch: This is called by available_models() after checking caches and in-flight requests. """ cfg = get_config() - headers = {"Referer": default_headers.get("HTTP-Referer", "https://nomyo.ai")} - if api_key is not None: - headers["Authorization"] = "Bearer " + api_key + headers = _auth_headers(endpoint, api_key) ep_base = endpoint.rstrip("/") - if is_llama_server(endpoint) and "/v1" not in endpoint: + if is_anthropic_endpoint(endpoint): + endpoint_url = f"{ep_base}/v1/models" + key = "data" + elif is_llama_server(endpoint) and "/v1" not in endpoint: endpoint_url = f"{ep_base}/v1/models" key = "data" elif "/v1" in endpoint or is_llama_server(endpoint): @@ -320,11 +346,12 @@ class fetch: Uses request coalescing to prevent cache stampede and stale-while-revalidate to serve requests immediately even when cache is stale (refreshing in background). """ - if is_ext_openai_endpoint(endpoint): - # External OpenAI-compatible backends (vLLM, OpenAI, Groq, …) keep - # every advertised model permanently resident — there is no - # /api/ps-style "loaded" subset to probe. Report the advertised set - # as the loaded set so choose_endpoint's "loaded & free" preference + if is_ext_openai_endpoint(endpoint) or is_anthropic_endpoint(endpoint): + # External OpenAI-compatible backends (vLLM, OpenAI, Groq, …) and + # native Anthropic endpoints keep every advertised model permanently + # resident — there is no /api/ps-style "loaded" subset to probe. + # Report the advertised set as the loaded set so choose_endpoint's + # "loaded & free" preference # treats them on par with Ollama/llama-server backends that have the # model hot, instead of relegating them to the free-slot fallback and # never using them when an Ollama box advertises the same model. @@ -396,9 +423,7 @@ class fetch: if _is_fresh(_available_error_cache[endpoint], 300): return [] - headers = {"Referer": default_headers.get("HTTP-Referer", "https://nomyo.ai")} - if api_key is not None: - headers["Authorization"] = "Bearer " + api_key + headers = _auth_headers(endpoint, api_key) request_url = f"{endpoint.rstrip('/')}/{route.lstrip('/')}" client: aiohttp.ClientSession = get_probe_session(endpoint) @@ -434,9 +459,7 @@ async def _raw_probe( (unlike `fetch.endpoint_details`, which returns [] on either). Returns `(ok, payload_or_error_message)`. """ - headers = {"Referer": default_headers.get("HTTP-Referer", "https://nomyo.ai")} - if api_key is not None: - headers["Authorization"] = "Bearer " + api_key + headers = _auth_headers(ep, api_key) url = f"{ep.rstrip('/')}/{route.lstrip('/')}" req_kwargs = {} if timeout is not None: @@ -459,6 +482,14 @@ async def _endpoint_health(ep: str, *, timeout: Optional[float] = None) -> dict: path (issue #83) is reported as `error` rather than `ok`. OpenAI-compatible endpoints use a single `/models` probe. """ + if is_anthropic_endpoint(ep): + ok, payload = await _raw_probe( + ep, "/v1/models", get_config().api_keys.get(ep), timeout=timeout, + ) + if ok: + return {"status": "ok", "version": "latest"} + return {"status": "error", "detail": str(payload)} + if is_openai_compatible(ep): ok, payload = await _raw_probe( ep, "/models", get_config().api_keys.get(ep), timeout=timeout, diff --git a/config.py b/config.py index 03d8e94..ea98ee5 100644 --- a/config.py +++ b/config.py @@ -27,6 +27,12 @@ class Config(BaseSettings): # workers). Same surface as llama_server_endpoints, but loaded models are read from # /running (not /v1/models status) and unload uses POST /api/models/unload/:model_id. llama_swap_endpoints: List[str] = Field(default_factory=list) + # List of native Anthropic Messages-API endpoints (e.g. https://api.anthropic.com). + # Configure the base URL WITHOUT a /v1 suffix; the router appends /v1/models and + # /v1/messages itself. Requests routed here are forwarded verbatim (no Messages⇄Chat + # translation); the endpoint's api_keys entry is sent as the x-api-key header. Their + # advertised models are treated as always-loaded, like external OpenAI endpoints. + anthropic_endpoints: List[str] = Field(default_factory=list) # Max concurrent connections per endpoint‑model pair, see OLLAMA_NUM_PARALLEL max_concurrent_connections: int = 1 # Per-endpoint overrides: {endpoint_url: {max_concurrent_connections: N}} diff --git a/config.yaml b/config.yaml index 51ebb1b..3e26597 100644 --- a/config.yaml +++ b/config.yaml @@ -16,6 +16,14 @@ llama_server_endpoints: llama_swap_endpoints: - http://192.168.0.52:8890/v1 +# Native Anthropic Messages-API endpoints (optional). Configure the base URL WITHOUT a +# /v1 suffix; the router appends /v1/models and /v1/messages itself. Requests routed to a +# model advertised here are forwarded verbatim (no Messages⇄Chat translation), with the +# matching api_keys entry sent as the x-api-key header. Advertised models are treated as +# always-loaded, like external OpenAI endpoints. +# anthropic_endpoints: +# - https://api.anthropic.com + # Maximum concurrent connections *per endpoint‑model pair* (equals to OLLAMA_NUM_PARALLEL) # This is the global default; individual endpoints can override it via endpoint_config below. max_concurrent_connections: 2 diff --git a/requests/anthropic.py b/requests/anthropic.py new file mode 100644 index 0000000..bef32b0 --- /dev/null +++ b/requests/anthropic.py @@ -0,0 +1,519 @@ +"""Translation between the Anthropic **Messages API** and **Chat Completions**. + +The router speaks Chat Completions to its local backends (Ollama, llama-server, +llama-swap). To expose ``/v1/messages`` transparently on top of that, this module +converts in both directions: + + * request: Anthropic ``system`` / ``messages`` / ``tools`` → chat ``messages`` / ``tools`` + * response: chat ``choices[0].message`` → Anthropic ``content`` blocks + * stream: chat completion deltas → Anthropic typed SSE events + +Pure functions / a stream-translator class — no I/O, mirroring ``requests/responses.py``. +The native passthrough path (configured ``anthropic_endpoints``) does not use this +module; it forwards the Anthropic wire format straight through. + +Reasoning: an inbound ``thinking`` block maps to the backend's ``reasoning_effort``; +a backend that streams ``reasoning_content`` is surfaced back as Anthropic +``thinking`` content blocks / ``thinking_delta`` events. +""" +import secrets + +import orjson + + +# --------------------------------------------------------------------------- +# Request direction: Anthropic Messages → Chat Completions +# --------------------------------------------------------------------------- +def _system_to_text(system): + """Flatten Anthropic ``system`` (string or text blocks) to a plain string.""" + if system is None: + return None + if isinstance(system, str): + return system + if isinstance(system, list): + parts = [] + for b in system: + if isinstance(b, dict) and b.get("type") == "text": + parts.append(b.get("text", "")) + elif isinstance(b, str): + parts.append(b) + return "\n\n".join(p for p in parts if p) or None + return None + + +def _image_block_to_chat(block): + """Convert an Anthropic ``image`` block to an OpenAI ``image_url`` part (or None).""" + src = block.get("source") or {} + stype = src.get("type") + if stype == "base64": + media = src.get("media_type", "image/png") + data = src.get("data", "") + return {"type": "image_url", "image_url": {"url": f"data:{media};base64,{data}"}} + if stype == "url" and src.get("url"): + return {"type": "image_url", "image_url": {"url": src["url"]}} + return None + + +def _tool_result_content_to_str(content): + """Flatten an Anthropic ``tool_result`` content (string or blocks) to a string.""" + if content is None: + return "" + if isinstance(content, str): + return content + if isinstance(content, list): + parts = [] + for b in content: + if isinstance(b, dict): + if b.get("type") == "text": + parts.append(b.get("text", "")) + elif b.get("type") == "image": + parts.append("[image]") + else: + parts.append(orjson.dumps(b).decode("utf-8")) + else: + parts.append(str(b)) + return "\n".join(parts) + return orjson.dumps(content).decode("utf-8") + + +def _user_content_to_chat(content): + """Split an Anthropic user ``content`` into ``(chat_content, tool_messages)``. + + ``tool_result`` blocks become standalone OpenAI ``role:"tool"`` messages; the + remaining text/image parts collapse to a chat ``content`` (string or list). + """ + tool_messages = [] + if content is None or isinstance(content, str): + return content, tool_messages + parts = [] + for b in content if isinstance(content, list) else []: + if not isinstance(b, dict): + parts.append({"type": "text", "text": str(b)}) + continue + btype = b.get("type") + if btype == "text": + parts.append({"type": "text", "text": b.get("text", "")}) + elif btype == "image": + img = _image_block_to_chat(b) + if img: + parts.append(img) + elif btype == "tool_result": + tool_messages.append({ + "role": "tool", + "tool_call_id": b.get("tool_use_id"), + "content": _tool_result_content_to_str(b.get("content")), + }) + # document / other blocks: no chat equivalent → skip + if not parts: + chat_content = None + elif len(parts) == 1 and parts[0].get("type") == "text": + chat_content = parts[0]["text"] + else: + chat_content = parts + return chat_content, tool_messages + + +def _assistant_content_to_chat(content): + """Convert an Anthropic assistant ``content`` to ``(text_or_parts, tool_calls)``.""" + if content is None or isinstance(content, str): + return content, None + text_parts = [] + tool_calls = [] + for b in content if isinstance(content, list) else []: + if not isinstance(b, dict): + continue + btype = b.get("type") + if btype == "text": + text_parts.append(b.get("text", "")) + elif btype == "tool_use": + tool_calls.append({ + "id": b.get("id"), + "type": "function", + "function": { + "name": b.get("name"), + "arguments": orjson.dumps(b.get("input") or {}).decode("utf-8"), + }, + }) + # thinking blocks in history have no chat equivalent → drop + text = "".join(text_parts) if text_parts else None + return text, (tool_calls or None) + + +def anthropic_messages_to_chat(system, messages): + """Build a Chat Completions ``messages`` list from Anthropic ``system`` + ``messages``.""" + chat = [] + sys_text = _system_to_text(system) + if sys_text: + chat.append({"role": "system", "content": sys_text}) + for m in messages or []: + role = m.get("role") + content = m.get("content") + if role == "user": + chat_content, tool_messages = _user_content_to_chat(content) + # Anthropic packs tool results into a user turn; OpenAI wants each as a + # separate tool message *before* any trailing user text. + chat.extend(tool_messages) + if chat_content is not None and chat_content != []: + chat.append({"role": "user", "content": chat_content}) + elif role == "assistant": + text, tool_calls = _assistant_content_to_chat(content) + msg = {"role": "assistant", "content": text} + if tool_calls: + msg["tool_calls"] = tool_calls + chat.append(msg) + else: + chat.append({"role": role or "user", "content": content}) + return chat + + +def tools_anthropic_to_chat(tools): + """Map Anthropic tool definitions → Chat Completions function tools.""" + if not tools: + return None + out = [] + for t in tools: + if not isinstance(t, dict): + continue + # Server tools (web_search, …) carry a ``type`` but no ``input_schema`` — + # local backends can't run them, so skip. + if "input_schema" not in t and t.get("type") not in (None, "custom"): + continue + fn = {"name": t.get("name")} + if t.get("description"): + fn["description"] = t["description"] + fn["parameters"] = t.get("input_schema") or {"type": "object", "properties": {}} + out.append({"type": "function", "function": fn}) + return out or None + + +def tool_choice_anthropic_to_chat(tool_choice): + """Map Anthropic ``tool_choice`` → Chat Completions ``tool_choice``.""" + if not isinstance(tool_choice, dict): + return None + ttype = tool_choice.get("type") + if ttype == "auto": + return "auto" + if ttype == "any": + return "required" + if ttype == "none": + return "none" + if ttype == "tool" and tool_choice.get("name"): + return {"type": "function", "function": {"name": tool_choice["name"]}} + return None + + +def _thinking_to_reasoning_effort(thinking): + """Map an Anthropic ``thinking`` config to an OpenAI ``reasoning_effort`` level.""" + if not isinstance(thinking, dict): + return None + if thinking.get("type") == "disabled": + return None + budget = thinking.get("budget_tokens") + if not isinstance(budget, int): + return "medium" # adaptive / enabled without an explicit budget + if budget < 2048: + return "low" + if budget < 8192: + return "medium" + return "high" + + +def anthropic_to_chat_send_params(payload, chat_messages, model): + """Assemble the Chat Completions request body from an Anthropic payload.""" + send = {"messages": chat_messages, "model": model} + if payload.get("max_tokens") is not None: + send["max_tokens"] = payload["max_tokens"] + opt = { + "temperature": payload.get("temperature"), + "top_p": payload.get("top_p"), + "stop": payload.get("stop_sequences"), + "tools": tools_anthropic_to_chat(payload.get("tools")), + "tool_choice": tool_choice_anthropic_to_chat(payload.get("tool_choice")), + "reasoning_effort": _thinking_to_reasoning_effort(payload.get("thinking")), + } + send.update({k: v for k, v in opt.items() if v is not None}) + return send + + +# --------------------------------------------------------------------------- +# Response direction: Chat Completions → Anthropic Messages +# --------------------------------------------------------------------------- +def new_message_id(): + return f"msg_{secrets.token_hex(24)}" + + +_STOP_REASON_MAP = { + "stop": "end_turn", + "length": "max_tokens", + "tool_calls": "tool_use", + "function_call": "tool_use", + "content_filter": "end_turn", +} + + +def finish_reason_to_stop_reason(finish_reason, has_tool_use=False): + if has_tool_use: + return "tool_use" + return _STOP_REASON_MAP.get(finish_reason, "end_turn") + + +def usage_chat_to_anthropic(usage, cache_read_tokens=0, cache_creation_tokens=0): + """Map chat usage → Anthropic usage, folding in nomyo-cache attribution.""" + prompt = (usage or {}).get("prompt_tokens") or 0 + completion = (usage or {}).get("completion_tokens") or 0 + return { + "input_tokens": prompt, + "output_tokens": completion, + "cache_creation_input_tokens": cache_creation_tokens, + "cache_read_input_tokens": cache_read_tokens, + } + + +def chat_message_to_content_blocks(message): + """Convert an assistant chat message (dict) into Anthropic content blocks. + + Ordering follows the Anthropic convention: thinking → text → tool_use. + """ + blocks = [] + reasoning = message.get("reasoning_content") or message.get("reasoning") + if reasoning: + blocks.append({"type": "thinking", "thinking": reasoning}) + content = message.get("content") + if content: + blocks.append({"type": "text", "text": content}) + for tc in message.get("tool_calls") or []: + fn = tc.get("function", {}) + try: + args = orjson.loads(fn.get("arguments") or "{}") + except (orjson.JSONDecodeError, TypeError): + args = {} + blocks.append({ + "type": "tool_use", + "id": tc.get("id") or f"toolu_{secrets.token_hex(12)}", + "name": fn.get("name"), + "input": args, + }) + return blocks + + +def build_message_object(*, message_id, model, content_blocks, stop_reason, + usage, stop_sequence=None): + """Assemble a full ``type:"message"`` body for a non-streaming reply.""" + return { + "id": message_id, + "type": "message", + "role": "assistant", + "model": model, + "content": content_blocks or [], + "stop_reason": stop_reason, + "stop_sequence": stop_sequence, + "usage": usage, + } + + +# --------------------------------------------------------------------------- +# Cache-hit replay: a finished message object → Anthropic SSE stream +# --------------------------------------------------------------------------- +def _sse(etype, payload): + body = {"type": etype, **payload} + return f"event: {etype}\ndata: {orjson.dumps(body).decode('utf-8')}\n\n".encode("utf-8") + + +def message_object_to_sse(msg): + """Render a *finished* message object as a valid Anthropic SSE event stream. + + Used to serve cache hits to streaming clients without a backend call. + """ + out = [] + start = {**msg, "content": [], "stop_reason": None, "stop_sequence": None, + "usage": {**(msg.get("usage") or {}), "output_tokens": 0}} + out.append(_sse("message_start", {"message": start})) + for i, block in enumerate(msg.get("content") or []): + btype = block.get("type") + if btype == "text": + out.append(_sse("content_block_start", { + "index": i, "content_block": {"type": "text", "text": ""}})) + out.append(_sse("content_block_delta", { + "index": i, "delta": {"type": "text_delta", "text": block.get("text", "")}})) + elif btype == "thinking": + out.append(_sse("content_block_start", { + "index": i, "content_block": {"type": "thinking", "thinking": ""}})) + out.append(_sse("content_block_delta", { + "index": i, + "delta": {"type": "thinking_delta", "thinking": block.get("thinking", "")}})) + elif btype == "tool_use": + out.append(_sse("content_block_start", { + "index": i, + "content_block": {"type": "tool_use", "id": block.get("id"), + "name": block.get("name"), "input": {}}})) + out.append(_sse("content_block_delta", { + "index": i, + "delta": {"type": "input_json_delta", + "partial_json": orjson.dumps(block.get("input") or {}).decode("utf-8")}})) + out.append(_sse("content_block_stop", {"index": i})) + out.append(_sse("message_delta", { + "delta": {"stop_reason": msg.get("stop_reason"), + "stop_sequence": msg.get("stop_sequence")}, + "usage": {"output_tokens": (msg.get("usage") or {}).get("output_tokens", 0)}})) + out.append(_sse("message_stop", {})) + return b"".join(out) + + +# --------------------------------------------------------------------------- +# Streaming direction: Chat Completions deltas → Anthropic typed SSE events +# --------------------------------------------------------------------------- +class ChatToMessagesStream: + """Translate a Chat Completions streaming generator into Anthropic events. + + Usage:: + + translator = ChatToMessagesStream(message_id, model) + async for sse_bytes in translator.events(chat_async_gen): + yield sse_bytes + # translator.content_blocks / usage / stop_reason now populated for storage + + Emits ``message_start`` → (``content_block_start`` → ``*_delta``* → + ``content_block_stop``)* → ``message_delta`` (stop_reason + usage) → + ``message_stop``. A single monotonically increasing ``index`` is assigned across + thinking / text / tool_use blocks. + """ + + def __init__(self, message_id, model, cache_read_tokens=0, cache_creation_tokens=0): + self.message_id = message_id + self.model = model + self.cache_read_tokens = cache_read_tokens + self.cache_creation_tokens = cache_creation_tokens + self.usage = None + self.stop_reason = "end_turn" + self.content_blocks = [] + + def _open_block(self, index, content_block): + return _sse("content_block_start", {"index": index, "content_block": content_block}) + + def _delta(self, index, delta): + return _sse("content_block_delta", {"index": index, "delta": delta}) + + def _stop(self, index): + return _sse("content_block_stop", {"index": index}) + + async def events(self, async_gen): + yield _sse("message_start", {"message": { + "id": self.message_id, "type": "message", "role": "assistant", + "model": self.model, "content": [], "stop_reason": None, + "stop_sequence": None, + "usage": {"input_tokens": 0, "output_tokens": 0, + "cache_creation_input_tokens": self.cache_creation_tokens, + "cache_read_input_tokens": self.cache_read_tokens}, + }}) + + next_index = 0 + # open block state: ("thinking"|"text", index) or None + open_kind = None + open_index = None + thinking_text = [] + text_parts = [] + finish_reason = None + # tool call state: chat tool-call index -> {index, id, name, args} + tc_state = {} + + def close_open(): + nonlocal open_kind, open_index + if open_kind is not None: + ev = self._stop(open_index) + open_kind, open_index = None, None + return ev + return None + + async for chunk in async_gen: + usage = getattr(chunk, "usage", None) + if usage is not None: + self.usage = { + "prompt_tokens": getattr(usage, "prompt_tokens", 0) or 0, + "completion_tokens": getattr(usage, "completion_tokens", 0) or 0, + } + choices = getattr(chunk, "choices", None) + if not choices: + continue + choice = choices[0] + if getattr(choice, "finish_reason", None): + finish_reason = choice.finish_reason + delta = choice.delta + + reasoning = getattr(delta, "reasoning_content", None) or getattr(delta, "reasoning", None) + if reasoning: + if open_kind != "thinking": + ev = close_open() + if ev: + yield ev + open_kind, open_index = "thinking", next_index + next_index += 1 + yield self._open_block(open_index, {"type": "thinking", "thinking": ""}) + thinking_text.append(reasoning) + yield self._delta(open_index, {"type": "thinking_delta", "thinking": reasoning}) + + content_piece = getattr(delta, "content", None) + if content_piece: + if open_kind != "text": + ev = close_open() + if ev: + yield ev + open_kind, open_index = "text", next_index + next_index += 1 + yield self._open_block(open_index, {"type": "text", "text": ""}) + text_parts.append(content_piece) + yield self._delta(open_index, {"type": "text_delta", "text": content_piece}) + + for tc in getattr(delta, "tool_calls", None) or []: + idx = tc.index + fn = getattr(tc, "function", None) + if idx not in tc_state: + ev = close_open() + if ev: + yield ev + block_index = next_index + next_index += 1 + state = { + "index": block_index, + "id": getattr(tc, "id", None) or f"toolu_{secrets.token_hex(12)}", + "name": (fn.name if fn else None), + "args": "", + } + tc_state[idx] = state + open_kind, open_index = "tool", block_index + yield self._open_block(block_index, { + "type": "tool_use", "id": state["id"], "name": state["name"], "input": {}}) + else: + state = tc_state[idx] + if getattr(tc, "id", None): + state["id"] = tc.id + if fn and fn.name: + state["name"] = fn.name + if fn and fn.arguments: + state["args"] += fn.arguments + yield self._delta(state["index"], { + "type": "input_json_delta", "partial_json": fn.arguments}) + + ev = close_open() + if ev: + yield ev + + # Assemble final content blocks (thinking → text → tool_use) for storage. + if thinking_text: + self.content_blocks.append({"type": "thinking", "thinking": "".join(thinking_text)}) + if text_parts: + self.content_blocks.append({"type": "text", "text": "".join(text_parts)}) + for idx in sorted(tc_state.keys()): + state = tc_state[idx] + try: + parsed = orjson.loads(state["args"]) if state["args"] else {} + except (orjson.JSONDecodeError, TypeError): + parsed = {} + self.content_blocks.append({ + "type": "tool_use", "id": state["id"], "name": state["name"], "input": parsed}) + + self.stop_reason = finish_reason_to_stop_reason(finish_reason, has_tool_use=bool(tc_state)) + out_tokens = (self.usage or {}).get("completion_tokens", 0) + yield _sse("message_delta", { + "delta": {"stop_reason": self.stop_reason, "stop_sequence": None}, + "usage": {"output_tokens": out_tokens}}) + yield _sse("message_stop", {}) diff --git a/router.py b/router.py index aca2d01..d6bb41f 100644 --- a/router.py +++ b/router.py @@ -293,6 +293,8 @@ from api.openai import router as openai_router app.include_router(openai_router) from api.responses import router as responses_router app.include_router(responses_router) +from api.messages import router as messages_router +app.include_router(messages_router) from api.ollama import router as ollama_router app.include_router(ollama_router) @@ -375,6 +377,13 @@ async def startup_event() -> None: if is_ext_openai_endpoint(ep): app_state["httpx_clients"][ep] = httpx.AsyncClient(timeout=30.0) + # Native Anthropic Messages-API endpoints are forwarded over httpx. Use a + # long read timeout so streamed completions aren't cut short. Closed on + # shutdown by the shared httpx_clients cleanup. + for ep in config.anthropic_endpoints: + app_state["httpx_clients"][ep] = httpx.AsyncClient( + timeout=httpx.Timeout(300.0, connect=15.0)) + # Create per-endpoint Unix socket sessions for .sock endpoints for ep in llama_endpoints(config): if _is_unix_socket_endpoint(ep): diff --git a/routing.py b/routing.py index 17a03bd..df2da84 100644 --- a/routing.py +++ b/routing.py @@ -33,6 +33,7 @@ from backends.normalize import ( is_ext_openai_endpoint, is_openai_compatible, is_llama_server, + is_anthropic_endpoint, llama_endpoints, get_tracking_model, ) @@ -97,7 +98,8 @@ async def choose_endpoint(model: str, reserve: bool = True, # 1️⃣ Gather advertised‑model sets for all endpoints concurrently # Include config.endpoints plus any llama-server / llama-swap endpoints llama_eps_extra = [ep for ep in llama_endpoints(config) if ep not in config.endpoints] - all_endpoints = config.endpoints + llama_eps_extra + anthropic_eps_extra = [ep for ep in config.anthropic_endpoints if ep not in config.endpoints] + all_endpoints = config.endpoints + llama_eps_extra + anthropic_eps_extra # Build the probe tasks in the SAME order as ``all_endpoints`` so the # gathered results stay aligned for the ``zip(all_endpoints, advertised_sets)`` @@ -107,7 +109,7 @@ async def choose_endpoint(model: str, reserve: bool = True, # before an Ollama one would inherit the Ollama model set and get a request # for a model it cannot serve, 404ing. See issue #128.) def _advertised_task(ep: str): - if is_openai_compatible(ep): + if is_openai_compatible(ep) or is_anthropic_endpoint(ep): return fetch.available_models(ep, config.api_keys.get(ep)) return fetch.available_models(ep) diff --git a/test/test_messages.py b/test/test_messages.py new file mode 100644 index 0000000..ab081c9 --- /dev/null +++ b/test/test_messages.py @@ -0,0 +1,440 @@ +"""Tests for the Anthropic Messages API support (api/messages.py + requests/anthropic.py). + +Covers the pure translation layer, the translated (Ollama-style) and native +(configured-anthropic-endpoint) backend paths, thinking→reasoning mapping, +streaming event shape, token counting, and the nomyo-cache reflection. +""" +from contextlib import ExitStack, contextmanager +from types import SimpleNamespace as NS +from unittest.mock import AsyncMock, MagicMock, patch + +import orjson +import pytest + +import router +from api import messages as api_messages +from requests import anthropic as at + + +# ────────────────────────────────────────────────────────────────────────────── +# Pure translation unit tests (no app / no I/O) +# ────────────────────────────────────────────────────────────────────────────── + +class TestRequestTranslation: + def test_system_string(self): + chat = at.anthropic_messages_to_chat("be brief", [{"role": "user", "content": "hi"}]) + assert chat[0] == {"role": "system", "content": "be brief"} + assert chat[1] == {"role": "user", "content": "hi"} + + def test_system_blocks_join(self): + chat = at.anthropic_messages_to_chat( + [{"type": "text", "text": "a"}, {"type": "text", "text": "b"}], + [{"role": "user", "content": "hi"}]) + assert chat[0] == {"role": "system", "content": "a\n\nb"} + + def test_image_base64_to_data_url(self): + chat = at.anthropic_messages_to_chat(None, [{ + "role": "user", "content": [ + {"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": "AAA"}}, + {"type": "text", "text": "what?"}, + ]}]) + parts = chat[0]["content"] + assert parts[0] == {"type": "image_url", "image_url": {"url": "data:image/png;base64,AAA"}} + assert parts[1] == {"type": "text", "text": "what?"} + + def test_tool_use_and_result_roundtrip(self): + chat = at.anthropic_messages_to_chat(None, [ + {"role": "assistant", "content": [ + {"type": "text", "text": "calling"}, + {"type": "tool_use", "id": "toolu_1", "name": "get", "input": {"x": 1}}]}, + {"role": "user", "content": [ + {"type": "tool_result", "tool_use_id": "toolu_1", "content": "42"}]}, + ]) + assert chat[0]["role"] == "assistant" + assert chat[0]["content"] == "calling" + assert chat[0]["tool_calls"][0]["id"] == "toolu_1" + assert chat[0]["tool_calls"][0]["function"]["arguments"] == '{"x":1}' + assert chat[1] == {"role": "tool", "tool_call_id": "toolu_1", "content": "42"} + + def test_tool_result_blocks_flattened(self): + chat = at.anthropic_messages_to_chat(None, [{ + "role": "user", "content": [ + {"type": "tool_result", "tool_use_id": "t1", + "content": [{"type": "text", "text": "line"}]}]}]) + assert chat[0] == {"role": "tool", "tool_call_id": "t1", "content": "line"} + + def test_tools_and_choice(self): + sp = at.anthropic_to_chat_send_params( + {"max_tokens": 10, "tools": [{"name": "get", "description": "d", + "input_schema": {"type": "object"}}], + "tool_choice": {"type": "tool", "name": "get"}}, + [], "m") + assert sp["tools"] == [{"type": "function", "function": { + "name": "get", "description": "d", "parameters": {"type": "object"}}}] + assert sp["tool_choice"] == {"type": "function", "function": {"name": "get"}} + + def test_tool_choice_variants(self): + assert at.tool_choice_anthropic_to_chat({"type": "auto"}) == "auto" + assert at.tool_choice_anthropic_to_chat({"type": "any"}) == "required" + assert at.tool_choice_anthropic_to_chat({"type": "none"}) == "none" + + def test_server_tools_dropped(self): + # web_search has a type but no input_schema → not runnable locally + assert at.tools_anthropic_to_chat([{"type": "web_search_20260209", "name": "web_search"}]) is None + + def test_stop_sequences_and_sampling(self): + sp = at.anthropic_to_chat_send_params( + {"max_tokens": 5, "stop_sequences": ["X"], "temperature": 0.2, "top_p": 0.9}, [], "m") + assert sp["stop"] == ["X"] + assert sp["temperature"] == 0.2 + assert sp["top_p"] == 0.9 + + def test_thinking_maps_to_reasoning_effort(self): + assert at.anthropic_to_chat_send_params( + {"max_tokens": 5, "thinking": {"type": "enabled", "budget_tokens": 1000}}, [], "m" + )["reasoning_effort"] == "low" + assert at.anthropic_to_chat_send_params( + {"max_tokens": 5, "thinking": {"type": "enabled", "budget_tokens": 4096}}, [], "m" + )["reasoning_effort"] == "medium" + assert at.anthropic_to_chat_send_params( + {"max_tokens": 5, "thinking": {"type": "enabled", "budget_tokens": 20000}}, [], "m" + )["reasoning_effort"] == "high" + assert "reasoning_effort" not in at.anthropic_to_chat_send_params( + {"max_tokens": 5, "thinking": {"type": "disabled"}}, [], "m") + + +class TestResponseTranslation: + def test_content_blocks_order(self): + blocks = at.chat_message_to_content_blocks({ + "role": "assistant", "reasoning_content": "hmm", "content": "answer", + "tool_calls": [{"id": "c1", "function": {"name": "f", "arguments": '{"a":1}'}}]}) + assert [b["type"] for b in blocks] == ["thinking", "text", "tool_use"] + assert blocks[0]["thinking"] == "hmm" + assert blocks[1]["text"] == "answer" + assert blocks[2] == {"type": "tool_use", "id": "c1", "name": "f", "input": {"a": 1}} + + def test_stop_reason_mapping(self): + assert at.finish_reason_to_stop_reason("stop") == "end_turn" + assert at.finish_reason_to_stop_reason("length") == "max_tokens" + assert at.finish_reason_to_stop_reason("tool_calls") == "tool_use" + assert at.finish_reason_to_stop_reason("stop", has_tool_use=True) == "tool_use" + + def test_usage_mapping(self): + u = at.usage_chat_to_anthropic({"prompt_tokens": 7, "completion_tokens": 3}, + cache_read_tokens=5) + assert u == {"input_tokens": 7, "output_tokens": 3, + "cache_creation_input_tokens": 0, "cache_read_input_tokens": 5} + + +# ────────────────────────────────────────────────────────────────────────────── +# Streaming translator (pure) +# ────────────────────────────────────────────────────────────────────────────── + +def _chunk(content=None, tool_calls=None, reasoning=None, finish_reason=None): + delta = NS(content=content, tool_calls=tool_calls) + if reasoning is not None: + delta.reasoning_content = reasoning + return NS(choices=[NS(delta=delta, finish_reason=finish_reason)], usage=None) + + +def _usage_chunk(p, c): + return NS(choices=[], usage=NS(prompt_tokens=p, completion_tokens=c)) + + +async def _collect(translator, gen): + frames = [] + async for sse in translator.events(gen): + frames.append(sse.decode()) + return _parse_sse("".join(frames)) + + +def _parse_sse(text): + out = [] + for frame in text.strip().split("\n\n"): + if not frame.strip(): + continue + etype = data = None + for line in frame.splitlines(): + if line.startswith("event: "): + etype = line[len("event: "):] + elif line.startswith("data: "): + data = orjson.loads(line[len("data: "):]) + out.append((etype, data)) + return out + + +class TestStreamTranslator: + async def test_text_stream(self): + async def gen(): + yield _chunk(content="Hel") + yield _chunk(content="lo", finish_reason="stop") + yield _usage_chunk(3, 5) + tr = at.ChatToMessagesStream("msg_1", "m") + events = await _collect(tr, gen()) + types = [e[0] for e in events] + assert types[0] == "message_start" + assert types[-1] == "message_stop" + assert "content_block_start" in types and "content_block_stop" in types + text = "".join(d["delta"]["text"] for t, d in events + if t == "content_block_delta" and d["delta"]["type"] == "text_delta") + assert text == "Hello" + md = [d for t, d in events if t == "message_delta"][0] + assert md["delta"]["stop_reason"] == "end_turn" + assert md["usage"]["output_tokens"] == 5 + assert tr.content_blocks == [{"type": "text", "text": "Hello"}] + + async def test_thinking_then_text(self): + async def gen(): + yield _chunk(reasoning="think ") + yield _chunk(reasoning="more") + yield _chunk(content="answer", finish_reason="stop") + tr = at.ChatToMessagesStream("msg_1", "m") + events = await _collect(tr, gen()) + # thinking block (index 0) then text block (index 1) + starts = [(d["index"], d["content_block"]["type"]) + for t, d in events if t == "content_block_start"] + assert starts == [(0, "thinking"), (1, "text")] + think = "".join(d["delta"]["thinking"] for t, d in events + if t == "content_block_delta" and d["delta"]["type"] == "thinking_delta") + assert think == "think more" + assert tr.content_blocks[0] == {"type": "thinking", "thinking": "think more"} + + async def test_tool_call_stream(self): + tc0 = NS(index=0, id="call_1", function=NS(name="lookup", arguments='{"q":')) + tc1 = NS(index=0, id=None, function=NS(name=None, arguments='"hi"}')) + + async def gen(): + yield _chunk(tool_calls=[tc0]) + yield _chunk(tool_calls=[tc1], finish_reason="tool_calls") + yield _usage_chunk(4, 2) + tr = at.ChatToMessagesStream("msg_1", "m") + events = await _collect(tr, gen()) + assert any(t == "content_block_start" and d["content_block"]["type"] == "tool_use" + for t, d in events) + partial = "".join(d["delta"]["partial_json"] for t, d in events + if t == "content_block_delta" and d["delta"]["type"] == "input_json_delta") + assert partial == '{"q":"hi"}' + md = [d for t, d in events if t == "message_delta"][0] + assert md["delta"]["stop_reason"] == "tool_use" + assert tr.content_blocks[-1] == {"type": "tool_use", "id": "call_1", + "name": "lookup", "input": {"q": "hi"}} + + +# ────────────────────────────────────────────────────────────────────────────── +# Cache-hit replay +# ────────────────────────────────────────────────────────────────────────────── + +class TestCacheReplay: + def test_message_object_to_sse_roundtrip(self): + msg = at.build_message_object( + message_id="msg_1", model="m", + content_blocks=[{"type": "text", "text": "hi"}], + stop_reason="end_turn", + usage=at.usage_chat_to_anthropic({"prompt_tokens": 2, "completion_tokens": 1})) + events = _parse_sse(at.message_object_to_sse(msg).decode()) + types = [e[0] for e in events] + assert types[0] == "message_start" + assert types[-1] == "message_stop" + text = "".join(d["delta"]["text"] for t, d in events + if t == "content_block_delta" and d["delta"]["type"] == "text_delta") + assert text == "hi" + + +# ────────────────────────────────────────────────────────────────────────────── +# Route-level tests +# ────────────────────────────────────────────────────────────────────────────── + +@contextmanager +def _enter(*cms): + with ExitStack() as stack: + for cm in cms: + stack.enter_context(cm) + yield + + +def _fake_completion(content="hello", usage=(3, 5), reasoning=None, tool_calls=None, + finish_reason="stop"): + md = {"role": "assistant", "content": content} + if reasoning is not None: + md["reasoning_content"] = reasoning + if tool_calls is not None: + md["tool_calls"] = tool_calls + msg = MagicMock() + msg.model_dump.return_value = md + usage_obj = MagicMock() + usage_obj.model_dump.return_value = { + "prompt_tokens": usage[0], "completion_tokens": usage[1], "total_tokens": sum(usage)} + return NS(choices=[NS(message=msg, finish_reason=finish_reason)], usage=usage_obj) + + +def _patch_backend(native=False, endpoint="http://ollama:11434", cache=None): + return ( + patch.object(api_messages, "choose_endpoint", + AsyncMock(return_value=(endpoint, "test-model:latest"))), + patch.object(api_messages, "decrement_usage", AsyncMock()), + patch.object(api_messages, "is_anthropic_endpoint", return_value=native), + patch.object(api_messages, "_make_openai_client", return_value=MagicMock()), + patch.object(api_messages, "get_llm_cache", return_value=cache), + ) + + +class TestTranslatedRoute: + async def test_nonstream(self, client): + with _enter(*_patch_backend(native=False), + patch.object(api_messages, "create_chat_with_retries", + AsyncMock(return_value=_fake_completion("hello world")))): + resp = await client.post("/v1/messages", + json={"model": "test-model", "max_tokens": 100, + "messages": [{"role": "user", "content": "hi"}]}) + assert resp.status_code == 200 + body = resp.json() + assert body["type"] == "message" + assert body["role"] == "assistant" + assert body["content"] == [{"type": "text", "text": "hello world"}] + assert body["stop_reason"] == "end_turn" + assert body["usage"]["input_tokens"] == 3 and body["usage"]["output_tokens"] == 5 + assert body["id"].startswith("msg_") + + async def test_missing_max_tokens_400(self, client): + with _enter(*_patch_backend(native=False)): + resp = await client.post("/v1/messages", + json={"model": "m", "messages": [{"role": "user", "content": "hi"}]}) + assert resp.status_code == 400 + + async def test_nonstream_tool_use(self, client): + tc = [{"id": "c1", "function": {"name": "get", "arguments": '{"a":1}'}}] + with _enter(*_patch_backend(native=False), + patch.object(api_messages, "create_chat_with_retries", + AsyncMock(return_value=_fake_completion( + content=None, tool_calls=tc, finish_reason="tool_calls")))): + resp = await client.post("/v1/messages", + json={"model": "m", "max_tokens": 50, + "messages": [{"role": "user", "content": "call it"}]}) + body = resp.json() + assert body["stop_reason"] == "tool_use" + assert body["content"][0] == {"type": "tool_use", "id": "c1", "name": "get", "input": {"a": 1}} + + async def test_stream_event_sequence(self, client): + async def _text(): + yield _chunk(content="Hi", finish_reason="stop") + yield _usage_chunk(3, 2) + with _enter(*_patch_backend(native=False), + patch.object(api_messages, "create_chat_with_retries", + AsyncMock(return_value=_text()))): + resp = await client.post("/v1/messages", + json={"model": "m", "max_tokens": 50, "stream": True, + "messages": [{"role": "user", "content": "hi"}]}) + assert resp.headers["content-type"].startswith("text/event-stream") + events = _parse_sse(resp.content.decode()) + types = [e[0] for e in events] + assert types[0] == "message_start" and types[-1] == "message_stop" + text = "".join(d["delta"]["text"] for t, d in events + if t == "content_block_delta" and d["delta"]["type"] == "text_delta") + assert text == "Hi" + + async def test_thinking_passed_as_reasoning_effort(self, client): + captured = {} + + async def _spy(oclient, send_params, endpoint, model, tracking_model): + captured.update(send_params) + return _fake_completion("ok") + + with _enter(*_patch_backend(native=False), + patch.object(api_messages, "create_chat_with_retries", _spy)): + await client.post("/v1/messages", + json={"model": "m", "max_tokens": 50, + "thinking": {"type": "enabled", "budget_tokens": 20000}, + "messages": [{"role": "user", "content": "hi"}]}) + assert captured["reasoning_effort"] == "high" + + +class TestCacheRoute: + async def test_hit_reports_cache_read(self, client): + stored = at.build_message_object( + message_id="msg_old", model="m", + content_blocks=[{"type": "text", "text": "cached"}], stop_reason="end_turn", + usage=at.usage_chat_to_anthropic({"prompt_tokens": 9, "completion_tokens": 4})) + fake_cache = MagicMock() + fake_cache.get_chat = AsyncMock(return_value=orjson.dumps(stored)) + with _enter(*_patch_backend(native=False, cache=fake_cache)): + resp = await client.post("/v1/messages", + json={"model": "m", "max_tokens": 50, "nomyo": {"cache": True}, + "messages": [{"role": "user", "content": "hi"}]}) + body = resp.json() + assert body["content"] == [{"type": "text", "text": "cached"}] + assert body["usage"]["cache_read_input_tokens"] == 9 + assert body["usage"]["input_tokens"] == 0 + assert body["id"].startswith("msg_") and body["id"] != "msg_old" + + +class TestNativeRoute: + def _fake_client(self, *, post_return=None, stream_frames=None, status=200): + client = MagicMock() + if post_return is not None: + resp = MagicMock() + resp.status_code = status + resp.json.return_value = post_return + client.post = AsyncMock(return_value=resp) + if stream_frames is not None: + class _Stream: + async def __aenter__(self_): + return self_ + async def __aexit__(self_, *a): + return False + async def aiter_bytes(self_): + for f in stream_frames: + yield f + client.stream = MagicMock(return_value=_Stream()) + return client + + async def test_nonstream_passthrough(self, client): + upstream = {"id": "msg_upstream", "type": "message", "role": "assistant", + "content": [{"type": "text", "text": "native hi"}], + "stop_reason": "end_turn", + "usage": {"input_tokens": 2, "output_tokens": 3}} + fake = self._fake_client(post_return=upstream) + with _enter(*_patch_backend(native=True, endpoint="https://api.anthropic.com"), + patch.object(api_messages, "_anthropic_http_client", return_value=fake)): + resp = await client.post("/v1/messages", + json={"model": "claude-x", "max_tokens": 50, + "messages": [{"role": "user", "content": "hi"}]}) + assert resp.status_code == 200 + assert resp.json()["content"][0]["text"] == "native hi" + # request forwarded verbatim with stream disabled, nomyo stripped + sent = fake.post.call_args.kwargs["json"] + assert sent["stream"] is False and "nomyo" not in sent + headers = fake.post.call_args.kwargs["headers"] + assert headers["x-api-key"] and headers["anthropic-version"] + + async def test_stream_passthrough(self, client): + frames = [ + b'event: message_start\ndata: {"type":"message_start","message":{"usage":{"input_tokens":5}}}\n\n', + b'event: message_delta\ndata: {"type":"message_delta","usage":{"output_tokens":7}}\n\n', + b'event: message_stop\ndata: {"type":"message_stop"}\n\n', + ] + fake = self._fake_client(stream_frames=frames) + tracked = [] + with _enter(*_patch_backend(native=True, endpoint="https://api.anthropic.com"), + patch.object(api_messages, "_anthropic_http_client", return_value=fake), + patch.object(api_messages, "_track", + AsyncMock(side_effect=lambda *a: tracked.append(a)))): + resp = await client.post("/v1/messages", + json={"model": "claude-x", "max_tokens": 50, "stream": True, + "messages": [{"role": "user", "content": "hi"}]}) + body = resp.content.decode() + assert "message_start" in body and "message_stop" in body + # usage parsed out of the proxied stream for token tracking + assert tracked and tracked[0][2] == 5 and tracked[0][3] == 7 + + +class TestCountTokens: + async def test_local_estimate(self, client): + with _enter(patch.object(api_messages, "choose_endpoint", + AsyncMock(return_value=("http://ollama:11434", "m"))), + patch.object(api_messages, "is_anthropic_endpoint", return_value=False)): + resp = await client.post("/v1/messages/count_tokens", + json={"model": "m", + "messages": [{"role": "user", "content": "count me"}]}) + assert resp.status_code == 200 + assert isinstance(resp.json()["input_tokens"], int) + assert resp.json()["input_tokens"] > 0