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

Initial implementation of claude-max-bridge

Browse Source 
OpenAI-compatible FastAPI server wrapping the Claude Code CLI.
Translates /v1/chat/completions requests into claude --print subprocess
calls using stream-json input/output for multi-turn support and real
streaming deltas.

Features:
- Full multi-turn conversation support via --input-format stream-json
- Real-time streaming with --include-partial-messages delta tracking
- Rate limit (429) and auth error (401) detection from CLI stderr
- Configurable timeout, graceful subprocess cleanup on disconnect
- /v1/models endpoint and /health check
This commit is contained in:
Florian Krebs
2026-05-22 16:08:44 +02:00
parent 9862a9d585
commit 3a58fba521
5 changed files with 599 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+40
View File
@@ -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
Dockerfile
+13
View File
@@ -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"]
README.md
+90 -1
View File
@@ -1,3 +1,92 @@
# claude-max-bridge
OpenAI-compatible HTTP bridge over the Claude Code CLI — connects LiteLLM to the Claude Max subscription
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.
requirements.txt
+2
View File
@@ -0,0 +1,2 @@
fastapi
uvicorn[standard]
server.py
+454
View File
@@ -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},
}