server.py

"""
Dual OpenAI + Anthropic compatible API proxy powered by Claude Code CLI (Max subscription).

Provides two fully spec-compliant API surfaces:
  - /openai/v1/*   — OpenAI API compatible (chat completions, completions, models)
  - /anthropic/v1/* — Anthropic API compatible (messages, models)

Both route through `claude -p` using your Claude Max subscription auth.

Usage:
    source ~/claude-proxy-venv/bin/activate
    python server.py [--port 4000] [--host 0.0.0.0]
"""

import asyncio
import datetime
import hmac
import json
import logging
import os
import re
import time
import uuid
import argparse
from contextlib import asynccontextmanager
from typing import Optional

from dotenv import load_dotenv
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse, JSONResponse
from fastapi.middleware.cors import CORSMiddleware
import uvicorn

load_dotenv()  # Load .env file if present (before any os.environ reads)

LOG_LEVEL = os.environ.get("CLAUDE_PROXY_LOG_LEVEL", "INFO").upper()
logging.basicConfig(level=getattr(logging, LOG_LEVEL, logging.INFO))
logger = logging.getLogger("claude-proxy")

# Track active subprocesses for graceful shutdown
_active_processes: set[asyncio.subprocess.Process] = set()
_shutting_down = False


SEMAPHORE_ACQUIRE_TIMEOUT = 30  # seconds to wait for a semaphore slot before returning 429


@asynccontextmanager
async def lifespan(app: FastAPI):
    """Manage server lifecycle — init per-worker state, clean up on shutdown."""
    # Initialize per-worker globals from env vars (needed for multi-worker mode)
    global CLI_TIMEOUT_SECONDS, MAX_CONCURRENT_REQUESTS, _semaphore
    if os.environ.get("_CLAUDE_PROXY_TIMEOUT"):
        CLI_TIMEOUT_SECONDS = int(os.environ["_CLAUDE_PROXY_TIMEOUT"])
    if os.environ.get("_CLAUDE_PROXY_MAX_CONCURRENT"):
        MAX_CONCURRENT_REQUESTS = int(os.environ["_CLAUDE_PROXY_MAX_CONCURRENT"])
    if _semaphore is None:
        _semaphore = asyncio.Semaphore(MAX_CONCURRENT_REQUESTS)
    yield
    global _shutting_down
    _shutting_down = True
    drain_seconds = int(os.environ.get("CLAUDE_PROXY_DRAIN_SECONDS", "5"))
    logger.info(f"Draining for {drain_seconds}s, then cleaning up {len(_active_processes)} active subprocess(es)...")
    await asyncio.sleep(drain_seconds)
    for proc in list(_active_processes):
        if proc.returncode is None:
            proc.terminate()
    await asyncio.sleep(5)
    for proc in list(_active_processes):
        if proc.returncode is None:
            proc.kill()
    for proc in list(_active_processes):
        try:
            await asyncio.wait_for(proc.wait(), timeout=5)
        except asyncio.TimeoutError:
            logger.warning(f"Process {proc.pid} did not exit after kill")
    logger.info("Shutdown complete.")


app = FastAPI(title="Claude Max API Proxy", version="3.1.0", lifespan=lifespan)

