本文面向有 API 开发经验的工程师,覆盖环境配置、鉴权、基础调用、流式响应、多轮对话、错误处理全流程,代码示例可直接运行。


一、国内直连 Claude API 的核心问题

直接访问 api.anthropic.com 在国内会被拦截,稳定性极差。工程实践中有两种解法:

  1. 代理转发:本地或服务器挂代理,延迟高、不稳定、维护成本高
  2. 使用国内直连接入点:将 base_url 指向国内可访问的中转节点,API 格式与官方完全一致,代码零改动

本文所有示例均使用第二种方案。核心配置只有两个参数:

ANTHROPIC_BASE_URL=https://gw.claudeapi.com
ANTHROPIC_API_KEY=sk-你的Key

国内直连接入点可在 claudeapi.com 获取 API Key,支持支付宝 / 微信充值,无需境外信用卡。


二、模型选择

当前可用模型及适用场景:

模型 ID 定位 输入价格 输出价格
claude-opus-4-7 复杂推理、长上下文、高质量生成 $4.0/1M $20.0/1M
claude-opus-4-6 复杂推理、专业任务 $4.0/1M $20.0/1M
claude-sonnet-4-6 通用开发、Agent、日常任务(推荐 $2.4/1M $12.0/1M
claude-haiku-4-5-20251001 轻量请求、快速响应 $0.8/1M $4.0/1M

日常开发优先选 claude-sonnet-4-6,性价比最高。


三、环境配置

Python

pip install anthropic

推荐将密钥写入 .env 文件,不要硬编码在代码里:

# .env
ANTHROPIC_API_KEY=sk-你的Key
ANTHROPIC_BASE_URL=https://gw.claudeapi.com
# 加载 .env(需安装 python-dotenv)
from dotenv import load_dotenv
load_dotenv()

Node.js

npm install @anthropic-ai/sdk dotenv
# .env
ANTHROPIC_API_KEY=sk-你的Key
ANTHROPIC_BASE_URL=https://gw.claudeapi.com

Go

go get github.com/anthropics/anthropic-sdk-go

四、基础调用

Python

import os
import anthropic

client = anthropic.Anthropic(
    api_key=os.environ.get("ANTHROPIC_API_KEY"),
    base_url=os.environ.get("ANTHROPIC_BASE_URL", "https://gw.claudeapi.com"),
)

message = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[
        {"role": "user", "content": "用 Python 写一个快速排序,加注释"}
    ]
)

print(message.content[0].text)

Node.js

import Anthropic from "@anthropic-ai/sdk";
import "dotenv/config";

const client = new Anthropic({
  apiKey: process.env.ANTHROPIC_API_KEY,
  baseURL: process.env.ANTHROPIC_BASE_URL ?? "https://gw.claudeapi.com",
});

const message = await client.messages.create({
  model: "claude-sonnet-4-6",
  max_tokens: 1024,
  messages: [
    { role: "user", content: "用 Python 写一个快速排序,加注释" }
  ],
});

console.log(message.content[0].text);

Go

package main

import (
    "context"
    "fmt"
    "os"

    anthropic "github.com/anthropics/anthropic-sdk-go"
    "github.com/anthropics/anthropic-sdk-go/option"
)

func main() {
    client := anthropic.NewClient(
        option.WithAPIKey(os.Getenv("ANTHROPIC_API_KEY")),
        option.WithBaseURL(os.Getenv("ANTHROPIC_BASE_URL")),
    )

    message, err := client.Messages.New(context.Background(),
        anthropic.MessageNewParams{
            Model:     anthropic.F(anthropic.ModelClaudeSonnet46),
            MaxTokens: anthropic.F(int64(1024)),
            Messages: anthropic.F([]anthropic.MessageParam{
                anthropic.UserMessageParam(anthropic.NewTextBlock(
                    "用 Python 写一个快速排序,加注释",
                )),
            }),
        },
    )
    if err != nil {
        panic(err)
    }

    fmt.Println(message.Content[0].Text)
}

