Claude2API 部署完整总结

2025/9/1 逆向有效

项目简介

Claude2API 是一个将 Claude 网页服务转换为 OpenAI 兼容 API 的项目，支持识图、文件上传、思考输出等功能。

GitHub 项目: https://github.com/yushangxiao/claude2api

部署环境

系统: ubuntu22.04
服务器IP: YOUR_SERVER_IP
端口配置: 过盾代理(443)，API服务(8080)

部署步骤

1. 过盾测试

首先测试服务器是否能绕过 Cloudflare 防护：

python3 -c "
from curl_cffi import requests
session = requests.Session(impersonate='chrome110')
resp = session.get('https://claude.ai/')
print(f'Status: {resp.status_code}')
print(f'Title found: {\"<title>Claude</title>\" in resp.text}')
"

2. 部署过盾代理

2.1 创建代理脚本

sudo nano /opt/cf-bypass-proxy.py

#!/usr/bin/python3
from flask import Flask, request, Response, stream_with_context
from curl_cffi import requests
import json
import time
from threading import Thread, Event, Lock
from queue import Queue
from urllib.parse import urlparse
from datetime import datetime, timedelta

app = Flask(__name__)

# Session 管理
class SessionManager:
    def __init__(self, refresh_interval=3600):  # 默认每小时刷新
        self.session = None
        self.last_refresh = None
        self.refresh_interval = refresh_interval
        self.lock = Lock()
        self._create_session()
    
    def _create_session(self):
        """创建新的 session"""
        self.session = requests.Session(impersonate="chrome110")
        self.last_refresh = datetime.now()
        print(f"[{self.last_refresh}] Created new session")
    
    def get_session(self):
        """获取 session，必要时刷新"""
        with self.lock:
            if (datetime.now() - self.last_refresh).total_seconds() > self.refresh_interval:
                print(f"[{datetime.now()}] Session expired, creating new one...")
                self._create_session()
            return self.session

# 创建全局 session 管理器
session_manager = SessionManager(refresh_interval=3600)  # 每小时刷新

# 支持的域名映射
PROXY_DOMAINS = {
    'claude.ai': 'https://claude.ai',
    'www.claude.ai': 'https://claude.ai',
    # 保留原有的Perplexity域名以保持兼容性
    'www.perplexity.ai': 'https://www.perplexity.ai',
    'perplexity.ai': 'https://www.perplexity.ai',
    'api.cloudinary.com': 'https://api.cloudinary.com',
    'ppl-ai-file-upload.s3.amazonaws.com': 'https://ppl-ai-file-upload.s3.amazonaws.com',
    'pplx-res.cloudinary.com': 'https://pplx-res.cloudinary.com'
}

def heartbeat_generator(response_queue, stop_event):
    """生成心跳数据以保持连接活跃"""
    while not stop_event.is_set():
        time.sleep(2)
        response_queue.put(b' ')

@app.route('/', defaults={'path': ''}, methods=['GET', 'POST', 'PUT', 'DELETE', 'HEAD', 'OPTIONS'])
@app.route('/<path:path>', methods=['GET', 'POST', 'PUT', 'DELETE', 'HEAD', 'OPTIONS'])
def proxy(path):
    try:
        method = request.method
        headers = dict(request.headers)
        
        original_host = headers.get('Host', 'claude.ai')
        
        # 保留原始Cookie
        cookies = request.cookies
        
        # 删除可能导致问题的头部
        for h in ['Host', 'host', 'X-Real-Ip', 'X-Forwarded-For']:
            headers.pop(h, None)
        
        if original_host in PROXY_DOMAINS:
            base_url = PROXY_DOMAINS[original_host]
        else:
            base_url = 'https://claude.ai'
        
        target_url = f"{base_url}/{path}" if path else base_url
        if request.query_string:
            target_url += f"?{request.query_string.decode('utf-8')}"
        
        target_host = urlparse(base_url).netloc
        headers['Host'] = target_host
        
        # 添加Claude特定的头部
        if 'claude.ai' in target_host:
            headers['Origin'] = 'https://claude.ai'
            headers['Referer'] = 'https://claude.ai/'
            headers['anthropic-client-platform'] = 'web_claude_ai'
        
        print(f"Proxying {method} request from {original_host} to: {target_url}")
        
        # 更精确的SSE判断条件
        is_sse = (
            '/completion' in path and  # 必须是completion端点
            method == 'POST'  # 必须是POST请求
        ) or (
            'perplexity_ask' in path  # 保持原有的perplexity支持
        )
        
        print(f"Is SSE request: {is_sse}")
        
        if is_sse:
            return handle_sse_request(method, target_url, headers, cookies)
        else:
            return handle_normal_request(method, target_url, headers, cookies)
            
    except Exception as e:
        print(f"Exception: {str(e)}")
        import traceback
        traceback.print_exc()
        return Response(f"Error: {str(e)}", status=500)

