first commit

Author: Mng-dev-ai

.github/workflows/ci.yml

@@ -0,0 +1,34 @@
+name: CI
+
+on:
+  push:
+    branches: [main]
+  pull_request:
+    branches: [main]
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.13"
+
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install -e ".[test,dev]"
+
+      - name: Run ruff
+        run: ruff check anthropic_bridge/ tests/
+
+      - name: Run mypy
+        run: mypy anthropic_bridge/
+
+      - name: Run tests
+        env:
+          OPENROUTER_API_KEY: ${{ secrets.OPENROUTER_API_KEY }}
+        run: pytest tests/ -v

.gitignore

@@ -0,0 +1,8 @@
+__pycache__/
+*.pyc
+*.pyo
+*.egg-info/
+dist/
+build/
+.pytest_cache/
+.env

CLAUDE.md

@@ -0,0 +1,43 @@
+# CLAUDE.md
+
+This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
+
+## Project Overview
+
+anthropic-bridge is a proxy server that translates Anthropic Messages API requests into OpenRouter API format, enabling use of various LLM providers (Gemini, OpenAI, Grok, DeepSeek, Qwen, MiniMax) through an Anthropic-compatible interface.
+
+## Commands
+
+```bash
+# Install dependencies
+pip install -e ".[test,dev]"
+
+# Run server (requires OPENROUTER_API_KEY env var)
+OPENROUTER_API_KEY=your_key anthropic-bridge --port 8080 --host 127.0.0.1
+
+# Lint
+ruff check anthropic_bridge/ tests/
+
+# Type check
+mypy anthropic_bridge/
+
+# Run tests (requires OPENROUTER_API_KEY env var)
+OPENROUTER_API_KEY=your_key pytest tests/ -v
+```
+
+## Architecture
+
+**Request Flow**: Anthropic API request → `server.py` → `client.py` → OpenRouter API → SSE stream converted back to Anthropic format
+
+**Core Components**:
+- `server.py` - FastAPI app exposing `/v1/messages` endpoint that accepts Anthropic API format
+- `client.py` - `OpenRouterClient` handles request transformation and streams OpenRouter responses back as Anthropic SSE events
+- `transform.py` - Converts Anthropic messages/tools/tool_choice to OpenAI format for OpenRouter
+
+**Provider System** (`providers/`):
+- `BaseProvider` - Abstract base defining `process_text_content()`, `should_handle()`, and `prepare_request()` hooks
+- `ProviderRegistry` - Selects appropriate provider based on model ID
+- Provider implementations (Grok, Gemini, OpenAI, etc.) handle model-specific quirks like XML tool call parsing (Grok) or reasoning detail injection (Gemini)
+
+**Caching** (`cache.py`):
+- `ReasoningCache` persists Gemini reasoning details between tool call rounds to `~/.anthropic_bridge/cache/`

README.md

@@ -0,0 +1,83 @@
+# anthropic-bridge
+
+A proxy server that exposes an Anthropic Messages API-compatible endpoint while routing requests to various LLM providers through OpenRouter.
+
+## Features
+
+- Anthropic Messages API compatible (`/v1/messages`)
+- Streaming SSE responses
+- Tool/function calling support
+- Multi-round conversations
+- Support for multiple providers: Gemini, OpenAI, Grok, DeepSeek, Qwen, MiniMax
+- Extended thinking/reasoning support for compatible models
+- Reasoning cache for Gemini models across tool call rounds
+
+## Installation
+
+```bash
+pip install -e .
+
+# With development dependencies
+pip install -e ".[test,dev]"
+```
+
+## Usage
+
+Set your OpenRouter API key and start the server:
+
+```bash
+export OPENROUTER_API_KEY=your_key
+anthropic-bridge --port 8080 --host 127.0.0.1
+```
+
+Then point your Anthropic SDK client to `http://localhost:8080`:
+
+```python
+from anthropic import Anthropic
+
+client = Anthropic(
+    api_key="not-used",
+    base_url="http://localhost:8080"
+)
+
+response = client.messages.create(
+    model="google/gemini-2.5-pro-preview",  # Any OpenRouter model
+    max_tokens=1024,
+    messages=[{"role": "user", "content": "Hello!"}]
+)
+```
+
+## API Endpoints
+
+| Endpoint | Method | Description |
+|----------|--------|-------------|
+| `/` | GET | Health check |
+| `/health` | GET | Health check |
+| `/v1/messages` | POST | Anthropic Messages API |
+| `/v1/messages/count_tokens` | POST | Token counting (approximate) |
+
+## Configuration
+
+| Environment Variable | Required | Description |
+|---------------------|----------|-------------|
+| `OPENROUTER_API_KEY` | Yes | Your OpenRouter API key |
+
+| CLI Flag | Default | Description |
+|----------|---------|-------------|
+| `--port` | 8080 | Port to run on |
+| `--host` | 127.0.0.1 | Host to bind to |
+
+## Supported Models
+
+Any model available on OpenRouter can be used. Provider-specific optimizations exist for:
+
+- **Google Gemini** (`google/*`) - Reasoning detail caching
+- **OpenAI** (`openai/*`) - Extended thinking support
+- **xAI Grok** (`x-ai/*`) - XML tool call parsing
+- **DeepSeek** (`deepseek/*`)
+- **Qwen** (`qwen/*`)
+- **MiniMax** (`minimax/*`)
+
+## License
+
+MIT

