anthropic_proxy.py

"""Pont Anthropic -> OpenAI, pour brancher Claude Code sur un serveur vLLM.

Claude Code parle le protocole Anthropic (`POST /v1/messages`, SSE a evenements
nommes). vLLM ne sert que le protocole OpenAI. Ce module traduit dans les deux
sens, en streaming comme en une passe, avec les appels d'outils -- sans quoi un
agent ne peut rien faire.

Lance a cote de vLLM sur la meme machine ; ecoute sur PROXY_PORT et relaie vers
UPSTREAM.
"""

from __future__ import annotations

import asyncio
import json
import os
import re
import time
import uuid
from typing import Any

import httpx
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse, StreamingResponse

UPSTREAM = os.environ.get("VL_UPSTREAM", "http://127.0.0.1:8080")
MODEL = os.environ.get("VL_SERVED_NAME", "qwen")
# ---------------------------------------------------------------- SWAP (v70)
# Deux modeles a tour de role sur la meme carte. Le bootstrap ecrit
# `$VL_SWAP_DIR/modele_courant` (cle nom depot) a chaque lancement de vLLM ; le
# pont y lit le modele charge A CHAQUE REQUETE (il survit au swap). Une requete
# pour un modele non charge ecrit `modele_demande`, puis attend que le
# bootstrap ait relance vLLM (jusqu'a VL_SWAP_TIMEOUT s).
SWAP_DIR = os.environ.get("VL_SWAP_DIR", "/travail")
SWAP = dict(x.split(":", 1) for x in os.environ.get("VL_SWAP", "").split(",") if ":" in x)
SWAP_TIMEOUT = float(os.environ.get("VL_SWAP_TIMEOUT", "900"))


def _courant() -> tuple[str, str, str, str]:
    """(cle bootstrap, nom servi, depot, speculation) du modele charge."""
    try:
        with open(os.path.join(SWAP_DIR, "modele_courant"), encoding="utf-8") as f:
            champs = f.read().split()
        cle, nom, depot = champs[:3]
        return cle, nom, depot, (champs[3] if len(champs) > 3 else "off")
    except Exception:
        return os.environ.get("VL_MODEL_KEY", ""), MODEL, os.environ.get("VL_REAL_MODEL", MODEL), "off"


def _cle_demandee(model: object) -> tuple[str, str] | None:
    """`ornith` -> (cle, "") ; `ornith+dspark` -> (cle, "dspark") : la speculation
    voyage dans le nom du modele, pour changer de reglage sans recreer le pod."""
    if not isinstance(model, str):
        return None
    nom = model.removesuffix("[1m]")
    if nom.startswith("claude-"):
        nom = nom[len("claude-"):]
    nom, _, spec = nom.partition("+")
    cle = SWAP.get(nom)
    return (cle, spec) if cle else None
TIMEOUT = float(os.environ.get("VL_TIMEOUT", "1800"))
# Sortie maximale du modele. Claude Code demande couramment 64 k, ce que
# vLLM refuse d'un 400 portant sur max_tokens -- la requete entiere echoue
# alors qu'un plafonnement silencieux suffit.
MAX_OUTPUT = int(os.environ.get("VL_MAX_OUTPUT", "32768"))

# Claude Code refuse tout identifiant de modele qui ne commence pas par
# "claude-" : il valide le nom avant d'emettre la requete. On expose donc des
# alias conformes, et on ignore le nom recu pour router vers l'unique modele
# reellement charge -- le client choisit une etiquette, pas un moteur.
# Le PREMIER alias nomme le modele reellement charge : sans cela, un client qui
# voit "claude-kimi-k3" croit legitimement executer du Kimi alors que le moteur
# sert du Qwen. Les suivants sont des etiquettes de compatibilite, et tous
# routent vers l'unique modele charge.
_REAL = {"qwen": "claude-qwen3-coder-30b", "kimi": "claude-kimi-linear-48b"}
ALIASES = [
    # Alias NU, sans prefixe. Indispensable pour une fenetre > 200 k.
    # Claude Code 2.1.239 (fonction JFd du binaire) :
    #     let n = CLAUDE_CODE_MAX_CONTEXT_TOKENS;
    #     if (n !== undefined && n > 0 && !id.startsWith("claude-")) return n;
    #     return <defaut 200 000>
    # Autrement dit il REFUSE toute fenetre personnalisee sur un identifiant
    # commencant par "claude-" : avec `claude-flashnext`, ni
    # CLAUDE_CODE_MAX_CONTEXT_TOKENS ni CLAUDE_CODE_AUTO_COMPACT_WINDOW n'ont
    # le moindre effet, et /context affiche 200k quoi qu'on fasse.
    MODEL,
    _REAL.get(MODEL, f"claude-{MODEL}"),
    "claude-kimi-k3",
    "claude-kimi-k3-linear",
    "claude-qwen3-coder",
    "claude-sonnet-4-5",       # alias de compatibilite : certains clients
    "claude-3-5-haiku",        # codent en dur un modele "rapide" et un "lent"
]
# Variantes "[1m]". Claude Code deduit la fenetre de contexte du NOM du modele :
# un identifiant qu'il ne connait pas est suppose a 200 k, et l'auto-compactage
# se declenche a 200 k meme si le moteur en accepte 1 000 000. Le suffixe [1m]
# est sa convention pour la fenetre du million ; VERIFIE : avec lui,
# l'avertissement "auto-compact will keep this session within 200k" disparait.
for _n in SWAP:                       # les modeles interchangeables, tous annonces
    ALIASES += [_n, f"claude-{_n}", f"{_n}+dspark", f"{_n}+off"]
ALIASES = [a for pair in ((x, f"{x}[1m]") for x in ALIASES) for a in pair]
ALIASES = list(dict.fromkeys(ALIASES))   # dedoublonne en gardant l'ordre

app = FastAPI(title="anthropic-bridge")

_verrou_swap = asyncio.Lock()


