diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..d8b8460 --- /dev/null +++ b/.gitignore @@ -0,0 +1,40 @@ +# Python +__pycache__/ +*.py[cod] +*$py.class +*.so +*.egg +*.egg-info/ +dist/ +build/ +.eggs/ +.Python +pip-wheel-metadata/ + +# Virtual environments +.env +.venv +env/ +venv/ +ENV/ + +# Distribution / packaging +*.tar.gz +*.whl + +# Test / coverage +.tox/ +.coverage +.coverage.* +.cache +htmlcov/ +.pytest_cache/ + +# IDE +.idea/ +.vscode/ +*.swp +*.swo + +# Docker +*.log diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..74b4bed --- /dev/null +++ b/Dockerfile @@ -0,0 +1,13 @@ +FROM python:3.12-slim + +RUN pip install --no-cache-dir fastapi "uvicorn[standard]" + +COPY server.py /app/server.py +WORKDIR /app + +# claude binary and ~/.claude config dir are bind-mounted at runtime: +# -v /usr/local/bin/claude:/usr/local/bin/claude:ro +# -v /root/.claude:/root/.claude:ro + +EXPOSE 8000 +CMD ["uvicorn", "server:app", "--host", "0.0.0.0", "--port", "8000"] diff --git a/README.md b/README.md index 5346aec..26523d7 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,92 @@ # claude-max-bridge -OpenAI-compatible HTTP bridge over the Claude Code CLI — connects LiteLLM to the Claude Max subscription \ No newline at end of file +A small FastAPI server that exposes an **OpenAI-compatible `/v1/chat/completions` +endpoint** and translates each request into a `claude` CLI subprocess call. +This lets LiteLLM (or any OpenAI-compatible client) consume a Claude Max +subscription through the locally-installed `claude` binary without needing an +Anthropic API key. + +## How it works + +1. Client sends an OpenAI chat-completions request (streaming or non-streaming). +2. The bridge extracts the system prompt and conversation turns. +3. It spawns `claude --print --output-format stream-json --input-format stream-json ...` + and pipes the conversation as NDJSON to stdin. +4. It parses the NDJSON stdout, computes text deltas from partial assistant + messages, and re-emits them as OpenAI SSE chunks (or buffers for + non-streaming). +5. Usage totals from the `result` event are forwarded in the final response. + +Authentication is handled entirely by the `claude` binary; the bridge just +bind-mounts the existing `~/.claude` credentials directory. + +## Environment variables + +| Variable | Default | Description | +|---------------------|------------------------|--------------------------------------------------| +| `CLAUDE_BIN` | `/usr/local/bin/claude`| Path to the `claude` CLI binary | +| `CLAUDE_HOME` | `/root/.claude` | Path to the Claude config / credentials directory| +| `LOG_LEVEL` | `INFO` | Python logging level | +| `REQUEST_TIMEOUT_S` | `120` | Per-request subprocess timeout in seconds | + +## Running with Docker + +```bash +docker build -t claude-max-bridge . + +docker run -d \ + --name claude-max-bridge \ + -p 8000:8000 \ + -v /usr/local/bin/claude:/usr/local/bin/claude:ro \ + -v /root/.claude:/root/.claude:ro \ + claude-max-bridge +``` + +## Supported models + +- `claude-sonnet-4-6` (default fallback) +- `claude-opus-4-5` +- `claude-haiku-4-5` +- `claude-opus-4-0` +- `claude-sonnet-3-7` +- `claude-haiku-3-5` + +Unknown model names in requests fall back to `claude-sonnet-4-6`. + +## LiteLLM configuration + +Add these entries to your `litellm_config.yaml`: + +```yaml +model_list: + - model_name: claude-sonnet-4-6 + litellm_params: + model: openai/claude-sonnet-4-6 + api_base: http://claude-max-bridge:8000/v1 + api_key: "unused" # required by LiteLLM but ignored by bridge + + - model_name: claude-opus-4-5 + litellm_params: + model: openai/claude-opus-4-5 + api_base: http://claude-max-bridge:8000/v1 + api_key: "unused" +``` + +Replace `claude-max-bridge` with the actual hostname or IP where the bridge +container is running. + +## Endpoints + +| Method | Path | Description | +|--------|------------------------|------------------------------------------| +| GET | `/health` | Liveness check, always returns `200 OK` | +| GET | `/v1/models` | Lists supported models | +| POST | `/v1/chat/completions` | OpenAI-compatible chat completions | + +## Notes + +- `temperature`, `top_p`, and other sampling parameters are silently ignored + (logged at DEBUG level) because the CLI does not expose them. +- Rate-limit responses from the CLI are translated to HTTP `429`. +- Auth errors from the CLI are translated to HTTP `401`. +- The subprocess is killed on client disconnect or timeout. diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..364e2ee --- /dev/null +++ b/requirements.txt @@ -0,0 +1,2 @@ +fastapi +uvicorn[standard] diff --git a/server.py b/server.py new file mode 100644 index 0000000..267a02f --- /dev/null +++ b/server.py @@ -0,0 +1,454 @@ +""" +claude-max-bridge: OpenAI-compatible API server that translates requests +to Claude Code CLI subprocess calls, enabling use of a Claude Max subscription +via LiteLLM or any OpenAI-compatible client. +""" + +import asyncio +import json +import logging +import os +import time +import uuid +from typing import AsyncGenerator, Optional + +from fastapi import FastAPI, HTTPException, Request +from fastapi.responses import JSONResponse, StreamingResponse + +# --------------------------------------------------------------------------- +# Configuration +# --------------------------------------------------------------------------- + +CLAUDE_BIN = os.environ.get("CLAUDE_BIN", "/usr/local/bin/claude") +CLAUDE_HOME = os.environ.get("CLAUDE_HOME", "/root/.claude") +LOG_LEVEL = os.environ.get("LOG_LEVEL", "INFO") +REQUEST_TIMEOUT_S = float(os.environ.get("REQUEST_TIMEOUT_S", "120")) + +logging.basicConfig( + level=getattr(logging, LOG_LEVEL.upper(), logging.INFO), + format="%(asctime)s %(levelname)s %(name)s: %(message)s", +) +log = logging.getLogger("claude-max-bridge") + +# --------------------------------------------------------------------------- +# Supported models (pass-through to CLI --model flag) +# --------------------------------------------------------------------------- + +SUPPORTED_MODELS = [ + "claude-opus-4-5", + "claude-sonnet-4-6", + "claude-haiku-4-5", + "claude-opus-4-0", + "claude-sonnet-3-7", + "claude-haiku-3-5", +] + +DEFAULT_MODEL = "claude-sonnet-4-6" + +# --------------------------------------------------------------------------- +# App +# --------------------------------------------------------------------------- + +app = FastAPI(title="claude-max-bridge", version="1.0.0") + + +# --------------------------------------------------------------------------- +# Helper: build subprocess env +# --------------------------------------------------------------------------- + +def _subprocess_env() -> dict: + env = os.environ.copy() + env["HOME"] = "/root" + env["XDG_CONFIG_HOME"] = os.path.join(CLAUDE_HOME, "..") # ~/.claude lives at CLAUDE_HOME + # Ensure the claude binary can find its config + env["CLAUDE_CONFIG_DIR"] = CLAUDE_HOME + # Prevent interactive prompts + env["TERM"] = "dumb" + env.pop("DISPLAY", None) + return env + + +# --------------------------------------------------------------------------- +# Helper: parse OpenAI messages → (system_prompt, conversation_ndjson) +# --------------------------------------------------------------------------- + +def _parse_messages(messages: list[dict]) -> tuple[Optional[str], list[dict]]: + """ + Split OpenAI messages into: + - system_prompt: first system message content (str | None) + - turns: list of {type, message} dicts suitable for NDJSON stdin + """ + system_prompt: Optional[str] = None + turns: list[dict] = [] + + for msg in messages: + role = msg.get("role", "") + content = msg.get("content", "") + + # Normalise content: list-of-parts → plain string + if isinstance(content, list): + text_parts = [] + for part in content: + if isinstance(part, dict) and part.get("type") == "text": + text_parts.append(part.get("text", "")) + elif isinstance(part, str): + text_parts.append(part) + content = "".join(text_parts) + + if role == "system": + if system_prompt is None: + system_prompt = content + else: + # Additional system messages: prepend to next user turn or skip + log.debug("Ignoring additional system message: %s", content[:80]) + elif role == "user": + turns.append({"type": "user", "message": {"role": "user", "content": content}}) + elif role == "assistant": + turns.append({"type": "assistant", "message": {"role": "assistant", "content": content}}) + else: + log.debug("Ignoring unknown role: %s", role) + + return system_prompt, turns + + +# --------------------------------------------------------------------------- +# Helper: build CLI command +# --------------------------------------------------------------------------- + +def _build_command(model: str, system_prompt: Optional[str]) -> list[str]: + cmd = [ + CLAUDE_BIN, + "--print", + "--output-format", "stream-json", + "--input-format", "stream-json", + "--include-partial-messages", + "--no-session-persistence", + "--tools", "", + "--model", model, + ] + if system_prompt: + cmd += ["--system-prompt", system_prompt] + return cmd + + +# --------------------------------------------------------------------------- +# Core: run subprocess and stream parsed events +# --------------------------------------------------------------------------- + +async def _run_claude( + model: str, + system_prompt: Optional[str], + turns: list[dict], +) -> AsyncGenerator[dict, None]: + """ + Yields dicts with keys: + {"event": "delta", "text": "..."} + {"event": "done", "usage": {...}} + {"event": "error", "status": int, "message": "..."} + """ + cmd = _build_command(model, system_prompt) + log.debug("Spawning: %s", " ".join(cmd[:6]) + " ...") + env = _subprocess_env() + + proc: Optional[asyncio.subprocess.Process] = None + try: + proc = await asyncio.create_subprocess_exec( + *cmd, + stdin=asyncio.subprocess.PIPE, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + env=env, + ) + + # Write conversation turns as NDJSON then close stdin + stdin_data = "\n".join(json.dumps(t) for t in turns) + "\n" + proc.stdin.write(stdin_data.encode()) + await proc.stdin.drain() + proc.stdin.close() + + last_text = "" + usage: dict = {} + + # Read stdout line by line with overall timeout + async def _read_lines(): + nonlocal last_text, usage + assert proc is not None + assert proc.stdout is not None + async for raw_line in proc.stdout: + line = raw_line.decode("utf-8", errors="replace").strip() + if not line: + continue + try: + event = json.loads(line) + except json.JSONDecodeError: + log.warning("Non-JSON stdout line: %s", line[:200]) + continue + + etype = event.get("type", "") + + if etype == "assistant": + msg = event.get("message", {}) + content = msg.get("content", []) + # content is a list; find the first text block + current_text = "" + for block in content: + if isinstance(block, dict) and block.get("type") == "text": + current_text = block.get("text", "") + break + if current_text and current_text != last_text: + delta = current_text[len(last_text):] + last_text = current_text + if delta: + yield {"event": "delta", "text": delta} + + elif etype == "result": + raw_usage = event.get("usage", {}) + usage = { + "prompt_tokens": raw_usage.get("input_tokens", 0), + "completion_tokens": raw_usage.get("output_tokens", 0), + "total_tokens": ( + raw_usage.get("input_tokens", 0) + + raw_usage.get("output_tokens", 0) + ), + } + yield {"event": "done", "usage": usage} + + elif etype == "system": + subtype = event.get("subtype", "") + if subtype == "error": + err_msg = event.get("error", {}).get("message", "Unknown CLI error") + yield {"event": "error", "status": 500, "message": err_msg} + return + + try: + async with asyncio.timeout(REQUEST_TIMEOUT_S): + async for ev in _read_lines(): + yield ev + except asyncio.TimeoutError: + log.error("Subprocess timed out after %.0fs", REQUEST_TIMEOUT_S) + yield {"event": "error", "status": 504, "message": f"Claude CLI timed out after {REQUEST_TIMEOUT_S:.0f}s"} + return + + # Wait for process to finish and collect stderr + try: + _, stderr_bytes = await asyncio.wait_for( + proc.communicate(), timeout=10.0 + ) + except asyncio.TimeoutError: + stderr_bytes = b"" + + stderr_text = stderr_bytes.decode("utf-8", errors="replace") if stderr_bytes else "" + rc = proc.returncode + + if rc != 0: + log.error("Claude CLI exited %d. stderr: %s", rc, stderr_text[:500]) + stderr_lower = stderr_text.lower() + if "rate limit" in stderr_lower or (rc == 1 and "usage limit" in stderr_lower): + yield {"event": "error", "status": 429, "message": stderr_text or "Rate limit exceeded"} + elif "not logged in" in stderr_lower or "authentication" in stderr_lower: + yield {"event": "error", "status": 401, "message": "Claude CLI authentication error"} + else: + yield {"event": "error", "status": 500, "message": f"Claude CLI error (exit {rc}): {stderr_text[:300]}"} + + except Exception as exc: + log.exception("Unexpected error running Claude CLI") + yield {"event": "error", "status": 500, "message": str(exc)} + finally: + if proc is not None and proc.returncode is None: + log.debug("Killing lingering subprocess") + try: + proc.kill() + await asyncio.wait_for(proc.wait(), timeout=5.0) + except Exception: + pass + + +# --------------------------------------------------------------------------- +# Streaming response builder +# --------------------------------------------------------------------------- + +async def _stream_response( + request: Request, + model: str, + system_prompt: Optional[str], + turns: list[dict], +) -> 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): + # Check if client disconnected + if await request.is_disconnected(): + log.info("Client disconnected, stopping stream") + return + + if ev["event"] == "delta": + chunk = { + "id": completion_id, + "object": "chat.completion.chunk", + "created": created, + "model": model, + "choices": [ + { + "index": 0, + "delta": {"content": ev["text"]}, + "finish_reason": None, + } + ], + } + yield f"data: {json.dumps(chunk)}\n\n" + + elif ev["event"] == "done": + got_done = True + # Final chunk with finish_reason + final_chunk = { + "id": completion_id, + "object": "chat.completion.chunk", + "created": created, + "model": model, + "choices": [ + { + "index": 0, + "delta": {}, + "finish_reason": "stop", + } + ], + "usage": ev.get("usage", {}), + } + yield f"data: {json.dumps(final_chunk)}\n\n" + yield "data: [DONE]\n\n" + + elif ev["event"] == "error": + # Emit an error chunk — client will see it in the stream + err_chunk = { + "id": completion_id, + "object": "chat.completion.chunk", + "created": created, + "model": model, + "choices": [], + "error": {"type": "server_error", "message": ev["message"]}, + } + yield f"data: {json.dumps(err_chunk)}\n\n" + yield "data: [DONE]\n\n" + return + + if not got_done: + # Ensure stream is always terminated + yield "data: [DONE]\n\n" + + return _gen() + + +# --------------------------------------------------------------------------- +# Routes +# --------------------------------------------------------------------------- + +@app.get("/health") +async def health(): + return {"status": "ok"} + + +@app.get("/v1/models") +async def list_models(): + now = int(time.time()) + return { + "object": "list", + "data": [ + { + "id": m, + "object": "model", + "created": now, + "owned_by": "anthropic", + } + for m in SUPPORTED_MODELS + ], + } + + +@app.post("/v1/chat/completions") +async def chat_completions(request: Request): + try: + body = await request.json() + except Exception: + raise HTTPException(status_code=400, detail="Invalid JSON body") + + # --- Parse request --- + messages = body.get("messages", []) + if not messages: + raise HTTPException(status_code=400, detail="messages must not be empty") + + raw_model = body.get("model", DEFAULT_MODEL) + # Accept model aliases / prefixes gracefully + model = raw_model if raw_model in SUPPORTED_MODELS else DEFAULT_MODEL + if model != raw_model: + log.debug("Unknown model %r, falling back to %s", raw_model, model) + + stream = bool(body.get("stream", False)) + + # Warn about unsupported / silently-ignored params + _IGNORED = {"temperature", "top_p", "frequency_penalty", "presence_penalty", + "logit_bias", "n", "best_of", "logprobs", "top_logprobs", + "response_format", "seed", "stop", "user"} + for param in _IGNORED: + if param in body: + log.debug("Ignoring unsupported parameter: %s", param) + + system_prompt, turns = _parse_messages(messages) + + if not turns: + raise HTTPException(status_code=400, detail="No user/assistant turns found in messages") + + # --- Streaming --- + if stream: + gen = await _stream_response(request, model, system_prompt, turns) + return StreamingResponse( + gen, + media_type="text/event-stream", + headers={ + "Cache-Control": "no-cache", + "X-Accel-Buffering": "no", + }, + ) + + # --- Non-streaming: buffer all deltas --- + full_text = "" + usage: dict = {} + error_ev: Optional[dict] = None + + async for ev in _run_claude(model, system_prompt, turns): + if ev["event"] == "delta": + full_text += ev["text"] + elif ev["event"] == "done": + usage = ev.get("usage", {}) + elif ev["event"] == "error": + error_ev = ev + break + + if error_ev: + status_code = error_ev.get("status", 500) + err_type = { + 429: "rate_limit_error", + 401: "authentication_error", + }.get(status_code, "server_error") + return JSONResponse( + status_code=status_code, + content={"error": {"type": err_type, "message": error_ev["message"]}}, + ) + + completion_id = f"chatcmpl-{uuid.uuid4().hex}" + return { + "id": completion_id, + "object": "chat.completion", + "created": int(time.time()), + "model": model, + "choices": [ + { + "index": 0, + "message": {"role": "assistant", "content": full_text}, + "finish_reason": "stop", + } + ], + "usage": usage or {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0}, + }