如何用Python调用Claude API 初学者完整代码示例

一、环境准备

Node.js 版本要求

# 检查 Node.js 版本（需要 18.0.0 以上）
node --version

# 如果版本太低，到 nodejs.org 下载最新 LTS 版本
# 推荐使用 nvm 管理多版本：
# Mac/Linux: curl -o- https://raw.githubusercontent.com/nvm-sh/nvm/v0.39.0/install.sh | bash
# 然后: nvm install --lts && nvm use --lts

创建项目并安装 SDK

# 创建项目目录
mkdir claude-demo && cd claude-demo

# 初始化项目
npm init -y

# 安装 Anthropic SDK
npm install @anthropic-ai/sdk

# 如果要用 TypeScript（推荐）
npm install -D typescript @types/node ts-node
npx tsc --init

# 验证安装
node -e "const a = require('@anthropic-ai/sdk'); console.log('SDK 版本：', require('./node_modules/@anthropic-ai/sdk/package.json').version)"

设置 API Key

# 方式一：环境变量（推荐）
export ANTHROPIC_API_KEY="sk-ant-api03-你的key"  # Mac/Linux
$env:ANTHROPIC_API_KEY = "sk-ant-api03-你的key"  # Windows PowerShell

# 方式二：.env 文件（项目开发推荐）
npm install dotenv

# 创建 .env 文件
echo "ANTHROPIC_API_KEY=sk-ant-api03-你的key" > .env
echo ".env" >> .gitignore   # 防止泄漏到 git

二、基础调用

JavaScript 版本

// basic.js
require("dotenv").config();
const Anthropic = require("@anthropic-ai/sdk");

const client = new Anthropic();   // 自动读取 ANTHROPIC_API_KEY 环境变量

async function main() {
  const message = await client.messages.create({
    model:      "claude-haiku-4-5-20251001",  // 练习用最便宜的模型
    max_tokens: 256,
    messages: [
      {
        role:    "user",
        content: "你好！用一句话介绍你自己。"
      }
    ]
  });

  // 提取回复文本
  console.log(message.content[0].text);

  // 查看 token 用量
  console.log(`\n输入：${message.usage.input_tokens} tokens`);
  console.log(`输出：${message.usage.output_tokens} tokens`);
  console.log(`停止原因：${message.stop_reason}`);
}

main().catch(console.error);

node basic.js

TypeScript 版本

// basic.ts
import Anthropic from "@anthropic-ai/sdk";

const client = new Anthropic({
  apiKey: process.env.ANTHROPIC_API_KEY,  // 显式传入（可选，不传也会自动读取）
});

async function main(): Promise {
  const message = await client.messages.create({
    model:      "claude-haiku-4-5-20251001",
    max_tokens: 256,
    messages: [
      {
        role:    "user",
        content: "你好！用一句话介绍你自己。"
      }
    ]
  });

  // TypeScript 类型安全的方式提取文本
  const text = message.content
    .filter(block => block.type === "text")
    .map(block => block.text)
    .join("");

  console.log(text);
  console.log(`\n输入：${message.usage.input_tokens} tokens`);
  console.log(`输出：${message.usage.output_tokens} tokens`);
}

main().catch(console.error);

npx ts-node basic.ts

关键参数说明

三、设置角色（System Prompt）

// system.js
const Anthropic = require("@anthropic-ai/sdk");
const client = new Anthropic();

async function askExpert(question) {
  const response = await client.messages.create({
    model:      "claude-haiku-4-5-20251001",
    max_tokens: 512,
    system:     "你是一名资深的前端工程师，擅长 React 和 TypeScript。" +
                "回答时给出具体的代码示例，使用中文，代码用英文注释。",
    messages: [
      { role: "user", content: question }
    ]
  });

  return response.content[0].text;
}

(async () => {
  const answer = await askExpert("如何用 React hooks 实现一个防抖函数？");
  console.log(answer);
})();

四、流式输出

