Python FastAPI 全栈集成

你将构建什么

本教程将带你从零构建一个 FastAPI 后端应用，采用双客户端架构：

Profy（API Key） — 你的应用自己的 API Key，用于调用专家和 AI 模型
ProfyApp（OAuth） — 用户的 OAuth Token，用于向袋袋上报计费事件

前置条件

袋袋开发者账号

已注册袋袋账号并在开发者后台创建 App，获取 Client ID、Client Secret 和 API Key

开发环境

Python 3.10+、pip、基础 FastAPI 和 async/await 经验

Step 1: 项目初始化

mkdir profy-fastapi-demo && cd profy-fastapi-demo
python -m venv .venv && source .venv/bin/activate
pip install fastapi uvicorn profy-sdk sqlalchemy aiosqlite python-dotenv

创建 .env 文件存放凭证：

.env

# Profy 客户端（API Key）— 调用专家和 AI 模型
PROFY_API_KEY=sk-pro-your-api-key

# ProfyApp 客户端（OAuth）— 用户授权和计费
PROFY_APP_ID=your_app_client_id
PROFY_APP_SECRET=your_app_client_secret
PROFY_CALLBACK_URL=http://localhost:8000/callback

永远不要将 .env 提交到版本控制。确保 .gitignore 包含该文件。

Step 2: 初始化双客户端

config.py

import os

from dotenv import load_dotenv
from profy import Profy, ProfyApp

load_dotenv()

client = Profy(api_key=os.environ["PROFY_API_KEY"])

app_client = ProfyApp(
    client_id=os.environ["PROFY_APP_ID"],
    client_secret=os.environ["PROFY_APP_SECRET"],
)

CALLBACK_URL = os.environ["PROFY_CALLBACK_URL"]

两个客户端各司其职：client（Profy）持有你的 API Key，用于 AI 调用；app_client（ProfyApp）持有 OAuth 凭证，用于用户授权和计费。

Step 3: Token 存储

手动实现 Token 存储。生产环境使用数据库，这里用 dict 演示核心模式：

token_store.py

from profy import ProfyApp, OAuthToken

token_store: dict[str, OAuthToken] = {}


async def save_token(user_id: str, token: OAuthToken):
    token_store[user_id] = token


async def get_token(user_id: str) -> OAuthToken | None:
    return token_store.get(user_id)


async def delete_token(user_id: str):
    token_store.pop(user_id, None)


async def get_valid_token(user_id: str, app: ProfyApp) -> OAuthToken:
    stored = await get_token(user_id)
    if not stored:
        raise Exception("User not authenticated")

    if not stored.is_expired():
        return stored

    refreshed = await app.refresh(stored.refresh_token)
    await save_token(user_id, refreshed)
    return refreshed

生产环境必须用持久化存储（数据库 / Redis）。dict 在进程重启后会丢失所有 Token。参考 Token 管理最佳实践。

Step 4: OAuth 登录流程

main.py

import os
import uuid
from contextlib import asynccontextmanager

from fastapi import FastAPI, Cookie, Depends, Request
from fastapi.responses import JSONResponse, RedirectResponse, StreamingResponse

from config import app_client, client, CALLBACK_URL
from token_store import save_token, get_valid_token, delete_token

app = FastAPI(title="Profy FastAPI Demo")


@app.get("/login")
async def login():
    auth_url = app_client.authorization_url(
        redirect_uri=CALLBACK_URL,
        scope="events:write",
    )
    return RedirectResponse(auth_url)


@app.get("/callback")
async def callback(code: str):
    token = await app_client.exchange_code(code, CALLBACK_URL)

    user_id = str(uuid.uuid4())
    await save_token(user_id, token)

    response = RedirectResponse("/dashboard")
    response.set_cookie(
        "profy_user_id",
        user_id,
        httponly=True,
        samesite="lax",
        max_age=60 * 60 * 24 * 90,
    )
    return response

Step 5: 保护路由

通过 FastAPI 依赖注入自动加载并校验 Token：

dependencies.py

from fastapi import Cookie, HTTPException

from config import app_client
from token_store import get_valid_token
from profy import OAuthToken


async def get_current_token(profy_user_id: str = Cookie()) -> OAuthToken:
    try:
        return await get_valid_token(profy_user_id, app_client)
    except Exception:
        raise HTTPException(status_code=401, detail="未登录或 Token 已过期，请重新授权")