async def _assurer_modele(model: object) -> str:
    """Rend le nom servi pour `model`, en declenchant le swap s'il le faut."""
    dem = _cle_demandee(model)
    courant = _courant()
    if not dem:
        return courant[1]
    cle, spec = dem

    def satisfait(c):
        # sans "+spec" dans le nom, n'importe quelle speculation du bon modele convient
        return c[0] == cle and (not spec or c[3] == spec)

    if satisfait(courant):
        return courant[1]
    async with _verrou_swap:
        courant = _courant()
        if satisfait(courant):
            return courant[1]
        with open(os.path.join(SWAP_DIR, "modele_demande"), "w", encoding="utf-8") as f:
            f.write(f"{cle} {spec}\n")
        t0 = time.time()
        while time.time() - t0 < SWAP_TIMEOUT:
            await asyncio.sleep(3)
            c = _courant()
            if not satisfait(c):
                continue
            try:
                r = await _client.get("/health", timeout=3.0)
                if r.status_code == 200:
                    return c[1]
            except Exception:
                pass
        raise HTTPException(status_code=503, detail=f"swap vers {cle} non termine en {SWAP_TIMEOUT:.0f} s")


@app.middleware("http")
async def _guerir_bassin(request: Request, call_next):
    generation = _bassin_generation
    try:
        return await call_next(request)
    except httpx.PoolTimeout:
        await _reconstruire_bassin(generation)
        return JSONResponse(
            status_code=503,
            content={"type": "error",
                     "error": {"type": "overloaded_error",
                               "message": "upstream pool exhausted, "
                                          "connection pool rebuilt; retry"}})
    except (httpx.HTTPError, RuntimeError) as e:
        # v73 : mesure du 29/08 -- la saturation ne se presente pas toujours en
        # PoolTimeout. Un RuntimeError("client has been closed") apres une
        # reconstruction, ou un ReadError sur une connexion recyclee, sortait en
        # 500 brut sans jamais declencher la guerison. Meme remede, et le type
        # REEL est journalise pour qu'on ne rediagnostique plus a l'aveugle.
        import traceback
        print(f"[bassin?] {type(e).__name__}: {e} sur {request.url.path}",
              flush=True)
        traceback.print_exc()
        await _reconstruire_bassin(generation)
        return JSONResponse(
            status_code=503,
            content={"type": "error",
                     "error": {"type": "overloaded_error",
                               "message": "upstream relay failed "
                                          f"({type(e).__name__}), pool "
                                          "rebuilt; retry"}})

# Pool EXPLICITE. Mesure 22/08 nuit (5 agents Claude Code, ~150 k de contexte) :
# 189 flux ouverts pour 100 termines, 45 connexions amont pour 2 clients -- les
# flux abandonnes cote client (retry, coupure proxy) gardaient leur connexion
# amont, vLLM generait pour personne, et a 100 connexions (defaut httpx) le pool
# bloquait TOUT, /health compris : pont muet, processus vivant a 3 % CPU.
# Le timeout de pool transforme une saturation en 503 rapide au lieu d'un blocage.
def _nouveau_client() -> httpx.AsyncClient:
    return httpx.AsyncClient(
        base_url=UPSTREAM,
        timeout=httpx.Timeout(TIMEOUT, connect=10.0, pool=10.0),
        limits=httpx.Limits(max_connections=512, max_keepalive_connections=64))


_client = _nouveau_client()

# GUERISON DU BASSIN, ajoutee le 27/08 apres trois saturations en douze heures,
# a intervalle d'environ une heure et quart sous charge multi-agents.
#
# Ce que la mesure a montre, et qui change le diagnostic : au moment du blocage,
# `ss -tan | grep :8000` renvoyait **zero** socket. Le bassin n'est donc pas
# plein de connexions vivantes -- il est plein de PLACES comptabilisees et
# jamais rendues. Consequence : baisser VL_TIMEOUT de 1800 a 240 s n'a rien
# change, parce que la fuite n'est pas temporelle, elle est definitive par flux
# abandonne. Monter `max_connections` ne fait que retarder l'echeance.
#
# On ne repare donc pas la fuite ici (elle est dans httpx/httpcore, pas dans ce
# fichier) : on rend le pont capable d'en sortir seul. A la premiere PoolTimeout
# on remplace le client par un neuf et on ferme l'ancien en arriere-plan. Le
# compteur de generation evite que dix requetes simultanees reconstruisent dix
# fois : seule celle qui a vu la generation courante agit.
_bassin_verrou = asyncio.Lock()
_bassin_generation = 0


async def _reconstruire_bassin(generation_vue: int) -> None:
    global _client, _bassin_generation
    async with _bassin_verrou:
        if generation_vue != _bassin_generation:
            return
        ancien, _client = _client, _nouveau_client()
        _bassin_generation += 1
        print("[bassin] PoolTimeout -> client httpx reconstruit "
              "(generation %d)" % _bassin_generation, flush=True)
        try:
            await ancien.aclose()
        except Exception:
            pass


# ------------------------------------------------------------------ garde-fou
# Un modele de 30 a 48 milliards de parametres quantifie en 4 bits abrege : somme
# de reecrire un long fichier, il repond `// ... [previous content] ...` et
# considere le travail fait. Avec --permission-mode acceptEdits, cet abrege est
# ecrit sur le disque sans qu'un diff soit montre : 1841 lignes de source ont
# ainsi disparu. Le modele ne sait pas qu'il a detruit quelque chose, donc il
# rapporte un succes.
#
# C'est un defaut de capacite, pas de configuration : aucun reglage de vLLM ne
# le corrige. Ce qu'on peut faire, en revanche, c'est refuser l'ecriture. Le
# pont voit passer tous les arguments d'outil ; il est le dernier endroit ou
# l'on puisse transformer une destruction silencieuse en refus visible.

GUARD = os.environ.get("VL_GUARD", "on") != "off"

# Les outils dont un argument atterrit tel quel dans un fichier.
# Seules les ecritures de fichier ENTIER sont gardees. Les editions ciblees
# (Edit, MultiEdit, str_replace_*, NotebookEdit) portent legitimement des
# "..." dans old_string/new_string -- et le message du garde leur disait
# justement d'utiliser une edition ciblee : boucle de refus, observee le 22/08.
_GUARDED = {"write", "write_file", "create_file"}

# Marqueurs d'omission EXPLICITES seulement. Les motifs generiques "[...]" et
# "# ..." bloquaient des contenus legitimes (un .md avec une ligne "# ...", un
# script contenant "[...]") : faux positifs observes le 22/08 sur une vraie
# session Claude Code.
PLACEHOLDER = re.compile(
    r"\.\.\.\s*(\[|#|//|--)?\s*(previous|rest of|remaining|existing|unchanged|original)"
    r"|\[\s*\.\.\.\s*(previous|rest|remaining|existing|unchanged|original|snip)[^\]]*\]"
    r"|<\s*(unchanged|snip|elided)\s*>"
    r"|\(\s*(reste|suite) (du|des) ",
    re.I | re.M,
)