流式输出让用户看到逐字生成的效果，而不是等待完整响应。Node.js SDK 有两种流式方式：

方式一：.stream() 方法（推荐，更简洁）

// stream.js
const Anthropic = require("@anthropic-ai/sdk");
const client = new Anthropic();

async function streamChat(userMessage) {
  process.stdout.write("Claude：");

  // .stream() 返回一个流对象
  const stream = client.messages.stream({
    model:      "claude-sonnet-4-6",
    max_tokens: 1024,
    messages:   [{ role: "user", content: userMessage }],
  });

  // 逐块输出文本
  stream.on("text", (text) => {
    process.stdout.write(text);
  });

  // 等待流结束，获取完整统计
  const finalMessage = await stream.finalMessage();
  console.log();  // 换行
  console.log(`\n输入：${finalMessage.usage.input_tokens} tokens`);
  console.log(`输出：${finalMessage.usage.output_tokens} tokens`);
  console.log(`停止原因：${finalMessage.stop_reason}`);

  return finalMessage;
}

streamChat("用 100 字介绍一下 Node.js 的事件循环机制");

方式二：for await…of 迭代器（适合需要处理每个事件的场景）

// stream_events.js
const Anthropic = require("@anthropic-ai/sdk");
const client = new Anthropic();

async function streamWithEvents(userMessage) {
  const stream = await client.messages.stream({
    model:      "claude-sonnet-4-6",
    max_tokens: 1024,
    messages:   [{ role: "user", content: userMessage }],
  });

  // 迭代所有流事件
  for await (const event of stream) {
    switch (event.type) {
      case "message_start":
        // 请求开始
        break;

      case "content_block_delta":
        // 文本增量
        if (event.delta.type === "text_delta") {
          process.stdout.write(event.delta.text);
        }
        break;

      case "message_delta":
        // 消息结束时的 stop_reason 和用量
        if (event.usage) {
          console.log(`\n输出 tokens：${event.usage.output_tokens}`);
        }
        break;

      case "message_stop":
        console.log("\n--- 流结束 ---");
        break;
    }
  }
}

streamWithEvents("列举 JavaScript 的5个新特性");

TypeScript 类型安全的流式输出

// stream.ts
import Anthropic from "@anthropic-ai/sdk";

const client = new Anthropic();

async function streamChat(userMessage: string): Promise {
  let fullText = "";

  process.stdout.write("Claude：");

  const stream = client.messages.stream({
    model:      "claude-sonnet-4-6",
    max_tokens: 1024,
    messages:   [{ role: "user", content: userMessage }],
  });

  // TypeScript 自动推断类型
  for await (const chunk of stream) {
    if (
      chunk.type === "content_block_delta" &&
      chunk.delta.type === "text_delta"
    ) {
      process.stdout.write(chunk.delta.text);
      fullText += chunk.delta.text;
    }
  }

  const finalMsg = await stream.finalMessage();
  console.log(`\n\n共 ${finalMsg.usage.output_tokens} output tokens`);

  return fullText;
}

streamChat("简单介绍 TypeScript 的泛型").catch(console.error);

五、多轮对话

// conversation.js
const Anthropic = require("@anthropic-ai/sdk");
const readline  = require("readline");

const client = new Anthropic();

// 对话历史（每次调用都传完整历史）
const history = [];

async function chat(userInput) {
  // 加入用户消息
  history.push({ role: "user", content: userInput });

  // 控制历史长度（最多保留20条，防止 token 超限）
  const recentHistory = history.slice(-20);

  const response = await client.messages.create({
    model:      "claude-sonnet-4-6",
    max_tokens: 1024,
    system:     "你是一个友善的 AI 助手，记住整个对话的上下文。",
    messages:   recentHistory,
  });

  const reply = response.content
    .filter(b => b.type === "text")
    .map(b => b.text)
    .join("");

  // 加入 AI 回复
  history.push({ role: "assistant", content: reply });

  return reply;
}

