fkrebs/claude-max-bridge
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: effort levels, fallback model, cost/rate-limit metadata

Browse Source 
- Temperature→effort mapping (0-0.2→low, 0.2-0.4→medium, 0.4-0.7→high,
  0.7-0.9→xhigh, >=0.9→max); overridable via extra_body.effort
- --fallback-model support via extra_body.fallback_model
- Real cost reporting: x_claude_cost_usd in response JSON (from CLI
  total_cost_usd field); x_claude_rate_limit with utilization + resetsAt
- Model aliases: sonnet, opus, haiku passed directly to --model flag
- rate_limit_event parsed and forwarded in done event
This commit is contained in:
Florian Krebs
2026-05-22 16:46:06 +02:00
parent 5f8a98772d
commit ac0459ffa5
1 changed files with 136 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server.py
+136 -10
View File
@@ -41,10 +41,17 @@ SUPPORTED_MODELS = [
"claude-opus-4-0",
"claude-sonnet-3-7",
"claude-haiku-3-5",
# Short aliases accepted by the CLI
"sonnet",
"opus",
"haiku",
]
DEFAULT_MODEL = "claude-sonnet-4-6"
# Valid effort levels accepted by the CLI
_VALID_EFFORT_LEVELS = {"low", "medium", "high", "xhigh", "max"}
# ---------------------------------------------------------------------------
# App
# ---------------------------------------------------------------------------
@@ -111,11 +118,34 @@ def _parse_messages(messages: list[dict]) -> tuple[Optional[str], list[dict]]:
return system_prompt, turns
# ---------------------------------------------------------------------------
# Helper: map temperature float to effort level string
# ---------------------------------------------------------------------------
def _temperature_to_effort(temperature: float) -> str:
if temperature < 0.2:
return "low"
elif temperature < 0.4:
return "medium"
elif temperature < 0.7:
return "high"
elif temperature < 0.9:
return "xhigh"
else:
return "max"
# ---------------------------------------------------------------------------
# Helper: build CLI command
# ---------------------------------------------------------------------------
def _build_command(model: str, system_prompt: Optional[str]) -> list[str]:
def _build_command(
model: str,
system_prompt: Optional[str],
effort: Optional[str] = None,
temperature: Optional[float] = None,
fallback_model: Optional[str] = None,
) -> list[str]:
cmd = [
CLAUDE_BIN,
"--print",
@@ -129,6 +159,21 @@ def _build_command(model: str, system_prompt: Optional[str]) -> list[str]:
"--tools", "",
"--model", model,
]
# Determine effort level to pass to CLI
resolved_effort: Optional[str] = None
if effort is not None and effort in _VALID_EFFORT_LEVELS:
resolved_effort = effort
elif temperature is not None:
resolved_effort = _temperature_to_effort(temperature)
# If both are None, don't add --effort (let CLI use its default)
if resolved_effort is not None:
cmd += ["--effort", resolved_effort]
if fallback_model is not None:
cmd += ["--fallback-model", fallback_model]
if system_prompt:
cmd += ["--system-prompt", system_prompt]
return cmd
@@ -142,14 +187,17 @@ async def _run_claude(
model: str,
system_prompt: Optional[str],
turns: list[dict],
effort: Optional[str] = None,
temperature: Optional[float] = None,
fallback_model: Optional[str] = None,
) -> AsyncGenerator[dict, None]:
"""
Yields dicts with keys:
{"event": "delta", "text": "..."}
{"event": "done", "usage": {...}}
{"event": "done", "usage": {...}, "cost_usd": float | None, "rate_limit": dict | None}
{"event": "error", "status": int, "message": "..."}
"""
cmd = _build_command(model, system_prompt)
cmd = _build_command(model, system_prompt, effort=effort, temperature=temperature, fallback_model=fallback_model)
log.debug("Spawning: %s", " ".join(cmd[:6]) + " ...")
env = _subprocess_env()
@@ -171,10 +219,12 @@ async def _run_claude(
last_text = ""
usage: dict = {}
cost_usd: Optional[float] = None
rate_limit_info: Optional[dict] = None
# Read stdout line by line with overall timeout
async def _read_lines():
nonlocal last_text, usage
nonlocal last_text, usage, cost_usd, rate_limit_info
assert proc is not None
assert proc.stdout is not None
async for raw_line in proc.stdout:
@@ -214,7 +264,24 @@ async def _run_claude(
+ raw_usage.get("output_tokens", 0)
),
}
yield {"event": "done", "usage": usage}
# Extract cost from result event
raw_cost = event.get("total_cost_usd")
if raw_cost is not None:
try:
cost_usd = float(raw_cost)
except (TypeError, ValueError):
pass
yield {
"event": "done",
"usage": usage,
"cost_usd": cost_usd,
"rate_limit": rate_limit_info,
}
elif etype == "rate_limit_event":
# Capture rate limit info for the done event
rate_limit_info = event.get("rate_limit_info", event)
log.debug("Rate limit event: %s", rate_limit_info)
elif etype == "system":
subtype = event.get("subtype", "")
@@ -275,13 +342,19 @@ async def _stream_response(
model: str,
system_prompt: Optional[str],
turns: list[dict],
effort: Optional[str] = None,
temperature: Optional[float] = None,
fallback_model: Optional[str] = None,
) -> AsyncGenerator[str, None]:
completion_id = f"chatcmpl-{uuid.uuid4().hex}"
created = int(time.time())
async def _gen():
got_done = False
async for ev in _run_claude(model, system_prompt, turns):
async for ev in _run_claude(
model, system_prompt, turns,
effort=effort, temperature=temperature, fallback_model=fallback_model,
):
# Check if client disconnected
if await request.is_disconnected():
log.info("Client disconnected, stopping stream")
@@ -321,6 +394,25 @@ async def _stream_response(
"usage": ev.get("usage", {}),
}
yield f"data: {json.dumps(final_chunk)}\n\n"
# Emit cost/rate-limit metadata chunk before [DONE]
cost_usd = ev.get("cost_usd")
rate_limit = ev.get("rate_limit")
if cost_usd is not None or rate_limit is not None:
meta_chunk = {
"id": completion_id,
"object": "chat.completion.chunk",
"created": created,
"model": model,
"choices": [],
"usage": ev.get("usage", {}),
}
if cost_usd is not None:
meta_chunk["x_claude_cost_usd"] = cost_usd
if rate_limit is not None:
meta_chunk["x_claude_rate_limit"] = rate_limit
yield f"data: {json.dumps(meta_chunk)}\n\n"
yield "data: [DONE]\n\n"
elif ev["event"] == "error":
@@ -390,8 +482,27 @@ async def chat_completions(request: Request):
stream = bool(body.get("stream", False))
# Extract temperature (kept for effort mapping; otherwise ignored)
temperature: Optional[float] = None
if "temperature" in body:
try:
temperature = float(body["temperature"])
except (TypeError, ValueError):
pass
# Extract effort from extra_body or top-level body
extra_body = body.get("extra_body") or {}
effort: Optional[str] = extra_body.get("effort") or body.get("effort")
if effort is not None:
effort = str(effort)
# Extract fallback_model from extra_body
fallback_model: Optional[str] = extra_body.get("fallback_model")
if fallback_model is not None:
fallback_model = str(fallback_model)
# Warn about unsupported / silently-ignored params
_IGNORED = {"temperature", "top_p", "frequency_penalty", "presence_penalty",
_IGNORED = {"top_p", "frequency_penalty", "presence_penalty",
"logit_bias", "n", "best_of", "logprobs", "top_logprobs",
"response_format", "seed", "stop", "user"}
for param in _IGNORED:
@@ -405,7 +516,10 @@ async def chat_completions(request: Request):
# --- Streaming ---
if stream:
gen = await _stream_response(request, model, system_prompt, turns)
gen = await _stream_response(
request, model, system_prompt, turns,
effort=effort, temperature=temperature, fallback_model=fallback_model,
)
return StreamingResponse(
gen,
media_type="text/event-stream",
@@ -418,13 +532,20 @@ async def chat_completions(request: Request):
# --- Non-streaming: buffer all deltas ---
full_text = ""
usage: dict = {}
cost_usd: Optional[float] = None
rate_limit_info: Optional[dict] = None
error_ev: Optional[dict] = None
async for ev in _run_claude(model, system_prompt, turns):
async for ev in _run_claude(
model, system_prompt, turns,
effort=effort, temperature=temperature, fallback_model=fallback_model,
):
if ev["event"] == "delta":
full_text += ev["text"]
elif ev["event"] == "done":
usage = ev.get("usage", {})
cost_usd = ev.get("cost_usd")
rate_limit_info = ev.get("rate_limit")
elif ev["event"] == "error":
error_ev = ev
break
@@ -441,7 +562,7 @@ async def chat_completions(request: Request):
)
completion_id = f"chatcmpl-{uuid.uuid4().hex}"
return {
response_body: dict = {
"id": completion_id,
"object": "chat.completion",
"created": int(time.time()),
@@ -455,3 +576,8 @@ async def chat_completions(request: Request):
],
"usage": usage or {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0},
}
if cost_usd is not None:
response_body["x_claude_cost_usd"] = cost_usd
if rate_limit_info is not None:
response_body["x_claude_rate_limit"] = rate_limit_info
return response_body