1、在win11中安装git和nodejs

安装完成后,在git shell中查看 npm版本号

21565@xiaoxin MINGW64 ~
$ npm -v
10.2.3
21565@xiaoxin MINGW64 ~
$ node -v
v18.19.0

2、安装claude code
npm install -g @anthropic-ai/claude-code

3、在智普官网上免费实名注册

4、查看自己的apikeys,增加一个用于编码的key,比如名称叫 coder

https://bigmodel.cn/usercenter/proj-mgmt/apikeys

5、设置环境变量

export ANTHROPIC_BASE_URL="https://open.bigmodel.cn/api/anthropic"
export ANTHROPIC_AUTH_TOKEN="your Zhipu API key"

6、打开claude

# 创建您的工作目录,例如 `your-project`,

使用 `cd` 命令导航到您的项目 cd your-project

# 运行命令 `claude` 即可进入 Claude Code 交互界面 claude

7、claude会打印欢迎词,然后让你选择颜色风格(默认选dark即可)

8、输入指令,比如  “请编写logutil.py”  就可以看到claude自动生成了logutil.py文件

9、升级claude

npm update -g @anthropic-ai/claude-code  --用这个命令升级

升级后,claude-code新版本要求必须先登陆。可以通过下面的方法绕过

vi ~/.claude.json   增加下面的代码

"hasCompletedOnboarding": true,

 "env": {  

"ANTHROPIC_BASE_URL": "你的智谱API地址",

 "ANTHROPIC_AUTH_TOKEN": "你的智谱Token"  

}

