danvics/claude-code-proxy
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

first commit

Browse source
This commit is contained in:
Rahul Sengottuvelu 2025-03-18 20:35:36 -04:00
commit 5a52c511fb
7 changed files with 3841 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

38
.gitignore vendored Normal file
View file

@ -0,0 +1,38 @@
# Environment variables
.env
# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg
# Virtual environments
venv/
env/
ENV/
# Logs
*.log
# IDE specific files
.idea/
.vscode/
*.swp
*.swo

1
.python-version Normal file
View file

@ -0,0 +1 @@
3.10

85
README.md Normal file
View file

@ -0,0 +1,85 @@
# Claude-on-OpenAI Proxy
🚀 **Use Claude Code with OpenAI Models** 🚀
A proxy server that lets you use Claude Code with OpenAI models like GPT-4o.
## Why Use This?
- 🧠 **Leverage Claude Code with more models**: Use Claude Code's powerful coding interface but with OpenAI's more affordable models
- ⚡ **No code changes needed**: Your Claude clients work without modification
- 🔄 **Transparent model swapping**: Claude requests are automatically routed to OpenAI equivalents
## Quick Start
### Prerequisites
- OpenAI API key
### Setup
1. **Clone this repository**:
```bash
git clone https://github.com/1rgs/claude-code-openai.git
cd claude-code-openai
```
2. **Install UV**:
```bash
curl -LsSf https://astral.sh/uv/install.sh | sh
```
3. **Configure your API keys**:
Create a `.env` file with:
```
OPENAI_API_KEY=your-openai-key
```
4. **Start the proxy server**:
```bash
uv run uvicorn server:app --host 0.0.0.0 --port 8082
```
### Using with Claude Code
1. **Install Claude Code** (if you haven't already):
```bash
npm install -g @anthropic-ai/claude-code
```
2. **Connect to your proxy**:
```bash
ANTHROPIC_BASE_URL=http://localhost:8082 DISABLE_PROMPT_CACHING=1 claude
```
3. **That's it!** Your Claude Code client will now use OpenAI models through the proxy.
## Model Mapping
The proxy automatically maps Claude models to OpenAI models:
| Claude Model | OpenAI Model |
|--------------|--------------|
| haiku | gpt-4o-mini |
| sonnet | o3-mini |
You can customize these mappings in `server.py` by editing the `validate_model` function.
## How It Works
This proxy works by:
1. **Receiving requests** in Anthropic's API format
2. **Translating** the requests to OpenAI format via LiteLLM
3. **Sending** the translated request to OpenAI
4. **Converting** the response back to Anthropic format
5. **Returning** the formatted response to the client
The proxy handles both streaming and non-streaming responses, maintaining compatibility with all Claude clients.
## Contributing
Contributions are welcome! Please feel free to submit a Pull Request.

14
pyproject.toml Normal file
View file

@ -0,0 +1,14 @@
[project]
name = "anthropic-proxy"
version = "0.1.0"
description = "Proxy that translates between Anthropic API and LiteLLM"
readme = "README.md"
requires-python = ">=3.10"
dependencies = [
"fastapi[standard]>=0.115.11",
"uvicorn>=0.34.0",
"httpx>=0.25.0",
"pydantic>=2.0.0",
"litellm>=1.40.14",
"python-dotenv>=1.0.0",
]

1266
server.py Normal file
View file

File diff suppressed because it is too large Load diff

716
tests.py Normal file
View file

@ -0,0 +1,716 @@
#!/usr/bin/env python3
"""
Comprehensive test suite for Claude-on-OpenAI Proxy.
This script provides tests for both streaming and non-streaming requests,
with various scenarios including tool use, multi-turn conversations,
and content blocks.
Usage:
python tests.py # Run all tests
python tests.py --no-streaming # Skip streaming tests
python tests.py --simple # Run only simple tests
python tests.py --tools # Run tool-related tests only
"""
import os
import json
import time
import httpx
import argparse
import asyncio
import sys
from datetime import datetime
from typing import Dict, Any, List, Optional, Set
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
# Configuration
ANTHROPIC_API_KEY = os.environ.get("ANTHROPIC_API_KEY")
PROXY_API_KEY = os.environ.get("ANTHROPIC_API_KEY") # Using same key for proxy
ANTHROPIC_API_URL = "https://api.anthropic.com/v1/messages"
PROXY_API_URL = "http://localhost:8082/v1/messages"
ANTHROPIC_VERSION = "2023-06-01"
MODEL = "claude-3-sonnet-20240229" # Change to your preferred model
# Headers
anthropic_headers = {
"x-api-key": ANTHROPIC_API_KEY,
"anthropic-version": ANTHROPIC_VERSION,
"content-type": "application/json",
}
proxy_headers = {
"x-api-key": PROXY_API_KEY,
"anthropic-version": ANTHROPIC_VERSION,
"content-type": "application/json",
}
# Tool definitions
calculator_tool = {
"name": "calculator",
"description": "Evaluate mathematical expressions",
"input_schema": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "The mathematical expression to evaluate"
}
},
"required": ["expression"]
}
}
weather_tool = {
"name": "weather",
"description": "Get weather information for a location",
"input_schema": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "The city or location to get weather for"
},
"units": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "Temperature units"
}
},
"required": ["location"]
}
}
search_tool = {
"name": "search",
"description": "Search for information on the web",
"input_schema": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The search query"
}
},
"required": ["query"]
}
}
# Test scenarios
TEST_SCENARIOS = {
# Simple text response
"simple": {
"model": MODEL,
"max_tokens": 300,
"messages": [
{"role": "user", "content": "Hello, world! Can you tell me about Paris in 2-3 sentences?"}
]
},
# Basic tool use
"calculator": {
"model": MODEL,
"max_tokens": 300,
"messages": [
{"role": "user", "content": "What is 135 + 7.5 divided by 2.5?"}
],
"tools": [calculator_tool],
"tool_choice": {"type": "auto"}
},
# Multiple tools
"multi_tool": {
"model": MODEL,
"max_tokens": 500,
"temperature": 0.7,
"top_p": 0.95,
"system": "You are a helpful assistant that uses tools when appropriate. Be concise and precise.",
"messages": [
{"role": "user", "content": "I'm planning a trip to New York next week. What's the weather like and what are some interesting places to visit?"}
],
"tools": [weather_tool, search_tool],
"tool_choice": {"type": "auto"}
},
# Multi-turn conversation
"multi_turn": {
"model": MODEL,
"max_tokens": 500,
"messages": [
{"role": "user", "content": "Let's do some math. What is 240 divided by 8?"},
{"role": "assistant", "content": "To calculate 240 divided by 8, I'll perform the division:\n\n240 ÷ 8 = 30\n\nSo the result is 30."},
{"role": "user", "content": "Now multiply that by 4 and tell me the result."}
],
"tools": [calculator_tool],
"tool_choice": {"type": "auto"}
},
# Content blocks
"content_blocks": {
"model": MODEL,
"max_tokens": 500,
"messages": [
{"role": "user", "content": [
{"type": "text", "text": "I need to know the weather in Los Angeles and calculate 75.5 / 5. Can you help with both?"}
]}
],
"tools": [calculator_tool, weather_tool],
"tool_choice": {"type": "auto"}
},
# Simple streaming test
"simple_stream": {
"model": MODEL,
"max_tokens": 100,
"stream": True,
"messages": [
{"role": "user", "content": "Count from 1 to 5, with one number per line."}
]
},
# Tool use with streaming
"calculator_stream": {
"model": MODEL,
"max_tokens": 300,
"stream": True,
"messages": [
{"role": "user", "content": "What is 135 + 17.5 divided by 2.5?"}
],
"tools": [calculator_tool],
"tool_choice": {"type": "auto"}
}
}
# Required event types for Anthropic streaming responses
REQUIRED_EVENT_TYPES = {
"message_start",
"content_block_start",
"content_block_delta",
"content_block_stop",
"message_delta",
"message_stop"
}
# ================= NON-STREAMING TESTS =================
def get_response(url, headers, data):
"""Send a request and get the response."""
start_time = time.time()
response = httpx.post(url, headers=headers, json=data, timeout=30)
elapsed = time.time() - start_time
print(f"Response time: {elapsed:.2f} seconds")
return response
def compare_responses(anthropic_response, proxy_response, check_tools=False):
"""Compare the two responses to see if they're similar enough."""
anthropic_json = anthropic_response.json()
proxy_json = proxy_response.json()
print("\n--- Anthropic Response Structure ---")
print(json.dumps({k: v for k, v in anthropic_json.items() if k != "content"}, indent=2))
print("\n--- Proxy Response Structure ---")
print(json.dumps({k: v for k, v in proxy_json.items() if k != "content"}, indent=2))
# Basic structure verification with more flexibility
# The proxy might map values differently, so we're more lenient in our checks
assert proxy_json.get("role") == "assistant", "Proxy role is not 'assistant'"
assert proxy_json.get("type") == "message", "Proxy type is not 'message'"
# Check if stop_reason is reasonable (might be different between Anthropic and our proxy)
valid_stop_reasons = ["end_turn", "max_tokens", "stop_sequence", "tool_use", None]
assert proxy_json.get("stop_reason") in valid_stop_reasons, "Invalid stop reason"
# Check content exists and has valid structure
assert "content" in anthropic_json, "No content in Anthropic response"
assert "content" in proxy_json, "No content in Proxy response"
anthropic_content = anthropic_json["content"]
proxy_content = proxy_json["content"]
# Make sure content is a list and has at least one item
assert isinstance(anthropic_content, list), "Anthropic content is not a list"
assert isinstance(proxy_content, list), "Proxy content is not a list"
assert len(proxy_content) > 0, "Proxy content is empty"
# If we're checking for tool uses
if check_tools:
# Check if content has tool use
anthropic_tool = None
proxy_tool = None
# Find tool use in Anthropic response
for item in anthropic_content:
if item.get("type") == "tool_use":
anthropic_tool = item
break
# Find tool use in Proxy response
for item in proxy_content:
if item.get("type") == "tool_use":
proxy_tool = item
break
# At least one of them should have a tool use
if anthropic_tool is not None:
print("\n---------- ANTHROPIC TOOL USE ----------")
print(json.dumps(anthropic_tool, indent=2))
if proxy_tool is not None:
print("\n---------- PROXY TOOL USE ----------")
print(json.dumps(proxy_tool, indent=2))
# Check tool structure
assert proxy_tool.get("name") is not None, "Proxy tool has no name"
assert proxy_tool.get("input") is not None, "Proxy tool has no input"
print("\n✅ Both responses contain tool use")
else:
print("\n⚠️ Proxy response does not contain tool use, but Anthropic does")
elif proxy_tool is not None:
print("\n---------- PROXY TOOL USE ----------")
print(json.dumps(proxy_tool, indent=2))
print("\n⚠️ Proxy response contains tool use, but Anthropic does not")
else:
print("\n⚠️ Neither response contains tool use")
# Check if content has text
anthropic_text = None
proxy_text = None
for item in anthropic_content:
if item.get("type") == "text":
anthropic_text = item.get("text")
break
for item in proxy_content:
if item.get("type") == "text":
proxy_text = item.get("text")
break
# For tool use responses, there might not be text content
if check_tools and (anthropic_text is None or proxy_text is None):
print("\n⚠️ One or both responses don't have text content (expected for tool-only responses)")
return True
assert anthropic_text is not None, "No text found in Anthropic response"
assert proxy_text is not None, "No text found in Proxy response"
# Print the first few lines of each text response
max_preview_lines = 5
anthropic_preview = "\n".join(anthropic_text.strip().split("\n")[:max_preview_lines])
proxy_preview = "\n".join(proxy_text.strip().split("\n")[:max_preview_lines])
print("\n---------- ANTHROPIC TEXT PREVIEW ----------")
print(anthropic_preview)
print("\n---------- PROXY TEXT PREVIEW ----------")
print(proxy_preview)
# Check for some minimum text overlap - proxy might have different exact wording
# but should have roughly similar content
return True # We're not enforcing similarity, just basic structure
def test_request(test_name, request_data, check_tools=False):
"""Run a test with the given request data."""
print(f"\n{'='*20} RUNNING TEST: {test_name} {'='*20}")
# Log the request data
print(f"\nRequest data:\n{json.dumps({k: v for k, v in request_data.items() if k != 'messages'}, indent=2)}")
# Make copies of the request data to avoid modifying the original
anthropic_data = request_data.copy()
proxy_data = request_data.copy()
try:
# Send requests to both APIs
print("\nSending to Anthropic API...")
anthropic_response = get_response(ANTHROPIC_API_URL, anthropic_headers, anthropic_data)
print("\nSending to Proxy...")
proxy_response = get_response(PROXY_API_URL, proxy_headers, proxy_data)
# Check response codes
print(f"\nAnthropic status code: {anthropic_response.status_code}")
print(f"Proxy status code: {proxy_response.status_code}")
if anthropic_response.status_code != 200 or proxy_response.status_code != 200:
print("\n⚠️ One or both requests failed")
if anthropic_response.status_code != 200:
print(f"Anthropic error: {anthropic_response.text}")
if proxy_response.status_code != 200:
print(f"Proxy error: {proxy_response.text}")
return False
# Compare the responses
result = compare_responses(anthropic_response, proxy_response, check_tools=check_tools)
if result:
print(f"\n✅ Test {test_name} passed!")
return True
else:
print(f"\n❌ Test {test_name} failed!")
return False
except Exception as e:
print(f"\n❌ Error in test {test_name}: {str(e)}")
import traceback
traceback.print_exc()
return False
# ================= STREAMING TESTS =================
class StreamStats:
"""Track statistics about a streaming response."""
def __init__(self):
self.event_types = set()
self.event_counts = {}
self.first_event_time = None
self.last_event_time = None
self.total_chunks = 0
self.events = []
self.text_content = ""
self.content_blocks = {}
self.has_tool_use = False
self.has_error = False
self.error_message = ""
self.text_content_by_block = {}
def add_event(self, event_data):
"""Track information about each received event."""
now = datetime.now()
if self.first_event_time is None:
self.first_event_time = now
self.last_event_time = now
self.total_chunks += 1
# Record event type and increment count
if "type" in event_data:
event_type = event_data["type"]
self.event_types.add(event_type)
self.event_counts[event_type] = self.event_counts.get(event_type, 0) + 1
# Track specific event data
if event_type == "content_block_start":
block_idx = event_data.get("index")
content_block = event_data.get("content_block", {})
if content_block.get("type") == "tool_use":
self.has_tool_use = True
self.content_blocks[block_idx] = content_block
self.text_content_by_block[block_idx] = ""
elif event_type == "content_block_delta":
block_idx = event_data.get("index")
delta = event_data.get("delta", {})
if delta.get("type") == "text_delta":
text = delta.get("text", "")
self.text_content += text
# Also track text by block ID
if block_idx in self.text_content_by_block:
self.text_content_by_block[block_idx] += text
# Keep track of all events for debugging
self.events.append(event_data)
def get_duration(self):
"""Calculate the total duration of the stream in seconds."""
if self.first_event_time is None or self.last_event_time is None:
return 0
return (self.last_event_time - self.first_event_time).total_seconds()
def summarize(self):
"""Print a summary of the stream statistics."""
print(f"Total chunks: {self.total_chunks}")
print(f"Unique event types: {sorted(list(self.event_types))}")
print(f"Event counts: {json.dumps(self.event_counts, indent=2)}")
print(f"Duration: {self.get_duration():.2f} seconds")
print(f"Has tool use: {self.has_tool_use}")
# Print the first few lines of content
if self.text_content:
max_preview_lines = 5
text_preview = "\n".join(self.text_content.strip().split("\n")[:max_preview_lines])
print(f"Text preview:\n{text_preview}")
else:
print("No text content extracted")
if self.has_error:
print(f"Error: {self.error_message}")
async def stream_response(url, headers, data, stream_name):
"""Send a streaming request and process the response."""
print(f"\nStarting {stream_name} stream...")
stats = StreamStats()
error = None
try:
async with httpx.AsyncClient() as client:
# Add stream flag to ensure it's streamed
request_data = data.copy()
request_data["stream"] = True
start_time = time.time()
async with client.stream("POST", url, json=request_data, headers=headers, timeout=30) as response:
if response.status_code != 200:
error_text = await response.aread()
stats.has_error = True
stats.error_message = f"HTTP {response.status_code}: {error_text.decode('utf-8')}"
error = stats.error_message
print(f"Error: {stats.error_message}")
return stats, error
print(f"{stream_name} connected, receiving events...")
# Process each chunk
buffer = ""
async for chunk in response.aiter_text():
if not chunk.strip():
continue
# Handle multiple events in one chunk
buffer += chunk
events = buffer.split("\n\n")
# Process all complete events
for event_text in events[:-1]: # All but the last (possibly incomplete) event
if not event_text.strip():
continue
# Parse server-sent event format
if "data: " in event_text:
# Extract the data part
data_parts = []
for line in event_text.split("\n"):
if line.startswith("data: "):
data_part = line[len("data: "):]
# Skip the "[DONE]" marker
if data_part == "[DONE]":
break
data_parts.append(data_part)
if data_parts:
try:
event_data = json.loads("".join(data_parts))
stats.add_event(event_data)
except json.JSONDecodeError as e:
print(f"Error parsing event: {e}\nRaw data: {''.join(data_parts)}")
# Keep the last (potentially incomplete) event for the next iteration
buffer = events[-1] if events else ""
# Process any remaining complete events in the buffer
if buffer.strip():
lines = buffer.strip().split("\n")
data_lines = [line[len("data: "):] for line in lines if line.startswith("data: ")]
if data_lines and data_lines[0] != "[DONE]":
try:
event_data = json.loads("".join(data_lines))
stats.add_event(event_data)
except:
pass
elapsed = time.time() - start_time
print(f"{stream_name} stream completed in {elapsed:.2f} seconds")
except Exception as e:
stats.has_error = True
stats.error_message = str(e)
error = str(e)
print(f"Error in {stream_name} stream: {e}")
return stats, error
def compare_stream_stats(anthropic_stats, proxy_stats):
"""Compare the statistics from the two streams to see if they're similar enough."""
print("\n--- Stream Comparison ---")
# Required events
anthropic_missing = REQUIRED_EVENT_TYPES - anthropic_stats.event_types
proxy_missing = REQUIRED_EVENT_TYPES - proxy_stats.event_types
print(f"Anthropic missing event types: {anthropic_missing}")
print(f"Proxy missing event types: {proxy_missing}")
# Check if proxy has the required events
if proxy_missing:
print(f"⚠️ Proxy is missing required event types: {proxy_missing}")
else:
print("✅ Proxy has all required event types")
# Compare content
if anthropic_stats.text_content and proxy_stats.text_content:
anthropic_preview = "\n".join(anthropic_stats.text_content.strip().split("\n")[:5])
proxy_preview = "\n".join(proxy_stats.text_content.strip().split("\n")[:5])
print("\n--- Anthropic Content Preview ---")
print(anthropic_preview)
print("\n--- Proxy Content Preview ---")
print(proxy_preview)
# Compare tool use
if anthropic_stats.has_tool_use and proxy_stats.has_tool_use:
print("✅ Both have tool use")
elif anthropic_stats.has_tool_use and not proxy_stats.has_tool_use:
print("⚠️ Anthropic has tool use but proxy does not")
elif not anthropic_stats.has_tool_use and proxy_stats.has_tool_use:
print("⚠️ Proxy has tool use but Anthropic does not")
# Success as long as proxy has some content and no errors
return (not proxy_stats.has_error and
len(proxy_stats.text_content) > 0 or proxy_stats.has_tool_use)
async def test_streaming(test_name, request_data):
"""Run a streaming test with the given request data."""
print(f"\n{'='*20} RUNNING STREAMING TEST: {test_name} {'='*20}")
# Log the request data
print(f"\nRequest data:\n{json.dumps({k: v for k, v in request_data.items() if k != 'messages'}, indent=2)}")
# Make copies of the request data to avoid modifying the original
anthropic_data = request_data.copy()
proxy_data = request_data.copy()
if not anthropic_data.get("stream"):
anthropic_data["stream"] = True
if not proxy_data.get("stream"):
proxy_data["stream"] = True
check_tools = "tools" in request_data
try:
# Send streaming requests
anthropic_stats, anthropic_error = await stream_response(
ANTHROPIC_API_URL, anthropic_headers, anthropic_data, "Anthropic"
)
proxy_stats, proxy_error = await stream_response(
PROXY_API_URL, proxy_headers, proxy_data, "Proxy"
)
# Print statistics
print("\n--- Anthropic Stream Statistics ---")
anthropic_stats.summarize()
print("\n--- Proxy Stream Statistics ---")
proxy_stats.summarize()
# Compare the responses
if anthropic_error:
print(f"\n⚠️ Anthropic stream had an error: {anthropic_error}")
# If Anthropic errors, the test passes if proxy does anything useful
if not proxy_error and proxy_stats.total_chunks > 0:
print(f"\n✅ Test {test_name} passed! (Proxy worked even though Anthropic failed)")
return True
else:
print(f"\n❌ Test {test_name} failed! Both streams had errors.")
return False
if proxy_error:
print(f"\n❌ Test {test_name} failed! Proxy had an error: {proxy_error}")
return False
result = compare_stream_stats(anthropic_stats, proxy_stats)
if result:
print(f"\n✅ Test {test_name} passed!")
return True
else:
print(f"\n❌ Test {test_name} failed!")
return False
except Exception as e:
print(f"\n❌ Error in test {test_name}: {str(e)}")
import traceback
traceback.print_exc()
return False
# ================= MAIN =================
async def run_tests(args):
"""Run all tests based on command-line arguments."""
# Track test results
results = {}
# First run non-streaming tests
if not args.streaming_only:
print("\n\n=========== RUNNING NON-STREAMING TESTS ===========\n")
for test_name, test_data in TEST_SCENARIOS.items():
# Skip streaming tests
if test_data.get("stream"):
continue
# Skip tool tests if requested
if args.simple and "tools" in test_data:
continue
# Skip non-tool tests if tools_only
if args.tools_only and "tools" not in test_data:
continue
# Run the test
check_tools = "tools" in test_data
result = test_request(test_name, test_data, check_tools=check_tools)
results[test_name] = result
# Now run streaming tests
if not args.no_streaming:
print("\n\n=========== RUNNING STREAMING TESTS ===========\n")
for test_name, test_data in TEST_SCENARIOS.items():
# Only select streaming tests, or force streaming
if not test_data.get("stream") and not test_name.endswith("_stream"):
continue
# Skip tool tests if requested
if args.simple and "tools" in test_data:
continue
# Skip non-tool tests if tools_only
if args.tools_only and "tools" not in test_data:
continue
# Run the streaming test
result = await test_streaming(test_name, test_data)
results[f"{test_name}_streaming"] = result
# Print summary
print("\n\n=========== TEST SUMMARY ===========\n")
total = len(results)
passed = sum(1 for v in results.values() if v)
for test, result in results.items():
print(f"{test}: {'✅ PASS' if result else '❌ FAIL'}")
print(f"\nTotal: {passed}/{total} tests passed")
if passed == total:
print("\n🎉 All tests passed!")
return True
else:
print(f"\n⚠️ {total - passed} tests failed")
return False
async def main():
# Check that API key is set
if not ANTHROPIC_API_KEY:
print("Error: ANTHROPIC_API_KEY not set in .env file")
return
# Parse command-line arguments
parser = argparse.ArgumentParser(description="Test the Claude-on-OpenAI proxy")
parser.add_argument("--no-streaming", action="store_true", help="Skip streaming tests")
parser.add_argument("--streaming-only", action="store_true", help="Only run streaming tests")
parser.add_argument("--simple", action="store_true", help="Only run simple tests (no tools)")
parser.add_argument("--tools-only", action="store_true", help="Only run tool tests")
args = parser.parse_args()
# Run tests
success = await run_tests(args)
sys.exit(0 if success else 1)
if __name__ == "__main__":
asyncio.run(main())

1721
uv.lock Normal file
View file

File diff suppressed because it is too large Load diff