# Rappel injecte en tete de systeme. Ne remplace pas le garde-fou -- un modele
# qui abrege le fait souvent malgre la consigne -- mais reduit la frequence, et
# ne coute qu'une constante en tete de prefixe, donc le cache reste partage.
SYSTEM_GUARD = (
    "Quand tu ecris un fichier, l'argument `content` doit contenir le fichier "
    "INTEGRAL. N'ecris jamais de marqueur d'omission du type "
    "\"... [previous content] ...\", \"// ... rest of file ...\" ou "
    "\"(reste du fichier inchange)\" : ces marqueurs sont ecrits litteralement "
    "sur le disque et detruisent le fichier. Si le fichier est trop long pour "
    "etre reemis en entier, dis-le et utilise une edition ciblee."
)


def guard_violation(name: str, args: Any) -> str | None:
    """Renvoie la description de l'abreviation trouvee, ou None."""
    if not GUARD or (name or "").lower() not in _GUARDED:
        return None

    def walk(v: Any, path: str) -> str | None:
        if isinstance(v, str):
            m = PLACEHOLDER.search(v)
            return f"{path} contient {m.group(0).strip()!r}" if m else None
        if isinstance(v, dict):
            for k, sub in v.items():
                hit = walk(sub, f"{path}.{k}")
                if hit:
                    return hit
        elif isinstance(v, list):
            for i, sub in enumerate(v):
                hit = walk(sub, f"{path}[{i}]")
                if hit:
                    return hit
        return None

    return walk(args, name)


def guard_message(detail: str) -> str:
    return (
        "\n\n[garde-fou du pont] Appel d'outil bloque : " + detail + ".\n"
        "Le contenu propose abrege le fichier par un marqueur d'omission, ce qui "
        "l'aurait ecrase par une version incomplete. L'ecriture n'a PAS eu lieu et "
        "le fichier est intact.\n"
        "Reemets le fichier integral, ou fais une edition ciblee qui ne remplace "
        "que les lignes concernees."
    )


# --------------------------------------------------------------- Anthropic -> OpenAI

def _image_part(block: dict) -> dict | None:
    """Traduit un bloc `image` Anthropic vers la partie `image_url` d'OpenAI.

    Anthropic decrit l'image par une `source` typee : `base64` porte les octets
    et le type MIME separement, `url` porte un lien. OpenAI attend dans les deux
    cas UNE chaine dans `image_url.url` -- une URI de donnees pour le premier,
    le lien tel quel pour le second.

    Sans cette traduction, `_text_of` ignorait purement et simplement les blocs
    image : une capture collee dans Claude Code arrivait au modele comme un
    message vide, et le modele repondait a cote sans que rien ne signale la
    perte.
    """
    src = block.get("source") or {}
    kind = src.get("type")
    if kind == "base64":
        data = src.get("data")
        if not data:
            return None
        mime = src.get("media_type") or "image/png"
        return {"type": "image_url",
                "image_url": {"url": f"data:{mime};base64,{data}"}}
    if kind == "url" and src.get("url"):
        return {"type": "image_url", "image_url": {"url": src["url"]}}
    return None


def _text_of(content: Any) -> str:
    """Anthropic autorise une chaine ou une liste de blocs typés."""
    if isinstance(content, str):
        return content
    if not isinstance(content, list):
        return ""
    out = []
    for b in content:
        if isinstance(b, str):
            out.append(b)
        elif isinstance(b, dict) and b.get("type") == "text":
            out.append(b.get("text", ""))
    return "".join(out)


