团队里第一个接 Claude API 的人，代码通常长这样：在业务文件里 Anthropic(api_key="sk-xxx", base_url="...")，然后直接 messages.create。能跑，但很快会遇到四件麻烦事。

开发机连的入口和线上不一样，每换一次要改代码重新发版。某天上游偶发 529，请求直接抛异常打到用户面前。想知道这个月哪个功能吃掉了大部分调用量，翻不出数据，因为没人记过。新同事接手时不知道 /v1 该不该写，照着别人的代码复制，运行起来 404。

这四件事的根因是同一个：客户端构造和调用逻辑混在业务代码里，没有一层自己的接入封装。这篇从零写这一层，代码可以直接跑。

环境准备

mkdir claude_client && cd claude_client
python -m venv .venv
.venv\Scripts\activate          # Windows PowerShell
pip install anthropic==0.40.0 python-dotenv==1.0.1

有个前置条件：平台要在响应里回传 usage 字段（input_tokens / output_tokens），第三节的用量统计依赖它；另外接口需要兼容 Anthropic 原生协议，这样官方 SDK 的调用代码不用改。jiekou.vip 兼容原生协议并在控制台提供按 Key 的调用明细，接入前把这两项确认一下，拿到 Key 和文档给出的接口地址备用。

步骤一：把连接参数收进配置

新建 .env（记得加进 .gitignore）：

ANTHROPIC_API_KEY=sk-xxxxxx
ANTHROPIC_BASE_URL=按平台文档填写的接口地址
ANTHROPIC_MODEL=claude-sonnet-4-6
ANTHROPIC_TIMEOUT=60

然后写 config.py，只做一件事：把环境变量读成一个不可变的配置对象。

# config.py
import os
from dataclasses import dataclass
from dotenv import load_dotenv

load_dotenv()


@dataclass(frozen=True)
class Settings:
    api_key: str
    base_url: str | None
    model: str
    timeout: float

    @classmethod
    def from_env(cls) -> "Settings":
        key = os.environ.get("ANTHROPIC_API_KEY", "").strip()
        if not key:
            raise RuntimeError("ANTHROPIC_API_KEY 未设置，检查 .env 是否在当前目录")

        raw = os.environ.get("ANTHROPIC_BASE_URL", "").strip().rstrip("/")
        # SDK 自己会拼 /v1/messages，配置里再带 /v1 会变成 /v1/v1/messages -> 404
        if raw.endswith("/v1"):
            raw = raw[: -len("/v1")]

        return cls(
            api_key=key,
            base_url=raw or None,
            model=os.environ.get("ANTHROPIC_MODEL", "claude-sonnet-4-6"),
            timeout=float(os.environ.get("ANTHROPIC_TIMEOUT", "60")),
        )

    def masked(self) -> str:
        tail = self.api_key[-4:] if len(self.api_key) >= 4 else "****"
        return f"key=***{tail} base={self.base_url or '<official>'} model={self.model}"


if __name__ == "__main__":
    print(Settings.from_env().masked())

key=***9f2c base=https://your-gateway.example.com model=claude-sonnet-4-6

那段去掉尾部 /v1 的代码是这层封装里最值钱的三行。404 Not Found 的报错信息里完全看不出是路径重复，人肉排查这个坑平均要花半小时，而且每个新人都会踩一次。写死在配置层，之后 .env 里带不带 /v1 都不影响。

masked() 是为了让日志能打印配置又不泄露 Key——直接 print 整个 Settings 会把 key 写进日志文件。

步骤二：客户端工厂 + 重试策略

# client.py
import logging
from anthropic import Anthropic
from config import Settings

log = logging.getLogger(__name__)

_client: Anthropic | None = None
_settings: Settings | None = None


def get_settings() -> Settings:
    global _settings
    if _settings is None:
        _settings = Settings.from_env()
        log.info("settings loaded: %s", _settings.masked())
    return _settings


def get_client() -> Anthropic:
    """进程内复用同一个客户端，连接池不会反复重建。"""
    global _client
    if _client is None:
        s = get_settings()
        _client = Anthropic(
            api_key=s.api_key,
            base_url=s.base_url,
            timeout=s.timeout,
            max_retries=3,
        )
    return _client

max_retries=3 交给 SDK 内置的退避重试，它对 429 和 5xx 会自动指数退避并尊重 retry-after 头，比自己写 for i in range(3) 靠谱得多。自己写循环最常见的错误是把 400 参数错误也重试三遍——那种错重试一万次也不会成功，只是白等。