anthropic_bridge/__init__.py

@@ -0,0 +1,3 @@
+from .server import create_app
+
+__all__ = ["create_app"]

anthropic_bridge/__main__.py

@@ -0,0 +1,28 @@
+import argparse
+import os
+
+import uvicorn
+
+from .server import create_app
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Anthropic Bridge Server")
+    parser.add_argument("--port", type=int, default=8080, help="Port to run on")
+    parser.add_argument("--host", default="127.0.0.1", help="Host to bind to")
+
+    args = parser.parse_args()
+
+    api_key = os.environ.get("OPENROUTER_API_KEY", "")
+    if not api_key:
+        print("Error: OPENROUTER_API_KEY environment variable required")
+        raise SystemExit(1)
+
+    app = create_app(openrouter_api_key=api_key)
+
+    print(f"Starting Anthropic Bridge on {args.host}:{args.port}")
+    uvicorn.run(app, host=args.host, port=args.port, log_level="info")
+
+
+if __name__ == "__main__":
+    main()

anthropic_bridge/cache.py

@@ -0,0 +1,88 @@
+import json
+import time
+from pathlib import Path
+from threading import Lock
+from typing import Any
+
+DEFAULT_CACHE_DIR = Path.home() / ".anthropic_bridge" / "cache"
+DEFAULT_TTL_DAYS = 30
+
+
+class ReasoningCache:
+    def __init__(self, cache_dir: Path | None = None, ttl_days: int = DEFAULT_TTL_DAYS):
+        self._cache_dir = cache_dir or DEFAULT_CACHE_DIR
+        self._cache_file = self._cache_dir / "reasoning_details.json"
+        self._ttl_seconds = ttl_days * 24 * 60 * 60
+        self._lock = Lock()
+        self._memory_cache: dict[str, dict[str, Any]] = {}
+        self._loaded = False
+
+    def _ensure_loaded(self) -> None:
+        if self._loaded:
+            return
+        with self._lock:
+            if self._loaded:  # Double-checked locking for thread safety
+                return  # type: ignore[unreachable]
+            self._cache_dir.mkdir(parents=True, exist_ok=True)
+            if self._cache_file.exists():
+                try:
+                    data = json.loads(self._cache_file.read_text())
+                    self._memory_cache = data if isinstance(data, dict) else {}
+                except (json.JSONDecodeError, OSError):
+                    self._memory_cache = {}
+            self._loaded = True
+
+    def _save(self) -> None:
+        try:
+            self._cache_file.write_text(json.dumps(self._memory_cache, indent=2))
+        except OSError:
+            pass
+
+    def _cleanup_expired(self) -> None:
+        now = time.time()
+        expired = [
+            k
+            for k, v in self._memory_cache.items()
+            if now - v.get("timestamp", 0) > self._ttl_seconds
+        ]
+        for k in expired:
+            del self._memory_cache[k]
+
+    def get(self, tool_call_id: str) -> list[dict[str, Any]] | None:
+        self._ensure_loaded()
+        entry = self._memory_cache.get(tool_call_id)
+        if not entry:
+            return None
+        if time.time() - entry.get("timestamp", 0) > self._ttl_seconds:
+            with self._lock:
+                self._memory_cache.pop(tool_call_id, None)
+                self._save()
+            return None
+        return entry.get("data")
+
+    def set(self, tool_call_id: str, reasoning_details: list[dict[str, Any]]) -> None:
+        self._ensure_loaded()
+        with self._lock:
+            self._memory_cache[tool_call_id] = {
+                "timestamp": time.time(),
+                "data": reasoning_details,
+            }
+            self._cleanup_expired()
+            self._save()
+
+    def clear(self) -> None:
+        with self._lock:
+            self._memory_cache = {}
+            if self._cache_file.exists():
+                self._cache_file.unlink()
+
+
+# global instance
+_cache: ReasoningCache | None = None
+
+
+def get_reasoning_cache() -> ReasoningCache:
+    global _cache
+    if _cache is None:
+        _cache = ReasoningCache()
+    return _cache