_cors_origins = os.environ.get("CLAUDE_PROXY_CORS_ORIGINS", "*").split(",")
app.add_middleware(
    CORSMiddleware,
    allow_origins=_cors_origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


MAX_REQUEST_BODY_BYTES = int(os.environ.get("CLAUDE_PROXY_MAX_BODY_BYTES", 10 * 1024 * 1024))  # 10MB default
_API_KEY = os.environ.get("CLAUDE_PROXY_API_KEY", "").strip()
_AUTH_EXEMPT_PATHS = {"/health", "/", "/docs", "/openapi.json"}


@app.middleware("http")
async def log_requests(request: Request, call_next):
    request_id = request.headers.get("X-Request-ID", uuid.uuid4().hex[:12])
    request.state.request_id = request_id
    start = time.time()

    # ── API key authentication (when CLAUDE_PROXY_API_KEY is set) ──
    if _API_KEY and request.url.path not in _AUTH_EXEMPT_PATHS:
        is_anthropic = request.url.path.startswith("/anthropic/")
        # Extract key from the appropriate header
        if is_anthropic:
            provided_key = request.headers.get("x-api-key", "")
        else:
            auth_header = request.headers.get("authorization", "")
            provided_key = auth_header.removeprefix("Bearer ").strip() if auth_header.startswith("Bearer ") else ""

        if not provided_key or not hmac.compare_digest(provided_key, _API_KEY):
            if is_anthropic:
                return JSONResponse(status_code=401, content={"type": "error", "error": {"type": "authentication_error", "message": "Invalid API key. Set the x-api-key header."}})
            return JSONResponse(status_code=401, content={"error": {"message": "Incorrect API key provided.", "type": "invalid_api_key", "param": None, "code": "invalid_api_key"}})

    # Check body size via Content-Length header
    content_length = request.headers.get("content-length")
    try:
        if content_length and int(content_length) > MAX_REQUEST_BODY_BYTES:
            msg = f"Request body too large (max {MAX_REQUEST_BODY_BYTES // 1024 // 1024}MB)"
            if request.url.path.startswith("/anthropic/"):
                return JSONResponse(status_code=413, content={"type": "error", "error": {"type": "invalid_request_error", "message": msg}})
            return JSONResponse(status_code=413, content={"error": {"message": msg, "type": "invalid_request_error"}})
    except (ValueError, TypeError):
        pass  # Malformed Content-Length, let downstream handle it

    response = await call_next(request)
    elapsed = time.time() - start
    response.headers["X-Request-ID"] = request_id
    # Only log timing for non-streaming (streaming elapsed would just be time-to-first-byte)
    if response.headers.get("content-type", "").startswith("text/event-stream"):
        logger.info(f"[{request_id}] {request.method} {request.url.path} -> {response.status_code} (streaming)")
    else:
        logger.info(f"[{request_id}] {request.method} {request.url.path} -> {response.status_code} ({elapsed:.2f}s)")
    return response


# ═════════════════════════════════════════════════════════════════════════════
# SHARED: Model definitions & CLI interface
# ═════════════════════════════════════════════════════════════════════════════

_STARTUP_TIME = int(time.time())

CLAUDE_MODELS = {
    "claude-opus-4-6": {
        "id": "claude-opus-4-6",
        "display_name": "Claude Opus 4.6",
        "created": _STARTUP_TIME,
        "context_window": 200000,
        "max_output_tokens": 32000,
    },
    "claude-sonnet-4-6": {
        "id": "claude-sonnet-4-6",
        "display_name": "Claude Sonnet 4.6",
        "created": _STARTUP_TIME,
        "context_window": 200000,
        "max_output_tokens": 32000,
    },
    "claude-haiku-4-5": {
        "id": "claude-haiku-4-5",
        "display_name": "Claude Haiku 4.5",
        "created": _STARTUP_TIME,
        "context_window": 200000,
        "max_output_tokens": 32000,
    },
}

OPENAI_MODEL_ALIASES = {
    "gpt-4": "claude-sonnet-4-6",
    "gpt-4-turbo": "claude-sonnet-4-6",
    "gpt-4-turbo-preview": "claude-sonnet-4-6",
    "gpt-4o": "claude-sonnet-4-6",
    "gpt-4o-2024-05-13": "claude-sonnet-4-6",
    "gpt-4o-mini": "claude-haiku-4-5",
    "gpt-4o-mini-2024-07-18": "claude-haiku-4-5",
    "gpt-3.5-turbo": "claude-haiku-4-5",
    "gpt-3.5-turbo-0125": "claude-haiku-4-5",
    "o1": "claude-opus-4-6",
    "o1-preview": "claude-opus-4-6",
    "o1-mini": "claude-sonnet-4-6",
    "o3": "claude-opus-4-6",
    "o3-mini": "claude-sonnet-4-6",
    "o4-mini": "claude-sonnet-4-6",
}

DEFAULT_MODEL = "claude-sonnet-4-6"

# Subprocess safety limits (overridable via CLI args)
CLI_TIMEOUT_SECONDS = 300
CLI_LINE_TIMEOUT_SECONDS = 120
MAX_CONCURRENT_REQUESTS = 10
_semaphore: asyncio.Semaphore | None = None


def _get_semaphore() -> asyncio.Semaphore:
    global _semaphore
    if _semaphore is None:
        _semaphore = asyncio.Semaphore(MAX_CONCURRENT_REQUESTS)
    return _semaphore

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


def resolve_model_openai(requested: str) -> Optional[str]:
    if requested in CLAUDE_MODELS:
        return requested
    return OPENAI_MODEL_ALIASES.get(requested)


def resolve_model_anthropic(requested: str) -> Optional[str]:
    if requested in CLAUDE_MODELS:
        return requested
    return None


def _sanitize_cli_error(error: Exception) -> str:
    """Strip filesystem paths and env vars from CLI error messages for client safety."""
    msg = str(error)
    msg = re.sub(r'/(?:[\w.@-]+/)+[\w.@-]+', '[path]', msg)
    msg = re.sub(r'\b[A-Z_]{2,}=\S+', '[env]', msg)
    return msg[:500]


def _aggregate_input_tokens(usage: dict) -> int:
    """Calculate total input tokens including cache tokens."""
    return (
        usage.get("input_tokens", 0)
        + usage.get("cache_read_input_tokens", 0)
        + usage.get("cache_creation_input_tokens", 0)
    )


def _sse(event_type: str, data: dict) -> str:
    """Build an Anthropic SSE event string."""
    return f"event: {event_type}\ndata: {json.dumps(data)}\n\n"


# ─── Claude CLI interface (shared) ───────────────────────────────────────────

async def call_claude(
    prompt: str,
    model: str,
    system_prompt: Optional[str] = None,
) -> dict:
    """Non-streaming call to claude CLI. Returns parsed result dict."""
    cmd = [
        "claude", "-p",
        "--output-format", "json",
        "--model", model,
        "--tools", "",
        "--no-session-persistence",
    ]
    if system_prompt:
        cmd.extend(["--system-prompt", system_prompt])

    sem = _get_semaphore()
    try:
        await asyncio.wait_for(sem.acquire(), timeout=SEMAPHORE_ACQUIRE_TIMEOUT)
    except asyncio.TimeoutError:
        raise RuntimeError(f"Server at capacity ({MAX_CONCURRENT_REQUESTS} concurrent requests)")

    try:
        proc = await asyncio.create_subprocess_exec(
            *cmd,
            stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        _active_processes.add(proc)
        try:
            stdout, stderr = await asyncio.wait_for(
                proc.communicate(input=prompt.encode()),
                timeout=CLI_TIMEOUT_SECONDS,
            )
        except asyncio.TimeoutError:
            proc.kill()
            await proc.wait()
            raise RuntimeError(f"Claude CLI timed out after {CLI_TIMEOUT_SECONDS}s")
        except Exception:
            if proc.returncode is None:
                proc.kill()
                await proc.wait()
            raise
        finally:
            _active_processes.discard(proc)
    finally:
        sem.release()

    if proc.returncode != 0:
        error_msg = stderr.decode(errors='replace').strip()
        raise RuntimeError(f"claude CLI error (exit {proc.returncode}): {error_msg}")

    raw = stdout.decode(errors='replace').strip()
    try:
        result = json.loads(raw)
    except json.JSONDecodeError:
        logger.warning(f"Claude CLI returned non-JSON: {raw[:200]}")
        return {
            "content": raw,
            "usage": {"input_tokens": 0, "output_tokens": 0},
            "stop_reason": "end_turn",
            "model": model,
        }

    usage = result.get("usage", {})
    return {
        "content": result.get("result", ""),
        "usage": {
            "input_tokens": _aggregate_input_tokens(usage),
            "output_tokens": usage.get("output_tokens", 0),
            "cache_creation_input_tokens": usage.get("cache_creation_input_tokens", 0),
            "cache_read_input_tokens": usage.get("cache_read_input_tokens", 0),
        },
        "stop_reason": result.get("stop_reason", "end_turn"),
        "model": model,
    }


async def call_claude_streaming(
    prompt: str,
    model: str,
    system_prompt: Optional[str] = None,
):
    """Streaming call to claude CLI, yielding parsed JSON events."""
    cmd = [
        "claude", "-p",
        "--output-format", "stream-json",
        "--verbose",
        "--include-partial-messages",
        "--model", model,
        "--tools", "",
        "--no-session-persistence",
    ]
    if system_prompt:
        cmd.extend(["--system-prompt", system_prompt])

    sem = _get_semaphore()
    try:
        await asyncio.wait_for(sem.acquire(), timeout=SEMAPHORE_ACQUIRE_TIMEOUT)
    except asyncio.TimeoutError:
        raise RuntimeError(f"Server at capacity ({MAX_CONCURRENT_REQUESTS} concurrent requests)")
    try:
        proc = await asyncio.create_subprocess_exec(
            *cmd,
            stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.DEVNULL,
        )
        _active_processes.add(proc)
        try:
            proc.stdin.write(prompt.encode())
            await proc.stdin.drain()
            proc.stdin.close()
            await proc.stdin.wait_closed()

            while True:
                if _shutting_down:
                    break
                try:
                    line = await asyncio.wait_for(
                        proc.stdout.readline(),
                        timeout=CLI_LINE_TIMEOUT_SECONDS,
                    )
                except asyncio.TimeoutError:
                    logger.error(f"Claude CLI streaming timed out after {CLI_LINE_TIMEOUT_SECONDS}s idle")
                    break
                if not line:
                    break
                line = line.decode(errors='replace').strip()
                if not line:
                    continue
                try:
                    yield json.loads(line)
                except json.JSONDecodeError:
                    continue
        finally:
            if proc.returncode is None:
                proc.terminate()
                try:
                    await asyncio.wait_for(proc.wait(), timeout=3)
                except asyncio.TimeoutError:
                    proc.kill()
                    await proc.wait()
            _active_processes.discard(proc)
    finally:
        sem.release()


STREAM_QUEUE_SIZE = 64  # Max buffered events for backpressure
_QUEUE_SENTINEL = object()


async def _buffered_claude_stream(prompt, model, system_prompt):
    """Wraps call_claude_streaming with a bounded queue for backpressure.

    When the client reads slowly, the queue fills and the subprocess reader
    naturally pauses, preventing unbounded memory growth.
    """
    queue: asyncio.Queue = asyncio.Queue(maxsize=STREAM_QUEUE_SIZE)

    async def _reader():
        try:
            async for event in call_claude_streaming(prompt, model, system_prompt):
                await queue.put(event)
        except Exception as e:
            await queue.put(e)
        finally:
            await queue.put(_QUEUE_SENTINEL)

    reader_task = asyncio.create_task(_reader())
    try:
        while True:
            item = await queue.get()
            if item is _QUEUE_SENTINEL:
                break
            if isinstance(item, Exception):
                raise item
            yield item
    finally:
        reader_task.cancel()
        try:
            await reader_task
        except asyncio.CancelledError:
            pass


# ═════════════════════════════════════════════════════════════════════════════
# OPENAI API — /openai/v1/*
# ═════════════════════════════════════════════════════════════════════════════

def openai_error(status: int, message: str, error_type: str, code: str) -> JSONResponse:
    return JSONResponse(
        status_code=status,
        content={
            "error": {
                "message": message,
                "type": error_type,
                "param": None,
                "code": code,
            }
        },
    )


# ─── OpenAI message conversion ───────────────────────────────────────────────

def openai_extract_system(messages: list[dict]) -> tuple[Optional[str], list[dict]]:
    system_parts = []
    other = []
    for msg in messages:
        if msg.get("role") == "system":
            content = msg.get("content", "")
            if isinstance(content, list):
                content = "\n".join(
                    b.get("text", "") for b in content if isinstance(b, dict) and b.get("type") == "text"
                )
            elif not isinstance(content, str):
                content = str(content)
            system_parts.append(content)
        else:
            other.append(msg)
    return ("\n\n".join(system_parts) if system_parts else None), other


def openai_messages_to_prompt(messages: list[dict]) -> str:
    parts = []
    for msg in messages:
        role = msg.get("role", "user")
        content = msg.get("content") or ""
        if isinstance(content, list):
            text_parts = []
            for block in content:
                if isinstance(block, str):
                    text_parts.append(block)
                elif isinstance(block, dict):
                    if block.get("type") == "text":
                        text_parts.append(block.get("text", ""))
                    elif block.get("type") == "image_url":
                        text_parts.append("[Image provided]")
            content = "\n".join(text_parts)
        elif not isinstance(content, str):
            content = str(content)

        if role == "user":
            parts.append(content)
        elif role == "assistant":
            tool_calls = msg.get("tool_calls")
            if isinstance(tool_calls, list) and tool_calls:
                tc_text = "\n".join(
                    f'[Tool call: {tc.get("function", {}).get("name", "unknown")}'
                    f'({tc.get("function", {}).get("arguments", "{}")}) '
                    f'id={tc.get("id", "")}]'
                    for tc in tool_calls
                    if isinstance(tc, dict)
                )
                parts.append(
                    f"[Previous assistant response]: {content}\n{tc_text}"
                    if content
                    else f"[Previous assistant]:\n{tc_text}"
                )
            elif content:
                parts.append(f"[Previous assistant response]: {content}")
        elif role == "tool" or role == "function":
            tool_call_id = msg.get("tool_call_id", "") or msg.get("name", "")
            parts.append(f"[Tool result for {tool_call_id}]: {content}")
    return "\n\n".join(parts)


def _sanitize_tool_name(name: str) -> str:
    """Sanitize tool/function name to safe identifier characters."""
    return re.sub(r'[^a-zA-Z0-9_.-]', '', name)[:128]


def _sanitize_tool_text(text: str) -> str:
    """Sanitize tool description — collapse newlines to prevent prompt escape."""
    return re.sub(r'\s+', ' ', text.replace('\n', ' ').replace('\r', ' ')).strip()[:2048]


def openai_build_tools_system(tools: list[dict], tool_choice) -> str:
    if not tools:
        return ""
    lines = [
        "\n\nYou have access to the following functions. To call a function, respond with a JSON object "
        'in this exact format on its own line: {"tool_call": {"name": "<function_name>", "arguments": {<args>}}}',
        "",
    ]
    for tool in tools:
        if tool.get("type") == "function":
            fn = tool.get("function", {})
            lines.append(f'Function: {_sanitize_tool_name(fn.get("name", "unknown"))}')
            if fn.get("description"):
                lines.append(f'Description: {_sanitize_tool_text(fn["description"])}')
            if fn.get("parameters"):
                lines.append(f"Parameters: {json.dumps(fn['parameters'])}")
            lines.append("")
    if tool_choice == "none":
        lines.append("Do NOT call any functions. Respond normally.")
    elif tool_choice == "auto" or tool_choice is None:
        lines.append("Call functions if appropriate, otherwise respond normally.")
    elif tool_choice == "required":
        lines.append("You MUST call at least one function in your response.")
    elif isinstance(tool_choice, dict) and tool_choice.get("type") == "function":
        fn_name = tool_choice.get("function", {}).get("name", "")
        lines.append(f"You MUST call the function '{fn_name}'.")
    return "\n".join(lines)


def _extract_tool_json(text: str, key: str) -> list[tuple[int, int, dict]]:
    """Find JSON objects containing the given key, handling multi-line JSON.

    Returns list of (start_index, end_index, parsed_object) tuples.
    """
    prefix = '{"' + key + '"'
    results = []
    start = 0
    while True:
        idx = text.find(prefix, start)
        if idx == -1:
            break
        depth = 0
        in_string = False
        escape = False
        for i in range(idx, len(text)):
            c = text[i]
            if escape:
                escape = False
                continue
            if c == '\\' and in_string:
                escape = True
                continue
            if c == '"':
                in_string = not in_string
                continue
            if not in_string:
                if c == '{':
                    depth += 1
                elif c == '}':
                    depth -= 1
                    if depth == 0:
                        try:
                            obj = json.loads(text[idx:i + 1])
                            results.append((idx, i + 1, obj))
                        except json.JSONDecodeError:
                            pass
                        break
        start = idx + 1
    return results


def _remove_ranges(text: str, ranges: list[tuple[int, int]]) -> str:
    """Remove character ranges from text and return cleaned result."""
    if not ranges:
        return text
    parts = []
    prev = 0
    for s, e in sorted(ranges):
        parts.append(text[prev:s])
        prev = e
    parts.append(text[prev:])
    return "\n".join(line for line in "".join(parts).split("\n") if line.strip()).strip()


def openai_parse_tool_calls(text: str) -> tuple[str, list[dict]]:
    matches = _extract_tool_json(text, "tool_call")
    tool_calls = []
    ranges = []
    idx = 0
    for start, end, parsed in matches:
        tc = parsed.get("tool_call", {})
        if not isinstance(tc, dict) or not tc.get("name"):
            continue
        tool_calls.append({
            "index": idx,
            "id": f"call_{uuid.uuid4().hex[:24]}",
            "type": "function",
            "function": {
                "name": tc["name"],
                "arguments": json.dumps(tc.get("arguments", {})),
            },
        })
        idx += 1
        ranges.append((start, end))
    return _remove_ranges(text, ranges), tool_calls


# ─── OpenAI Models ───────────────────────────────────────────────────────────

def _build_openai_models_list():
    data = []
    for model_id, info in CLAUDE_MODELS.items():
        data.append({"id": model_id, "object": "model", "created": info["created"], "owned_by": "anthropic"})
    for alias, target in OPENAI_MODEL_ALIASES.items():
        data.append({"id": alias, "object": "model", "created": _STARTUP_TIME, "owned_by": "anthropic", "parent": target})
    return {"object": "list", "data": data}


_OPENAI_MODELS_RESPONSE = _build_openai_models_list()


@app.get("/openai/v1/models")
@app.get("/v1/models")
async def openai_list_models():
    return _OPENAI_MODELS_RESPONSE


@app.get("/openai/v1/models/{model_id}")
@app.get("/v1/models/{model_id}")
async def openai_retrieve_model(model_id: str):
    if model_id in CLAUDE_MODELS:
        info = CLAUDE_MODELS[model_id]
        return {"id": model_id, "object": "model", "created": info["created"], "owned_by": "anthropic"}
    if model_id in OPENAI_MODEL_ALIASES:
        target = OPENAI_MODEL_ALIASES[model_id]
        return {"id": model_id, "object": "model", "created": _STARTUP_TIME, "owned_by": "anthropic", "parent": target}
    return openai_error(404, f"The model '{model_id}' does not exist", "invalid_request_error", "model_not_found")


# ─── OpenAI Chat Completions ─────────────────────────────────────────────────

@app.post("/openai/v1/chat/completions")
@app.post("/v1/chat/completions")
async def openai_chat_completions(request: Request):
    try:
        body = await request.json()
    except Exception:
        return openai_error(400, "Invalid JSON in request body", "invalid_request_error", "invalid_json")
    messages = body.get("messages")
    if not messages:
        return openai_error(400, "'messages' is required", "invalid_request_error", "missing_messages")

    requested_model = body.get("model", DEFAULT_MODEL)
    model = resolve_model_openai(requested_model)
    if model is None:
        return openai_error(404, f"Model '{requested_model}' not found. Use /openai/v1/models to list available models.", "invalid_request_error", "model_not_found")
    stream = body.get("stream", False)
    tools = body.get("tools", [])
    tool_choice = body.get("tool_choice", "auto")
    n = body.get("n", 1)

    if n != 1:
        return openai_error(400, "Only n=1 is supported", "invalid_request_error", "unsupported_n")

    system_prompt, conversation_messages = openai_extract_system(messages)
    if tools:
        system_prompt = (system_prompt or "") + openai_build_tools_system(tools, tool_choice)

    prompt = openai_messages_to_prompt(conversation_messages)
    if not prompt.strip():
        return openai_error(400, "No user content provided", "invalid_request_error", "empty_prompt")

    if stream:
        if not _get_semaphore()._value:
            return openai_error(429, f"Server at capacity ({MAX_CONCURRENT_REQUESTS} concurrent requests)", "rate_limit_error", "rate_limit_exceeded")
        return StreamingResponse(
            _openai_stream_chat(request, prompt, model, requested_model, system_prompt, bool(tools)),
            media_type="text/event-stream",
            headers={"Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no"},
        )

    try:
        result = await call_claude(prompt, model, system_prompt)
    except RuntimeError as e:
        logger.error(f"Claude CLI error: {e}")
        if "at capacity" in str(e):
            return openai_error(429, str(e), "rate_limit_error", "rate_limit_exceeded")
        return openai_error(502, _sanitize_cli_error(e), "server_error", "claude_cli_error")

    content = result["content"]
    finish_reason = STOP_REASON_TO_OPENAI.get(result["stop_reason"], "stop")

    tool_calls_list = []
    if tools:
        content, tool_calls_list = openai_parse_tool_calls(content)
        if tool_calls_list:
            finish_reason = "tool_calls"

    message = {"role": "assistant", "content": content or None}
    if tool_calls_list:
        message["tool_calls"] = tool_calls_list
        if not content:
            message["content"] = None

    prompt_tokens = result["usage"]["input_tokens"]
    completion_tokens = result["usage"]["output_tokens"]

    return {
        "id": f"chatcmpl-{uuid.uuid4().hex[:24]}",
        "object": "chat.completion",
        "created": int(time.time()),
        "model": requested_model,
        "choices": [{
            "index": 0,
            "message": message,
            "logprobs": None,
            "finish_reason": finish_reason,
        }],
        "usage": {
            "prompt_tokens": prompt_tokens,
            "completion_tokens": completion_tokens,
            "total_tokens": prompt_tokens + completion_tokens,
        },
        "system_fingerprint": f"claude-proxy-{model}",
    }


_TOOL_PREFIXES = ('{"tool_call"', '{"tool_use"')


class _StreamToolBuffer:
    """Buffers text line-by-line, withholding lines that look like tool call JSON."""

    def __init__(self):
        self._line_buf = ""
        self._deferred: list[str] = []
        self._in_deferred = False
        self._deferred_depth = 0

    def add_text(self, text: str) -> list[str]:
        """Add text chunk, return list of content strings safe to yield."""
        self._line_buf += text
        to_yield = []
        while '\n' in self._line_buf:
            line, self._line_buf = self._line_buf.split('\n', 1)
            result = self._process_line(line)
            if result is not None:
                to_yield.append(result + "\n")
        return to_yield

    def flush(self) -> list[str]:
        """Flush remaining buffer. Returns content strings safe to yield."""
        to_yield = []
        if self._line_buf.strip():
            result = self._process_line(self._line_buf)
            if result is not None:
                to_yield.append(result)
        self._line_buf = ""
        # If still mid-deferred block (unbalanced braces), move incomplete block to content
        if self._in_deferred and self._deferred:
            incomplete = self._deferred.pop()
            to_yield.append(incomplete)
            self._in_deferred = False
            self._deferred_depth = 0
        return to_yield

    def _process_line(self, line: str) -> str | None:
        stripped = line.strip()
        if self._in_deferred:
            self._deferred[-1] += '\n' + line
            self._deferred_depth += stripped.count('{') - stripped.count('}')
            if self._deferred_depth <= 0:
                self._in_deferred = False
            return None
        if stripped.startswith(_TOOL_PREFIXES):
            self._deferred.append(line)
            self._deferred_depth = stripped.count('{') - stripped.count('}')
            if self._deferred_depth > 0:
                self._in_deferred = True
            return None
        return line

    @property
    def deferred_text(self) -> str:
        return "\n".join(self._deferred)


async def _openai_stream_chat(request, prompt, model, requested_model, system_prompt, has_tools):
    chunk_id = f"chatcmpl-{uuid.uuid4().hex[:24]}"
    created = int(time.time())

    def make_chunk(delta, finish_reason=None):
        return f"data: {json.dumps({'id': chunk_id, 'object': 'chat.completion.chunk', 'created': created, 'model': requested_model, 'system_fingerprint': f'claude-proxy-{model}', 'choices': [{'index': 0, 'delta': delta, 'logprobs': None, 'finish_reason': finish_reason}]})}\n\n"

    yield make_chunk({"role": "assistant", "content": ""})

    tool_buf = _StreamToolBuffer() if has_tools else None
    stop_reason = "stop"
    chunk_count = 0

    try:
        async for event in _buffered_claude_stream(prompt, model, system_prompt):
            chunk_count += 1
            if chunk_count % 10 == 0 and await request.is_disconnected():
                logger.info("Client disconnected during OpenAI streaming")
                break
            et = event.get("type")
            if et == "stream_event":
                inner = event.get("event", {})
                if inner.get("type") == "content_block_delta":
                    delta = inner.get("delta", {})
                    if delta.get("type") == "text_delta":
                        text = delta.get("text", "")
                        if text:
                            if tool_buf:
                                for chunk_text in tool_buf.add_text(text):
                                    yield make_chunk({"content": chunk_text})
                            else:
                                yield make_chunk({"content": text})
                elif inner.get("type") == "message_delta":
                    sr = inner.get("delta", {}).get("stop_reason", "end_turn")
                    stop_reason = STOP_REASON_TO_OPENAI.get(sr, "stop")
            elif et == "result":
                stop_reason = STOP_REASON_TO_OPENAI.get(event.get("stop_reason", "end_turn"), "stop")
    except (RuntimeError, asyncio.TimeoutError, OSError) as e:
        logger.error(f"OpenAI streaming error: {e}")
        yield make_chunk({"content": f"\n\n[Error: {_sanitize_cli_error(e)}]"})

    # Flush buffer and parse tool calls from deferred lines
    if tool_buf:
        for chunk_text in tool_buf.flush():
            yield make_chunk({"content": chunk_text})

    tool_calls = []
    if tool_buf and tool_buf.deferred_text:
        _, tool_calls = openai_parse_tool_calls(tool_buf.deferred_text)
        if tool_calls:
            stop_reason = "tool_calls"
        else:
            # Deferred lines didn't parse as tools — emit them as content
            yield make_chunk({"content": tool_buf.deferred_text})

    final_delta = {"tool_calls": tool_calls} if tool_calls else {}
    yield make_chunk(final_delta, finish_reason=stop_reason)
    yield "data: [DONE]\n\n"


# ─── OpenAI Legacy Completions ────────────────────────────────────────────────

@app.post("/openai/v1/completions")
@app.post("/v1/completions")
async def openai_completions(request: Request):
    try:
        body = await request.json()
    except Exception:
        return openai_error(400, "Invalid JSON in request body", "invalid_request_error", "invalid_json")
    prompt_text = body.get("prompt", "")
    if isinstance(prompt_text, list):
        prompt_text = "\n".join(prompt_text)
    if not prompt_text:
        return openai_error(400, "'prompt' is required", "invalid_request_error", "missing_prompt")

    requested_model = body.get("model", DEFAULT_MODEL)
    model = resolve_model_openai(requested_model)
    if model is None:
        return openai_error(404, f"Model '{requested_model}' not found. Use /openai/v1/models to list available models.", "invalid_request_error", "model_not_found")
    stream = body.get("stream", False)
    suffix = body.get("suffix")
    system_prompt = f"After your response, append this suffix: {suffix}" if suffix else None

    if stream:
        if not _get_semaphore()._value:
            return openai_error(429, f"Server at capacity ({MAX_CONCURRENT_REQUESTS} concurrent requests)", "rate_limit_error", "rate_limit_exceeded")
        return StreamingResponse(
            _openai_stream_completion(request, prompt_text, model, requested_model, system_prompt),
            media_type="text/event-stream",
            headers={"Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no"},
        )

    try:
        result = await call_claude(prompt_text, model, system_prompt)
    except RuntimeError as e:
        logger.error(f"Claude CLI error: {e}")
        if "at capacity" in str(e):
            return openai_error(429, str(e), "rate_limit_error", "rate_limit_exceeded")
        return openai_error(502, _sanitize_cli_error(e), "server_error", "claude_cli_error")

    content = result["content"]
    if suffix:
        content += suffix

    prompt_tokens = result["usage"]["input_tokens"]
    completion_tokens = result["usage"]["output_tokens"]

    return {
        "id": f"cmpl-{uuid.uuid4().hex[:24]}",
        "object": "text_completion",
        "created": int(time.time()),
        "model": requested_model,
        "choices": [{"text": content, "index": 0, "logprobs": None, "finish_reason": STOP_REASON_TO_OPENAI.get(result["stop_reason"], "stop")}],
        "usage": {"prompt_tokens": prompt_tokens, "completion_tokens": completion_tokens, "total_tokens": prompt_tokens + completion_tokens},
        "system_fingerprint": f"claude-proxy-{model}",
    }


async def _openai_stream_completion(request, prompt, model, requested_model, system_prompt):
    chunk_id = f"cmpl-{uuid.uuid4().hex[:24]}"
    created = int(time.time())

    def make_chunk(text, finish_reason=None):
        return f"data: {json.dumps({'id': chunk_id, 'object': 'text_completion', 'created': created, 'model': requested_model, 'choices': [{'text': text, 'index': 0, 'logprobs': None, 'finish_reason': finish_reason}]})}\n\n"

    chunk_count = 0
    try:
        async for event in _buffered_claude_stream(prompt, model, system_prompt):
            chunk_count += 1
            if chunk_count % 10 == 0 and await request.is_disconnected():
                logger.info("Client disconnected during OpenAI completion streaming")
                break
            if event.get("type") == "stream_event":
                inner = event.get("event", {})
                if inner.get("type") == "content_block_delta":
                    delta = inner.get("delta", {})
                    if delta.get("type") == "text_delta":
                        text = delta.get("text", "")
                        if text:
                            yield make_chunk(text)
    except (RuntimeError, asyncio.TimeoutError, OSError) as e:
        logger.error(f"OpenAI streaming error: {e}")
        yield make_chunk(f"\n\n[Error: {_sanitize_cli_error(e)}]")

    yield make_chunk("", finish_reason="stop")
    yield "data: [DONE]\n\n"


# ─── OpenAI Unsupported endpoints ────────────────────────────────────────────

_OAI_UNSUPPORTED = "This endpoint is not supported. Claude does not provide this capability."

@app.post("/openai/v1/embeddings")
@app.post("/v1/embeddings")
async def openai_embeddings(request: Request):
    return openai_error(501, f"Embeddings: {_OAI_UNSUPPORTED}", "invalid_request_error", "unsupported_endpoint")

@app.post("/openai/v1/images/generations")
@app.post("/v1/images/generations")
async def openai_images_gen(request: Request):
    return openai_error(501, f"Image generation: {_OAI_UNSUPPORTED}", "invalid_request_error", "unsupported_endpoint")

@app.post("/openai/v1/images/edits")
@app.post("/v1/images/edits")
async def openai_images_edit(request: Request):
    return openai_error(501, f"Image editing: {_OAI_UNSUPPORTED}", "invalid_request_error", "unsupported_endpoint")

@app.post("/openai/v1/audio/transcriptions")
@app.post("/v1/audio/transcriptions")
async def openai_audio_transcribe(request: Request):
    return openai_error(501, f"Audio transcription: {_OAI_UNSUPPORTED}", "invalid_request_error", "unsupported_endpoint")

@app.post("/openai/v1/audio/translations")
@app.post("/v1/audio/translations")
async def openai_audio_translate(request: Request):
    return openai_error(501, f"Audio translation: {_OAI_UNSUPPORTED}", "invalid_request_error", "unsupported_endpoint")

@app.post("/openai/v1/audio/speech")
@app.post("/v1/audio/speech")
async def openai_audio_speech(request: Request):
    return openai_error(501, f"Text-to-speech: {_OAI_UNSUPPORTED}", "invalid_request_error", "unsupported_endpoint")

@app.post("/openai/v1/fine_tuning/jobs")
@app.post("/v1/fine_tuning/jobs")
async def openai_fine_tuning(request: Request):
    return openai_error(501, f"Fine-tuning: {_OAI_UNSUPPORTED}", "invalid_request_error", "unsupported_endpoint")

@app.post("/openai/v1/moderations")
@app.post("/v1/moderations")
async def openai_moderations(request: Request):
    return openai_error(501, f"Moderations: {_OAI_UNSUPPORTED}", "invalid_request_error", "unsupported_endpoint")


# ═════════════════════════════════════════════════════════════════════════════
# ANTHROPIC API — /anthropic/v1/*
# ═════════════════════════════════════════════════════════════════════════════

def anthropic_error(status: int, error_type: str, message: str) -> JSONResponse:
    """Return an Anthropic-format error response."""
    return JSONResponse(
        status_code=status,
        content={
            "type": "error",
            "error": {
                "type": error_type,
                "message": message,
            },
        },
    )


# ─── Anthropic message conversion ────────────────────────────────────────────

def anthropic_extract_system(body: dict) -> Optional[str]:
    """Extract system prompt from Anthropic request (top-level 'system' field)."""
    system = body.get("system")
    if system is None:
        return None
    if isinstance(system, str):
        return system
    if isinstance(system, list):
        return "\n\n".join(
            b.get("text", "") for b in system if isinstance(b, dict) and b.get("type") == "text"
        )
    return str(system)


def anthropic_content_to_text(content) -> str:
    """Convert Anthropic content (string or block array) to plain text."""
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        parts = []
        for block in content:
            if isinstance(block, str):
                parts.append(block)
            elif isinstance(block, dict):
                btype = block.get("type", "")
                if btype == "text":
                    parts.append(block.get("text", ""))
                elif btype == "image":
                    parts.append("[Image provided]")
                elif btype == "tool_use":
                    parts.append(
                        f'[Tool call: {block.get("name", "unknown")}'
                        f'({json.dumps(block.get("input", {}))}) '
                        f'id={block.get("id", "")}]'
                    )
                elif btype == "tool_result":
                    tool_content = block.get("content", "")
                    if isinstance(tool_content, list):
                        tool_content = "\n".join(
                            b.get("text", "") for b in tool_content
                            if isinstance(b, dict) and b.get("type") == "text"
                        )
                    parts.append(f'[Tool result for {block.get("tool_use_id", "")}]: {tool_content}')
        return "\n".join(parts)
    return str(content)


def anthropic_messages_to_prompt(messages: list[dict]) -> str:
    """Convert Anthropic-format messages to a prompt string."""
    parts = []
    for msg in messages:
        role = msg.get("role", "user")
        text = anthropic_content_to_text(msg.get("content", ""))
        if role == "user":
            parts.append(text)
        elif role == "assistant":
            parts.append(f"[Previous assistant response]: {text}")
    return "\n\n".join(parts)


def anthropic_build_tools_system(tools: list[dict], tool_choice: Optional[dict]) -> str:
    """Build system prompt section for Anthropic-format tools."""
    if not tools:
        return ""
    lines = [
        "\n\nYou have access to the following tools. To use a tool, respond with a JSON object "
        'in this exact format on its own line: {"tool_use": {"name": "<tool_name>", "input": {<args>}}}',
        "",
    ]
    for tool in tools:
        lines.append(f'Tool: {_sanitize_tool_name(tool.get("name", "unknown"))}')
        if tool.get("description"):
            lines.append(f'Description: {_sanitize_tool_text(tool["description"])}')
        if tool.get("input_schema"):
            lines.append(f"Input schema: {json.dumps(tool['input_schema'])}")
        lines.append("")

    if tool_choice:
        tc_type = tool_choice.get("type", "auto")
        if tc_type == "none":
            lines.append("Do NOT use any tools. Respond normally.")
        elif tc_type == "auto":
            lines.append("Use tools if appropriate, otherwise respond normally.")
        elif tc_type == "any":
            lines.append("You MUST use at least one tool in your response.")
        elif tc_type == "tool":
            lines.append(f'You MUST use the tool \'{tool_choice.get("name", "")}\'.')
    else:
        lines.append("Use tools if appropriate, otherwise respond normally.")

    return "\n".join(lines)


def anthropic_parse_tool_use(text: str) -> tuple[str, list[dict]]:
    """Parse tool_use JSON objects from response text."""
    matches = _extract_tool_json(text, "tool_use")
    tool_uses = []
    ranges = []
    for start, end, parsed in matches:
        tu = parsed.get("tool_use", {})
        if not isinstance(tu, dict) or not tu.get("name"):
            continue
        tool_uses.append({
            "type": "tool_use",
            "id": f"toolu_{uuid.uuid4().hex[:24]}",
            "name": tu["name"],
            "input": tu.get("input", {}),
        })
        ranges.append((start, end))
    return _remove_ranges(text, ranges), tool_uses


# ─── Anthropic Models ────────────────────────────────────────────────────────

def _build_anthropic_models_list():
    model_keys = list(CLAUDE_MODELS.keys())
    data = [{"id": mid, "type": "model", "display_name": CLAUDE_MODELS[mid]["display_name"], "created_at": "2024-01-01T00:00:00Z"} for mid in model_keys]
    return {"data": data, "has_more": False, "first_id": model_keys[0], "last_id": model_keys[-1]}


_ANTHROPIC_MODELS_RESPONSE = _build_anthropic_models_list()


@app.get("/anthropic/v1/models")
async def anthropic_list_models():
    """List models in Anthropic API format."""
    return _ANTHROPIC_MODELS_RESPONSE


@app.get("/anthropic/v1/models/{model_id}")
async def anthropic_retrieve_model(model_id: str):
    """Retrieve a model in Anthropic API format."""
    if model_id not in CLAUDE_MODELS:
        return anthropic_error(404, "not_found_error", f"model '{model_id}' not found")
    info = CLAUDE_MODELS[model_id]
    return {
        "id": model_id,
        "type": "model",
        "display_name": info["display_name"],
        "created_at": "2024-01-01T00:00:00Z",
    }


# ─── Anthropic Messages ──────────────────────────────────────────────────────

@app.post("/anthropic/v1/messages")
async def anthropic_messages(request: Request):
    """Anthropic Messages API — fully spec-compliant."""
    try:
        body = await request.json()
    except Exception:
        return anthropic_error(400, "invalid_request_error", "Invalid JSON in request body")

    # ── Validate required fields ──
    model_requested = body.get("model")
    if not model_requested:
        return anthropic_error(400, "invalid_request_error", "'model' is required")

    messages = body.get("messages")
    if not messages:
        return anthropic_error(400, "invalid_request_error", "'messages' is required")

    max_tokens = body.get("max_tokens")
    if max_tokens is None:
        return anthropic_error(400, "invalid_request_error", "'max_tokens' is required")
    if not isinstance(max_tokens, int) or max_tokens <= 0:
        return anthropic_error(400, "invalid_request_error", "'max_tokens' must be a positive integer")

    model = resolve_model_anthropic(model_requested)
    if model is None:
        return anthropic_error(404, "not_found_error", f"Model '{model_requested}' not found. Use /anthropic/v1/models to list available models.")
    stream = body.get("stream", False)
    tools = body.get("tools", [])
    tool_choice = body.get("tool_choice")
    stop_sequences = body.get("stop_sequences", [])

    # ── Build prompt ──
    system_prompt = anthropic_extract_system(body)

    if tools:
        tools_section = anthropic_build_tools_system(tools, tool_choice)
        system_prompt = (system_prompt or "") + tools_section

    if stop_sequences:
        stop_note = f"\n\nIMPORTANT: Stop generating immediately when you are about to output any of these sequences: {json.dumps(stop_sequences)}"
        system_prompt = (system_prompt or "") + stop_note

    prompt = anthropic_messages_to_prompt(messages)
    if not prompt.strip():
        return anthropic_error(400, "invalid_request_error", "No user content provided in messages")

    # ── Streaming ──
    if stream:
        if not _get_semaphore()._value:
            return anthropic_error(429, "rate_limit_error", f"Server at capacity ({MAX_CONCURRENT_REQUESTS} concurrent requests)")
        return StreamingResponse(
            _anthropic_stream_messages(request, prompt, model, model_requested, system_prompt, bool(tools)),
            media_type="text/event-stream",
            headers={"Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no"},
        )

    # ── Non-streaming ──
    try:
        result = await call_claude(prompt, model, system_prompt)
    except RuntimeError as e:
        logger.error(f"Claude CLI error: {e}")
        if "at capacity" in str(e):
            return anthropic_error(429, "rate_limit_error", str(e))
        return anthropic_error(502, "api_error", _sanitize_cli_error(e))

    content_text = result["content"]
    stop_reason = result["stop_reason"]

    # Build content blocks
    content_blocks = []
    if tools:
        clean_text, tool_uses = anthropic_parse_tool_use(content_text)
        if clean_text:
            content_blocks.append({"type": "text", "text": clean_text})
        content_blocks.extend(tool_uses)
        if tool_uses:
            stop_reason = "tool_use"
    else:
        content_blocks.append({"type": "text", "text": content_text})

    input_tokens = result["usage"]["input_tokens"]
    output_tokens = result["usage"]["output_tokens"]

    return {
        "id": f"msg_{uuid.uuid4().hex[:24]}",
        "type": "message",
        "role": "assistant",
        "content": content_blocks,
        "model": model_requested,
        "stop_reason": stop_reason,
        "stop_sequence": None,
        "usage": {
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "cache_creation_input_tokens": result["usage"].get("cache_creation_input_tokens", 0),
            "cache_read_input_tokens": result["usage"].get("cache_read_input_tokens", 0),
        },
        "created_at": datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
    }


async def _anthropic_stream_messages(request, prompt, model, model_requested, system_prompt, has_tools):
    """Generator for Anthropic SSE streaming messages."""
    msg_id = f"msg_{uuid.uuid4().hex[:24]}"

    # ── message_start ──
    created_at = datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    yield _sse("message_start", {"type": "message_start", "message": {"id": msg_id, "type": "message", "role": "assistant", "content": [], "model": model_requested, "stop_reason": None, "stop_sequence": None, "usage": {"input_tokens": 0, "output_tokens": 0}, "created_at": created_at}})

    # ── content_block_start ──
    yield _sse("content_block_start", {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}})

    # ── ping ──
    yield _sse("ping", {"type": "ping"})

    tool_buf = _StreamToolBuffer() if has_tools else None
    stop_reason = "end_turn"
    input_tokens = 0
    output_tokens = 0
    chunk_count = 0

    def _emit_text_delta(text):
        return _sse("content_block_delta", {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": text}})

    try:
        async for event in _buffered_claude_stream(prompt, model, system_prompt):
            chunk_count += 1
            if chunk_count % 10 == 0 and await request.is_disconnected():
                logger.info("Client disconnected during Anthropic streaming")
                break
            et = event.get("type")

            if et == "stream_event":
                inner = event.get("event", {})
                inner_type = inner.get("type")

                if inner_type == "content_block_delta":
                    delta = inner.get("delta", {})
                    if delta.get("type") == "text_delta":
                        text = delta.get("text", "")
                        if text:
                            if tool_buf:
                                for chunk_text in tool_buf.add_text(text):
                                    yield _emit_text_delta(chunk_text)
                            else:
                                yield _emit_text_delta(text)

                elif inner_type == "message_start":
                    msg_data = inner.get("message", {})
                    usage = msg_data.get("usage", {})
                    input_tokens = _aggregate_input_tokens(usage)

                elif inner_type == "message_delta":
                    delta_data = inner.get("delta", {})
                    stop_reason = delta_data.get("stop_reason", "end_turn")
                    usage = inner.get("usage", {})
                    output_tokens = usage.get("output_tokens", output_tokens)

            elif et == "result":
                stop_reason = event.get("stop_reason", "end_turn")
                usage = event.get("usage", {})
                input_tokens = _aggregate_input_tokens(usage)
                output_tokens = usage.get("output_tokens", output_tokens)

    except (RuntimeError, asyncio.TimeoutError, OSError) as e:
        logger.error(f"Anthropic streaming error: {e}")
        yield _emit_text_delta(f'[Error: {_sanitize_cli_error(e)}]')

    # Flush buffer and parse tool uses from deferred lines
    if tool_buf:
        for chunk_text in tool_buf.flush():
            yield _emit_text_delta(chunk_text)

    parsed_tool_uses = []
    if tool_buf and tool_buf.deferred_text:
        _, parsed_tool_uses = anthropic_parse_tool_use(tool_buf.deferred_text)
        if parsed_tool_uses:
            stop_reason = "tool_use"
        else:
            # Deferred lines didn't parse as tools — emit them as content
            yield _emit_text_delta(tool_buf.deferred_text)

    # ── content_block_stop ──
    yield _sse("content_block_stop", {"type": "content_block_stop", "index": 0})

    # Emit tool use content blocks
    for i, tu in enumerate(parsed_tool_uses, start=1):
        yield _sse("content_block_start", {"type": "content_block_start", "index": i, "content_block": {"type": "tool_use", "id": tu["id"], "name": tu["name"], "input": {}}})
        yield _sse("content_block_delta", {"type": "content_block_delta", "index": i, "delta": {"type": "input_json_delta", "partial_json": json.dumps(tu["input"])}})
        yield _sse("content_block_stop", {"type": "content_block_stop", "index": i})

    # ── message_delta ──
    yield _sse("message_delta", {"type": "message_delta", "delta": {"stop_reason": stop_reason, "stop_sequence": None}, "usage": {"output_tokens": output_tokens}})

    # ── message_stop ──
    yield _sse("message_stop", {"type": "message_stop"})


# ─── Anthropic Token Counting ────────────────────────────────────────────────

@app.post("/anthropic/v1/messages/count_tokens")
async def anthropic_count_tokens(request: Request):
    """Approximate token count (Anthropic-format)."""
    try:
        body = await request.json()
    except Exception:
        return anthropic_error(400, "invalid_request_error", "Invalid JSON in request body")
    messages = body.get("messages", [])
    system = anthropic_extract_system(body) or ""

    total_text = system
    for msg in messages:
        total_text += " " + anthropic_content_to_text(msg.get("content", ""))

    # Rough approximation: ~4 chars per token
    approx_tokens = max(1, len(total_text) // 4)
    return {
        "input_tokens": approx_tokens,
    }


# ═════════════════════════════════════════════════════════════════════════════
# HEALTH & ROOT
# ═════════════════════════════════════════════════════════════════════════════

_cli_status_cache: dict = {"ok": None, "checked_at": 0, "version": None}


@app.get("/health")
async def health():
    sem = _get_semaphore()
    status = {
        "status": "ok",
        "version": "3.1.0",
        "backend": "claude-code-cli",
        "active_requests": MAX_CONCURRENT_REQUESTS - sem._value,
        "max_concurrent_per_worker": MAX_CONCURRENT_REQUESTS,
        "worker_pid": os.getpid(),
        "endpoints": {
            "openai": "/openai/v1",
            "anthropic": "/anthropic/v1",
        },
    }
    # Check CLI availability (cached for 30s)
    now = time.time()
    if now - _cli_status_cache["checked_at"] > 30:
        try:
            proc = await asyncio.create_subprocess_exec(
                "claude", "--version",
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.DEVNULL,
            )
            _active_processes.add(proc)
            try:
                stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=5)
                _cli_status_cache["ok"] = True
                _cli_status_cache["version"] = stdout.decode(errors='replace').strip()
            finally:
                _active_processes.discard(proc)
        except Exception:
            _cli_status_cache["ok"] = False
            _cli_status_cache["version"] = None
        _cli_status_cache["checked_at"] = now
    if _cli_status_cache["version"]:
        status["cli_version"] = _cli_status_cache["version"]
    if _cli_status_cache["ok"] is False:
        status["status"] = "degraded"
        status["cli_available"] = False
    return status


@app.get("/")
async def root():
    return {
        "service": "Claude Max API Proxy",
        "version": "3.1.0",
        "endpoints": {
            "openai": {
                "base_url": "/openai/v1",
                "chat_completions": "/openai/v1/chat/completions",
                "completions": "/openai/v1/completions",
                "models": "/openai/v1/models",
            },
            "anthropic": {
                "base_url": "/anthropic/v1",
                "messages": "/anthropic/v1/messages",
                "models": "/anthropic/v1/models",
                "count_tokens": "/anthropic/v1/messages/count_tokens",
            },
            "health": "/health",
        },
        "note": "OpenAI endpoints are also available at /v1/* for backwards compatibility.",
    }


# ═════════════════════════════════════════════════════════════════════════════
# MAIN
# ═════════════════════════════════════════════════════════════════════════════

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Claude Max Dual API Proxy")
    parser.add_argument("--port", type=int, default=4000, help="Port to listen on (1-65535)")
    parser.add_argument("--host", type=str, default="0.0.0.0", help="Host to bind to")
    parser.add_argument("--timeout", type=int, default=300, help="CLI timeout in seconds (default: 300)")
    parser.add_argument("--max-concurrent", type=int, default=10, help="Max concurrent CLI requests (default: 10)")
    parser.add_argument("--log-level", type=str, default=None, choices=["DEBUG", "INFO", "WARNING", "ERROR"],
                        help="Log level (default: INFO, or CLAUDE_PROXY_LOG_LEVEL env var)")
    parser.add_argument("--no-access-log", action="store_true",
                        help="Disable uvicorn access log (or set CLAUDE_PROXY_ACCESS_LOG=false)")
    parser.add_argument("--workers", type=int, default=1,
                        help="Number of uvicorn workers (default: 1)")
    args = parser.parse_args()

    if not 1 <= args.port <= 65535:
        parser.error(f"Port must be between 1 and 65535, got {args.port}")

    if args.log_level:
        logging.getLogger().setLevel(getattr(logging, args.log_level))
        logger.setLevel(getattr(logging, args.log_level))

    # Set globals for single-worker mode, and env vars for multi-worker mode
    global CLI_TIMEOUT_SECONDS, MAX_CONCURRENT_REQUESTS, _semaphore
    CLI_TIMEOUT_SECONDS = args.timeout
    MAX_CONCURRENT_REQUESTS = args.max_concurrent
    _semaphore = asyncio.Semaphore(MAX_CONCURRENT_REQUESTS)
    os.environ["_CLAUDE_PROXY_TIMEOUT"] = str(args.timeout)
    os.environ["_CLAUDE_PROXY_MAX_CONCURRENT"] = str(args.max_concurrent)

    total_capacity = args.workers * args.max_concurrent
    print("╔══════════════════════════════════════════════════════════╗")
    print("║  Claude Max API Proxy v3.0                              ║")
    print(f"║  Listening on http://{args.host}:{args.port:<5}                      ║")
    print("║                                                          ║")
    print(f"║  OpenAI API:    http://<host>:{args.port}/openai/v1          ║")
    print(f"║  Anthropic API: http://<host>:{args.port}/anthropic/v1      ║")
    print(f"║  Legacy compat: http://<host>:{args.port}/v1                ║")
    print("║                                                          ║")
    print(f"║  Workers: {args.workers} | Timeout: {args.timeout}s                       ║")
    print(f"║  Concurrent: {args.max_concurrent}/worker | Total capacity: {total_capacity:<3}      ║")
    print("║  Backend: Claude Code CLI (Max subscription)             ║")
    print("╚══════════════════════════════════════════════════════════╝")
    access_log = not args.no_access_log and os.environ.get("CLAUDE_PROXY_ACCESS_LOG", "true").lower() != "false"
    if args.workers > 1:
        uvicorn.run("server:app", host=args.host, port=args.port,
                    access_log=access_log, workers=args.workers)
    else:
        uvicorn.run(app, host=args.host, port=args.port, access_log=access_log)