Walkthrough ｜手写 Claude API Client¶

难度：⭐⭐⭐⭐ 时间：~2h 目标：理解 Anthropic SDK 的核心，写一个简化版

1. Claude API Client 是什么¶

Claude API Client = 与 Anthropic API 通信的层。 - HTTP 请求 - 流式响应 - Token 累加 - 错误处理 - Retry

2. 目标¶

手写一个简化版 Claude API Client： - HTTP POST /v1/messages - 流式 SSE - Basic auth - Retry - Token 累加

3. 完整代码¶

// mini-claude-client.ts

interface Message {
  role: 'user' | 'assistant' | 'system'
  content: string
}

interface Tool {
  name: string
  description: string
  input_schema: any
}

interface Usage {
  input_tokens: number
  output_tokens: number
  cache_creation_input_tokens?: number
  cache_read_input_tokens?: number
}

interface Response {
  id: string
  content: Array<{ type: string; text?: string; id?: string; name?: string; input?: any }>
  usage: Usage
  stop_reason: string
  model: string
}

export class MiniClaudeClient {
  private apiKey: string
  private baseURL: string
  private maxRetries: number

  constructor(apiKey: string, baseURL: string = 'https://api.anthropic.com') {
    this.apiKey = apiKey
    this.baseURL = baseURL
    this.maxRetries = 3
  }

  async createMessage(params: {
    model: string
    max_tokens: number
    system?: string
    messages: Message[]
    tools?: Tool[]
    stream?: boolean
  }): Promise<Response> {
    let attempt = 0
    let lastError: Error | null = null

    while (attempt < this.maxRetries) {
      try {
        const response = await this.sendRequest(params)
        return response
      } catch (e) {
        lastError = e as Error
        attempt++

        // Exponential backoff
        const wait = Math.min(1000 * 2 ** attempt, 10000)
        await new Promise((r) => setTimeout(r, wait))
      }
    }

    throw lastError || new Error('Max retries exceeded')
  }

  private async sendRequest(params: any): Promise<Response> {
    const response = await fetch(`${this.baseURL}/v1/messages`, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'x-api-key': this.apiKey,
        'anthropic-version': '2023-06-01',
      },
      body: JSON.stringify(params),
    })

    if (!response.ok) {
      throw new Error(`HTTP ${response.status}: ${await response.text()}`)
    }

    return await response.json()
  }

  async *streamMessage(params: any): AsyncGenerator<any> {
    const response = await fetch(`${this.baseURL}/v1/messages`, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'x-api-key': this.apiKey,
        'anthropic-version': '2023-06-01',
      },
      body: JSON.stringify({ ...params, stream: true }),
    })

    if (!response.ok) {
      throw new Error(`HTTP ${response.status}`)
    }

    const reader = response.body!.getReader()
    const decoder = new TextDecoder()

    let buffer = ''

    while (true) {
      const { done, value } = await reader.read()
      if (done) break

      buffer += decoder.decode(value, { stream: true })

      // SSE format: "event: ...\ndata: ...\n\n"
      const lines = buffer.split('\n\n')
      buffer = lines.pop() || ''

      for (const line of lines) {
        if (line.startsWith('data: ')) {
          const data = line.slice(6)
          if (data === '[DONE]') return

          try {
            yield JSON.parse(data)
          } catch (e) {
            // skip
          }
        }
      }
    }
  }

  // 累加 token
  accumulateUsage(usage: Usage): Usage {
    if (!this.totalUsage) this.totalUsage = { input_tokens: 0, output_tokens: 0 }
    this.totalUsage.input_tokens += usage.input_tokens
    this.totalUsage.output_tokens += usage.output_tokens
    if (usage.cache_creation_input_tokens) {
      this.totalUsage.cache_creation_input_tokens = 
        (this.totalUsage.cache_creation_input_tokens || 0) + usage.cache_creation_input_tokens
    }
    if (usage.cache_read_input_tokens) {
      this.totalUsage.cache_read_input_tokens = 
        (this.totalUsage.cache_read_input_tokens || 0) + usage.cache_read_input_tokens
    }
    return this.totalUsage
  }
}

~120 行。

4. 使用示例¶

const client = new MiniClaudeClient(process.env.ANTHROPIC_API_KEY!)

const response = await client.createMessage({
  model: 'claude-sonnet-4-6',
  max_tokens: 1024,
  messages: [{ role: 'user', content: 'Hello' }],
})

console.log(response.content[0].text)
console.log('Tokens:', response.usage)

简单。

5. 流式使用¶

for await (const event of client.streamMessage({
  model: 'claude-sonnet-4-6',
  max_tokens: 1024,
  messages: [{ role: 'user', content: 'Tell a story' }],
})) {
  if (event.type === 'content_block_delta' && event.delta?.text) {
    process.stdout.write(event.delta.text)
  }
}

流式。

6. 4 大关键设计¶

6.1 Retry + Exponential backoff¶

const wait = Math.min(1000 * 2 ** attempt, 10000)

1s → 2s → 4s → 8s → 10s (max)。

6.2 SSE 解析¶

// SSE: "event: ...\ndata: ...\n\n"
const lines = buffer.split('\n\n')
for (const line of lines) {
  if (line.startsWith('data: ')) {
    yield JSON.parse(line.slice(6))
  }
}

SSE 标准。

6.3 Token 累加¶

this.totalUsage.input_tokens += usage.input_tokens

4 维 token。

6.4 流式 + 缓存¶

// prompt cache 取决于 static 前缀
// stream 模式不变

缓存。

7. 6 个扩展¶

7.1 System prompt 缓存¶

const systemPrompt = `You are ...`
// 同样内容 → cache 命中 → cost 节省 90%

content hash。

7.2 Beta features¶

headers['anthropic-beta'] = 'prompt-caching-2024-07-31'

Beta header。

7.3 Custom retry policy¶

retryOn: (response) => response.status === 429 || response.status === 500

自定义。

7.4 Telemetry¶

logEvent('tengu_api_call', { model, tokens })

telemetry。

7.5 Streaming cancel¶

const controller = new AbortController()
const stream = client.streamMessage(params, { signal: controller.signal })

cancel。

7.6 Multiple API keys (rotation)¶

class ClientWithKeyRotation {
  private keys: string[]
  private currentIdx = 0
  // rotate on 429
}

rotation。

8. 对比真实 SDK¶

维度	Mini	真实
行数	120	3419
Stream	基础	完整
Retry	简单	复杂（含 jitter）
Auth	单 key	rotation
缓存	无	自动
Telemetry	无	完整
Beta features	推测	完整

简化 vs 真实。

9. 5 个关键洞察¶

HTTP + JSON 是基础
SSE 是流式协议
Retry + backoff 是必备
Token 累加 是 cost tracking 基础
prompt cache 取决于 static 前缀

10. 5 个练习¶

加 streaming —— 用 ReadableStream
加 retry with jitter —— 防止 thundering herd
加 token 累加 —— 持久化
加 system prompt 缓存 —— content hash
加 custom retry policy —— 按 status code 决定

5 步。

11. 总结¶

手写 Claude API Client = 理解 HTTP + SSE + Retry + Token。

核心： - 120 行简化版 - HTTP + SSE - 3 次 retry - 4 维 token

下一步： - 看真实 SDK - 加 streaming - 加 telemetry

Walkthrough ｜ 手写 Claude API Client¶