你有没有想过：把 Claude Code 的智能编码能力，和 DeepSeek 的高性价比模型结合起来，再包装成一个支持流式输出的 API 服务？

简单说就是：Claude Code 当“大脑”，DeepSeek 当“算力”，Python + SSE 当“嘴巴”——让 AI 边想边说，逐字输出。

今天这篇，手把手教你实现这个组合。

一、为什么是这个组合？

先看三个组件的定位：

组合之后的效果：

• ✅ 低成本：DeepSeek V4-Pro 比原生 Claude 模型便宜得多
• ✅ 高兼容：DeepSeek 提供完全兼容 Claude 的 API，Claude Code 无需改代码
• ✅ 流式输出：SSE 协议让 AI“边想边说”，用户体验接近 ChatGPT
• ✅ 可扩展：封装成 API 后，可以被任何前端/后端调用

一句话：用 Claude Code 的习惯，花 DeepSeek 的钱，还能自己控制服务。

二、第二步：Claude Code 对接 DeepSeek V4-Pro

2.1 核心原理

DeepSeek 提供了完全兼容 Anthropic 协议的 API 接口。

这意味着：Claude Code 以为自己在调用 Claude，实际上调用的是 DeepSeek。

只需要修改环境配置，无需改任何代码。

2.2 获取 DeepSeek API Key

1. 访问 DeepSeek 官网
2. 注册/登录
3. 进入控制台，创建 API Key
4. 保存备用

3.3 配置 Claude Code

Claude Code 的配置文件位置：

• macOS/Linux：~/.claude/settings.json
• Windows：%USERPROFILE%.claudesettings.json

写入以下配置：

{
  "env": {
    "ANTHROPIC_BASE_URL": "https://api.deepseek.com/anthropic",
    "ANTHROPIC_AUTH_TOKEN": "${DEEPSEEK_API_KEY}",
    "API_TIMEOUT_MS": "3000000",
    "ANTHROPIC_MODEL": "deepseek-v4-pro",
    "ANTHROPIC_SMALL_FAST_MODEL": "deepseek-v4-flash",
    "ANTHROPIC_DEFAULT_SONNET_MODEL": "deepseek-v4-pro",
    "ANTHROPIC_DEFAULT_OPUS_MODEL": "deepseek-v4-pro",
    "ANTHROPIC_DEFAULT_HAIKU_MODEL": "deepseek-v4-flash",
    "CLAUDE_CODE_SUBAGENT_MODEL": "deepseek-v4-pro",
    "CLAUDE_CODE_EFFORT_LEVEL": "max"
  },
  "model": "deepseek-v4-pro"
}

配置解读：

• ANTHROPIC_BASE_URL：指向 DeepSeek 的兼容端点
• ANTHROPIC_AUTH_TOKEN：你的 DeepSeek API Key
• ANTHROPIC_MODEL：默认使用 V4-Pro（复杂任务）
• ANTHROPIC_SMALL_FAST_MODEL：轻量任务用 V4-Flash（省钱）
• API_TIMEOUT_MS：300 秒超时，避免长推理中断

2.4 设置环境变量

为了避免 API Key 硬编码在配置文件里：

export DEEPSEEK_API_KEY="sk-xxxxx"

或在项目根目录创建 .env 文件：

DEEPSEEK_API_KEY=sk-xxxxx

2.5 验证配置

重启 Claude Code：

claude

向它提问：“你是什么模型？”

如果回复“我是 DeepSeek V4-Pro”，配置成功。

三、第三步：Python 对接 DeepSeek API

3.1 环境准备

# 创建虚拟环境
python -m venv deepseek-sse
source deepseek-sse/bin/activate  # Windows: deepseek-sseScriptsactivate

# 安装依赖
pip install fastapi uvicorn sse-starlette python-dotenv httpx

3.2 DeepSeek API 基础调用

DeepSeek 的 API 与 OpenAI 兼容，使用方式几乎一致：

import os
import requests
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.getenv("DEEPSEEK_API_KEY")
BASE_URL = "https://api.deepseek.com/v1/chat/completions"