def to_openai(body: dict) -> dict:
    msgs: list[dict] = []

    # `system` est un champ separe chez Anthropic, un message de role chez OpenAI.
    # Le rappel anti-abreviation n'est ajoute qu'en presence d'outils : sans
    # outil, aucun contenu n'atteint le disque et la consigne serait du bruit.
    sys = _text_of(body.get("system") or "")
    if GUARD and body.get("tools"):
        sys = (sys + "\n\n" + SYSTEM_GUARD) if sys else SYSTEM_GUARD
    if sys:
        msgs.append({"role": "system", "content": sys})

    for m in body.get("messages", []):
        role = m.get("role", "user")
        content = m.get("content")

        # Claude Code place un SECOND message systeme en fin de conversation
        # (rappel de contexte). Le gabarit Qwen n'accepte le role systeme qu'en
        # tete et vLLM refuse toute la requete :
        #   400 "System message must be at the beginning."
        # Le pont ne verifiait pas le statut amont, donc l'echec ressortait en
        # reponse VIDE avec stop_reason=end_turn -- Claude Code affichait
        # "Cogitated for 0s" et rien d'autre.
        #
        # On le convertit en message utilisateur plutot que de le fusionner dans
        # le systeme de tete : ce rappel change a chaque tour, et le remonter en
        # tete invaliderait le prefixe partage, donc le cache -- soit 68 % des
        # blocs, mesures.
        if role == "system" and msgs and msgs[0].get("role") == "system":
            txt = _text_of(content)
            if txt:
                msgs.append({"role": "user", "content": txt})
            continue

        if isinstance(content, list):
            # Un tour d'assistant peut melanger du texte et des tool_use ; un tour
            # d'utilisateur porte les tool_result. OpenAI separe les deux en
            # `tool_calls` sur l'assistant et en messages de role `tool`.
            texts, calls, results, images, thinks = [], [], [], [], []
            for b in content:
                if not isinstance(b, dict):
                    continue
                t = b.get("type")
                if t == "text":
                    texts.append(b.get("text", ""))
                elif t == "thinking":
                    # Claude Code renvoie le raisonnement du tour precedent dans
                    # l'historique (boucles d'outils). Le gabarit Qwen3.5 sait le
                    # garder pour le DERNIER tour d'assistant via
                    # `reasoning_content` -- exactement la semantique Anthropic.
                    # Avant : bloc ignore en silence, le modele perdait son plan
                    # entre deux appels d'outil.
                    thinks.append(b.get("thinking", ""))
                elif t == "image":
                    part = _image_part(b)
                    if part:
                        images.append(part)
                elif t == "tool_use":
                    calls.append({
                        "id": b.get("id") or f"call_{uuid.uuid4().hex[:8]}",
                        "type": "function",
                        "function": {
                            "name": b.get("name", ""),
                            "arguments": json.dumps(b.get("input") or {}),
                        },
                    })
                elif t == "tool_result":
                    # Un resultat d'outil peut porter des images (capture rendue
                    # par un outil). Le role `tool` d'OpenAI n'accepte que du
                    # texte : on extrait les images pour les rattacher au tour
                    # utilisateur, sinon elles disparaissent en silence.
                    rc = b.get("content")
                    if isinstance(rc, list):
                        for sub in rc:
                            if isinstance(sub, dict) and sub.get("type") == "image":
                                part = _image_part(sub)
                                if part:
                                    images.append(part)
                    results.append({
                        "role": "tool",
                        "tool_call_id": b.get("tool_use_id", ""),
                        "content": _text_of(rc) or "",
                    })

            if role == "assistant":
                a: dict[str, Any] = {"role": "assistant", "content": "".join(texts) or None}
                if thinks:
                    a["reasoning_content"] = chr(10).join(t for t in thinks if t)
                if calls:
                    a["tool_calls"] = calls
                msgs.append(a)
            else:
                if images:
                    # Contenu multipart : OpenAI n'accepte les images que dans
                    # une LISTE de parties, jamais dans une chaine.
                    parts: list[dict] = []
                    joined = "".join(texts)
                    if joined:
                        parts.append({"type": "text", "text": joined})
                    parts.extend(images)
                    msgs.append({"role": "user", "content": parts})
                elif texts:
                    msgs.append({"role": "user", "content": "".join(texts)})
                msgs.extend(results)
        else:
            msgs.append({"role": role, "content": content or ""})

    out: dict[str, Any] = {
        "model": _courant()[1],
        "messages": msgs,
        "max_tokens": min(int(body.get("max_tokens") or 4096), MAX_OUTPUT),
        "stream": bool(body.get("stream")),
    }
    for src, dst in (("temperature", "temperature"), ("top_p", "top_p"),
                     ("stop_sequences", "stop")):
        if body.get(src) is not None:
            out[dst] = body[src]

    # Le raisonnement est ACTIF par defaut sur ce modele et consomme des jetons
    # avant le premier caractere de reponse : une requete a max_tokens=40 revient
    # avec content vide et stop_reason=max_tokens. Le protocole Anthropic exprime
    # la coupure par `thinking: {"type": "disabled"}` ; le gabarit Qwen l'attend
    # sous la forme enable_thinking=false, qui fait prefixer un bloc <think> deja
    # ferme. Sans cette traduction, le champ etait recu puis ignore en silence :
    # la case "raisonnement" de la console ne changeait rien.
    # `thinking` a plusieurs formes. La documentation du protocole de passerelle
    # precise que Claude Code envoie `{"type": "adaptive"}` aux modeles recents
    # ET "traite les noms de modeles qu'il ne reconnait pas, tels les alias de
    # passerelle, comme des modeles actuels qui recoivent le champ" -- donc nous.
    # Seul "disabled" doit couper le raisonnement ; "adaptive" et "enabled" le
    # laissent actif, qui est le defaut du gabarit.
    th = body.get("thinking")
    if isinstance(th, dict) and th.get("type") == "disabled":
        out["chat_template_kwargs"] = {"enable_thinking": False}
    elif isinstance(th, dict) and th.get("budget_tokens"):
        # Le niveau de raisonnement de Claude Code arrive ici, en jetons. vLLM
        # 0.27.1 l'applique via `thinking_token_budget` (sampling params) : il
        # ferme le bloc de raisonnement a ce nombre de jetons. Mesure : sans
        # ce champ, budget 512 et budget 8000 donnaient la meme chose.
        try:
            out["thinking_token_budget"] = max(1, int(th["budget_tokens"]))
        except (TypeError, ValueError):
            pass
    # Niveau d'effort (output_config.effort) -> plafond de jetons de raisonnement,
    # memes seuils que le patch natif (vllm_anthropic_effort_patch.py). Le budget
    # explicite ci-dessus gagne ; `disabled` aussi.
    _eff = (body.get("output_config") or {}).get("effort") if isinstance(body.get("output_config"), dict) else None
    _BUDGET = {"low": 1024, "medium": 4096, "high": 16384, "xhigh": 32768, "max": None}
    if _eff in _BUDGET and "thinking_token_budget" not in out and "chat_template_kwargs" not in out:
        if _BUDGET[_eff] is not None:
            out["thinking_token_budget"] = _BUDGET[_eff]

    if body.get("tools"):
        out["tools"] = [{
            "type": "function",
            "function": {
                "name": t["name"],
                "description": t.get("description", ""),
                "parameters": t.get("input_schema") or {"type": "object", "properties": {}},
            },
        } for t in body["tools"] if t.get("name")]
        tc = body.get("tool_choice") or {}
        kind = tc.get("type") if isinstance(tc, dict) else None
        if kind == "none":
            out["tool_choice"] = "none"
        elif kind == "any":
            out["tool_choice"] = "required"
        elif kind == "tool" and tc.get("name"):
            out["tool_choice"] = {"type": "function", "function": {"name": tc["name"]}}
        else:
            out["tool_choice"] = "auto"
    return out


_STOP = {"stop": "end_turn", "length": "max_tokens", "tool_calls": "tool_use"}


# --------------------------------------------------------------- OpenAI -> Anthropic

def _usage_of(usage: dict) -> dict:
    """vLLM expose les jetons servis par le cache de prefixe dans
    `prompt_tokens_details.cached_tokens` ; Anthropic les appelle
    `cache_read_input_tokens`. Sans cette traduction Claude Code affichait
    0 % de cache alors que vLLM en servait 68 %."""
    u = {"input_tokens": usage.get("prompt_tokens", 0) or 0,
         "output_tokens": usage.get("completion_tokens", 0) or 0}
    det = usage.get("prompt_tokens_details") or {}
    cached = det.get("cached_tokens") if isinstance(det, dict) else None
    if cached:
        u["cache_read_input_tokens"] = int(cached)
        u["cache_creation_input_tokens"] = 0
    return u


def _stop_of(choice: dict) -> tuple[str, str | None]:
    """finish_reason=stop couvre deux cas Anthropic : end_turn, ou
    stop_sequence quand vLLM rapporte la chaine d'arret dans `stop_reason`."""
    fr = choice.get("finish_reason") or "stop"
    sr = choice.get("stop_reason")
    if fr == "stop" and isinstance(sr, str) and sr:
        return "stop_sequence", sr
    return _STOP.get(fr, "end_turn"), None


