Merge pull request 'chore(deps): update dependency pillow to v12.3.0 [security]' (#142) from renovate/pypi-pillow-vulnerability into main

Reviewed-on: https://bitfreedom.net/code/code/nomyo-ai/nomyo-router/pulls/142

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.

api/messages.py

@@ -0,0 +1,330 @@
+"""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 {}
+    # Whole prompt served from cache: fold the stored uncached + read tokens into read.
+    read = (u.get("input_tokens", 0) or 0) + (u.get("cache_read_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)})

api/ollama.py

@@ -167,31 +167,38 @@ async def proxy(request: Request):
 
     _affinity_key = _conversation_fingerprint(model, None, prompt)
     endpoint, tracking_model = await choose_endpoint(model, affinity_key=_affinity_key)
-    use_openai = is_openai_compatible(endpoint)
+    # _guarded_stream's finally releases the reservation once we hand off; until
-    if use_openai:
+    # then any failure during request building / client construction (including
-        if ":latest" in model:
+    # CancelledError on client disconnect) must release it or the counter leaks.
-            model = model.split(":latest")
+    try:
-            model = model[0]
+        use_openai = is_openai_compatible(endpoint)
-        params = {
+        if use_openai:
-            "prompt": prompt,
+            if ":latest" in model:
-            "model": model,
+                model = model.split(":latest")
-        }
+                model = model[0]
-
+            params = {
-        optional_params = {
+                "prompt": prompt,
-            "stream": stream,
+                "model": model,
-            "max_tokens": options.get("num_predict") if options and "num_predict" in options else None,
-            "frequency_penalty": options.get("frequency_penalty") if options and "frequency_penalty" in options else None,
-            "presence_penalty": options.get("presence_penalty") if options and "presence_penalty" in options else None,
-            "seed": options.get("seed") if options and "seed" in options else None,
-            "stop": options.get("stop") if options and "stop" in options else None,
-            "top_p": options.get("top_p") if options and "top_p" in options else None,
-            "temperature": options.get("temperature") if options and "temperature" in options else None,
-            "suffix": suffix,
             }
-        params.update({k: v for k, v in optional_params.items() if v is not None})
+
-        oclient = _make_openai_client(endpoint, default_headers=default_headers, api_key=config.api_keys.get(endpoint, "no-key"))
+            optional_params = {
-    else:
+                "stream": stream,
-        client = get_ollama_client(endpoint)
+                "max_tokens": options.get("num_predict") if options and "num_predict" in options else None,
+                "frequency_penalty": options.get("frequency_penalty") if options and "frequency_penalty" in options else None,
+                "presence_penalty": options.get("presence_penalty") if options and "presence_penalty" in options else None,
+                "seed": options.get("seed") if options and "seed" in options else None,
+                "stop": options.get("stop") if options and "stop" in options else None,
+                "top_p": options.get("top_p") if options and "top_p" in options else None,
+                "temperature": options.get("temperature") if options and "temperature" in options else None,
+                "suffix": suffix,
+                }
+            params.update({k: v for k, v in optional_params.items() if v is not None})
+            oclient = _make_openai_client(endpoint, default_headers=default_headers, api_key=config.api_keys.get(endpoint, "no-key"))
+        else:
+            client = get_ollama_client(endpoint)
+    except BaseException:
+        await decrement_usage(endpoint, tracking_model)
+        raise
 
     # 4. Async generator body (error handling + cleanup handled by _guarded_stream)
     async def stream_generate_response():
@@ -336,59 +343,70 @@ async def chat_proxy(request: Request):
         opt = False
     _affinity_key = _conversation_fingerprint(model, messages, None)
     endpoint, tracking_model = await choose_endpoint(model, affinity_key=_affinity_key)
-    use_openai = is_openai_compatible(endpoint)
+    # Releasing the reservation is owned by _guarded_stream's finally once we hand
-    if use_openai:
+    # off to the streaming generator. Until then, any failure during request
-        if ":latest" in model:
+    # building / client construction (including CancelledError on client
-            model = model.split(":latest")
+    # disconnect) must release it here or the usage counter leaks.
-            model = model[0]
+    try:
-        if messages:
+        use_openai = is_openai_compatible(endpoint)
-            if any("images" in m for m in messages):
+        if use_openai:
-                messages = await asyncio.to_thread(transform_images_to_data_urls, messages)
+            if ":latest" in model:
-            messages = transform_tool_calls_to_openai(messages)
+                model = model.split(":latest")
-            messages = _strip_assistant_prefill(messages)
+                model = model[0]
-        params = {
+            if messages:
-            "messages": messages,
+                if any("images" in m for m in messages):
-            "model": model,
+                    messages = await asyncio.to_thread(transform_images_to_data_urls, messages)
-            }
+                messages = transform_tool_calls_to_openai(messages)
-        optional_params = {
+                messages = _strip_assistant_prefill(messages)
-            "tools": tools,
+            params = {
-            "stream": stream,
+                "messages": messages,
-            "stream_options": {"include_usage": True} if stream else None,
+                "model": model,
-            "max_tokens": options.get("num_predict") if options and "num_predict" in options else None,
+                }
-            "frequency_penalty": options.get("frequency_penalty") if options and "frequency_penalty" in options else None,
+            optional_params = {
-            "presence_penalty": options.get("presence_penalty") if options and "presence_penalty" in options else None,
+                "tools": tools,
-            "seed": options.get("seed") if options and "seed" in options else None,
+                "stream": stream,
-            "stop": options.get("stop") if options and "stop" in options else None,
+                "stream_options": {"include_usage": True} if stream else None,
-            "top_p": options.get("top_p") if options and "top_p" in options else None,
+                "max_tokens": options.get("num_predict") if options and "num_predict" in options else None,
-            "temperature": options.get("temperature") if options and "temperature" in options else None,
+                "frequency_penalty": options.get("frequency_penalty") if options and "frequency_penalty" in options else None,
-            "logprobs": logprobs if logprobs is not None else (options.get("logprobs") if options and "logprobs" in options else None),
+                "presence_penalty": options.get("presence_penalty") if options and "presence_penalty" in options else None,
-            "top_logprobs": top_logprobs if top_logprobs is not None else (options.get("top_logprobs") if options and "top_logprobs" in options else None),
+                "seed": options.get("seed") if options and "seed" in options else None,
-            "response_format": {"type": "json_schema", "json_schema": _format} if _format is not None else None
+                "stop": options.get("stop") if options and "stop" in options else None,
-            }
+                "top_p": options.get("top_p") if options and "top_p" in options else None,
-        params.update({k: v for k, v in optional_params.items() if v is not None})
+                "temperature": options.get("temperature") if options and "temperature" in options else None,
-        oclient = _make_openai_client(endpoint, default_headers=default_headers, api_key=config.api_keys.get(endpoint, "no-key"))
+                "logprobs": logprobs if logprobs is not None else (options.get("logprobs") if options and "logprobs" in options else None),
-    else:
+                "top_logprobs": top_logprobs if top_logprobs is not None else (options.get("top_logprobs") if options and "top_logprobs" in options else None),
-        client = get_ollama_client(endpoint)
+                "response_format": {"type": "json_schema", "json_schema": _format} if _format is not None else None
+                }
+            params.update({k: v for k, v in optional_params.items() if v is not None})
+            oclient = _make_openai_client(endpoint, default_headers=default_headers, api_key=config.api_keys.get(endpoint, "no-key"))
+        else:
+            client = get_ollama_client(endpoint)
+    except BaseException:
+        await decrement_usage(endpoint, tracking_model)
+        raise
     # For OpenAI endpoints: make the API call in handler scope
     # (try/except inside async generators is unreliable with Starlette's streaming)
     start_ts = None
     async_gen = None
     if use_openai:
         start_ts = time.perf_counter()
-        # Proactive trim: only for small-ctx models we've already seen run out of space
-        _lookup_model = _normalize_llama_model_name(model) if is_llama_server(endpoint) else model
-        _known_nctx = _endpoint_nctx.get((endpoint, _lookup_model))
-        if _known_nctx and _known_nctx <= _CTX_TRIM_SMALL_LIMIT:
-            _pre_target = int((_known_nctx - _known_nctx // 4) / 1.2)
-            _pre_est = _count_message_tokens(params.get("messages", []))
-            if _pre_est > _pre_target:
-                _pre_msgs = params.get("messages", [])
-                _pre_trimmed = _trim_messages_for_context(_pre_msgs, _known_nctx, target_tokens=_pre_target)
-                _dropped = len(_pre_msgs) - len(_pre_trimmed)
-                print(f"[ctx-pre] n_ctx={_known_nctx} est={_pre_est} target={_pre_target} dropped={_dropped}", flush=True)
-                params = {**params, "messages": _pre_trimmed}
         try:
+            # Proactive trim: only for small-ctx models we've already seen run out of space
+            _lookup_model = _normalize_llama_model_name(model) if is_llama_server(endpoint) else model
+            _known_nctx = _endpoint_nctx.get((endpoint, _lookup_model))
+            if _known_nctx and _known_nctx <= _CTX_TRIM_SMALL_LIMIT:
+                _pre_target = int((_known_nctx - _known_nctx // 4) / 1.2)
+                _pre_est = _count_message_tokens(params.get("messages", []))
+                if _pre_est > _pre_target:
+                    _pre_msgs = params.get("messages", [])
+                    _pre_trimmed = _trim_messages_for_context(_pre_msgs, _known_nctx, target_tokens=_pre_target)
+                    _dropped = len(_pre_msgs) - len(_pre_trimmed)
+                    print(f"[ctx-pre] n_ctx={_known_nctx} est={_pre_est} target={_pre_target} dropped={_dropped}", flush=True)
+                    params = {**params, "messages": _pre_trimmed}
             async_gen = await oclient.chat.completions.create(**params)
+        except asyncio.CancelledError:
+            await decrement_usage(endpoint, tracking_model)
+            raise
         except Exception as e:
             _e_str = str(e)
             print(f"[chat_proxy] caught {type(e).__name__}: {_e_str[:200]}")
@@ -595,14 +613,21 @@ async def _handle_embedding_request(
 
     # 2. Endpoint logic
     endpoint, tracking_model = await choose_endpoint(model)
-    use_openai = is_openai_compatible(endpoint)
+    # _guarded_stream's finally releases the reservation once we hand off; until
-    if use_openai:
+    # then any failure during client construction (including CancelledError on
-        if ":latest" in model:
+    # client disconnect) must release it or the counter leaks.
-            model = model.split(":latest")
+    try:
-            model = model[0]
+        use_openai = is_openai_compatible(endpoint)
-        client = _make_openai_client(endpoint, api_key=config.api_keys.get(endpoint, "no-key"))
+        if use_openai:
-    else:
+            if ":latest" in model:
-        client = get_ollama_client(endpoint)
+                model = model.split(":latest")
+                model = model[0]
+            client = _make_openai_client(endpoint, api_key=config.api_keys.get(endpoint, "no-key"))
+        else:
+            client = get_ollama_client(endpoint)
+    except BaseException:
+        await decrement_usage(endpoint, tracking_model)
+        raise
 
     # 3. Async generator body (error handling + cleanup handled by _guarded_stream)
     async def stream_embedding_response():

api/openai.py

@@ -61,8 +61,10 @@ async def create_chat_with_retries(oclient, send_params, endpoint, model, tracki
         request reroutes, then re-raise
       * ``image input is not supported`` → strip images and retry
 
-    On unrecoverable failure the endpoint usage counter is decremented and the
+    The caller owns the usage reservation taken by ``choose_endpoint``: this
-    exception is re-raised. Returns the established async generator / response.
+    function never decrements it. On unrecoverable failure the exception is
+    re-raised so the caller's guard releases the slot exactly once. Returns the
+    established async generator / response.
     """
     config = get_config()
     try:
@@ -74,12 +76,8 @@ async def create_chat_with_retries(oclient, send_params, endpoint, model, tracki
         if "does not support tools" in _e_str:
             # Model doesn't support tools — retry without them
             print(f"[ochat] retry: no tools", flush=True)
-            try:
+            params_without_tools = {k: v for k, v in send_params.items() if k != "tools"}
-                params_without_tools = {k: v for k, v in send_params.items() if k != "tools"}
+            async_gen = await oclient.chat.completions.create(**params_without_tools)
-                async_gen = await oclient.chat.completions.create(**params_without_tools)
-            except Exception:
-                await decrement_usage(endpoint, tracking_model)
-                raise
         elif _is_ctx_err:
             # Backend context limit hit — apply sliding-window trim (context-shift at message level)
             err_body = getattr(e, "body", {}) or {}
@@ -97,7 +95,6 @@ async def create_chat_with_retries(oclient, send_params, endpoint, model, tracki
                     actual_tokens = int(_m.group(1))
             print(f"[ctx-trim] n_ctx={n_ctx_limit} actual={actual_tokens}", flush=True)
             if not n_ctx_limit:
-                await decrement_usage(endpoint, tracking_model)
                 raise
             if n_ctx_limit <= _CTX_TRIM_SMALL_LIMIT:
                 _endpoint_nctx[(endpoint, model)] = n_ctx_limit
@@ -108,7 +105,6 @@ async def create_chat_with_retries(oclient, send_params, endpoint, model, tracki
                 trimmed_messages = _trim_messages_for_context(msgs_to_trim, n_ctx_limit, target_tokens=cal_target)
             except Exception as _helper_exc:
                 print(f"[ctx-trim] helper crash: {type(_helper_exc).__name__}: {str(_helper_exc)[:100]}", flush=True)
-                await decrement_usage(endpoint, tracking_model)
                 raise
             dropped = len(msgs_to_trim) - len(trimmed_messages)
             print(f"[ctx-trim] target={cal_target} dropped={dropped} remaining={len(trimmed_messages)} retrying-1", flush=True)
@@ -121,14 +117,9 @@ async def create_chat_with_retries(oclient, send_params, endpoint, model, tracki
                     # Still too large — tool definitions likely consuming too many tokens, strip them too
                     print(f"[ctx-trim] retry-1 still exceeded, stripping tools retrying-2", flush=True)
                     params_no_tools = {k: v for k, v in send_params.items() if k not in ("tools", "tool_choice")}
-                    try:
+                    async_gen = await oclient.chat.completions.create(**{**params_no_tools, "messages": trimmed_messages})
-                        async_gen = await oclient.chat.completions.create(**{**params_no_tools, "messages": trimmed_messages})
+                    print(f"[ctx-trim] retry-2 ok", flush=True)
-                        print(f"[ctx-trim] retry-2 ok", flush=True)
-                    except Exception:
-                        await decrement_usage(endpoint, tracking_model)
-                        raise
                 else:
-                    await decrement_usage(endpoint, tracking_model)
                     raise
         elif _is_backend_connection_error(e):
             # Upstream connection failed (e.g. llama-server in router mode
@@ -136,18 +127,12 @@ async def create_chat_with_retries(oclient, send_params, endpoint, model, tracki
             # next request reroutes; the client will retry this one.
             print(f"[ochat] backend connection error → marking ({endpoint}, {model}) unhealthy", flush=True)
             await _mark_backend_unhealthy(endpoint, model, _e_str)
-            await decrement_usage(endpoint, tracking_model)
             raise
         elif "image input is not supported" in _e_str:
             # Model doesn't support images — strip and retry
             print(f"[openai_chat_completions_proxy] Model {model} doesn't support images, retrying with text-only messages")
-            try:
+            async_gen = await oclient.chat.completions.create(**{**send_params, "messages": _strip_images_from_messages(send_params.get("messages", []))})
-                async_gen = await oclient.chat.completions.create(**{**send_params, "messages": _strip_images_from_messages(send_params.get("messages", []))})
-            except Exception:
-                await decrement_usage(endpoint, tracking_model)
-                raise
         else:
-            await decrement_usage(endpoint, tracking_model)
             raise
     return async_gen
 
@@ -195,13 +180,14 @@ async def openai_embedding_proxy(request: Request):
 
     # 2. Endpoint logic
     endpoint, tracking_model = await choose_endpoint(model)
-    if is_openai_compatible(endpoint):
+    # The finally below releases the reservation for every exit — success, error,
-        api_key = config.api_keys.get(endpoint, "no-key")
+    # or CancelledError — so client construction is kept inside the guarded block.
-    else:
-        api_key = "ollama"
-    oclient = _make_openai_client(endpoint, default_headers=default_headers, api_key=api_key)
-
     try:
+        if is_openai_compatible(endpoint):
+            api_key = config.api_keys.get(endpoint, "no-key")
+        else:
+            api_key = "ollama"
+        oclient = _make_openai_client(endpoint, default_headers=default_headers, api_key=api_key)
         async_gen = await oclient.embeddings.create(input=doc, model=model)
         result = async_gen.model_dump()
         for item in result.get("data", []):
@@ -350,23 +336,30 @@ async def openai_chat_completions_proxy(request: Request):
 
     # Make the API call in handler scope — try/except inside async generators is unreliable
     # with Starlette's streaming machinery, so we resolve errors here before the generator starts.
-    send_params = params
+    # The reservation taken by choose_endpoint is released by stream_ochat_response's finally
-    if not is_ext_openai_endpoint(endpoint):
+    # once we hand off; until then, any failure here (including CancelledError on client
-        resolved_msgs = await _normalize_images_in_messages(params.get("messages", []))
+    # disconnect during a cold model load) must release it or the counter leaks.
-        send_params = {**params, "messages": resolved_msgs}
+    try:
-    # Proactive trim: only for small-ctx models we've already seen run out of space
+        send_params = params
-    _lookup_model = _normalize_llama_model_name(model) if is_llama_server(endpoint) else model
+        if not is_ext_openai_endpoint(endpoint):
-    _known_nctx = _endpoint_nctx.get((endpoint, _lookup_model))
+            resolved_msgs = await _normalize_images_in_messages(params.get("messages", []))
-    if _known_nctx and _known_nctx <= _CTX_TRIM_SMALL_LIMIT:
+            send_params = {**params, "messages": resolved_msgs}
-        _pre_target = int(((_known_nctx - _known_nctx // 4)) / 1.2)
+        # Proactive trim: only for small-ctx models we've already seen run out of space
-        _pre_est = _count_message_tokens(send_params.get("messages", []))
+        _lookup_model = _normalize_llama_model_name(model) if is_llama_server(endpoint) else model
-        if _pre_est > _pre_target:
+        _known_nctx = _endpoint_nctx.get((endpoint, _lookup_model))
-            _pre_msgs = send_params.get("messages", [])
+        if _known_nctx and _known_nctx <= _CTX_TRIM_SMALL_LIMIT:
-            _pre_trimmed = _trim_messages_for_context(_pre_msgs, _known_nctx, target_tokens=_pre_target)
+            _pre_target = int(((_known_nctx - _known_nctx // 4)) / 1.2)
-            _dropped = len(_pre_msgs) - len(_pre_trimmed)
+            _pre_est = _count_message_tokens(send_params.get("messages", []))
-            print(f"[ctx-pre] n_ctx={_known_nctx} est={_pre_est} target={_pre_target} dropped={_dropped}", flush=True)
+            if _pre_est > _pre_target:
-            send_params = {**send_params, "messages": _pre_trimmed}
+                _pre_msgs = send_params.get("messages", [])
-    async_gen = await create_chat_with_retries(oclient, send_params, endpoint, model, tracking_model)
+                _pre_trimmed = _trim_messages_for_context(_pre_msgs, _known_nctx, target_tokens=_pre_target)
+                _dropped = len(_pre_msgs) - len(_pre_trimmed)
+                print(f"[ctx-pre] n_ctx={_known_nctx} est={_pre_est} target={_pre_target} dropped={_dropped}", flush=True)
+                send_params = {**send_params, "messages": _pre_trimmed}
+        async_gen = await create_chat_with_retries(oclient, send_params, endpoint, model, tracking_model)
+    except BaseException:
+        await decrement_usage(endpoint, tracking_model)
+        raise
 
     # 4. Async generator — only streams the already-established async_gen
     async def stream_ochat_response():
@@ -547,12 +540,17 @@ async def openai_completions_proxy(request: Request):
     # 2. Endpoint logic
     _affinity_key = _conversation_fingerprint(model, None, prompt)
     endpoint, tracking_model = await choose_endpoint(model, affinity_key=_affinity_key)
-    oclient = _make_openai_client(endpoint, default_headers=default_headers, api_key=config.api_keys.get(endpoint, "no-key"))
 
     # 3. Async generator that streams completions data and decrements the counter
-    # Make the API call in handler scope (try/except inside async generators is unreliable)
+    # Make the API call in handler scope (try/except inside async generators is unreliable).
+    # The reservation is released by stream_ocompletions_response's finally once we hand off;
+    # until then any failure here — including CancelledError on client disconnect — releases it.
     try:
+        oclient = _make_openai_client(endpoint, default_headers=default_headers, api_key=config.api_keys.get(endpoint, "no-key"))
         async_gen = await oclient.completions.create(**params)
+    except asyncio.CancelledError:
+        await decrement_usage(endpoint, tracking_model)
+        raise
     except Exception as e:
         if _is_backend_connection_error(e):
             print(f"[ocompl] backend connection error → marking ({endpoint}, {model}) unhealthy", flush=True)
@@ -667,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': []}
 
@@ -704,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'])},
@@ -775,36 +789,38 @@ async def rerank_proxy(request: Request):
             ),
         )
 
-    if ":latest" in model:
+    # The finally below releases the reservation for every exit (success, error,
-        model = model.split(":latest")[0]
+    # or CancelledError), so request building and session lookup stay inside it.
-
-    # Build upstream rerank request body – forward only recognised fields
-    upstream_payload: dict = {"model": model, "query": query, "documents": documents}
-    for optional_key in ("top_n", "return_documents", "max_tokens_per_doc"):
-        if optional_key in payload:
-            upstream_payload[optional_key] = payload[optional_key]
-
-    # Determine upstream URL:
-    #   llama-server / llama-swap expose /v1/rerank (base already contains /v1)
-    #   External OpenAI endpoints expose /rerank under their /v1 base
-    if is_llama_server(endpoint):
-        # llama-server / llama-swap: endpoint may or may not already contain /v1
-        if "/v1" in endpoint:
-            rerank_url = f"{endpoint}/rerank"
-        else:
-            rerank_url = f"{endpoint}/v1/rerank"
-    else:
-        # External OpenAI-compatible: ep2base gives us the /v1 base
-        rerank_url = f"{ep2base(endpoint)}/rerank"
-
-    api_key = config.api_keys.get(endpoint, "no-key")
-    headers = {
-        "Content-Type": "application/json",
-        "Authorization": f"Bearer {api_key}",
-    }
-
-    client: aiohttp.ClientSession = get_session(endpoint)
     try:
+        if ":latest" in model:
+            model = model.split(":latest")[0]
+
+        # Build upstream rerank request body – forward only recognised fields
+        upstream_payload: dict = {"model": model, "query": query, "documents": documents}
+        for optional_key in ("top_n", "return_documents", "max_tokens_per_doc"):
+            if optional_key in payload:
+                upstream_payload[optional_key] = payload[optional_key]
+
+        # Determine upstream URL:
+        #   llama-server / llama-swap expose /v1/rerank (base already contains /v1)
+        #   External OpenAI endpoints expose /rerank under their /v1 base
+        if is_llama_server(endpoint):
+            # llama-server / llama-swap: endpoint may or may not already contain /v1
+            if "/v1" in endpoint:
+                rerank_url = f"{endpoint}/rerank"
+            else:
+                rerank_url = f"{endpoint}/v1/rerank"
+        else:
+            # External OpenAI-compatible: ep2base gives us the /v1 base
+            rerank_url = f"{ep2base(endpoint)}/rerank"
+
+        api_key = config.api_keys.get(endpoint, "no-key")
+        headers = {
+            "Content-Type": "application/json",
+            "Authorization": f"Bearer {api_key}",
+        }
+
+        client: aiohttp.ClientSession = get_session(endpoint)
         async with client.post(rerank_url, json=upstream_payload, headers=headers) as resp:
             response_bytes = await resp.read()
             if resp.status >= 400:

api/responses.py

@@ -140,9 +140,9 @@ async def _run_to_completion(*, native, oclient, endpoint, model, tracking_model
                              send_params, native_params):
     """Drive the backend to completion (no client streaming).
 
-    Returns ``(output_items, usage)`` where usage is responses-shaped. Caller is
+    Returns ``(output_items, usage)`` where usage is responses-shaped. The caller
-    responsible for ``decrement_usage`` (translated failures self-decrement inside
+    owns the usage reservation and must release it (this function and
-    ``create_chat_with_retries``)."""
+    ``create_chat_with_retries`` never decrement)."""
     if native:
         resp_obj = await oclient.responses.create(stream=False, **native_params)
         data = resp_obj.model_dump()
@@ -209,38 +209,46 @@ async def openai_responses_proxy(request: Request):
                 return StreamingResponse(_served_cached(), media_type="text/event-stream")
             return JSONResponse(content=resp_obj)
 
-    # Endpoint selection (reserves a slot — must be released exactly once).
+    # Endpoint selection (reserves a slot — must be released exactly once). The
+    # release is owned by the per-branch finally (_bg_run / _stream / the
+    # non-streaming try) once we hand off; any failure during client/param
+    # construction (including CancelledError on client disconnect) must release
+    # it here or the usage counter leaks.
     _affinity_key = _conversation_fingerprint(model, messages, None)
     endpoint, tracking_model = await choose_endpoint(model, affinity_key=_affinity_key)
-    oclient = _make_openai_client(endpoint, default_headers=default_headers,
+    try:
-                                  api_key=config.api_keys.get(endpoint, "no-key"))
+        oclient = _make_openai_client(endpoint, default_headers=default_headers,
-    native = is_ext_openai_endpoint(endpoint)
+                                      api_key=config.api_keys.get(endpoint, "no-key"))
+        native = is_ext_openai_endpoint(endpoint)
 
-    # Build backend params for both shapes.
+        # Build backend params for both shapes.
-    send_params = {"messages": messages, "model": model}
+        send_params = {"messages": messages, "model": model}
-    _opt = {
+        _opt = {
-        "temperature": payload.get("temperature"),
+            "temperature": payload.get("temperature"),
-        "top_p": payload.get("top_p"),
+            "top_p": payload.get("top_p"),
-        "max_tokens": payload.get("max_output_tokens"),
+            "max_tokens": payload.get("max_output_tokens"),
-        "tools": tools_responses_to_chat(tools),
+            "tools": tools_responses_to_chat(tools),
-        "tool_choice": payload.get("tool_choice"),
+            "tool_choice": payload.get("tool_choice"),
-        "response_format": _text_format_to_response_format(payload.get("text")),
+            "response_format": _text_format_to_response_format(payload.get("text")),
-    }
+        }
-    send_params.update({k: v for k, v in _opt.items() if v is not None})
+        send_params.update({k: v for k, v in _opt.items() if v is not None})
 
-    native_instructions, native_input = messages_to_responses_input(messages)
+        native_instructions, native_input = messages_to_responses_input(messages)
-    native_params = {"model": model, "input": native_input, "store": False}
+        native_params = {"model": model, "input": native_input, "store": False}
-    _nopt = {
+        _nopt = {
-        "instructions": native_instructions,
+            "instructions": native_instructions,
-        "temperature": payload.get("temperature"),
+            "temperature": payload.get("temperature"),
-        "top_p": payload.get("top_p"),
+            "top_p": payload.get("top_p"),
-        "max_output_tokens": payload.get("max_output_tokens"),
+            "max_output_tokens": payload.get("max_output_tokens"),
-        "tools": tools,
+            "tools": tools,
-        "tool_choice": payload.get("tool_choice"),
+            "tool_choice": payload.get("tool_choice"),
-        "text": payload.get("text"),
+            "text": payload.get("text"),
-        "reasoning": payload.get("reasoning"),
+            "reasoning": payload.get("reasoning"),
-    }
+        }
-    native_params.update({k: v for k, v in _nopt.items() if v is not None})
+        native_params.update({k: v for k, v in _nopt.items() if v is not None})
+    except BaseException:
+        await decrement_usage(endpoint, tracking_model)
+        raise
 
     async def _persist(status, output_items=None, usage=None, error=None, insert=False):
         if not store:
@@ -275,30 +283,37 @@ async def openai_responses_proxy(request: Request):
 
     # ---- background: run detached, return queued immediately --------------
     if background:
-        await _persist("queued", insert=True)
+        # Once the task is created, _bg_run's finally owns the release. Guard the
+        # pre-task setup so a failure there (queued persist, task creation, or a
+        # client disconnect) still releases the reservation.
+        try:
+            await _persist("queued", insert=True)
 
-        async def _bg_run():
+            async def _bg_run():
-            try:
+                try:
-                await get_db().update_response_status(response_id, "in_progress")
+                    await get_db().update_response_status(response_id, "in_progress")
-                output_items, usage = await _run_to_completion(
+                    output_items, usage = await _run_to_completion(
-                    native=native, oclient=oclient, endpoint=endpoint, model=model,
+                        native=native, oclient=oclient, endpoint=endpoint, model=model,
-                    tracking_model=tracking_model, send_params=send_params,
+                        tracking_model=tracking_model, send_params=send_params,
-                    native_params=native_params)
+                        native_params=native_params)
-                await _track(usage)
+                    await _track(usage)
-                await _persist("completed", output_items=output_items, usage=usage)
+                    await _persist("completed", output_items=output_items, usage=usage)
-                await _cache_store(output_items, usage)
+                    await _cache_store(output_items, usage)
-            except asyncio.CancelledError:
+                except asyncio.CancelledError:
-                await get_db().update_response_status(response_id, "cancelled")
+                    await get_db().update_response_status(response_id, "cancelled")
-                raise
+                    raise
-            except Exception as e:
+                except Exception as e:
-                await get_db().update_response_status(
+                    await get_db().update_response_status(
-                    response_id, "failed",
+                        response_id, "failed",
-                    error={"message": str(e)[:500], "type": type(e).__name__})
+                        error={"message": str(e)[:500], "type": type(e).__name__})
-            finally:
+                finally:
-                await decrement_usage(endpoint, tracking_model)
+                    await decrement_usage(endpoint, tracking_model)
-                _background_tasks.pop(response_id, None)
+                    _background_tasks.pop(response_id, None)
 
-        task = asyncio.create_task(_bg_run())
+            task = asyncio.create_task(_bg_run())
+        except BaseException:
+            await decrement_usage(endpoint, tracking_model)
+            raise
         _background_tasks[response_id] = task
         queued = build_response_object(response_id=response_id, model=model, output_items=[],
                                        status="queued", created_at=created_at,
@@ -308,18 +323,25 @@ async def openai_responses_proxy(request: Request):
 
     # ---- streaming sync ----------------------------------------------------
     if stream:
-        if native:
+        # _stream's finally owns the release once iteration starts. Establishing
-            source = await oclient.responses.create(stream=True, **native_params)
+        # the source can fail (or be cancelled) before that — release here, since
-            translator = _NativeStream(response_id)
+        # create_chat_with_retries no longer self-decrements.
-        else:
+        try:
-            source = await create_chat_with_retries(
+            if native:
-                oclient, {**send_params, "stream": True,
+                source = await oclient.responses.create(stream=True, **native_params)
-                          "stream_options": {"include_usage": True}},
+                translator = _NativeStream(response_id)
-                endpoint, model, tracking_model)
+            else:
-            translator = ChatToResponsesStream(
+                source = await create_chat_with_retries(
-                response_id, model, created_at=created_at,
+                    oclient, {**send_params, "stream": True,
-                previous_response_id=previous_response_id, instructions=instructions,
+                              "stream_options": {"include_usage": True}},
-                metadata=metadata)
+                    endpoint, model, tracking_model)
+                translator = ChatToResponsesStream(
+                    response_id, model, created_at=created_at,
+                    previous_response_id=previous_response_id, instructions=instructions,
+                    metadata=metadata)
+        except BaseException:
+            await decrement_usage(endpoint, tracking_model)
+            raise
 
         async def _stream():
             await _persist("in_progress", insert=True)

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
+    # External OpenAI / native Anthropic endpoints are not shown in PS, keep as-is
-    if is_ext_openai_endpoint(endpoint):
+    if is_ext_openai_endpoint(endpoint) or is_anthropic_endpoint(endpoint):
         return model
 
     # llama-server / llama-swap endpoints use normalized names in PS

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")}
+        headers = _auth_headers(endpoint, api_key)
-        if api_key is not None:
-            headers["Authorization"] = "Bearer " + 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,8 +346,18 @@ 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):
+        if is_ext_openai_endpoint(endpoint) or is_anthropic_endpoint(endpoint):
-            return set()
+            # 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.
+            # See issue #128. (llama-server / llama-swap are NOT ext-openai — they
+            # keep their real resident-subset detection below.)
+            return await fetch.available_models(endpoint, get_config().api_keys.get(endpoint))
 
         # Check loaded models cache with lock protection
         async with _loaded_models_cache_lock:
@@ -387,9 +423,7 @@ class fetch:
                     if _is_fresh(_available_error_cache[endpoint], 300):
                         return []
 
-        headers = {"Referer": default_headers.get("HTTP-Referer", "https://nomyo.ai")}
+        headers = _auth_headers(endpoint, api_key)
-        if api_key is not None:
-            headers["Authorization"] = "Bearer " + api_key
 
         request_url = f"{endpoint.rstrip('/')}/{route.lstrip('/')}"
         client: aiohttp.ClientSession = get_probe_session(endpoint)
@@ -425,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")}
+    headers = _auth_headers(ep, api_key)
-    if api_key is not None:
-        headers["Authorization"] = "Bearer " + api_key
     url = f"{ep.rstrip('/')}/{route.lstrip('/')}"
     req_kwargs = {}
     if timeout is not None:
@@ -450,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,

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}}

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

requests/anthropic.py

@@ -0,0 +1,558 @@
+"""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 _cached_prompt_tokens(usage):
+    """Read ``prompt_tokens_details.cached_tokens`` from a chat usage (dict or SDK obj).
+
+    OpenAI-compatible backends with automatic prefix caching (vLLM, recent
+    llama-server) report the reused-prefix token count here; backends that don't
+    populate it yield 0.
+    """
+    if not usage:
+        return 0
+    details = usage.get("prompt_tokens_details") if isinstance(usage, dict) \
+        else getattr(usage, "prompt_tokens_details", None)
+    if details is None:
+        return 0
+    if isinstance(details, dict):
+        return details.get("cached_tokens") or 0
+    return getattr(details, "cached_tokens", 0) or 0
+
+
+def usage_chat_to_anthropic(usage, cache_read_tokens=0, cache_creation_tokens=0):
+    """Map chat usage → Anthropic usage.
+
+    ``cached_tokens`` reported by the backend (automatic prefix-cache reuse) plus any
+    explicit ``cache_read_tokens`` become ``cache_read_input_tokens``; ``input_tokens``
+    is the remaining uncached prompt so ``input + cache_read`` equals the prompt total.
+
+    ``cache_creation_input_tokens`` stays at ``cache_creation_tokens`` (0 by default):
+    local backends do automatic KV-prefix reuse with no explicit ``cache_control``
+    write/breakpoint concept, so there is no honest "tokens written" figure to report —
+    that value is only real on the native Anthropic passthrough path.
+    """
+    prompt = (usage or {}).get("prompt_tokens") or 0 if isinstance(usage, dict) \
+        else (getattr(usage, "prompt_tokens", 0) or 0)
+    completion = (usage or {}).get("completion_tokens") or 0 if isinstance(usage, dict) \
+        else (getattr(usage, "completion_tokens", 0) or 0)
+    cache_read = _cached_prompt_tokens(usage) + cache_read_tokens
+    return {
+        "input_tokens": max(prompt - cache_read, 0),
+        "output_tokens": completion,
+        "cache_creation_input_tokens": cache_creation_tokens,
+        "cache_read_input_tokens": cache_read,
+    }
+
+
+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,
+                    "prompt_tokens_details": {"cached_tokens": _cached_prompt_tokens(usage)},
+                }
+            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))
+        final_usage = usage_chat_to_anthropic(
+            self.usage, cache_read_tokens=self.cache_read_tokens,
+            cache_creation_tokens=self.cache_creation_tokens)
+        yield _sse("message_delta", {
+            "delta": {"stop_reason": self.stop_reason, "stop_sequence": None},
+            "usage": {
+                "input_tokens": final_usage["input_tokens"],
+                "output_tokens": final_usage["output_tokens"],
+                "cache_read_input_tokens": final_usage["cache_read_input_tokens"],
+                "cache_creation_input_tokens": final_usage["cache_creation_input_tokens"],
+            }})
+        yield _sse("message_stop", {})

requests/chat.py

@@ -44,38 +44,40 @@ async def _make_chat_request(model: str, messages: list, tools=None, stream: boo
     """
     config = get_config()
     endpoint, tracking_model = await choose_endpoint(model)  # selects and atomically reserves
-    use_openai = is_openai_compatible(endpoint)
+    # The finally below releases the reservation on every exit — success, error,
-    if use_openai:
+    # or CancelledError — so request building and client construction stay inside it.
-        if ":latest" in model:
-            model = model.split(":latest")[0]
-        if messages:
-            if any("images" in m for m in messages):
-                messages = await asyncio.to_thread(transform_images_to_data_urls, messages)
-            messages = transform_tool_calls_to_openai(messages)
-            messages = _strip_assistant_prefill(messages)
-        params = {
-            "messages": messages,
-            "model": model,
-        }
-        optional_params = {
-            "tools": tools,
-            "stream": stream,
-            "stream_options": {"include_usage": True} if stream else None,
-            "max_tokens": options.get("num_predict") if options and "num_predict" in options else None,
-            "frequency_penalty": options.get("frequency_penalty") if options and "frequency_penalty" in options else None,
-            "presence_penalty": options.get("presence_penalty") if options and "presence_penalty" in options else None,
-            "seed": options.get("seed") if options and "seed" in options else None,
-            "stop": options.get("stop") if options and "stop" in options else None,
-            "top_p": options.get("top_p") if options and "top_p" in options else None,
-            "temperature": options.get("temperature") if options and "temperature" in options else None,
-            "response_format": {"type": "json_schema", "json_schema": format} if format is not None else None
-        }
-        params.update({k: v for k, v in optional_params.items() if v is not None})
-        oclient = _make_openai_client(endpoint, default_headers=default_headers, api_key=config.api_keys.get(endpoint, "no-key"))
-    else:
-        client = ollama.AsyncClient(host=endpoint)
-
     try:
+        use_openai = is_openai_compatible(endpoint)
+        if use_openai:
+            if ":latest" in model:
+                model = model.split(":latest")[0]
+            if messages:
+                if any("images" in m for m in messages):
+                    messages = await asyncio.to_thread(transform_images_to_data_urls, messages)
+                messages = transform_tool_calls_to_openai(messages)
+                messages = _strip_assistant_prefill(messages)
+            params = {
+                "messages": messages,
+                "model": model,
+            }
+            optional_params = {
+                "tools": tools,
+                "stream": stream,
+                "stream_options": {"include_usage": True} if stream else None,
+                "max_tokens": options.get("num_predict") if options and "num_predict" in options else None,
+                "frequency_penalty": options.get("frequency_penalty") if options and "frequency_penalty" in options else None,
+                "presence_penalty": options.get("presence_penalty") if options and "presence_penalty" in options else None,
+                "seed": options.get("seed") if options and "seed" in options else None,
+                "stop": options.get("stop") if options and "stop" in options else None,
+                "top_p": options.get("top_p") if options and "top_p" in options else None,
+                "temperature": options.get("temperature") if options and "temperature" in options else None,
+                "response_format": {"type": "json_schema", "json_schema": format} if format is not None else None
+            }
+            params.update({k: v for k, v in optional_params.items() if v is not None})
+            oclient = _make_openai_client(endpoint, default_headers=default_headers, api_key=config.api_keys.get(endpoint, "no-key"))
+        else:
+            client = ollama.AsyncClient(host=endpoint)
+
         if use_openai:
             start_ts = time.perf_counter()
             try:

requirements.txt

@@ -1,31 +1,31 @@
-aiohappyeyeballs==2.6.2
+aiohappyeyeballs==2.7.1
 aiohttp==3.14.1
 aiosignal==1.4.0
 annotated-types==0.7.0
-anyio==4.14.0
+anyio==4.14.2
 async-timeout==5.0.1
 attrs==26.1.0
-certifi==2026.5.20
+certifi==2026.6.17
-click==8.4.1
+click==8.4.2
 distro==1.9.0
 exceptiongroup==1.3.1
-fastapi==0.137.1
+fastapi==0.139.0
 fastapi-sse==1.1.1
 frozenlist==1.8.0
 h11==0.16.0
 httpcore==1.0.9
 httpx==0.28.1
 idna==3.18
-jiter==0.15.0
+jiter==0.16.0
 multidict==6.7.1
 ollama==0.6.2
-openai==2.41.1
+openai==2.45.0
 orjson>=3.11.5
 numpy>=1.26
-pillow==12.2.0
+pillow==12.3.0
 propcache==0.5.2
 pydantic==2.13.4
-pydantic-settings==2.14.1
+pydantic-settings==2.14.2
 pydantic_core==2.46.4
 python-dotenv==1.2.2
 PyYAML==6.0.3
@@ -33,9 +33,9 @@ sniffio==1.3.1
 starlette>=1.0.1
 truststore==0.10.4
 tiktoken==0.13.0
-tqdm==4.68.2
+tqdm==4.68.4
 typing-inspection==0.4.2
-typing_extensions==4.15.0
+typing_extensions==4.16.0
 uvicorn==0.49.0
 uvloop
 yarl==1.24.2

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):

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,11 +98,22 @@ 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
 
-    tag_tasks = [fetch.available_models(ep) for ep in config.endpoints if not is_openai_compatible(ep)]
+    # Build the probe tasks in the SAME order as ``all_endpoints`` so the
-    tag_tasks += [fetch.available_models(ep, config.api_keys.get(ep)) for ep in config.endpoints if is_openai_compatible(ep)]
+    # gathered results stay aligned for the ``zip(all_endpoints, advertised_sets)``
-    tag_tasks += [fetch.available_models(ep, config.api_keys.get(ep)) for ep in llama_eps_extra]
+    # below. (A previous partition into non-OpenAI / OpenAI groups reordered the
+    # tasks relative to ``all_endpoints``, pairing each endpoint with another
+    # endpoint's advertised models — so an OpenAI-compatible backend listed
+    # 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) or is_anthropic_endpoint(ep):
+            return fetch.available_models(ep, config.api_keys.get(ep))
+        return fetch.available_models(ep)
+
+    tag_tasks = [_advertised_task(ep) for ep in all_endpoints]
     advertised_sets = await asyncio.gather(*tag_tasks)
 
     # 2️⃣  Filter endpoints that advertise the requested model
@@ -311,15 +323,23 @@ async def choose_endpoint(model: str, reserve: bool = True,
         if reserve:
             usage_counts[selected][tracking_model] += 1
             snapshot = _capture_snapshot()
-    if snapshot is not None:
+    # The slot is now reserved. Any failure (including CancelledError on client
-        await _distribute_snapshot(snapshot)
+    # disconnect) between here and `return` would otherwise leak it — the caller
-    # Record / refresh affinity *after* releasing usage_lock.
+    # never receives (endpoint, tracking_model) and so can never decrement it.
-    if reserve and config.conversation_affinity and affinity_key:
+    try:
-        expires_at = time.monotonic() + config.conversation_affinity_ttl
+        if snapshot is not None:
-        async with _affinity_lock:
+            await _distribute_snapshot(snapshot)
-            _affinity_map[affinity_key] = (selected, model, expires_at)
+        # Record / refresh affinity *after* releasing usage_lock.
-            if len(_affinity_map) > _AFFINITY_MAX_ENTRIES:
+        if reserve and config.conversation_affinity and affinity_key:
-                now = time.monotonic()
+            expires_at = time.monotonic() + config.conversation_affinity_ttl
-                for k in [k for k, v in _affinity_map.items() if v[2] < now]:
+            async with _affinity_lock:
-                    _affinity_map.pop(k, None)
+                _affinity_map[affinity_key] = (selected, model, expires_at)
+                if len(_affinity_map) > _AFFINITY_MAX_ENTRIES:
+                    now = time.monotonic()
+                    for k in [k for k, v in _affinity_map.items() if v[2] < now]:
+                        _affinity_map.pop(k, None)
+    except BaseException:
+        if reserve:
+            await decrement_usage(selected, tracking_model)
+        raise
     return selected, tracking_model

test/test_choose_endpoint.py

@@ -279,6 +279,57 @@ class TestChooseEndpointModelNaming:
         assert ep == EP1
 
 
+class TestChooseEndpointAdvertisedAlignment:
+    """Regression for issue #128 — advertised-model sets must stay aligned with
+    their endpoints regardless of config ordering. A previous implementation
+    partitioned the probe tasks (non-OpenAI first, then OpenAI-compatible) while
+    the endpoint list kept config order, so an OpenAI-compatible endpoint listed
+    *before* an Ollama one inherited the Ollama model set and received requests
+    for models it does not serve (404)."""
+
+    OPENAI_EP = "http://vllm:8000/v1"
+
+    async def test_openai_endpoint_before_ollama_does_not_misroute(self):
+        # OpenAI-compatible endpoint deliberately listed FIRST in config.
+        cfg = _make_cfg([self.OPENAI_EP, EP1])
+
+        async def available(ep, *_):
+            if ep == self.OPENAI_EP:
+                return {"vllm-only-model"}
+            return {"qwen2.5-coder:1.5b-base"}  # only the Ollama box advertises this
+
+        async def loaded(ep):
+            return {"qwen2.5-coder:1.5b-base"} if ep == EP1 else set()
+
+        with (
+            patch.object(router, "config", cfg),
+            patch.object(router.fetch, "available_models", side_effect=available),
+            patch.object(router.fetch, "loaded_models", side_effect=loaded),
+        ):
+            ep, _ = await router.choose_endpoint("qwen2.5-coder:1.5b-base")
+        # Must route to the endpoint that actually advertises the model.
+        assert ep == EP1
+
+    async def test_openai_endpoint_before_ollama_still_routes_its_own_model(self):
+        cfg = _make_cfg([self.OPENAI_EP, EP1])
+
+        async def available(ep, *_):
+            if ep == self.OPENAI_EP:
+                return {"vllm-only-model"}
+            return {"qwen2.5-coder:1.5b-base"}
+
+        async def loaded(ep):
+            return {"qwen2.5-coder:1.5b-base"} if ep == EP1 else set()
+
+        with (
+            patch.object(router, "config", cfg),
+            patch.object(router.fetch, "available_models", side_effect=available),
+            patch.object(router.fetch, "loaded_models", side_effect=loaded),
+        ):
+            ep, _ = await router.choose_endpoint("vllm-only-model")
+        assert ep == self.OPENAI_EP
+
+
 class TestChooseEndpointLoadBalancing:
     async def test_random_selection_among_idle(self):
         cfg = _make_cfg([EP1, EP2, EP3])

test/test_fetch.py

@@ -260,12 +260,20 @@ class TestFetchLoadedModels:
             models = await router.fetch.loaded_models(MOCK_OLLAMA_EP)
         assert models == set()
 
-    async def test_ext_openai_always_empty(self):
+    async def test_ext_openai_reports_advertised_as_loaded(self):
+        # Issue #128: external OpenAI-compatible backends (vLLM, OpenAI, …) keep
+        # every advertised model permanently resident, so loaded_models mirrors
+        # the advertised set rather than returning empty (which made them lose
+        # choose_endpoint's "loaded & free" preference to Ollama boxes).
         ext_ep = "https://api.openai.com/v1"
         cfg = _make_cfg(ollama_eps=[ext_ep], llama_eps=[])
-        with patch.object(router, "config", cfg):
+        with patch.object(router, "config", cfg), mock_probe() as m:
+            m.add_get(
+                f"{ext_ep}/models",
+                payload={"data": [{"id": "gpt-4o"}, {"id": "gpt-4o-mini"}]},
+            )
             models = await router.fetch.loaded_models(ext_ep)
-        assert models == set()
+        assert models == {"gpt-4o", "gpt-4o-mini"}
 
     async def test_caches_result(self):
         cfg = _make_cfg(ollama_eps=[MOCK_OLLAMA_EP], llama_eps=[])

test/test_messages.py

@@ -0,0 +1,477 @@
+"""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_no_cache(self):
+        u = at.usage_chat_to_anthropic({"prompt_tokens": 7, "completion_tokens": 3})
+        assert u == {"input_tokens": 7, "output_tokens": 3,
+                     "cache_creation_input_tokens": 0, "cache_read_input_tokens": 0}
+
+    def test_usage_mapping_cached_tokens_subtracted(self):
+        # Backend reports 5 of 7 prompt tokens served from its prefix cache.
+        u = at.usage_chat_to_anthropic({
+            "prompt_tokens": 7, "completion_tokens": 3,
+            "prompt_tokens_details": {"cached_tokens": 5}})
+        assert u == {"input_tokens": 2, "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, cached=None):
+    usage = NS(prompt_tokens=p, completion_tokens=c)
+    if cached is not None:
+        usage.prompt_tokens_details = NS(cached_tokens=cached)
+    return NS(choices=[], usage=usage)
+
+
+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_cached_tokens_surface_as_cache_read(self):
+        async def gen():
+            yield _chunk(content="hi", finish_reason="stop")
+            yield _usage_chunk(10, 2, cached=6)
+        tr = at.ChatToMessagesStream("msg_1", "m")
+        events = await _collect(tr, gen())
+        md = [d for t, d in events if t == "message_delta"][0]
+        assert md["usage"]["cache_read_input_tokens"] == 6
+        assert md["usage"]["input_tokens"] == 4  # 10 prompt − 6 cached
+        assert md["usage"]["cache_creation_input_tokens"] == 0
+
+    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", cached=None):
+    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_dump = {"prompt_tokens": usage[0], "completion_tokens": usage[1],
+                  "total_tokens": sum(usage)}
+    if cached is not None:
+        usage_dump["prompt_tokens_details"] = {"cached_tokens": cached}
+    usage_obj = MagicMock()
+    usage_obj.model_dump.return_value = usage_dump
+    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_nonstream_cached_tokens(self, client):
+        with _enter(*_patch_backend(native=False),
+                    patch.object(api_messages, "create_chat_with_retries",
+                                 AsyncMock(return_value=_fake_completion(
+                                     "hi", usage=(10, 4), cached=6)))):
+            resp = await client.post("/v1/messages",
+                                     json={"model": "test-model", "max_tokens": 100,
+                                           "messages": [{"role": "user", "content": "hi"}]})
+        u = resp.json()["usage"]
+        assert u["cache_read_input_tokens"] == 6
+        assert u["input_tokens"] == 4
+        assert u["cache_creation_input_tokens"] == 0
+
+    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