五、流式响应(Streaming)

面向用户的产品场景几乎都需要流式输出,避免用户等待。

Python

import os
import anthropic

client = anthropic.Anthropic(
    api_key=os.environ.get("ANTHROPIC_API_KEY"),
    base_url="https://gw.claudeapi.com",
)

with client.messages.stream(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[{"role": "user", "content": "详细解释一下 TCP 三次握手"}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

print()  # 换行

Node.js

import Anthropic from "@anthropic-ai/sdk";

const client = new Anthropic({
  apiKey: process.env.ANTHROPIC_API_KEY,
  baseURL: "https://gw.claudeapi.com",
});

const stream = await client.messages.stream({
  model: "claude-sonnet-4-6",
  max_tokens: 1024,
  messages: [{ role: "user", content: "详细解释一下 TCP 三次握手" }],
});

for await (const chunk of stream) {
  if (
    chunk.type === "content_block_delta" &&
    chunk.delta.type === "text_delta"
  ) {
    process.stdout.write(chunk.delta.text);
  }
}

六、多轮对话

Claude API 是无状态的,多轮对话需要在客户端维护 messages 数组,每次请求带上完整历史。

Python 封装示例

import os
import anthropic

class ClaudeChat:
    def __init__(self, model="claude-sonnet-4-6", system=None):
        self.client = anthropic.Anthropic(
            api_key=os.environ.get("ANTHROPIC_API_KEY"),
            base_url="https://gw.claudeapi.com",
        )
        self.model = model
        self.system = system
        self.history = []

    def chat(self, user_input: str) -> str:
        self.history.append({"role": "user", "content": user_input})

        kwargs = {
            "model": self.model,
            "max_tokens": 2048,
            "messages": self.history,
        }
        if self.system:
            kwargs["system"] = self.system

        response = self.client.messages.create(**kwargs)
        assistant_reply = response.content[0].text

        self.history.append({"role": "assistant", "content": assistant_reply})
        return assistant_reply

    def reset(self):
        self.history = []


# 使用示例
chat = ClaudeChat(system="你是一个资深 Python 工程师,回答简洁直接。")

print(chat.chat("什么是装饰器?"))
print(chat.chat("给我一个实际项目中常用的例子"))
print(chat.chat("如何给装饰器传参数?"))

七、带 System Prompt 的工程集成

实际项目中,通常需要通过 System Prompt 定义模型角色和行为边界。

import os
import anthropic

client = anthropic.Anthropic(
    api_key=os.environ.get("ANTHROPIC_API_KEY"),
    base_url="https://gw.claudeapi.com",
)

SYSTEM_PROMPT = """
你是一个代码审查助手。
- 只关注代码质量、安全性和性能问题
- 以 Markdown 格式输出,分「问题」和「建议」两部分
- 不需要夸赞代码,直接指出问题
- 若代码无明显问题,说「未发现明显问题」即可
"""

def review_code(code: str, language: str = "python") -> str:
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=2048,
        system=SYSTEM_PROMPT,
        messages=[
            {
                "role": "user",
                "content": f"请审查以下 {language} 代码:\n\n```{language}\n{code}\n```"
            }
        ],
    )
    return response.content[0].text


# 示例
code = """
def get_user(user_id):
    query = f"SELECT * FROM users WHERE id = {user_id}"
    return db.execute(query)
"""

print(review_code(code))

八、错误处理

生产环境必须处理以下几类错误:

import os
import time
import anthropic
from anthropic import RateLimitError, APIStatusError, APIConnectionError

client = anthropic.Anthropic(
    api_key=os.environ.get("ANTHROPIC_API_KEY"),
    base_url="https://gw.claudeapi.com",
)

def call_with_retry(messages, max_retries=3, model="claude-sonnet-4-6"):
    for attempt in range(max_retries):
        try:
            response = client.messages.create(
                model=model,
                max_tokens=1024,
                messages=messages,
            )
            return response.content[0].text

        except RateLimitError:
            # 429:请求过快,等待后重试
            wait = 2 ** attempt
            print(f"触发限流,{wait}s 后重试(第 {attempt + 1} 次)")
            time.sleep(wait)

        except APIStatusError as e:
            if e.status_code == 401:
                raise ValueError("API Key 无效,请检查配置") from e
            elif e.status_code == 529:
                # 529:服务过载,短暂等待
                time.sleep(5)
            else:
                raise

        except APIConnectionError:
            # 网络问题,重试
            if attempt == max_retries - 1:
                raise
            time.sleep(2)

    raise RuntimeError(f"重试 {max_retries} 次后仍失败")

常见错误码速查:

错误码 原因 处理方式
401 API Key 无效或过期 检查 Key 是否正确,是否有余额
429 请求频率超限 指数退避重试
529 服务端过载 等待 5-10s 后重试
网络超时 连接问题 检查 base_url 配置,重试

九、工程最佳实践

1. 密钥管理

# ❌ 不要这样做
client = anthropic.Anthropic(api_key="sk-xxxxxxxx")

# ✅ 从环境变量读取
client = anthropic.Anthropic(
    api_key=os.environ["ANTHROPIC_API_KEY"],
    base_url=os.environ.get("ANTHROPIC_BASE_URL", "https://gw.claudeapi.com"),
)

2. 超时设置

client = anthropic.Anthropic(
    api_key=os.environ["ANTHROPIC_API_KEY"],
    base_url="https://gw.claudeapi.com",
    timeout=60.0,        # 总超时 60s
    max_retries=2,       # SDK 自动重试次数
)

3. Token 用量监控

response = client.messages.create(...)

# 响应中包含用量信息
print(f"输入 Token:{response.usage.input_tokens}")
print(f"输出 Token:{response.usage.output_tokens}")

4. 异步调用(高并发场景)

import asyncio
import anthropic

async def async_call(prompt: str) -> str:
    client = anthropic.AsyncAnthropic(
        api_key=os.environ["ANTHROPIC_API_KEY"],
        base_url="https://gw.claudeapi.com",
    )
    response = await client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}],
    )
    return response.content[0].text