def handle_normal_request(method, target_url, headers, cookies):
    """处理普通请求"""
    session = session_manager.get_session()
    
    try:
        # 打印请求详情以便调试
        print(f"Request headers: {json.dumps(headers, indent=2)}")
        
        if method in ["POST", "PUT", "PATCH", "DELETE"]:
            raw_data = request.get_data()
            if raw_data:
                # 打印请求体以便调试
                try:
                    print(f"Request body: {json.dumps(json.loads(raw_data), indent=2)}")
                except:
                    print(f"Request body (raw): {raw_data[:200]}...")
                
                resp = session.request(
                    method, 
                    target_url, 
                    data=raw_data, 
                    headers=headers, 
                    cookies=cookies,
                    timeout=30
                )
            else:
                resp = session.request(
                    method, 
                    target_url, 
                    headers=headers, 
                    cookies=cookies,
                    timeout=30
                )
        else:
            resp = session.request(
                method, 
                target_url, 
                headers=headers, 
                cookies=cookies,
                timeout=30
            )
        
        print(f"Response status: {resp.status_code}")
        
        # 打印响应头和内容以便调试
        print(f"Response headers: {json.dumps(dict(resp.headers), indent=2)}")
        try:
            print(f"Response body: {json.dumps(json.loads(resp.text), indent=2)}")
        except:
            print(f"Response body (raw): {resp.text[:200]}...")
        
        response_headers = {}
        for key, value in resp.headers.items():
            if key.lower() not in ['content-encoding', 'transfer-encoding', 'connection']:
                response_headers[key] = value
        
        return Response(
            resp.content,
            status=resp.status_code,
            headers=response_headers
        )
    except Exception as e:
        print(f"Error in normal request: {str(e)}")
        return Response(f"Request failed: {str(e)}", status=500)

def handle_sse_request(method, target_url, headers, cookies):
    """处理 SSE (Server-Sent Events) 请求"""
    def generate():
        response_queue = Queue()
        stop_event = Event()
        
        heartbeat_thread = Thread(target=heartbeat_generator, args=(response_queue, stop_event))
        heartbeat_thread.daemon = True
        heartbeat_thread.start()
        
        try:
            session = session_manager.get_session()
            
            # 打印请求详情以便调试
            print(f"SSE Request headers: {json.dumps(headers, indent=2)}")
            
            if method in ["POST", "PUT", "PATCH"]:
                raw_data = request.get_data()
                
                # 打印请求体以便调试
                try:
                    print(f"SSE Request body: {json.dumps(json.loads(raw_data), indent=2)}")
                except:
                    print(f"SSE Request body (raw): {raw_data[:200]}...")
                
                resp = session.request(
                    method, 
                    target_url, 
                    data=raw_data, 
                    headers=headers, 
                    cookies=cookies,
                    stream=True, 
                    timeout=60
                )
            else:
                resp = session.request(
                    method, 
                    target_url, 
                    headers=headers, 
                    cookies=cookies,
                    stream=True, 
                    timeout=60
                )
            
            print(f"SSE Response status: {resp.status_code}")
            print(f"SSE Response headers: {json.dumps(dict(resp.headers), indent=2)}")
            
            # 如果不是200状态码，返回错误信息
            if resp.status_code != 200:
                stop_event.set()
                error_content = resp.content
                try:
                    error_json = json.loads(error_content)
                    error_message = error_json.get('error', {}).get('message', str(error_content))
                    yield f"data: {{\"error\": \"{error_message}\"}}\n\n".encode()
                except:
                    yield f"data: {{\"error\": \"Request failed with status {resp.status_code}\"}}\n\n".encode()
                return
            
            # 正常处理流式响应
            for chunk in resp.iter_content(chunk_size=1024):
                if chunk:
                    yield chunk
                
                while not response_queue.empty():
                    response_queue.get()
            
        except Exception as e:
            print(f"Error in SSE request: {str(e)}")
            yield f"data: {{\"error\": \"SSE request failed: {str(e)}\"}}\n\n".encode()
        finally:
            stop_event.set()
            heartbeat_thread.join(timeout=1)
    
    response_headers = {
        'Content-Type': 'text/event-stream',
        'Cache-Control': 'no-cache',
        'Connection': 'keep-alive',
        'X-Accel-Buffering': 'no'
    }
    
    return Response(
        stream_with_context(generate()),
        status=200,
        headers=response_headers
    )

if __name__ == '__main__':
    import ssl
    import os
    
    if not os.path.exists('/opt/cert.pem'):
        print("Generating self-signed certificate...")
        domains = list(PROXY_DOMAINS.keys())
        san_list = ','.join([f'DNS:{domain}' for domain in domains])
        os.system(f'openssl req -x509 -newkey rsa:4096 -nodes -out /opt/cert.pem -keyout /opt/key.pem -days 365 -subj "/CN=proxy.local" -addext "subjectAltName={san_list}"')
    
    context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    context.load_cert_chain('/opt/cert.pem', '/opt/key.pem')
    
    app.run(host='0.0.0.0', port=443, ssl_context=context, threaded=True)

2.2 创建系统服务

sudo nano /etc/systemd/system/cf-bypass-proxy.service