def chat_sync(prompt: str) -> str:
    """同步调用 DeepSeek API"""
    payload = {
        "model": "deepseek-chat",
        "messages": [
            {"role": "system", "content": "你是一个专业的编程助手"},
            {"role": "user", "content": prompt}
        ],
        "temperature": 0.3,
        "max_tokens": 2000
    }
    
    response = requests.post(
        BASE_URL,
        headers={
            "Authorization": f"Bearer {API_KEY}",
            "Content-Type": "application/json"
        },
        json=payload,
        timeout=30
    )
    
    response.raise_for_status()
    return response.json()["choices"][0]["message"]["content"]

3.3 流式调用（Streaming）

流式调用的关键是设置 stream=True，然后逐块读取 delta.content：

def chat_stream(prompt: str):
    """流式调用 DeepSeek API"""
    payload = {
        "model": "deepseek-chat",
        "messages": [
            {"role": "system", "content": "你是一个专业的编程助手"},
            {"role": "user", "content": prompt}
        ],
        "stream": True,  # 关键！
        "temperature": 0.3,
    }
    
    with requests.post(
        BASE_URL,
        headers={
            "Authorization": f"Bearer {API_KEY}",
            "Content-Type": "application/json"
        },
        json=payload,
        stream=True,  # 关键！
        timeout=60
    ) as response:
        response.raise_for_status()
        
        for line in response.iter_lines(decode_unicode=True):
            if not line:
                continue
            if line.startswith("data: "):
                chunk = line[6:]  # 去掉 "data: " 前缀
                if chunk == "[DONE]":
                    break
                try:
                    import json
                    obj = json.loads(chunk)
                    delta = obj["choices"][0]["delta"]
                    content = delta.get("content", "")
                    if content:
                        yield content
                except json.JSONDecodeError:
                    pass

四、第四步：封装成 SSE 接口服务

4.1 SSE 是什么？

SSE（Server-Sent Events）是一种基于 HTTP 的单向推送协议，超级适合 AI 流式输出。相比 WebSocket，它更轻量，不需要双向通信。

SSE 的消息格式很简单：

data: 你
data: 好
data: [DONE]

每条消息以 data: 开头，以两个换行符

结尾。

5.2 用 FastAPI 实现 SSE 接口

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from fastapi.middleware.cors import CORSMiddleware
import asyncio
import json

app = FastAPI()

# 配置 CORS（前端调用需要）
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

async def sse_generator(prompt: str):
    """SSE 生成器，逐块推送 AI 回复"""
    try:
        # 调用 DeepSeek 流式接口
        for chunk in chat_stream(prompt):
            # SSE 格式：data: {content}


            yield f"data: {json.dumps({'content': chunk, 'done': False})}

"
            await asyncio.sleep(0.02)  # 模拟打字效果
        
        # 发送结束信号
        yield f"data: {json.dumps({'done': True})}

"
        
    except Exception as e:
        yield f"data: {json.dumps({'error': str(e), 'done': True})}

"

@app.get("/chat")
async def chat_stream_endpoint(prompt: str):
    """流式对话接口"""
    return StreamingResponse(
        sse_generator(prompt),
        media_type="text/event-stream"
    )

@app.post("/chat")
async def chat_stream_post(request: Request):
    """POST 版本，支持更复杂的请求体"""
    body = await request.json()
    prompt = body.get("prompt", "")
    return StreamingResponse(
        sse_generator(prompt),
        media_type="text/event-stream"
    )

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

4.3 启动服务

python main.py

服务启动在 http://localhost:8000。

测试流式接口：

curl "http://localhost:8000/chat?prompt=用Python写一个快速排序"

你会看到数据逐块推送出来。

五、前端怎么用？

5.1 使用 EventSource（原生 SSE）

// 创建 SSE 连接
const eventSource = new EventSource('http://localhost:8000/chat?prompt=写一个快速排序');

let fullResponse = '';

eventSource.onmessage = (event) => {
    const data = JSON.parse(event.data);
    
    if (data.done) {
        console.log('对话完成:', fullResponse);
        eventSource.close();
    } else {
        fullResponse += data.content;
        // 实时显示在页面上
        document.getElementById('output').innerHTML += data.content;
    }
};

eventSource.onerror = (error) => {
    console.error('SSE 错误:', error);
    eventSource.close();
};

5.2 使用 Fetch API（更灵活）

EventSource 不支持 POST 请求，如果需要传复杂参数，用 Fetch：