我的环境中还有另外一个ai ide把这个文件改了,导致连不上智谱。把settings.json暂时改名,claude就可以连上智谱了。
cat ~/.claude/settings.json
{
  "env": {
    "ANTHROPIC_AUTH_TOKEN": "",
    "ANTHROPIC_BASE_URL": "http://127.0.0.1:3456",

2026-5-3

购买的智谱token到期了,切换成公司内的模型吧。公司内的模型时openai格式的,需要转换为nthropic格式。

先是下载了cc-switch的rpm版本(CC-Switch-v3.14.1-Linux-x86_64.rpm),发现本地的动态库不匹配。

然后又下载了cc-switch的独立包CC-Switch-v3.14.1-Linux-x86_64.AppImage,发现还是运行不了,说glibc版本不匹配

然后找了一个 one-api的docker镜像

还有claude-code-proxy这个轻量级python工具,这个走通了,记录一下

1、先安装python依赖  pip install claude-code-proxy

2、git clone https://github.com/fuergaosi233/claude-code-proxy.git

cd claude-code-proxy

cp .env.example .env

修改 .env中的 url,token,

3、uv run start_proxy.py

本地环境变量中有proxy设置,要都unset掉

4、然后运行claude,发现claude会一直超时

5、把claude-code-proxy的打印日志调到debug级别,手工访问
http://127.0.0.1:8082/,能通。

6、把claude的env变量都修改了,还是不行

7、把claude调的dubug模式,grep log日志,还是没有8082的踪迹

8、用 lsof -i -P -n | grep claude 这个命令,发现claude根本就没有连接8082
claude    1024967   17u  IPv4 19124435      0t0  TCP 10.137.239.107:49890->47.110.175.20:443 (SYN_SENT)
claude    1024967   20u  IPv4 19124707      0t0  TCP 10.137.239.107:49900->47.110.175.20:443 (SYN_SENT)
9、说明.claude/settings.json中的设置没有生效。再检查,发现文件名写错了setting.json,少了个s

10、修改文件名后,就ok了

$ claude
 ▐▛███▜▌   Claude Code v2.1.126
▝▜█████▛▘  Sonnet 4.6 · API Usage Billing
  ▘▘ ▝▝    ~/code/claudecode2

❯ hello                                                                                                                                                
        
● Hello! How can I help you with your software engineering tasks today?                                                                                
                                                                                                                                                     
✻ Cogitated for 2s                                                                                                                                     
   
❯ 今天是星期几                                                                                                                                         
                                                                                                                                                     
● 今天是 2026 年 5 月 3 日,星期日。                                                                                                                   
                                                                                                                                                       
✻ Worked for 1s
                                                                                                                                                       
❯ 你是哪个大模型                                                                                                                                       
                                                                                                                                                       
● 我是 Claude Sonnet 4.6,由 Anthropic 开发。这是我在 Claude Code(一个命令行工具)中的实例。                                                          
                                                                                                                                                       
✻ Worked for 1s
                                                                                                                                                       
❯ 你知道郭靖的母亲是谁吗                                                                                                                               
                                                                                                                                                       
● 郭靖是金庸武侠小说《射雕英雄传》中的主角。他的母亲是李萍。                                                                                           
                                                                                                                                                       
  李萍是一位坚强的女性,在丈夫郭啸天死后,她独自一人在蒙古大漠中将郭靖抚养成人。虽然她不会武功,但品德高尚,教导郭靖要正直、忠诚、爱国。郭靖的许多品格 
  特质都深受母亲影响。                                                                                                                                 
                                                                                                                                                       
✻ Crunched for 2s                 

2026-5-3 17:00

"""
Anthropic 格式代理服务

接收 Anthropic API 格式的请求,转发到 OpenAI 兼容的 API 端点,
并将响应转换回 Anthropic 格式返回。

可以用于欺骗 Claude Code,让它以为在调用真正的 Anthropic API。

使用方法:
    python anthropic_proxy.py

然后在 Claude Code 中设置:
    export ANTHROPIC_BASE_URL=http://localhost:8889
    export ANTHROPIC_AUTH_TOKEN=123456
"""

from dataclasses import dataclass, field
from typing import Dict, Any, List, Optional, Generator
from flask import Flask, request, jsonify, Response
import requests
import json
import time
import uuid
import logging
from functools import wraps


@dataclass
class ProxyConfig:
    """代理配置"""
    backend_base_url: str = "https://xxx/v1"
    backend_api_key: str = "bf434796-0fa8-4f16-a3d3-cf20fffa5928"
    backend_model: str = "xxx"
    auth_token: str = "123456"
    request_timeout: int = 120


@dataclass
class AnthropicRequest:
    """Anthropic API 请求数据类"""
    messages: List[Dict[str, Any]] = field(default_factory=list)
    system: Optional[str] = None
    model: str = "claude-3-5-sonnet-20241022"
    max_tokens: int = 1024
    temperature: float = 0.7
    stream: bool = False
    tools: List[Dict[str, Any]] = field(default_factory=list)

    @classmethod
    def from_dict(cls, data: Dict[str, Any], default_model: str) -> "AnthropicRequest":
        return cls(
            messages=data.get("messages", []),
            system=data.get("system"),
            model=data.get("model", default_model),
            max_tokens=data.get("max_tokens", 1024),
            temperature=data.get("temperature", 0.7),
            stream=data.get("stream", False),
            tools=data.get("tools", [])
        )


class SSEBuilder:
    """SSE 事件构建器"""

    @staticmethod
    def escape_json_string(s: str) -> str:
        return json.dumps(s, ensure_ascii=False)[1:-1]

    @classmethod
    def message_start(cls, message_id: str, model: str) -> str:
        return f'event: message_start\ndata: {{"type":"message_start","message":{{"id":"{message_id}","type":"message","role":"assistant","content":[],"model":"{model}","stop_reason":null,"stop_sequence":null,"usage":{{"input_tokens":0,"output_tokens":0}}}}}}\n\n'

    @classmethod
    def content_block_start_text(cls, index: int) -> str:
        return f'event: content_block_start\ndata: {{"type":"content_block_start","index":{index},"content_block":{{"type":"text","text":""}}}}\n\n'

    @classmethod
    def content_block_start_tool_use(cls, index: int, tool_id: str, tool_name: str) -> str:
        escaped_name = cls.escape_json_string(tool_name)
        return f'event: content_block_start\ndata: {{"type":"content_block_start","index":{index},"content_block":{{"type":"tool_use","id":"{tool_id}","name":"{escaped_name}","input":{{}}}}}}\n\n'

    @classmethod
    def ping(cls) -> str:
        return f'event: ping\ndata: {{"type":"ping"}}\n\n'

    @classmethod
    def content_block_delta_text(cls, index: int, text: str) -> str:
        escaped = cls.escape_json_string(text)
        return f'event: content_block_delta\ndata: {{"type":"content_block_delta","index":{index},"delta":{{"type":"text_delta","text":"{escaped}"}}}}\n\n'

    @classmethod
    def content_block_delta_json(cls, index: int, partial_json: str) -> str:
        escaped = cls.escape_json_string(partial_json)
        return f'event: content_block_delta\ndata: {{"type":"content_block_delta","index":{index},"delta":{{"type":"input_json_delta","partial_json":"{escaped}"}}}}\n\n'

    @classmethod
    def content_block_stop(cls, index: int) -> str:
        return f'event: content_block_stop\ndata: {{"type":"content_block_stop","index":{index}}}\n\n'

    @classmethod
    def message_delta(cls, stop_reason: str, output_tokens: int, cache_read_tokens: int = 0) -> str:
        return f'event: message_delta\ndata: {{"type":"message_delta","delta":{{"stop_reason":"{stop_reason}","stop_sequence":null}},"usage":{{"output_tokens":{output_tokens},"cache_read_input_tokens":{cache_read_tokens}}}}}\n\n'

    @classmethod
    def message_stop(cls) -> str:
        return 'event: message_stop\ndata: {"type":"message_stop"}\n\n'

    @classmethod
    def error(cls, message: str) -> str:
        escaped = cls.escape_json_string(message)
        return f'event: error\ndata: {{"type":"error","error":{{"type":"error","message":"{escaped}"}}}}\n\n'


class AnthropicProxy:
    """Anthropic 到 OpenAI 代理"""

    STOP_REASON_MAP = {
        "stop": "end_turn",
        "length": "max_tokens",
        "tool_calls": "tool_use",
        "function_call": "tool_use",
        "content_filter": "end_turn",
    }

    def __init__(self, config: ProxyConfig):
        self.config = config
        self.logger = logging.getLogger(__name__)

    def _convert_finish_reason(self, reason: Optional[str], has_tool_calls: bool) -> str:
        if has_tool_calls:
            return "tool_use"
        return self.STOP_REASON_MAP.get(reason, "end_turn")

    def convert_tools(self, tools: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        return [{
            "type": "function",
            "function": {
                "name": t.get("name"),
                "description": t.get("description", ""),
                "parameters": t.get("input_schema", {})
            }
        } for t in tools]

    def convert_messages(
        self,
        messages: List[Dict[str, Any]],
        system: Optional[str] = None
    ) -> List[Dict[str, Any]]:
        result = []
        if system:
            result.append({"role": "system", "content": system})

        for msg in messages:
            role = msg.get("role")
            content = msg.get("content")

            if isinstance(content, list):
                result.extend(self._convert_content_blocks(content, role))
            elif isinstance(content, str):
                result.append({"role": role, "content": content})

        return result

    def _convert_content_blocks(
        self,
        blocks: List[Dict[str, Any]],
        role: str
    ) -> List[Dict[str, Any]]:
        result = []
        text_parts = []
        tool_uses = []
        tool_results = []

        for block in blocks:
            block_type = block.get("type", "")

            if block_type == "text":
                text_parts.append(block.get("text", ""))
            elif block_type == "tool_use":
                tool_uses.append({
                    "id": block.get("id"),
                    "type": "function",
                    "function": {
                        "name": block.get("name"),
                        "arguments": json.dumps(block.get("input", {}), ensure_ascii=False)
                    }
                })
            elif block_type == "tool_result":
                content = block.get("content", "")
                if isinstance(content, list):
                    content = "".join(b.get("text", "") for b in content if b.get("type") == "text")
                tool_results.append({
                    "role": "tool",
                    "content": content,
                    "tool_call_id": block.get("tool_use_id")
                })

        if role == "assistant":
            msg = {"role": "assistant"}
            if tool_uses:
                msg["content"] = "".join(text_parts) or None
                msg["tool_calls"] = tool_uses
            elif text_parts:
                msg["content"] = "".join(text_parts)
            else:
                msg["content"] = None
            result.append(msg)

        elif role == "user":
            result.extend(tool_results)
            if text_parts:
                result.append({"role": "user", "content": "".join(text_parts)})

        return result

    def convert_response(
        self,
        openai_response: Dict[str, Any],
        model: str
    ) -> Dict[str, Any]:
        choice = openai_response.get("choices", [{}])[0]
        message = choice.get("message", {})
        tool_calls = message.get("tool_calls", [])

        content = []
        if message.get("content"):
            content.append({"type": "text", "text": message["content"]})

        for tc in tool_calls:
            func = tc.get("function", {})
            try:
                args = json.loads(func.get("arguments", "{}"))
            except (json.JSONDecodeError, TypeError):
                args = {}
            content.append({
                "type": "tool_use",
                "id": tc.get("id"),
                "name": func.get("name"),
                "input": args
            })

        if not content:
            content = [{"type": "text", "text": ""}]

        usage = openai_response.get("usage", {})
        return {
            "id": f"msg_{openai_response.get('id', str(uuid.uuid4()))}",
            "type": "message",
            "role": "assistant",
            "content": content,
            "model": model,
            "stop_reason": self._convert_finish_reason(choice.get("finish_reason"), bool(tool_calls)),
            "stop_sequence": None,
            "usage": {
                "input_tokens": usage.get("prompt_tokens", 0),
                "output_tokens": usage.get("completion_tokens", 0)
            }
        }

    def build_backend_payload(self, req: AnthropicRequest) -> Dict[str, Any]:
        messages = self.convert_messages(req.messages, req.system)
        payload = {
            "model": self.config.backend_model,
            "messages": messages,
            "max_tokens": req.max_tokens,
            "temperature": req.temperature,
            "stream": req.stream
        }
        if req.tools:
            payload["tools"] = self.convert_tools(req.tools)
        return payload

    def stream(self, payload: Dict[str, str], model: str) -> Response:
        def generate() -> Generator[str, None, None]:
            message_id = f"msg_{str(uuid.uuid4())}"
            output_tokens = 0
            tool_info: Dict[int, Dict[str, str]] = {}  # index -> {id, name}
            has_tool = False
            max_tool_index = 0

            yield SSEBuilder.message_start(message_id, model)
            yield SSEBuilder.content_block_start_text(0)
            yield SSEBuilder.ping()
            yield SSEBuilder.content_block_delta_text(0, "")

            try:
                response = requests.post(
                    f"{self.config.backend_base_url}/chat/completions",
                    headers={
                        "Authorization": f"Bearer {self.config.backend_api_key}",
                        "Content-Type": "application/json"
                    },
                    json=payload,
                    timeout=self.config.request_timeout,
                    stream=True
                )

                for line in response.iter_lines():
                    if not line:
                        continue

                    line_str = line.decode('utf-8')
                    if not line_str.startswith('data: '):
                        continue

                    data_str = line_str[6:]
                    if data_str.strip() == '[DONE]':
                        break

                    try:
                        chunk = json.loads(data_str)
                        choices = chunk.get("choices", [])
                        if not choices:
                            continue
                        choice = choices[0]
                        delta = choice.get("delta", {})
                        finish_reason = choice.get("finish_reason")

                        if delta.get("content"):
                            text = delta["content"]
                            yield SSEBuilder.content_block_delta_text(0, text)
                            output_tokens += len(text) // 4

                        for tc in delta.get("tool_calls", []):
                            tc_index = tc.get("index", 0)
                            func = tc.get("function", {})

                            if tc_index not in tool_info:
                                has_tool = True
                                tool_info[tc_index] = {
                                    "id": tc.get("id", f"tool_{tc_index + 1}"),
                                    "name": func.get("name", ""),
                                    "arguments": ""
                                }
                                max_tool_index = max(max_tool_index, tc_index)

                                yield SSEBuilder.content_block_start_tool_use(
                                    tc_index + 1,
                                    tool_info[tc_index]["id"],
                                    tool_info[tc_index]["name"]
                                )

                            if func.get("arguments"):
                                tool_info[tc_index]["arguments"] += func["arguments"]
                                yield SSEBuilder.content_block_delta_json(
                                    tc_index + 1, func["arguments"]
                                )

                        if finish_reason:
                            stop_reason = self._convert_finish_reason(finish_reason, has_tool)
                            yield SSEBuilder.content_block_stop(0)
                            for i in range(max_tool_index + 1):
                                yield SSEBuilder.content_block_stop(i + 1)
                            yield SSEBuilder.message_delta(stop_reason, output_tokens)
                            yield SSEBuilder.message_stop()

                    except json.JSONDecodeError:
                        continue

            except Exception as e:
                self.logger.error(f"Stream error: {e}")
                yield SSEBuilder.error(str(e))

        return Response(generate(), mimetype='text/event-stream')


def create_app(config: ProxyConfig = None) -> Flask:
    """创建 Flask 应用"""
    if config is None:
        config = ProxyConfig()

    proxy = AnthropicProxy(config)

    logging.basicConfig(
        level=logging.DEBUG,
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
        datefmt='%Y-%m-%d %H:%M:%S'
    )
    logging.getLogger('urllib3').setLevel(logging.WARNING)
    logging.getLogger('requests').setLevel(logging.WARNING)
    logging.getLogger('werkzeug').setLevel(logging.WARNING)

    app = Flask(__name__)

    def require_auth(f):
        @wraps(f)
        def decorated(*args, **kwargs):
            token = request.headers.get("x-api-key") or \
                    request.headers.get("Authorization", "").replace("Bearer ", "")
            if token != config.auth_token:
                return jsonify({
                    "detail": "Invalid API key. Please provide a valid Anthropic API key."
                }), 401
            return f(*args, **kwargs)
        return decorated

    @app.route("/v1/messages", methods=["POST"])
    @app.route("/messages", methods=["POST"])
    @require_auth
    def messages():
        start_time = time.time()

        try:
            data = request.get_json()
        except Exception as e:
            logging.error(f"Invalid JSON: {e}")
            return jsonify({"error": f"Invalid JSON: {str(e)}"}), 400

        logging.info(f"=== Request ===")
        logging.info(f"Method: {request.method}")
        logging.info(f"Headers: {dict(request.headers)}")
        logging.info(f"Model: {data.get('model')}")
        logging.info(f"System: {data.get('system', 'None')[:100] if data.get('system') else 'None'}")
        logging.info(f"Messages: {len(data.get('messages', []))}")
        logging.info(f"Stream: {data.get('stream', False)}")
        logging.info(f"Tools: {len(data.get('tools', []))}")
        logging.info(f"Request body: {json.dumps(data, ensure_ascii=False)[:500]}")
        logging.info(f"=============")

        req = AnthropicRequest.from_dict(data, config.backend_model)
        payload = proxy.build_backend_payload(req)

        if req.stream:
            logging.info(f"Streaming request started")
            return proxy.stream(payload, req.model)

        try:
            logging.info(f"Sending to backend: {config.backend_base_url}/chat/completions")
            logging.debug(f"Backend payload: {json.dumps(payload, ensure_ascii=False)[:1000]}")
            response = requests.post(
                f"{config.backend_base_url}/chat/completions",
                headers={
                    "Authorization": f"Bearer {config.backend_api_key}",
                    "Content-Type": "application/json"
                },
                json=payload,
                timeout=config.request_timeout
            )
            logging.info(f"Backend response status: {response.status_code}")
            logging.debug(f"Backend response body: {response.text[:500]}")
            response.raise_for_status()
            result = proxy.convert_response(response.json(), req.model)
            logging.info(f"Converted response: {json.dumps(result, ensure_ascii=False)[:300]}")

            elapsed = time.time() - start_time
            logging.info(f"Request completed in {elapsed:.2f}s")

            return jsonify(result)

        except requests.exceptions.RequestException as e:
            logging.error(f"Backend request failed: {e}")
            error_msg = str(e)
            if hasattr(e, 'response') and e.response:
                try:
                    error_msg = e.response.json()
                except:
                    error_msg = e.response.text
            return jsonify({"error": error_msg}), 502

    @app.route("/v1/models", methods=["GET"])
    def models():
        return jsonify({
            "object": "list",
            "data": [{
                "id": config.backend_model,
                "object": "model",
                "created": 1704067200,
                "owned_by": "zzz"
            }]
        })

    @app.route("/health", methods=["GET"])
    def health():
        return jsonify({"status": "healthy"})

    return app


if __name__ == "__main__":
    print("=" * 60)
    print("Anthropic Proxy Server")
    print("=" * 60)
    config = ProxyConfig()
    print(f"Backend: {config.backend_base_url}")
    print(f"Model: {config.backend_model}")
    print(f"Auth Token: {config.auth_token}")
    print("=" * 60)
    print("Endpoints:")
    print("  POST /v1/messages  - Chat completion")
    print("  GET  /v1/models    - List models")
    print("  GET  /health       - Health check")
    print("=" * 60)
    print("\nUsage:")
    print("  export ANTHROPIC_BASE_URL=http://localhost:8889")
    print("  export ANTHROPIC_AUTH_TOKEN=123456")
    print("=" * 60)

    app = create_app()
    app.run(host="0.0.0.0", port=8889, debug=False)

Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