def to_anthropic(oai: dict, req_model: str) -> dict:
    choice = (oai.get("choices") or [{}])[0]
    msg = choice.get("message") or {}
    blocks: list[dict] = []

    # Le raisonnement arrive ici en non-flux ; en flux il est deja traduit en
    # bloc thinking. Meme forme dans les deux cas, signature comprise.
    reasoning = msg.get("reasoning_content") or msg.get("reasoning")
    if reasoning:
        blocks.append({"type": "thinking", "thinking": reasoning,
                       "signature": "vllm-" + uuid.uuid4().hex[:16]})
    if msg.get("content"):
        blocks.append({"type": "text", "text": msg["content"]})
    elif reasoning and not msg.get("tool_calls"):
        # meme repli qu'en flux : pas de texte, pas d'outil -> le raisonnement
        blocks.append({"type": "text", "text": str(reasoning).strip()})

    blocked = False
    for c in msg.get("tool_calls") or []:
        fn = c.get("function") or {}
        try:
            args = json.loads(fn.get("arguments") or "{}")
        except json.JSONDecodeError:
            # Un modele quantifie peut emettre du JSON legerement casse ; mieux
            # vaut transmettre la chaine brute que faire tomber la requete.
            args = {"_raw": fn.get("arguments", "")}
        name = fn.get("name", "")
        detail = guard_violation(name, args)
        if detail:
            blocked = True
            blocks.append({"type": "text", "text": guard_message(detail)})
            continue
        blocks.append({
            "type": "tool_use",
            "id": c.get("id") or f"toolu_{uuid.uuid4().hex[:16]}",
            "name": name,
            "input": args,
        })

    usage = oai.get("usage") or {}
    if blocked and not any(b["type"] == "tool_use" for b in blocks):
        # Plus aucun outil a executer : le tour se termine sur l'explication.
        return {
            "id": oai.get("id") or f"msg_{uuid.uuid4().hex[:24]}",
            "type": "message", "role": "assistant", "model": req_model,
            "content": blocks, "stop_reason": "end_turn", "stop_sequence": None,
            "usage": _usage_of(usage),
        }
    stop_reason, stop_seq = _stop_of(choice)
    return {
        "id": oai.get("id") or f"msg_{uuid.uuid4().hex[:24]}",
        "type": "message",
        "role": "assistant",
        "model": req_model,
        "content": blocks,
        "stop_reason": stop_reason,
        "stop_sequence": stop_seq,
        "usage": _usage_of(usage),
    }


def _err_of(body: str, status: int) -> dict:
    """Renvoie l'objet d'erreur amont intact, ou en fabrique un equivalent.

    Le libelle compte : c'est sur lui que le client decide s'il peut retenter.
    """
    try:
        d = json.loads(body)
        e = d.get("error")
        if isinstance(e, dict) and e.get("message"):
            return {"type": e.get("type") or "api_error", "message": e["message"]}
    except (json.JSONDecodeError, AttributeError):
        pass
    return {"type": "api_error", "message": body or f"upstream {status}"}


def _sse(event: str, data: dict) -> bytes:
    return f"event: {event}\ndata: {json.dumps(data, separators=(',', ':'))}\n\n".encode()


async def _flux_gueri(gen):
    """v73 : les exceptions nees dans un generateur SSE ne traversent pas le
    middleware _guerir_bassin. On les attrape ici : evenement `error` au client
    (sa logique de reprise lit le libelle) et reconstruction du bassin."""
    generation = _bassin_generation
    try:
        async for morceau in gen:
            yield morceau
    except (httpx.HTTPError, RuntimeError) as e:
        import traceback
        print(f"[bassin?] flux: {type(e).__name__}: {e}", flush=True)
        traceback.print_exc()
        asyncio.ensure_future(_reconstruire_bassin(generation))
        yield _sse("error", {
            "type": "error",
            "error": {"type": "overloaded_error",
                      "message": f"upstream stream failed "
                                 f"({type(e).__name__}); retry"}})