async function streamChat(prompt) {
    const response = await fetch('http://localhost:8000/chat', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ prompt })
    });
    
    const reader = response.body.getReader();
    const decoder = new TextDecoder();
    
    while (true) {
        const { done, value } = await reader.read();
        if (done) break;
        
        const chunk = decoder.decode(value);
        // 解析 SSE 格式
        const lines = chunk.split('
');
        for (const line of lines) {
            if (line.startsWith('data: ')) {
                const data = JSON.parse(line.slice(6));
                if (!data.done) {
                    // 实时更新 UI
                    updateUI(data.content);
                }
            }
        }
    }
}

六、进阶：让服务支持对话记忆

6.1 问题

当前实现是无状态的——每次请求都是全新对话，AI 不记得之前聊过什么。

6.2 解决方案：会话管理

用 conversation_id 维护历史：

# 简单的内存存储（生产环境用 Redis）
conversations = {}

@app.post("/chat")
async def chat_with_memory(request: Request):
    body = await request.json()
    prompt = body.get("prompt")
    conversation_id = body.get("conversation_id")
    
    # 获取或创建会话历史
    if conversation_id not in conversations:
        conversations[conversation_id] = []
    
    # 添加用户消息
    conversations[conversation_id].append({
        "role": "user",
        "content": prompt
    })
    
    # 构建完整消息列表（包含历史）
    messages = [
        {"role": "system", "content": "你是一个专业的编程助手"}
    ] + conversations[conversation_id]
    
    # 流式调用，同时收集完整回复
    full_response = []
    async def generate_with_memory():
        nonlocal full_response
        for chunk in chat_stream_with_messages(messages):
            full_response.append(chunk)
            yield f"data: {json.dumps({'content': chunk})}

"
        
        # 保存 AI 回复到历史
        conversations[conversation_id].append({
            "role": "assistant",
            "content": "".join(full_response)
        })
        yield "data: [DONE]

"
    
    return StreamingResponse(generate_with_memory(), media_type="text/event-stream")

七、注意事项与避坑指南

7.1 Claude Code + DeepSeek 的限制

不支持图片输入

DeepSeek V4-Pro 目前是纯文本模型。在 Claude Code 里发送图片，只会收到占位标记，无法识别。

临时方案：

• 需要图片分析时切换回原生模型
• 纯代码任务使用 DeepSeek

7.2 长上下文稳定性

虽然 DeepSeek 支持百万 token 上下文，但超长会话叠加多轮工具调用可能出现不稳定。

提议：

• 复杂任务分阶段执行
• 适时用 /clear 重置会话
• 设置合理的 API_TIMEOUT_MS（推荐 300 秒以上）

7.3 CORS 配置

如果前端和后端不同源，需要配置 CORS：

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],  # 前端地址
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

7.4 生产环境注意事项

• API Key 管理：用环境变量，不要硬编码
• 限流：防止滥用，可以加 slowapi 等限流中间件
• 日志：记录请求和错误，便于排查
• 超时：SSE 连接可能长时间保持，注意网关的超时配置

八、完整项目结构

deepseek-sse-service/
├── .env                 # API Key 配置
├── requirements.txt     # Python 依赖
├── main.py              # FastAPI 主程序
├── deepseek_client.py   # DeepSeek API 封装
├── static/              # 前端测试页面
│   └── index.html
└── README.md

requirements.txt：

fastapi==0.104.1
uvicorn==0.24.0
sse-starlette==1.6.5
python-dotenv==1.0.0
httpx==0.25.0

写在最后

回顾一下整个链路：

用户请求 → FastAPI (SSE) → DeepSeek API → 流式返回 → 前端实时渲染
                ↑
         Claude Code 配置
         (可选，用于复杂任务编排)

你得到了什么？

1. 一个完整的 AI 对话服务：支持流式输出、对话记忆
2. 低成本：DeepSeek 价格远低于原生 Claude
3. 可扩展：封装成 API 后，可接入任何前端/后端
4. Claude Code 兼容：配置一次，后续所有 Skills 都能用

下一步可以做的事情：

• 接入 Redis 做分布式会话管理
• 添加模型切换（一键换 Claude/GPT/GLM）
• 封装成 Docker 镜像，一键部署

代码已经可以跑了——你目前就可以打开终端，把这篇文章变成现实。