[Unit]
Description=Cloudflare Bypass Proxy Service
After=network.target

[Service]
Type=simple
User=root
WorkingDirectory=/opt
ExecStart=/usr/bin/python3 /opt/cf-bypass-proxy.py
Restart=always

[Install]
WantedBy=multi-user.target

2.3 安装依赖并启动服务

sudo apt update
sudo apt install python3-pip -y
sudo pip3 install curl-cffi flask --break-system-packages

sudo systemctl daemon-reload
sudo systemctl start cf-bypass-proxy
sudo systemctl enable cf-bypass-proxy
sudo systemctl status cf-bypass-proxy

3. 部署 Docker 服务

3.1 安装 Docker

curl -fsSL https://get.docker.com -o get-docker.sh
sudo sh get-docker.sh

3.2 创建配置文件

mkdir /etc/claude2api
cd /etc/claude2api
sudo nano docker-compose.yml

services:
  claude2api:
    image: ghcr.io/yushangxiao/claude2api:latest
    container_name: claude2api
    network_mode: host
    environment:
      - SESSIONS=sk-ant-sid01-YOUR_SESSION_KEY_HERE （可用逗号分隔塞多个）
      - ADDRESS=0.0.0.0:8080
      - APIKEY=sk-YOUR_API_KEY_HERE
      - MAX_CHAT_HISTORY_LENGTH=1000000
      - NO_ROLE_PREFIX=false
      - CHAT_DELETE=true
      - PROMPT_DISABLE_ARTIFACTS=true
    extra_hosts:
      - "claude.ai:127.0.0.1"
    restart: unless-stopped

3.3 启动容器

docker compose up -d

4. 配置证书

每次重启容器后需要重新配置证书：

4.1 获取宿主机证书

sudo cat /opt/cert.pem

4.2 进入容器并添加证书

sudo docker exec -it claude2api /bin/sh

在容器内执行：

cat >> /etc/ssl/certs/ca-certificates.crt << 'EOF'
[将宿主机证书内容完整复制到这里]
EOF
exit

4.3 重启容器

sudo docker restart claude2api

遇到的问题及解决方案

现象: API 返回 403 错误，日志显示 "Failed to get org ID: unexpected status code: 403"

原因: Claude 的 sessionKey 过期或无效

解决方案:

登录 claude.ai
F12 -> Application -> Cookies -> 复制新的 sessionKey
更新 docker-compose.yml 中的 SESSIONS 配置
重启容器

问题2: 重启容器后证书失效

现象: API 返回 "tls: failed to verify certificate: x509: certificate signed by unknown authority"

原因: 容器重启后证书配置丢失，无法信任过盾代理的自签名证书

解决方案: 每次容器重启后都需要重新执行第4步的证书配置流程

问题3: 模型输出为空

现象: API不报错，但是content字段为空

原因: 免费账号只支持sonnet4，即使返回可用模型列表有其他模型也不能使用

解决方案: 只使用sonnet4

测试验证

API 健康检查

curl -X GET http://YOUR_SERVER_IP:8080/v1/models -H "Authorization: Bearer YOUR_API_KEY"

非流式请求测试

curl -X POST http://YOUR_SERVER_IP:8080/v1/chat/completions \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -d '{
    "model": "claude-sonnet-4-20250514",
    "messages": [
      {
        "role": "user",
        "content": "What is the capital of France?"
      }
    ],
    "stream": false
  }'

流式请求测试

curl -X POST http://YOUR_SERVER_IP:8080/v1/chat/completions \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -d '{
    "model": "claude-sonnet-4-20250514",
    "messages": [
      {
        "role": "user",
        "content": "Hello"
      }
    ],
    "stream": true
  }'

Claude2API 部署完整总结

项目简介

部署环境

部署步骤

1. 过盾测试

2. 部署过盾代理

2.1 创建代理脚本

2.2 创建系统服务

2.3 安装依赖并启动服务

3. 部署 Docker 服务

3.1 安装 Docker

3.2 创建配置文件

3.3 启动容器

4. 配置证书

4.1 获取宿主机证书

4.2 进入容器并添加证书

4.3 重启容器

遇到的问题及解决方案

问题2: 重启容器后证书失效

问题3: 模型输出为空

测试验证

API 健康检查

非流式请求测试

流式请求测试

MCP进程管理实现给llm看服务器输出

syd233

syd233.blog

Claude2API 部署完整总结

项目简介

部署环境

部署步骤

1. 过盾测试

2. 部署过盾代理

2.1 创建代理脚本

2.2 创建系统服务

2.3 安装依赖并启动服务

3. 部署 Docker 服务

3.1 安装 Docker

3.2 创建配置文件

3.3 启动容器

4. 配置证书

4.1 获取宿主机证书

4.2 进入容器并添加证书

4.3 重启容器

遇到的问题及解决方案

问题1: Session Cookie 过期

问题2: 重启容器后证书失效

问题3: 模型输出为空

测试验证

API 健康检查

非流式请求测试

流式请求测试

MCP进程管理实现给llm看服务器输出

syd233

syd233.blog