From ac0459ffa5e4bf5542da3e9b314697d2c52c98a0 Mon Sep 17 00:00:00 2001 From: Florian Krebs Date: Fri, 22 May 2026 16:46:06 +0200 Subject: [PATCH] feat: effort levels, fallback model, cost/rate-limit metadata MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- server.py | 146 ++++++++++++++++++++++++++++++++++++++++++++++++++---- 1 file changed, 136 insertions(+), 10 deletions(-) diff --git a/server.py b/server.py index 93ba9dc..37bf7f6 100644 --- a/server.py +++ b/server.py @@ -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