【实践原创】使用 FastAPI 实现 Coze 流式聊天 SSE 接口

使用 FastAPI 实现 Coze 流式聊天 SSE 接口

在开发 AI 助手或聊天应用时，我们通常希望服务端能够实时向前端推送消息，让用户看到逐字打字效果。本文演示如何使用FastAPI + Coze Python SDK（cozepy）实现流式聊天 SSE 接口，并提供curl测试方法。

功能特点

1. 流式输出：前端可以实时接收聊天增量消息。
2. SSE 格式：便于浏览器或 Go/Node 前端解析。
3. 兼容不同版本 Coze SDK：处理可能缺失的异常类。
4. 可直接使用curl测试：无需前端即可验证接口。

技术栈

• Python 3.10+
• FastAPI
• uvicorn（ASGI 服务）
• cozepy（Coze 官方 Python SDK）
• SSE 流式推送

完整示例代码

importosfromtypingimportOptional,List,Dict,AnyfromfastapiimportFastAPI,HTTPExceptionfromfastapi.responsesimportStreamingResponsefrompydanticimportBaseModelfromcozepyimportCoze,TokenAuth,Message,ChatEventType,COZE_CN_BASE_URL# ===========================# 兼容不同版本的cozepy异常类# ===========================try:fromcozepyimportCozeAPIError,CozeAuthErrorexceptImportError:classCozeAPIError(Exception):passclassCozeAuthError(Exception):pass# ===========================# 初始化FastAPI应用# ===========================app=FastAPI(title="Coze Stream Chat API")# ===========================# 全局配置与Coze客户端初始化# ===========================COZE_API_TOKEN=os.getenv("COZE_API_TOKEN","你的默认Token")COZE_API_BASE=COZE_CN_BASE_URL BOT_VERSION="1756277832"coze_client:Optional[Coze]=Nonedefinit_coze_client():"""初始化Coze客户端"""globalcoze_clientifcoze_client:returncoze_clienttry:coze_client=Coze(auth=TokenAuth(token=COZE_API_TOKEN),base_url=COZE_API_BASE)returncoze_clientexceptExceptionase:raiseHTTPException(status_code=500,detail=f"Coze客户端初始化失败：{str(e)}")init_coze_client()# ===========================# 定义请求体模型# ===========================classChatRequest(BaseModel):user_id:strbot_id:strstream:bool=Trueadditional_messages:List[Dict[str,Any]]conversation_id:Optional[str]=Nonebot_version:Optional[str]=BOT_VERSION# ===========================# 流式聊天接口# ===========================@app.post("/api/coze-chat")asyncdefcoze_chat(request:ChatRequest):try:# 构建 Coze 消息importjson messages=[]formsginrequest.additional_messages:ifmsg.get("role")=="user"andmsg.get("content_type")=="text":content_list=json.loads(msg.get("content","[]"))text="".join([item.get("text","")foritemincontent_list])messages.append(Message.build_user_question_text(text))# 调用流式接口stream=coze_client.chat.stream(bot_id=request.bot_id,user_id=request.user_id,conversation_id=request.conversation_idorNone,publish_status="published_online",bot_version=request.bot_version,auto_save_history=False,additional_messages=messages)# SSE 流生成器asyncdefgenerate_stream():try:foreventinstream:ifnotevent:continue# 消息增量ifevent.event==ChatEventType.CONVERSATION_MESSAGE_DELTA:content=event.message.content.strip()ifevent.message.contentelse""ifcontent:yieldf"data:{json.dumps({'type':'delta','content':content})}\n\n"# 聊天完成elifevent.event==ChatEventType.CONVERSATION_CHAT_COMPLETED:usage=event.chat.usage.token_countifhasattr(event.chat,'usage')else0conv_id=event.chat.conversation_idifhasattr(event.chat,'conversation_id')else""yieldf"data:{json.dumps({'type':'completed','token_count':usage,'conversation_id':conv_id})}\n\n"yield"data: [DONE]\n\n"exceptExceptionase:yieldf"data:{json.dumps({'type':'error','message':str(e)})}\n\n"returnStreamingResponse(generate_stream(),media_type="text/event-stream",headers={"Cache-Control":"no-cache","Connection":"keep-alive","Access-Control-Allow-Origin":"*"})exceptCozeAuthErrorase:raiseHTTPException(status_code=401,detail=f"认证失败：{str(e)}")exceptCozeAPIErrorase:raiseHTTPException(status_code=502,detail=f"Coze API错误：{str(e)}")exceptExceptionase:raiseHTTPException(status_code=500,detail=f"服务器错误：{str(e)}")# ===========================# 启动服务# ===========================if__name__=="__main__":importuvicorn uvicorn.run(app,host="0.0.0.0",port=8000)

使用方法

1. 安装依赖：

pipinstallfastapi uvicorn cozepy

2. 设置环境变量（可选）：

exportCOZE_API_TOKEN="你的CozeToken"

3. 启动服务：

python main.py

服务将监听http://0.0.0.0:8000。

使用curl测试接口

你可以使用curl来实时查看 SSE 流：

# 测试Python服务curl-X POST -H"Content-Type: application/json"-d'{ "user_id": "123", "bot_id": "7579834670624407602", "stream": true, "additional_messages": [ { "role": "user", "type": "question", "content_type": "text", "content": "[{\"type\":\"text\",\"text\":\"你好\"}]" } ] }'http://localhost:8000/api/coze-chat

参数说明：

• -N/--no-buffer：禁用输出缓存，实时显示流式数据。
• -X POST：发送 POST 请求。
• -d：传递 JSON 请求体。

执行后，你会看到类似以下输出（SSE 流）：

data: {"type": "delta", "content": "你"} data: {"type": "delta", "content": "好"} data: {"type": "delta", "content": "，Coze!"} data: {"type": "completed", "token_count": 12, "conversation_id": "conv_123"} data: [DONE]

前端示例（实时渲染打字机效果）

<divid="chat"></div><script>constchatDiv=document.getElementById("chat");constevtSource=newEventSource("http://localhost:8000/api/coze-chat");evtSource.onmessage=(e)=>{if(e.data==="[DONE]"){console.log("聊天结束");return;}constdata=JSON.parse(e.data);if(data.type==="delta"){chatDiv.innerHTML+=data.content;}elseif(data.type==="completed"){console.log("聊天完成, token_count:",data.token_count);}};evtSource.onerror=()=>console.log("连接错误或关闭");</script>

效果：消息逐字符显示，模拟 AI 打字机输出。

总结

• 通过 FastAPI 可以快速实现 Coze 流式聊天接口。
• SSE 格式让前端无需轮询即可接收消息增量。
• 使用curl或前端 JS 均可实时验证流式输出。
• 可扩展为 AI 聊天助手、客服机器人或协作工具。