# 并发调用示例
async def main():
    prompts = ["解释递归", "解释闭包", "解释生成器"]
    results = await asyncio.gather(*[async_call(p) for p in prompts])
    for r in results:
        print(r[:100])

asyncio.run(main())

十、完整可运行 Demo

将以下内容保存为 demo.py,配置好 .env 后直接运行:

import os
import anthropic
from dotenv import load_dotenv

load_dotenv()

def main():
    client = anthropic.Anthropic(
        api_key=os.environ["ANTHROPIC_API_KEY"],
        base_url=os.environ.get("ANTHROPIC_BASE_URL", "https://gw.claudeapi.com"),
    )

    print("=== 基础调用 ===")
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=256,
        messages=[{"role": "user", "content": "用一句话介绍你自己"}],
    )
    print(response.content[0].text)

    print("\n=== 流式输出 ===")
    with client.messages.stream(
        model="claude-sonnet-4-6",
        max_tokens=512,
        messages=[{"role": "user", "content": "列出 5 个 Python 最佳实践"}],
    ) as stream:
        for text in stream.text_stream:
            print(text, end="", flush=True)
    print()

    print("\n=== Token 用量 ===")
    print(f"输入:{response.usage.input_tokens} tokens")
    print(f"输出:{response.usage.output_tokens} tokens")

if __name__ == "__main__":
    main()

总结

步骤 要点
接入点 base_url 设为 https://gw.claudeapi.com,国内直连,无需代理
模型选择 日常开发用 claude-sonnet-4-6,复杂任务用 claude-opus-4-6/4-7
多轮对话 客户端维护 messages 数组,每次请求带完整历史
错误处理 必须处理 401 / 429 / 529,生产环境加重试逻辑
密钥安全 从环境变量读取,不要硬编码

Logo

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。

更多推荐