async def stream_anthropic(payload: dict, req_model: str, request: Request | None = None):
    """Traduit le flux OpenAI en evenements Anthropic.

    Le point delicat est l'indexation des blocs : Anthropic numerote chaque bloc
    de contenu et exige un content_block_start/stop apparie, alors qu'OpenAI
    emet des deltas plats. On ouvre donc un bloc texte a la volee, et un bloc
    tool_use par appel, en fermant le precedent.

    Les arguments d'outil, eux, sont accumules et n'emis qu'une fois complets.
    Il le faut : un argument diffuse fragment par fragment ne peut pas etre
    valide, et le client aurait deja commence a ecrire le fichier quand le
    marqueur d'omission apparait.

    Mais accumuler veut dire n'envoyer AUCUN octet pendant toute la generation
    de l'appel, et le proxy de Runpod coupe une connexion inactive vers 125 s.
    MESURE : une reecriture de 1354 lignes par l'outil Write mourait a 126 s
    avec zero token et stop=None, alors que la meme reecriture en prose passait
    (le texte, lui, est diffuse au fil de l'eau). Un test voisin est passe de
    justesse a 123,6 s. On emet donc un `ping` periodique tant qu'on accumule :
    le protocole Anthropic le prevoit, les clients l'ignorent, et la connexion
    reste ouverte sans que le garde-fou perde sa capacite a valider avant
    emission.
    """
    mid = f"msg_{uuid.uuid4().hex[:24]}"
    yield _sse("message_start", {
        "type": "message_start",
        "message": {"id": mid, "type": "message", "role": "assistant",
                    "model": req_model, "content": [], "stop_reason": None,
                    "stop_sequence": None,
                    "usage": {"input_tokens": 0, "output_tokens": 0}},
    })
    yield _sse("ping", {"type": "ping"})

    idx = -1
    text_open = False
    think_open = False
    n_text = 0
    n_think = 0
    think_buf: list[str] = []   # raisonnement accumule, pour le repli "reponse vide"
    last_usage_sent = -1
    last_usage_ts = 0.0
    # Horodatage du dernier octet REELLEMENT envoye au client, pour savoir quand
    # le silence devient dangereux.
    last_out = time.monotonic()
    KEEPALIVE_S = 15.0
    tools: dict[int, dict] = {}         # index OpenAI -> {id, name, args}
    stop_reason = "end_turn"
    out_tokens = 0
    in_tokens = 0
    cached_tokens = 0
    stop_seq: str | None = None

    async with _client.stream("POST", "/v1/chat/completions", json=payload) as r:
        # Un 4xx amont renvoie du JSON d'erreur, pas des lignes "data:". Sans ce
        # controle, chaque ligne est ignoree, le flux se termine vide et le
        # client recoit un message sans contenu avec stop_reason=end_turn : une
        # panne parfaitement silencieuse, indiscernable d'un modele muet.
        if r.status_code >= 400:
            body = (await r.aread()).decode("utf-8", "ignore")[:800]
            roles = "/".join(m.get("role", "?") for m in payload.get("messages", []))
            print(f"[amont {r.status_code}] {body}", flush=True)
            print(f"[amont] roles={roles}", flush=True)
            # Le corps d'erreur repart TEL QUEL, dans un evenement SSE `error`.
            # La documentation du protocole de passerelle est explicite : Claude
            # Code se remet tout seul de certains refus -- champ `thinking`,
            # signatures de raisonnement, et justement les messages systeme en
            # milieu de conversation -- mais "la logique de reprise s'appuie sur
            # le libelle de l'erreur amont", et "une passerelle qui enveloppe
            # les erreurs dans son propre format casse la reprise meme si elle
            # preserve le code de statut". Emballer l'erreur dans un bloc texte,
            # comme je le faisais, empechait donc cette reprise.
            yield _sse("error", {"type": "error", "error": _err_of(body, r.status_code)})
            return
        # Le `ping` periodique ne suffisait pas : il ne se declenchait qu'a
        # l'interieur de cette boucle, donc uniquement quand vLLM emettait deja
        # quelque chose. Or pendant le PREREMPLISSAGE vLLM n'envoie rien -- et
        # un prompt de plusieurs centaines de milliers de jetons met plusieurs
        # minutes a etre calcule. Le flux restait muet, et le proxy Runpod
        # coupait vers 125 s.
        #
        # MESURE : une aiguille a ~700 k jetons echouait a 126,5 s avec zero
        # jeton, alors que la meme a ~358 k passait en 94,4 s. Le plafond
        # n'etait pas le modele mais le silence : aucune requete au-dela de
        # ~500 k ne pouvait aboutir, quel que soit le contexte annonce.
        #
        # On decouple donc la lecture amont de l'emission aval : une tache
        # pompe les lignes dans une file, et l'absence de ligne pendant
        # KEEPALIVE_S produit un `ping` au lieu d'un silence.
        file: asyncio.Queue = asyncio.Queue()

        async def _pompe() -> None:
            try:
                async for _l in r.aiter_lines():
                    await file.put(_l)
            finally:
                await file.put(None)

        _tache = asyncio.create_task(_pompe())
        try:
          while True:
            try:
                line = await asyncio.wait_for(file.get(), timeout=KEEPALIVE_S)
            except asyncio.TimeoutError:
                # Client parti pendant le prefill ? On ferme l'amont (vLLM
                # annule alors la generation) au lieu de pomper dans le vide.
                if request is not None and await request.is_disconnected():
                    print("[deconnexion] client parti pendant l'attente, amont ferme", flush=True)
                    break
                yield _sse("ping", {"type": "ping"})
                last_out = time.monotonic()
                continue
            if line is None:
                break
            # Le `yield` ne leve pas toujours a la deconnexion (le proxy Runpod
            # garde parfois sa connexion ouverte) : on verifie explicitement.
            if request is not None and (out_tokens % 32 == 0) and await request.is_disconnected():
                print(f"[deconnexion] client parti apres {out_tokens} jetons, amont ferme", flush=True)
                break
            if not line.startswith("data: "):
                continue
            chunk = line[6:].strip()
            if chunk == "[DONE]":
                break
            try:
                d = json.loads(chunk)
            except json.JSONDecodeError:
                continue

            if d.get("usage"):
                in_tokens = d["usage"].get("prompt_tokens", in_tokens)
                out_tokens = d["usage"].get("completion_tokens", out_tokens)
                _det = d["usage"].get("prompt_tokens_details") or {}
                if isinstance(_det, dict) and _det.get("cached_tokens"):
                    cached_tokens = int(_det["cached_tokens"])

            ch = (d.get("choices") or [{}])[0]
            delta = ch.get("delta") or {}

            # Comptage en direct facon Anthropic : vLLM (continuous_usage_stats)
            # renvoie le total cumule sur chaque chunk (lu ci-dessus). On le
            # relaie dans un message_delta periodique pour que le compteur du
            # client monte en continu au lieu de sauter au total a la fin.
            if out_tokens and out_tokens != last_usage_sent and (time.monotonic() - last_usage_ts) >= 0.5:
                yield _sse("message_delta", {"type": "message_delta",
                           "delta": {"stop_reason": None, "stop_sequence": None},
                           "usage": {"input_tokens": in_tokens, "output_tokens": out_tokens}})
                last_usage_sent = out_tokens; last_usage_ts = time.monotonic(); last_out = time.monotonic()

            # vLLM range le raisonnement dans un champ separe. Sans ce relais,
            # il est simplement perdu : le client ne peut ni l'afficher ni savoir
            # pourquoi la reponse tarde.
            if (delta.get("reasoning_content") or delta.get("reasoning")):
                if not think_open:
                    if text_open:
                        yield _sse("content_block_stop",
                                   {"type": "content_block_stop", "index": idx})
                        text_open = False
                    idx += 1
                    think_open = True
                    yield _sse("content_block_start", {
                        "type": "content_block_start", "index": idx,
                        "content_block": {"type": "thinking", "thinking": ""}})
                yield _sse("content_block_delta", {
                    "type": "content_block_delta", "index": idx,
                    "delta": {"type": "thinking_delta",
                              "thinking": (delta.get("reasoning_content") or delta.get("reasoning"))}})
                n_think += len((delta.get("reasoning_content") or delta.get("reasoning")))
                think_buf.append(delta.get("reasoning_content") or delta.get("reasoning"))
                last_out = time.monotonic()

            if delta.get("content"):
                if think_open:
                    yield _sse("content_block_delta", {
                        "type": "content_block_delta", "index": idx,
                        "delta": {"type": "signature_delta",
                                  "signature": "vllm-" + uuid.uuid4().hex[:16]}})
                    yield _sse("content_block_stop",
                               {"type": "content_block_stop", "index": idx})
                    think_open = False
                if not text_open:
                    idx += 1
                    text_open = True
                    yield _sse("content_block_start", {
                        "type": "content_block_start", "index": idx,
                        "content_block": {"type": "text", "text": ""}})
                yield _sse("content_block_delta", {
                    "type": "content_block_delta", "index": idx,
                    "delta": {"type": "text_delta", "text": delta["content"]}})
                n_text += len(delta["content"])
                last_out = time.monotonic()

            for tc in delta.get("tool_calls") or []:
                i = tc.get("index", 0)
                fn = tc.get("function") or {}
                t = tools.setdefault(i, {"id": None, "name": "", "args": ""})
                if tc.get("id"):
                    t["id"] = tc["id"]
                if fn.get("name"):
                    t["name"] = fn["name"]
                if fn.get("arguments"):
                    t["args"] += fn["arguments"]

            # Le silence de l'accumulation est ce qui tuait la connexion.
            if time.monotonic() - last_out > KEEPALIVE_S:
                yield _sse("ping", {"type": "ping"})
                last_out = time.monotonic()

            if ch.get("finish_reason"):
                stop_reason, stop_seq = _stop_of(ch)
        finally:
            _tache.cancel()

    if think_open:
        # Structure Anthropic complete : la signature precede la fermeture du
        # bloc. Claude Code la renvoie telle quelle dans l'historique ; le pont
        # ne la verifie pas, mais un client strict l'attend.
        yield _sse("content_block_delta", {
            "type": "content_block_delta", "index": idx,
            "delta": {"type": "signature_delta",
                      "signature": "vllm-" + uuid.uuid4().hex[:16]}})
        yield _sse("content_block_stop", {"type": "content_block_stop", "index": idx})
        think_open = False
    if text_open:
        yield _sse("content_block_stop", {"type": "content_block_stop", "index": idx})
        text_open = False

    emitted = 0
    for i in sorted(tools):
        t = tools[i]
        try:
            args = json.loads(t["args"] or "{}")
        except json.JSONDecodeError:
            args = {"_raw": t["args"]}

        detail = guard_violation(t["name"], args)
        if detail:
            print(f"[garde-fou] appel bloque : {detail}", flush=True)
            idx += 1
            yield _sse("content_block_start", {
                "type": "content_block_start", "index": idx,
                "content_block": {"type": "text", "text": ""}})
            yield _sse("content_block_delta", {
                "type": "content_block_delta", "index": idx,
                "delta": {"type": "text_delta", "text": guard_message(detail)}})
            yield _sse("content_block_stop",
                       {"type": "content_block_stop", "index": idx})
            continue

        idx += 1
        emitted += 1
        yield _sse("content_block_start", {
            "type": "content_block_start", "index": idx,
            "content_block": {"type": "tool_use",
                              "id": t["id"] or f"toolu_{uuid.uuid4().hex[:16]}",
                              "name": t["name"], "input": {}}})
        # Un seul fragment : l'argument est deja complet et valide.
        yield _sse("content_block_delta", {
            "type": "content_block_delta", "index": idx,
            "delta": {"type": "input_json_delta",
                      "partial_json": json.dumps(args, separators=(",", ":"))}})
        yield _sse("content_block_stop", {"type": "content_block_stop", "index": idx})

    # Annoncer `tool_use` sans avoir emis d'outil ferait attendre au client un
    # resultat qui ne viendra jamais.
    if stop_reason == "tool_use" and emitted == 0:
        stop_reason = "end_turn"

    # Repli "reponse vide" (22/08 nuit) : la compaction automatique de Claude
    # Code a echoue sur "summarization produced empty response" -- le modele
    # avait tout ecrit dans le raisonnement (1 679 car.) et rien en texte. Un
    # client qui attend du texte ne sait rien faire d'un bloc thinking seul :
    # on rend alors le raisonnement comme texte, clairement marque.
    if n_text == 0 and emitted == 0 and "".join(think_buf).strip():
        idx += 1
        yield _sse("content_block_start", {"type": "content_block_start", "index": idx,
                                           "content_block": {"type": "text", "text": ""}})
        yield _sse("content_block_delta", {"type": "content_block_delta", "index": idx,
                                           "delta": {"type": "text_delta", "text": "".join(think_buf).strip()}})
        yield _sse("content_block_stop", {"type": "content_block_stop", "index": idx})
        n_text = len("".join(think_buf).strip())
        print("[repli] reponse sans texte ni outil : raisonnement rendu en texte", flush=True)

    # Trace compacte d'une requete. Sans elle, un client qui n'affiche rien est
    # indiscernable d'un modele qui ne repond rien : les deux donnent un 200 OK
    # dans le journal d'acces.
    if os.environ.get("VL_TRACE", "on") != "off":
        print(f"[trace] entree={in_tokens} sortie={out_tokens} "
              f"texte={n_text}c raisonnement={n_think}c outils={emitted} "
              f"stop={stop_reason}", flush=True)

    _u = {"input_tokens": in_tokens, "output_tokens": out_tokens}
    if cached_tokens:
        _u["cache_read_input_tokens"] = cached_tokens
        _u["cache_creation_input_tokens"] = 0
    yield _sse("message_delta", {
        "type": "message_delta",
        "delta": {"stop_reason": stop_reason, "stop_sequence": stop_seq},
        "usage": _u})
    yield _sse("message_stop", {"type": "message_stop"})