// 交互式命令行聊天
async function startChat() {
  const rl = readline.createInterface({
    input:  process.stdin,
    output: process.stdout,
  });

  console.log("开始聊天（输入 exit 退出）\n");

  const ask = () => {
    rl.question("你：", async (input) => {
      if (input.trim() === "exit") {
        console.log("再见！");
        rl.close();
        return;
      }

      try {
        const reply = await chat(input);
        console.log(`\nClaude：${reply}\n`);
      } catch (err) {
        console.error("出错了：", err.message);
      }

      ask();   // 继续下一轮
    });
  };

  ask();
}

startChat();

六、Tool Use（工具调用）

// tools.js
const Anthropic = require("@anthropic-ai/sdk");
const client = new Anthropic();

// 定义工具
const tools = [
  {
    name:        "get_weather",
    description: "获取指定城市的当前天气",
    input_schema: {
      type:       "object",
      properties: {
        city: {
          type:        "string",
          description: "城市名称，如 '北京'、'上海'"
        }
      },
      required: ["city"]
    }
  },
  {
    name:        "calculate",
    description: "执行数学计算",
    input_schema: {
      type:       "object",
      properties: {
        expression: {
          type:        "string",
          description: "数学表达式，如 '(100 + 200) * 1.08'"
        }
      },
      required: ["expression"]
    }
  }
];

// 工具执行函数
function executeTool(name, input) {
  switch (name) {
    case "get_weather":
      // 实际项目替换为真实 API
      return JSON.stringify({
        city:        input.city,
        temperature: 22,
        condition:   "晴",
        humidity:    "45%"
      });

    case "calculate":
      try {
        const result = Function(`"use strict"; return (${input.expression})`)();
        return String(result);
      } catch {
        return "计算出错，请检查表达式格式";
      }

    default:
      return JSON.stringify({ error: `未知工具：${name}` });
  }
}

// 带工具调用的完整对话循环
async function chatWithTools(userMessage) {
  const messages = [{ role: "user", content: userMessage }];

  while (true) {
    const response = await client.messages.create({
      model:      "claude-sonnet-4-6",
      max_tokens: 1024,
      tools,
      messages,
    });

    console.log(`stop_reason: ${response.stop_reason}`);

    // 自然结束
    if (response.stop_reason === "end_turn") {
      const finalText = response.content
        .filter(b => b.type === "text")
        .map(b => b.text)
        .join("");
      return finalText;
    }

    // 需要调用工具
    if (response.stop_reason === "tool_use") {
      // 把 AI 的回复加入消息历史
      messages.push({ role: "assistant", content: response.content });

      // 执行所有工具调用
      const toolResults = [];
      for (const block of response.content) {
        if (block.type === "tool_use") {
          console.log(`调用工具：${block.name}(${JSON.stringify(block.input)})`);

          const result = executeTool(block.name, block.input);
          console.log(`工具结果：${result}`);

          toolResults.push({
            type:        "tool_result",
            tool_use_id: block.id,
            content:     result,
          });
        }
      }

      // 把工具结果放入 user 消息发回给 Claude
      messages.push({ role: "user", content: toolResults });
      // 继续循环，让 Claude 基于工具结果生成最终回复
    }
  }
}

// 测试
(async () => {
  const result = await chatWithTools(
    "北京今天天气怎么样？顺便帮我算一下 (1200 + 800) * 0.85 等于多少"
  );
  console.log("\n最终回复：\n", result);
})();

七、生产级封装

// claude-client.js
// 生产级封装：含重试、错误处理、日志

require("dotenv").config();
const Anthropic = require("@anthropic-ai/sdk");

class ClaudeClient {
  constructor({
    apiKey      = process.env.ANTHROPIC_API_KEY,
    defaultModel= "claude-sonnet-4-6",
    maxRetries  = 3,
    timeout     = 60_000,
  } = {}) {
    if (!apiKey) throw new Error("ANTHROPIC_API_KEY 未设置");

    this.client = new Anthropic({
      apiKey,
      maxRetries: 0,   // 关闭 SDK 内置重试，用我们自己的
      timeout,
    });
    this.defaultModel = defaultModel;
    this.maxRetries   = maxRetries;
  }