Step 6: 上报计费事件

main.py (续)

from profy import AuthenticationError, InsufficientBalanceError, OAuthToken
from dependencies import get_current_token
from pydantic import BaseModel
from token_store import save_token


class ReportRequest(BaseModel):
    event_name: str
    idempotency_key: str | None = None
    metadata: dict[str, str] | None = None


@app.post("/api/report-event")
async def report_event(
    body: ReportRequest,
    token: OAuthToken = Depends(get_current_token),
    profy_user_id: str = Cookie(),
):
    try:
        result = await app_client.report_event(
            body.event_name,
            token=token,
            idempotency_key=body.idempotency_key or str(uuid.uuid4()),
            metadata=body.metadata,
            on_refresh=lambda rotated: save_token(profy_user_id, rotated),
        )
        return result
    except AuthenticationError:
        return JSONResponse(
            status_code=401,
            content={"error": "登录已过期，请重新授权"},
        )
    except InsufficientBalanceError:
        return JSONResponse(
            status_code=402,
            content={"error": "余额不足，请充值后重试"},
        )

on_refresh 回调在 Token 自动刷新时触发。因为 Profy 的 Refresh Token 是一次性的（rotation），必须在回调中持久化新 Token。

Step 7: 调用 AI

使用 Profy（API Key）客户端调用专家或 AI 模型——这与用户 OAuth Token 无关：

main.py (续)

import json


class ChatRequest(BaseModel):
    message: str
    expert_id: str | None = None


@app.post("/api/chat")
async def chat(body: ChatRequest):
    if body.expert_id:
        result = await client.agents.run(body.expert_id, body.message)
        return {"text": result.text, "usage": result.usage.__dict__}

    response = await client.chat.completions.create(
        model="deepseek-chat",
        messages=[
            {"role": "system", "content": "你是一个专业的写作助手。"},
            {"role": "user", "content": body.message},
        ],
        temperature=0.7,
        max_tokens=2000,
    )
    return {"text": response.text, "usage": response.usage.__dict__}


@app.post("/api/chat-stream")
async def chat_stream(body: ChatRequest):
    async def stream_response():
        stream = await client.chat.completions.create(
            model="deepseek-chat",
            messages=[
                {"role": "system", "content": "你是一个专业的写作助手。"},
                {"role": "user", "content": body.message},
            ],
            stream=True,
            temperature=0.7,
            max_tokens=2000,
        )
        async for chunk in stream:
            if chunk.text:
                yield f"data: {json.dumps({'content': chunk.text})}\n\n"
        yield "data: [DONE]\n\n"

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

Step 8: 错误处理中间件

main.py (续)

from profy import AuthenticationError, InsufficientBalanceError, ProfyApiError


@app.exception_handler(AuthenticationError)
async def handle_auth_error(request: Request, exc: AuthenticationError):
    return JSONResponse(status_code=401, content={"error": "认证失败，请检查 API Key 或重新授权"})


@app.exception_handler(InsufficientBalanceError)
async def handle_balance_error(request: Request, exc: InsufficientBalanceError):
    return JSONResponse(status_code=402, content={"error": "余额不足，请充值后重试"})


@app.exception_handler(ProfyApiError)
async def handle_profy_error(request: Request, exc: ProfyApiError):
    return JSONResponse(status_code=502, content={"error": f"袋袋服务异常: {exc}"})

值得写进代码的契约限制

计费端点在服务端强校验这几项。提前在自己这边挡一道，免得线上才发现是 400：

项目	数值	超出后
`event` 名称长度	100 字符	`400 invalid_request`
`idempotency_key` 长度	128 字符	`400 invalid_request`
必填字段	`event` + `idempotency_key`	`400 invalid_request`——幂等键在服务端不是可选项
单 App × 单用户日消耗上限	默认 500 积分	`429 daily_cap_exceeded`，响应体带 `today_spent` 与 `daily_limit`
重复的 `idempotency_key`	返回首次扣费结果	不二次扣费，重试安全
API Key 前缀	`sk-pro-`	不带该前缀直接判为无效
上报所需 OAuth scope	`events:write`	缺 scope 授权失败

SDK 在你不传 idempotency_key 时会自动生成一个——但每次重试都生成新 UUID 等于没有幂等。幂等键要从业务单元派生（例如 f"summarize:{document_id}"），这样超时后的重试会收敛到同一次扣费，而不是给用户扣两次。