# ------------------------------------------------------------------------ routes

@app.post("/v1/messages")
async def messages(request: Request):
    body = await request.json()
    req_model = body.get("model", MODEL)
    payload = to_openai(body)

    if payload.get("stream"):
        payload["stream_options"] = {"include_usage": True, "continuous_usage_stats": True}

        async def flux_avec_swap():
            # Le proxy RunPod coupe a ~120 s toute reponse sans octet ; un swap
            # de modele en dure 2 a 6. On tient la ligne avec des pings SSE
            # pendant l'attente, puis on enchaine sur le flux normal.
            tache = asyncio.ensure_future(_assurer_modele(req_model))
            while not tache.done():
                await asyncio.wait({tache}, timeout=15)
                if not tache.done():
                    yield _sse("ping", {"type": "ping"})
            try:
                payload["model"] = tache.result()
            except HTTPException as e:
                yield _sse("error", {"type": "error",
                                     "error": {"type": "overloaded_error", "message": str(e.detail)}})
                return
            async for morceau in stream_anthropic(payload, req_model, request):
                yield morceau

        return StreamingResponse(_flux_gueri(flux_avec_swap()),
                                 media_type="text/event-stream")
    payload["model"] = await _assurer_modele(req_model)

    r = await _client.post("/v1/chat/completions", json=payload)
    if r.status_code != 200:
        return JSONResponse(status_code=r.status_code,
                            content={"type": "error",
                                     "error": {"type": "api_error", "message": r.text[:800]}})
    return JSONResponse(to_anthropic(r.json(), req_model))


