import logging from aiohttp import web, ClientSession, ClientError import urllib.parse logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) PORT = 9090 REVE_API_URL = "https://preview.reve.art/api/misc/chat" TOKEN_FILE = "token.txt" MODEL_MAPPING = { "claude-3-7-sonnet-20250219": "llm_claude_sonnet_3_7", "claude-3-7-sonnet-latest": "llm_claude_sonnet_3_7", "claude-3-5-sonnet-20241022": "llm_claude_sonnet_3_5_v2", "claude-3-5-sonnet-latest": "llm_claude_sonnet_3_5_v2", "claude-3-5-sonnet-20240620": "llm_claude_sonnet_3_5", "claude-3-5-haiku-20241022": "llm_claude_haiku_3_5_v2", "claude-3-5-haiku-latest": "llm_claude_haiku_3_5_v2" } def read_authorization_cookie() -> str: try: with open(TOKEN_FILE, 'r') as f: content = f.read().strip() if not content: logger.error(f"Cookie file '{TOKEN_FILE}' is empty. Please create it with your Reve authorization token.") exit(1) # Whole cookie (name=value) if content.startswith("c1."): logger.info("Detected the raw c1. cookie name=value") # Only decode if we don't see '=' in the token if '=' not in content: content = urllib.parse.unquote(content) parts = content.split('=', 1) if len(parts) == 2: cookie_value = parts[1] v2_index = cookie_value.find('v2.') if v2_index >= 0: return cookie_value[v2_index:] else: logger.error("Could not find 'v2.' in the cookie value") exit(1) else: logger.error("Invalid c1. cookie format. Expected name=value format") exit(1) # Netscape cookie file lines = content.splitlines() is_cookie_file = False auth_value = None for line in lines: if line.startswith('#') or not line.strip(): continue fields = line.split('\t') if len(fields) >= 6 and ("TRUE" in fields or "FALSE" in fields): is_cookie_file = True if len(fields) >= 7 and fields[5] == "auth": logger.info("Detected the Netscape cookie file") auth_value = fields[6] v2_index = auth_value.find('.v2.') if v2_index > 0: auth_value = auth_value[v2_index + 1:] break if is_cookie_file: if auth_value: return auth_value else: logger.error("Could not find 'auth' cookie in the Netscape cookie file.") exit(1) else: # Just the token logger.info("Detected the raw token") return content except FileNotFoundError: logger.error(f"Cookie file '{TOKEN_FILE}' not found. Please create it with your Reve authorization token.") exit(1) AUTH_TOKEN = read_authorization_cookie() if not AUTH_TOKEN.startswith("Bearer "): AUTH_TOKEN = f"Bearer {AUTH_TOKEN}" logger.info(f"Using Reve API token: {AUTH_TOKEN}") def convert_anthropic_to_reve_request(anthropic_req): model = anthropic_req.get("model", "claude-3-7-sonnet-latest") reve_model = MODEL_MAPPING.get(model, "llm_claude_sonnet_3_7") system_prompt = anthropic_req.get("system", "") if isinstance(system_prompt, list): system_text_chunks = [] for item in system_prompt: if isinstance(item, dict) and item.get("type") == "text": system_text_chunks.append(item.get("text", "")) elif isinstance(item, str): system_text_chunks.append(item) system_prompt = "\n".join(system_text_chunks) max_tokens = anthropic_req.get("max_tokens", 8192) temperature = anthropic_req.get("temperature", 1.0) temperature_percent = int(temperature * 100) conversation = [] for msg in anthropic_req.get("messages", []): role = msg.get("role", "user") content = msg.get("content", "") multi_content = [] if isinstance(content, str): multi_content = [{"text": content}] elif isinstance(content, list): for item in content: if isinstance(item, str): multi_content.append({"text": item}) elif isinstance(item, dict): if item.get("type") == "text": multi_content.append({"text": item.get("text", "")}) elif item.get("type") == "image": # Doesn't work for some reason # {"error_code":"UPSTREAM_ERROR","message":"Model inference failed","instance_id":"error-redacted"} src = item["source"] multi_content.append({ "image": { "type": src["type"], "media_type": src["type"], "data": src["data"] } }) conversation.append({ "role": role, "multi_content": multi_content }) reve_req = { "model": reve_model, "max_length": max_tokens, "system_prompt": system_prompt, "temperature_percent": temperature_percent, "conversation": conversation } return reve_req def convert_reve_to_anthropic_response(reve_resp, reve_model): content = reve_resp.get("response", "") anthropic_model = next((anthr_model for anthr_model, rev_model in MODEL_MAPPING.items() if rev_model == reve_model), "claude-3-7-sonnet-20250219") anthropic_resp = { "id": "msg_abcdef12", "type": "message", "role": "assistant", "content": [ { "type": "text", "text": content } ], "model": anthropic_model, "stop_reason": reve_resp.get("stop_reason", "end_turn"), "stop_sequence": None, "usage": { "input_tokens": reve_resp.get("prompt_tokens", 0), "output_tokens": reve_resp.get("completion_tokens", 0) } } return anthropic_resp async def handle_anthropic_request(request): try: anthropic_req = await request.json() if anthropic_req.get("stream", False): return web.json_response({ "error": { "type": "invalid_request_error", "message": "Streaming is not supported by this proxy." } }, status=400) reve_req = convert_anthropic_to_reve_request(anthropic_req) reve_model = reve_req['model'] logger.info(f"Doing request with model {reve_model}") try: async with ClientSession() as session: async with session.post( REVE_API_URL, json=reve_req, headers={ "Content-Type": "application/json; charset=utf-8", "Authorization": AUTH_TOKEN } ) as response: if response.status >= 400: error_text = await response.text() logger.error(f"API error ({response.status}): {error_text}") return web.Response( text=error_text, status=response.status, content_type='application/json' ) reve_resp = await response.json() logger.info(f"Received Reve response.") return web.json_response(convert_reve_to_anthropic_response(reve_resp, reve_model)) except ClientError as e: logger.error(f"Error calling Reve API: {str(e)}") return web.json_response({"error": str(e)}, status=500) except Exception as e: logger.error(f"Error processing request: {str(e)}") return web.json_response({"error": "Internal server error"}, status=500) def create_app(): app = web.Application() app.router.add_post('/v1/messages', handle_anthropic_request) return app if __name__ == "__main__": app = create_app() logger.info(f"Starting async proxy server on port {PORT}") # Edit the host to 0.0.0.0if you want it to be exposed. web.run_app(app, port=PORT, host='127.0.0.1')