失败与对策

现象	HTTP	原因	对策
看起来没问题却报 `invalid_request`	400	缺 `idempotency_key`，或字段超长	两个字段都传；event ≤ 100、key ≤ 128
OAuth 正常但 AI 调用 401	401	用错客户端——AI 调用走 API Key 不走用户 token	AI 用 `client`，计费用 `app_client`
AI 正常但计费 401	401	用户 token 过期且刷新失败	让该用户重新走一次授权
一发版 token 全丢	—	`dict` 存储是进程内的	上多 worker 前先换数据库或 Redis
刷新只成功一次，之后一直失败	—	refresh token 单次有效（轮换），你没保存轮换后的那个	在 `on_refresh` 回调里持久化
`daily_cap_exceeded`	429	该用户触到单 App 日上限（默认 500 积分）	读响应体的 `today_spent` / `daily_limit` 告知用户
`InsufficientBalanceError`	402	用户没积分了	引导充值，别盲目重试
用户为同一个动作被扣了两次	—	每次重试都换了幂等键	幂等键从业务单元派生，不要用 `uuid4()`
平台故障期间全部 502	502	上游失败经 `ProfyApiError` 透出	让请求失败，绝不把这次任务标记成已计费

完整项目结构

profy-fastapi-demo/
├── main.py               # FastAPI 应用入口
├── config.py              # 双客户端初始化
├── token_store.py         # Token 存储
├── dependencies.py        # 路由依赖（Token 校验）
├── .env                   # 环境变量
└── requirements.txt

requirements.txt：

requirements.txt

fastapi>=0.115
uvicorn>=0.34
profy-sdk>=1.0
sqlalchemy>=2.0
aiosqlite>=0.20
python-dotenv>=1.0

启动开发服务器：

uvicorn main:app --reload --port 8000

部署建议

多 Worker 部署

生产环境使用 gunicorn main:app -k uvicorn.workers.UvicornWorker -w 4。多 Worker 下 dict token store 不共享——必须用数据库或 Redis。

环境变量管理

生产环境通过 Kubernetes Secret 或云平台环境变量注入凭证，不要使用 .env 文件。Cookie 需设置 secure=True。

数据库迁移

用 SQLAlchemy + Alembic 管理 Token 表。生产环境不要用 create_all。

反向代理

在 Nginx / Caddy 后面运行，处理 HTTPS 终止。确保 X-Forwarded-Proto 正确传递。

下一步

Next.js 全栈集成

TypeScript 版本的双客户端全栈应用

按次计费实战

ProfyApp reportEvent 按次扣费

Token 管理最佳实践

并发刷新、安全存储、降级策略

SDK 完整指南

双语言 API 详解

开始使用

认证与密钥

API 能力指南

SDK

集成

Cookbook

参考

Python FastAPI 全栈集成

你将构建什么

前置条件

袋袋开发者账号

开发环境

Step 1: 项目初始化

Step 2: 初始化双客户端

Step 3: Token 存储

Step 4: OAuth 登录流程

Step 5: 保护路由

Step 6: 上报计费事件

Step 7: 调用 AI

Step 8: 错误处理中间件

值得写进代码的契约限制

失败与对策

完整项目结构

部署建议

多 Worker 部署

环境变量管理

数据库迁移

反向代理

下一步

Next.js 全栈集成

按次计费实战

Token 管理最佳实践

SDK 完整指南

​你将构建什么

​前置条件

袋袋开发者账号

开发环境

​Step 1: 项目初始化

​Step 2: 初始化双客户端

​Step 3: Token 存储

​Step 4: OAuth 登录流程

​Step 5: 保护路由

​Step 6: 上报计费事件

​Step 7: 调用 AI

​Step 8: 错误处理中间件

​值得写进代码的契约限制

​失败与对策

​完整项目结构

​部署建议

多 Worker 部署

环境变量管理

数据库迁移

反向代理

​下一步

Next.js 全栈集成

按次计费实战

Token 管理最佳实践

SDK 完整指南

你将构建什么

前置条件

Step 1: 项目初始化

Step 2: 初始化双客户端

Step 3: Token 存储

Step 4: OAuth 登录流程

Step 5: 保护路由

Step 6: 上报计费事件

Step 7: 调用 AI

Step 8: 错误处理中间件

值得写进代码的契约限制

失败与对策

完整项目结构

部署建议

下一步