  // 判断错误是否可以重试
  _isRetryable(error) {
    if (error instanceof Anthropic.RateLimitError)      return true;
    if (error instanceof Anthropic.InternalServerError) return true;
    if (error instanceof Anthropic.APIConnectionError)  return true;
    if (error instanceof Anthropic.APIConnectionTimeoutError) return true;
    if (error instanceof Anthropic.APIStatusError) {
      return [500, 502, 503, 529].includes(error.status);
    }
    return false;
  }

  // 指数退避
  _backoff(attempt) {
    const base  = Math.min(1000 * Math.pow(2, attempt), 30_000);
    const jitter= Math.random() * 0.4 + 0.8;  // 0.8–1.2 倍随机
    return Math.floor(base * jitter);
  }

  // 带重试的调用
  async complete({
    messages,
    system     = "",
    model      = this.defaultModel,
    maxTokens  = 1024,
    temperature,
  }) {
    let lastError;

    for (let attempt = 0; attempt <= this.maxRetries; attempt++) {
      try {
        const params = {
          model,
          max_tokens: maxTokens,
          messages,
        };
        if (system)      params.system      = system;
        if (temperature !== undefined) params.temperature = temperature;

        const response = await this.client.messages.create(params);

        return {
          text:    response.content.filter(b => b.type === "text").map(b => b.text).join(""),
          usage:   response.usage,
          model:   response.model,
          stop:    response.stop_reason,
        };

      } catch (error) {
        lastError = error;

        // 不可重试的错误直接抛出
        if (!this._isRetryable(error)) throw error;

        if (attempt < this.maxRetries) {
          // 优先读 Retry-After 响应头
          const retryAfter = error.headers?.["retry-after"];
          const wait = retryAfter
            ? parseFloat(retryAfter) * 1000
            : this._backoff(attempt);

          console.warn(`[Claude] 第 ${attempt + 1} 次失败，${(wait/1000).toFixed(1)}s 后重试...`);
          await new Promise(r => setTimeout(r, wait));
        }
      }
    }

    throw lastError;
  }

  // 流式输出（带重试）
  async *stream({
    messages,
    system    = "",
    model     = this.defaultModel,
    maxTokens = 1024,
  }) {
    for (let attempt = 0; attempt <= this.maxRetries; attempt++) {
      try {
        const params = { model, max_tokens: maxTokens, messages };
        if (system) params.system = system;

        const s = this.client.messages.stream(params);

        for await (const chunk of s) {
          if (
            chunk.type === "content_block_delta" &&
            chunk.delta.type === "text_delta"
          ) {
            yield chunk.delta.text;
          }
        }
        return;   // 成功，退出重试循环

      } catch (error) {
        if (!this._isRetryable(error) || attempt >= this.maxRetries) throw error;
        const wait = this._backoff(attempt);
        console.warn(`[Claude] 流失败，${(wait/1000).toFixed(1)}s 后重试...`);
        await new Promise(r => setTimeout(r, wait));
      }
    }
  }
}

module.exports = { ClaudeClient };


// ── 使用示例 ──────────────────────────────────────

(async () => {
  const claude = new ClaudeClient({
    defaultModel: "claude-sonnet-4-6",
    maxRetries:   3,
  });

  // 普通调用
  const result = await claude.complete({
    system:   "你是一个简洁的助手，回答不超过100字。",
    messages: [{ role: "user", content: "什么是闭包？" }],
    maxTokens: 256,
  });
  console.log(result.text);

  // 流式调用
  process.stdout.write("\n流式输出：");
  for await (const chunk of claude.stream({
    messages: [{ role: "user", content: "数到5" }],
    maxTokens: 64,
  })) {
    process.stdout.write(chunk);
  }
  console.log();
})();