@app.post("/v1/messages/count_tokens")
async def count_tokens(request: Request):
    """Claude Code interroge ce point avant d'envoyer. Une estimation suffit :
    il s'en sert pour decider de compacter, pas pour facturer."""
    body = await request.json()
    # Compte REEL par vLLM (/tokenize en forme chat : gabarit, systeme et
    # outils compris). L'estimation len/4 ignorait systeme et outils : 13
    # jetons pour un appel qui en pesait bien plus. Repli sur l'estimation si
    # le tokenizer amont ne repond pas.
    try:
        payload = to_openai(body)
        req = {"model": payload.get("model"), "messages": payload.get("messages", []),
               "add_generation_prompt": True}
        if payload.get("tools"):
            req["tools"] = payload["tools"]
        r = await _client.post("/tokenize", json=req, timeout=60)
        if r.status_code == 200:
            n = r.json().get("count")
            if isinstance(n, int) and n > 0:
                return JSONResponse({"input_tokens": n})
    except Exception:                                            # noqa: BLE001
        pass
    n = len(json.dumps(body.get("messages", []))) + len(json.dumps(body.get("system", ""))) \
        + len(json.dumps(body.get("tools", [])))
    return JSONResponse({"input_tokens": max(1, n // 4)})


# ------------------------------------------------------------------- console
# Une page servie par le pont lui-meme, donc de meme origine que /v1/messages :
# un fichier ouvert depuis le disque, ou une page hebergee ailleurs, se ferait
# refuser la requete. C'est aussi la raison pour laquelle le pod n'a besoin que
# d'un seul port ouvert.

@app.get("/")
async def console():
    from fastapi.responses import HTMLResponse, PlainTextResponse
    path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "console.html")
    try:
        with open(path, encoding="utf-8") as fh:
            return HTMLResponse(fh.read())
    except OSError:
        return PlainTextResponse(
            "console.html absente a cote du pont.\n"
            "Le bootstrap la telecharge depuis le Hub au demarrage.\n"
            f"Attendue ici : {path}\n", status_code=503)


@app.get("/health")
async def health():
    try:
        r = await _client.get("/health", timeout=5)
        ok = r.status_code == 200
    except Exception:
        ok = False
    return JSONResponse({"status": "ok" if ok else "upstream_down",
                         "upstream": UPSTREAM, "model": MODEL, "ts": time.time()},
                        status_code=200 if ok else 503)


@app.get("/v1/models")
async def models(request: Request):
    """Deux protocoles sur la meme route.

    Un client Anthropic attend `{"data":[{"type":"model","id":...}]}` avec des
    identifiants en "claude-*". Un client OpenAI ou un banc attend la reponse
    de vLLM, dont il lit `max_model_len`. On distingue sur l'en-tete
    `anthropic-version`, et on renvoie a chacun ce qu'il sait lire.
    """
    if "anthropic-version" not in request.headers:
        try:
            r = await _client.get("/v1/models", timeout=2.0)
            if r.status_code == 200:
                return JSONResponse(r.json())
        except Exception:
            pass

    ctx = None
    try:
        r = await _client.get("/v1/models", timeout=2.0)
        if r.status_code == 200:
            ctx = (r.json().get("data") or [{}])[0].get("max_model_len")
    except Exception:
        pass

    data = [{
        "type": "model",
        "id": a,
        # Le nom affiche porte le depot exact : c'est la seule facon pour un
        # utilisateur de savoir quel modele repond derriere une etiquette.
        "display_name": f"{a} -> {_courant()[2]}",
        "created_at": "2026-01-01T00:00:00Z",
        **({"context_window": ctx} if ctx else {}),
    } for a in ALIASES]
    return JSONResponse({"data": data, "has_more": False,
                         "first_id": data[0]["id"], "last_id": data[-1]["id"]})


@app.get("/v1/models/{model_id}")
async def model_detail(model_id: str):
    return JSONResponse({"type": "model", "id": model_id,
                         "display_name": f"{model_id} ({MODEL})",
                         "created_at": "2026-01-01T00:00:00Z"})

# ------------------------------------------------------------------ passe-plat
# Le pod n'expose qu'un port. Le pont le prend (c'est l'API que Claude Code
# consomme) et relaie ces routes vers vLLM, pour que les outils de mesure et les
# clients OpenAI restent joignables de l'exterieur.

def _real_model(body: dict) -> dict:
    """Traduit un alias claude-* vers le nom servi par vLLM.

    Sans ca, un harnais de mesure qui reprend l'identifiant vu dans /v1/models
    (`claude-ornith`) recoit un 404 "model not found" de vLLM, releve comme une
    route absente alors que seul le nom etait inconnu.
    """
    m = body.get("model")
    if isinstance(m, str) and (m in ALIASES or _cle_demandee(m)):
        body = {**body, "model": _courant()[1]}
    return body


@app.post("/v1/chat/completions")
async def oai_chat(request: Request):
    body = await request.json()
    await _assurer_modele(body.get("model"))
    body = _real_model(body)
    if body.get("stream"):
        async def gen():
            async with _client.stream("POST", "/v1/chat/completions", json=body) as r:
                async for chunk in r.aiter_raw():
                    yield chunk
        return StreamingResponse(_flux_gueri(gen()),
                                 media_type="text/event-stream")
    r = await _client.post("/v1/chat/completions", json=body)
    return JSONResponse(status_code=r.status_code, content=r.json())


@app.post("/v1/completions")
async def oai_completions(request: Request):
    body = await request.json()
    await _assurer_modele(body.get("model"))
    body = _real_model(body)
    r = await _client.post("/v1/completions", json=body)
    return JSONResponse(status_code=r.status_code, content=r.json())


@app.get("/metrics")
async def metrics():
    from fastapi.responses import PlainTextResponse
    r = await _client.get("/metrics")
    return PlainTextResponse(r.text, status_code=r.status_code)