单例是为了复用底层 HTTP 连接池。每次调用都 Anthropic(...) 会重建连接，高频场景下光 TLS 握手就能吃掉可观的延迟。

步骤三：包一层带用量记录的调用函数

这是整层封装的核心。业务代码只调 ask()，不碰 SDK。

# api.py
import json
import time
import pathlib
from datetime import datetime, timezone
from client import get_client, get_settings

USAGE_LOG = pathlib.Path("usage.jsonl")


def ask(
    prompt: str,
    *,
    scene: str,
    system: str | None = None,
    max_tokens: int = 1024,
    model: str | None = None,
) -> str:
    """发一次请求，返回文本，并把用量按场景记一行。

    scene 是必填的关键字参数：月底想知道哪个功能吃了多少量，全靠它。
    """
    s = get_settings()
    client = get_client()

    kwargs = {
        "model": model or s.model,
        "max_tokens": max_tokens,
        "messages": [{"role": "user", "content": prompt}],
    }
    if system:
        kwargs["system"] = system

    t0 = time.perf_counter()
    resp = client.messages.create(**kwargs)
    elapsed_ms = int((time.perf_counter() - t0) * 1000)

    _record(
        {
            "ts": datetime.now(timezone.utc).isoformat(timespec="seconds"),
            "scene": scene,
            "model": resp.model,
            "input_tokens": resp.usage.input_tokens,
            "output_tokens": resp.usage.output_tokens,
            "elapsed_ms": elapsed_ms,
            "stop_reason": resp.stop_reason,
        }
    )
    return resp.content[0].text


def _record(row: dict) -> None:
    with USAGE_LOG.open("a", encoding="utf-8") as f:
        f.write(json.dumps(row, ensure_ascii=False) + "\n")

跑一下：

from api import ask

print(ask("用三句话解释什么是 API。", scene="doc_qa",
          system="你是一个简洁的中文助手。"))

API 是软件之间约定好的对话方式，规定了怎么提问、怎么回答。
它让你不用了解对方内部怎么实现，照着约定发请求就能拿到结果。
日常用的地图、支付、登录，背后都是一次次 API 调用。

usage.jsonl 里同时多了一行：

{"ts":"2026-08-14T03:24:11+00:00","scene":"doc_qa","model":"claude-sonnet-4-6","input_tokens":38,"output_tokens":96,"elapsed_ms":1472,"stop_reason":"end_turn"}

三个设计点值得说明。

scene 设成必填的关键字参数，是故意给调用方添麻烦——一旦允许不传，所有人都不会传，三个月后这份日志就只是一堆无法归类的数字。强制传参的代价是每次多写八个字符，收益是月底能一行命令拉出分布。

记 resp.model 而不是记请求里传的 model。两者可能不一致：平台做了模型别名映射、或者你传了个已下线的名字被兜底到别的模型，只有从响应里读才知道真正跑的是哪个。

记 stop_reason。它等于 max_tokens 说明输出被截断了——用户看到的是半句话，但程序没有任何异常。这个字段是排查"回答莫名断掉"的唯一线索。

步骤四：流式和多轮走同一层

流式输出复用同一个客户端，用量在流结束后从 final message 里取：

# stream_api.py
import time
from client import get_client, get_settings
from api import _record
from datetime import datetime, timezone


def ask_stream(prompt: str, *, scene: str, max_tokens: int = 1024):
    s = get_settings()
    t0 = time.perf_counter()

    with get_client().messages.stream(
        model=s.model,
        max_tokens=max_tokens,
        messages=[{"role": "user", "content": prompt}],
    ) as stream:
        for text in stream.text_stream:
            yield text
        final = stream.get_final_message()

    _record({
        "ts": datetime.now(timezone.utc).isoformat(timespec="seconds"),
        "scene": scene + ":stream",
        "model": final.model,
        "input_tokens": final.usage.input_tokens,
        "output_tokens": final.usage.output_tokens,
        "elapsed_ms": int((time.perf_counter() - t0) * 1000),
        "stop_reason": final.stop_reason,
    })

for chunk in ask_stream("写一首关于秋天的短诗。", scene="poem"):
    print(chunk, end="", flush=True)