八、Express 集成示例

// server.js
require("dotenv").config();
const express    = require("express");
const Anthropic  = require("@anthropic-ai/sdk");

const app    = express();
const client = new Anthropic();

app.use(express.json());

// 普通接口
app.post("/api/chat", async (req, res) => {
  const { message, system } = req.body;
  if (!message) return res.status(400).json({ error: "message 不能为空" });

  try {
    const response = await client.messages.create({
      model:      "claude-sonnet-4-6",
      max_tokens: 1024,
      system:     system || "你是一个专业的 AI 助手。",
      messages:   [{ role: "user", content: message }],
    });
    res.json({
      content:      response.content[0].text,
      inputTokens:  response.usage.input_tokens,
      outputTokens: response.usage.output_tokens,
    });
  } catch (err) {
    res.status(500).json({ error: err.message });
  }
});

// 流式接口（SSE）
app.post("/api/chat/stream", async (req, res) => {
  const { message } = req.body;
  if (!message) return res.status(400).json({ error: "message 不能为空" });

  // 设置 SSE 响应头
  res.setHeader("Content-Type",      "text/event-stream");
  res.setHeader("Cache-Control",     "no-cache");
  res.setHeader("X-Accel-Buffering", "no");    // 关闭 Nginx 缓冲
  res.flushHeaders();

  let closed = false;
  req.on("close", () => { closed = true; });

  try {
    const stream = client.messages.stream({
      model:      "claude-sonnet-4-6",
      max_tokens: 2048,
      messages:   [{ role: "user", content: message }],
    });

    for await (const chunk of stream) {
      if (closed) break;
      if (
        chunk.type === "content_block_delta" &&
        chunk.delta.type === "text_delta"
      ) {
        res.write(`data: ${JSON.stringify({ text: chunk.delta.text })}\n\n`);
      }
    }

    res.write(`data: ${JSON.stringify({ done: true })}\n\n`);
    res.end();

  } catch (err) {
    if (!closed) {
      res.write(`data: ${JSON.stringify({ error: err.message })}\n\n`);
      res.end();
    }
  }
});

app.listen(3000, () => console.log("服务启动：http://localhost:3000"));

npm install express
node server.js

# 测试普通接口
curl -X POST http://localhost:3000/api/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "你好"}'

# 测试流式接口
curl -X POST http://localhost:3000/api/chat/stream \
  -H "Content-Type: application/json" \
  -d '{"message": "数到5"}'

常见问题

Q：用 ES Module（import/export）还是 CommonJS（require）？
两种都支持。CommonJS（require）在 Node.js 里直接用，不需要额外配置；ES Module（import）需要在 package.json 里加 "type": "module" 或把文件改为 .mjs。TypeScript 项目推荐 ES Module 风格，tsconfig.json 设 "module": "commonjs" 编译后可兼容两种。

Q：流式输出的回调和 async iterator 哪个更好用？
简单场景用事件回调（stream.on("text", ...)），需要更精细控制每个事件类型时用 for await...of。本文两种都给出了示例，按需选择。如果接入 Express 或 Fastify 做 SSE 接口，for await...of 写法更清晰。

Q：多轮对话的历史记录存在内存里，服务重启就丢失了怎么办？
生产环境用 Redis 或数据库存储历史，session_id 作为 Key，每次请求时从存储里读取并传给 Claude API。本文的内存方案适合单用户 CLI 工具和快速原型，多用户 Web 应用需要持久化存储。

总结

Node.js 调用 Claude API 的核心要点：max_tokens 必须传（没有默认值）；system 是独立参数不在 messages 里；流式输出用 .stream() 配合 for await...of；多轮对话需要自己维护消息历史数组并完整传入每次请求；工具调用完成后要把结果放在 user 角色的消息里传回。生产环境在初始化 Anthropic 客户端时关闭内置重试、用自己的指数退避逻辑，便于控制日志和监控。