flush=True 不能省，否则终端缓冲会让逐字输出看起来仍是一次性吐出来的，白做流式。

多轮对话要自己维护历史，因为接口是无状态的：

# conversation.py
from client import get_client, get_settings


class Conversation:
    def __init__(self, system: str | None = None, max_turns: int = 10):
        self.client = get_client()
        self.model = get_settings().model
        self.system = system
        self.max_turns = max_turns
        self.messages: list[dict] = []

    def ask(self, text: str) -> str:
        self.messages.append({"role": "user", "content": text})

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

        resp = self.client.messages.create(**kwargs)
        reply = resp.content[0].text
        self.messages.append({"role": "assistant", "content": reply})
        self._truncate()
        return reply

    def _truncate(self) -> None:
        keep = self.max_turns * 2
        if len(self.messages) > keep:
            self.messages = self.messages[-keep:]

conv = Conversation(system="你是一个简洁的中文助手。")
print(conv.ask("我叫小明。"))
print(conv.ask("我叫什么名字？"))

你好小明，有什么可以帮你的？
你叫小明。

第二问能答出来，说明历史确实带上去了。

_truncate 不是可选项。历史无限增长时，每轮的输入都比上一轮更长，输入量随轮数呈平方增长，几十轮之后就会撞上上下文上限直接报错。这里用最简单的滑动窗口；如果开场几轮有关键设定，可以改成"保留首轮 + 最近 N 轮"，或者把早期消息压成一段摘要再塞回去。

步骤五：把用量日志读成一张表

前面那份 usage.jsonl 现在可以用了：

# report.py
import json
import pathlib
from collections import defaultdict

LOG = pathlib.Path("usage.jsonl")


def main() -> None:
    stat = defaultdict(lambda: {"n": 0, "in": 0, "out": 0, "ms": []})

    with LOG.open(encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            row = json.loads(line)
            s = stat[row["scene"]]
            s["n"] += 1
            s["in"] += row["input_tokens"]
            s["out"] += row["output_tokens"]
            s["ms"].append(row["elapsed_ms"])

    print(f"{'场景':<16}{'次数':>6}{'输入':>10}{'输出':>10}{'耗时P90':>10}")
    for scene, s in sorted(stat.items(), key=lambda kv: -kv[1]["in"]):
        ms = sorted(s["ms"])
        p90 = ms[int(len(ms) * 0.9)] if ms else 0
        print(f"{scene:<16}{s['n']:>6}{s['in']:>10,}{s['out']:>10,}{p90:>8}ms")


if __name__ == "__main__":
    main()

场景                次数        输入        输出    耗时P90
doc_qa             412    486,231     52,940    2140ms
poem:stream         88      3,504     26,112    3260ms
title_gen          950     71,250     19,000     820ms

按输入量排序之后一眼能看出问题：title_gen 调用次数最多（950 次），但真正吃量的是 doc_qa——412 次调用带来近五十万输入。如果只统计调用次数，优化方向会完全找反。这类场景通常是把整份文档反复塞进上下文，加一层缓存或者先做检索裁剪，效果比优化那 950 次标题生成明显得多。

耗时看 P90 不看平均值。平均值会被大量快请求拉低，而用户体验取决于慢的那一批。

几个实践中的注意点

配置层的 /v1 处理不要省。 这是新人接入时最高频的一个 404，写在代码里一次，之后谁都不会再踩。

scene 必填。 允许不传就等于没有这份统计。

用量日志按天切分或者定期归档。 单个 jsonl 长到几十万行之后 report.py 会变慢，按 usage-YYYYMMDD.jsonl 命名即可。

别在这层写业务逻辑。 这层只负责"怎么把请求发出去、怎么记下来"。prompt 模板、结果解析、重试之外的降级策略都放业务层，否则很快会长成一个谁都不敢改的巨型工具类。

Key 不进日志。 打印配置一律走 masked()。

小结

一层合格的接入封装就五件事：连接参数外置到 .env（换环境不改代码）、配置层吃掉 /v1 这类坑（新人不再踩 404）、客户端单例 + SDK 内置重试（别自己写循环）、调用带 scene 记 usage（月底才有数据可查）、流式和多轮复用同一层（不留第二套调用姿势）。

加起来两百来行，一次写好之后长期不用动。下一篇写怎么在这层上加 prompt 模板管理和结果的结构化校验，让业务代码连 prompt 字符串都不用拼。