构建按 Token 计费的 AI 工具

你将构建什么

一个 AI 写作助手：用户输入主题，AI 生成文章草稿。每次调用按实际 token 消耗从用户积分中扣费——你不需要管理 API Key，不需要搭建计费系统，袋袋平台全部代理。最终效果：

用户通过 OAuth 授权你的 App
App 调用袋袋的 OpenAI 兼容端点生成内容
袋袋自动统计 token 用量并从用户积分扣费
创作者（你）按分成比例获得钻石收益

METERED vs PER_USE

维度	METERED（按量）	PER_USE（按次）
计费单位	实际 token 消耗	每次调用固定价格
典型场景	对话、文章生成、翻译	导出报告、图片处理
调用方式	`POST /openapi/v1/events/chat`	`profy.reportEvent()`
价格可预测性	按量浮动	完全固定
适合	输出长度不确定的 AI 场景	明确单次操作

本教程使用 METERED 模式。如果你的场景是固定价格的单次操作，参考 SDK 快速开始中的 reportEvent() 用法。

前置条件

已有袋袋开发者账号，且创作者审核已通过
在 Studio 创建了一个 App，计费类型选择 METERED
获取 App 的 clientId 和 clientSecret
配置了 OAuth 回调地址

Step 1: OAuth 授权

通过 OAuth 获取用户的 Access Token，后续所有 AI 调用都通过这个 Token 鉴权和计费。

import { ProfyApp } from "@profy-ai/sdk";

const profy = new ProfyApp({
  clientId: process.env.PROFY_APP_ID!,
  clientSecret: process.env.PROFY_APP_SECRET!,
});

const token = await profy.exchangeCode(code, redirectUri);
// token.accessToken — 1 小时有效
// token.refreshToken — 90 天有效，使用后轮换

import httpx

async def exchange_code(code: str, redirect_uri: str) -> dict:
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            "https://app.profy.cn/oauth/token",
            json={
                "grant_type": "authorization_code",
                "code": code,
                "redirect_uri": redirect_uri,
                "client_id": PROFY_APP_ID,
                "client_secret": PROFY_APP_SECRET,
            },
        )
        resp.raise_for_status()
        return resp.json()

OAuth 完整流程（授权页跳转、回调处理、Token 存储）参考 SDK 快速开始。

Step 2: 调用 AI 模型（非流式）

拿到 Access Token 后，直接调用袋袋的 OpenAI 兼容端点。请求格式与 OpenAI /v1/chat/completions 完全一致。

const PROFY_CHAT_URL = "https://app.profy.cn/openapi/v1/events/chat";

async function chat(accessToken: string, prompt: string) {
  const res = await fetch(PROFY_CHAT_URL, {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
      Authorization: `Bearer ${accessToken}`,
    },
    body: JSON.stringify({
      model: "deepseek-chat",
      messages: [
        { role: "system", content: "你是一个专业的写作助手。" },
        { role: "user", content: prompt },
      ],
      temperature: 0.7,
      max_tokens: 2000,
    }),
  });

  if (!res.ok) {
    throw new Error(`Chat failed: ${res.status} ${await res.text()}`);
  }

  const data = await res.json();
  return data.choices[0].message.content;
}

import httpx

PROFY_CHAT_URL = "https://app.profy.cn/openapi/v1/events/chat"

async def chat(access_token: str, prompt: str) -> str:
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            PROFY_CHAT_URL,
            headers={"Authorization": f"Bearer {access_token}"},
            json={
                "model": "deepseek-chat",
                "messages": [
                    {"role": "system", "content": "你是一个专业的写作助手。"},
                    {"role": "user", "content": prompt},
                ],
                "temperature": 0.7,
                "max_tokens": 2000,
            },
            timeout=60.0,
        )
        resp.raise_for_status()
        return resp.json()["choices"][0]["message"]["content"]

Access Token 是用户级别的凭证，袋袋会根据这个 Token 识别用户并从其积分中扣费。不要将不同用户的 Token 混用。

Step 3: 流式响应

设置 stream: true 即可获取 SSE 流式响应，适合实时显示 AI 输出。

async function* chatStream(accessToken: string, prompt: string) {
  const res = await fetch(PROFY_CHAT_URL, {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
      Authorization: `Bearer ${accessToken}`,
    },
    body: JSON.stringify({
      model: "deepseek-chat",
      messages: [
        { role: "system", content: "你是一个专业的写作助手。" },
        { role: "user", content: prompt },
      ],
      stream: true,
      temperature: 0.7,
      max_tokens: 2000,
    }),
  });

  if (!res.ok) {
    throw new Error(`Chat failed: ${res.status} ${await res.text()}`);
  }

  const reader = res.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 });
    const lines = buffer.split("\n");
    buffer = lines.pop()!;

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

      const chunk = JSON.parse(data);
      const content = chunk.choices[0]?.delta?.content;
      if (content) yield content;
    }
  }
}

// 使用
for await (const text of chatStream(token.accessToken, "写一篇关于 AI 的文章")) {
  process.stdout.write(text);
}

import httpx
from collections.abc import AsyncIterator

async def chat_stream(access_token: str, prompt: str) -> AsyncIterator[str]:
    async with httpx.AsyncClient() as client:
        async with client.stream(
            "POST",
            PROFY_CHAT_URL,
            headers={"Authorization": f"Bearer {access_token}"},
            json={
                "model": "deepseek-chat",
                "messages": [
                    {"role": "system", "content": "你是一个专业的写作助手。"},
                    {"role": "user", "content": prompt},
                ],
                "stream": True,
                "temperature": 0.7,
                "max_tokens": 2000,
            },
            timeout=60.0,
        ) as resp:
            resp.raise_for_status()
            async for line in resp.aiter_lines():
                if not line.startswith("data: "):
                    continue
                data = line[6:]
                if data == "[DONE]":
                    return
                import json
                chunk = json.loads(data)
                content = chunk["choices"][0].get("delta", {}).get("content")
                if content:
                    yield content

# 使用
async for text in chat_stream(access_token, "写一篇关于 AI 的文章"):
    print(text, end="", flush=True)

Step 4: 集成 OpenAI SDK

袋袋的聊天端点兼容 OpenAI 协议，可以直接使用 OpenAI 官方 SDK，只需修改 baseURL 和 apiKey。

import OpenAI from "openai";

const openai = new OpenAI({
  apiKey: token.accessToken,
  baseURL: "https://app.profy.cn/openapi/v1/events",
});

// 非流式
const completion = await openai.chat.completions.create({
  model: "deepseek-chat",
  messages: [
    { role: "system", content: "你是一个专业的写作助手。" },
    { role: "user", content: "写一段产品描述" },
  ],
  temperature: 0.7,
  max_tokens: 2000,
});

console.log(completion.choices[0].message.content);

// 流式
const stream = await openai.chat.completions.create({
  model: "deepseek-chat",
  messages: [{ role: "user", content: "写一篇短文" }],
  stream: true,
});

for await (const chunk of stream) {
  const content = chunk.choices[0]?.delta?.content;
  if (content) process.stdout.write(content);
}

from openai import AsyncOpenAI

client = AsyncOpenAI(
    api_key=access_token,
    base_url="https://app.profy.cn/openapi/v1/events",
)

# 非流式
completion = await client.chat.completions.create(
    model="deepseek-chat",
    messages=[
        {"role": "system", "content": "你是一个专业的写作助手。"},
        {"role": "user", "content": "写一段产品描述"},
    ],
    temperature=0.7,
    max_tokens=2000,
)

print(completion.choices[0].message.content)

# 流式
stream = await client.chat.completions.create(
    model="deepseek-chat",
    messages=[{"role": "user", "content": "写一篇短文"}],
    stream=True,
)

async for chunk in stream:
    content = chunk.choices[0].delta.content
    if content:
        print(content, end="", flush=True)

使用 OpenAI SDK 可以复用其完善的类型定义、自动重试和流式处理能力。推荐在生产环境中使用这种方式。

Step 5: Token 过期自动处理

Access Token 有效期 1 小时。封装一个自动刷新的调用函数，避免每次手动检查。

import { ProfyApp } from "@profy-ai/sdk";

interface TokenPair {
  accessToken: string;
  refreshToken: string;
  expiresAt: number;
}

class ProfyChat {
  private profy: ProfyApp;
  private token: TokenPair;

  constructor(profy: ProfyApp, token: TokenPair) {
    this.profy = profy;
    this.token = token;
  }

  private async getValidToken(): Promise<string> {
    if (Date.now() >= this.token.expiresAt - 60_000) {
      this.token = await this.profy.refreshToken(this.token.refreshToken);
    }
    return this.token.accessToken;
  }

  async chat(messages: Array<{ role: string; content: string }>) {
    const accessToken = await this.getValidToken();

    const res = await fetch(PROFY_CHAT_URL, {
      method: "POST",
      headers: {
        "Content-Type": "application/json",
        Authorization: `Bearer ${accessToken}`,
      },
      body: JSON.stringify({ model: "deepseek-chat", messages }),
    });

    if (res.status === 401) {
      this.token = await this.profy.refreshToken(this.token.refreshToken);
      return this.chat(messages);
    }

    if (!res.ok) throw new Error(`Chat failed: ${res.status}`);
    return res.json();
  }
}

import time
import httpx

class ProfyChat:
    def __init__(self, token: dict, client_id: str, client_secret: str):
        self.token = token
        self.client_id = client_id
        self.client_secret = client_secret

    async def _refresh_if_needed(self):
        if time.time() * 1000 >= self.token["expires_at"] - 60_000:
            async with httpx.AsyncClient() as client:
                resp = await client.post(
                    "https://app.profy.cn/oauth/token",
                    json={
                        "grant_type": "refresh_token",
                        "refresh_token": self.token["refresh_token"],
                        "client_id": self.client_id,
                        "client_secret": self.client_secret,
                    },
                )
                resp.raise_for_status()
                self.token = resp.json()

    async def chat(self, messages: list[dict]) -> dict:
        await self._refresh_if_needed()

        async with httpx.AsyncClient() as client:
            resp = await client.post(
                PROFY_CHAT_URL,
                headers={"Authorization": f"Bearer {self.token['access_token']}"},
                json={"model": "deepseek-chat", "messages": messages},
                timeout=60.0,
            )

            if resp.status_code == 401:
                await self._refresh_if_needed()
                return await self.chat(messages)

            resp.raise_for_status()
            return resp.json()

Step 6: 错误处理与重试

HTTP 状态码	含义	处理方式
400	请求参数错误或模型不可用	检查 `model` 名称和请求体格式
401	Token 过期或无效	用 Refresh Token 续期后重试
402	用户积分不足	提示用户充值，不要重试
502	上游模型服务异常	指数退避重试（最多 3 次）

async function chatWithRetry(
  profyChat: ProfyChat,
  messages: Array<{ role: string; content: string }>,
  maxRetries = 3,
) {
  for (let attempt = 0; attempt <= maxRetries; attempt++) {
    try {
      return await profyChat.chat(messages);
    } catch (err: any) {
      const status = err.status ?? err.statusCode;

      if (status === 402) {
        throw new Error("用户积分不足，请充值后重试");
      }

      if (status === 502 && attempt < maxRetries) {
        const delay = Math.min(1000 * 2 ** attempt, 10_000);
        await new Promise((r) => setTimeout(r, delay));
        continue;
      }

      throw err;
    }
  }
}

import asyncio

async def chat_with_retry(
    profy_chat: ProfyChat,
    messages: list[dict],
    max_retries: int = 3,
) -> dict:
    for attempt in range(max_retries + 1):
        try:
            return await profy_chat.chat(messages)
        except httpx.HTTPStatusError as exc:
            if exc.response.status_code == 402:
                raise ValueError("用户积分不足，请充值后重试") from exc

            if exc.response.status_code == 502 and attempt < max_retries:
                delay = min(1.0 * 2**attempt, 10.0)
                await asyncio.sleep(delay)
                continue

            raise

收到 402 时不要重试——用户余额不足不会因为重试而改变。向用户展示充值入口。

可用模型

袋袋平台管理员配置了哪些模型可用。你的 App 可以调用的模型取决于平台配置。查询方式：

API: GET /openapi/v1/meters 返回当前可用的 Meter 配置
Studio: 在 App 设置页面的「计费配置」中查看

常见模型包括 deepseek-chat、deepseek-reasoner、qwen-plus 等。具体可用列表以平台配置为准。

完整示例

一个 AI 写作助手后端，整合了 OAuth、Token 刷新、流式输出和错误处理。

import express from "express";
import OpenAI from "openai";
import { ProfyApp } from "@profy-ai/sdk";

const app = express();
app.use(express.json());

const profy = new ProfyApp({
  clientId: process.env.PROFY_APP_ID!,
  clientSecret: process.env.PROFY_APP_SECRET!,
  onTokenRefresh: (newToken) => {
    // 持久化到数据库
  },
});

const tokenStore = new Map<string, any>();

app.get("/auth/callback", async (req, res) => {
  const { code } = req.query;
  const token = await profy.exchangeCode(code as string, process.env.REDIRECT_URI!);
  const userId = "user-from-session";
  tokenStore.set(userId, token);
  res.redirect("/chat");
});

app.post("/api/chat", async (req, res) => {
  const userId = "user-from-session";
  let token = tokenStore.get(userId);
  if (!token) return res.status(401).json({ error: "未授权" });

  if (Date.now() >= token.expiresAt - 60_000) {
    token = await profy.refreshToken(token.refreshToken);
    tokenStore.set(userId, token);
  }

  const openai = new OpenAI({
    apiKey: token.accessToken,
    baseURL: "https://app.profy.cn/openapi/v1/events",
  });

  try {
    const stream = await openai.chat.completions.create({
      model: "deepseek-chat",
      messages: [
        { role: "system", content: "你是一个专业的写作助手。" },
        ...req.body.messages,
      ],
      stream: true,
      temperature: 0.7,
      max_tokens: 2000,
    });

    res.setHeader("Content-Type", "text/event-stream");
    res.setHeader("Cache-Control", "no-cache");

    for await (const chunk of stream) {
      const content = chunk.choices[0]?.delta?.content;
      if (content) {
        res.write(`data: ${JSON.stringify({ content })}\n\n`);
      }
    }

    res.write("data: [DONE]\n\n");
    res.end();
  } catch (err: any) {
    const status = err.status ?? 500;
    if (status === 402) {
      return res.status(402).json({ error: "积分不足，请充值" });
    }
    return res.status(status).json({ error: err.message });
  }
});

app.listen(3000);

import os
import time

import httpx
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse, RedirectResponse, StreamingResponse
from openai import AsyncOpenAI

app = FastAPI()

PROFY_BASE = os.environ["PROFY_BASE_URL"]  # https://app.profy.cn
CLIENT_ID = os.environ["PROFY_APP_ID"]
CLIENT_SECRET = os.environ["PROFY_APP_SECRET"]
REDIRECT_URI = os.environ["PROFY_CALLBACK_URL"]

token_store: dict[str, dict] = {}


async def exchange_code(code: str) -> dict:
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            f"{PROFY_BASE}/oauth/token",
            json={
                "grant_type": "authorization_code",
                "code": code,
                "redirect_uri": REDIRECT_URI,
                "client_id": CLIENT_ID,
                "client_secret": CLIENT_SECRET,
            },
        )
        resp.raise_for_status()
        return resp.json()


async def refresh_if_needed(user_id: str) -> str:
    token = token_store[user_id]
    if time.time() >= token["expires_at"] - 60:
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                f"{PROFY_BASE}/oauth/token",
                json={
                    "grant_type": "refresh_token",
                    "refresh_token": token["refresh_token"],
                    "client_id": CLIENT_ID,
                    "client_secret": CLIENT_SECRET,
                },
            )
            resp.raise_for_status()
            token_store[user_id] = resp.json()
    return token_store[user_id]["access_token"]


@app.get("/auth/callback")
async def callback(code: str):
    token = await exchange_code(code)
    user_id = "user-from-session"
    token_store[user_id] = token
    return RedirectResponse("/chat")


@app.post("/api/chat")
async def chat(request: Request):
    user_id = "user-from-session"
    if user_id not in token_store:
        return JSONResponse({"error": "未授权"}, status_code=401)

    access_token = await refresh_if_needed(user_id)

    body = await request.json()

    client = AsyncOpenAI(
        api_key=access_token,
        base_url=f"{PROFY_BASE}/openapi/v1/events",
    )

    async def stream_response():
        try:
            stream = await client.chat.completions.create(
                model="deepseek-chat",
                messages=[
                    {"role": "system", "content": "你是一个专业的写作助手。"},
                    *body["messages"],
                ],
                stream=True,
                temperature=0.7,
                max_tokens=2000,
            )
            async for chunk in stream:
                content = chunk.choices[0].delta.content
                if content:
                    yield f"data: {{'content': '{content}'}}\n\n"
            yield "data: [DONE]\n\n"
        except Exception as exc:
            yield f"data: {{'error': '{exc}'}}\n\n"

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

下一步

PER_USE 计费教程

固定价格按次扣费，适合导出报告等确定性操作

专家调用

调用平台上已发布的 AI 专家，获取 SSE 流式回复

Events API 参考

AI 模型调用端点完整字段说明

应用上架市场

完成开发后，将你的 App 提交到袋袋市场

开始使用

认证与密钥

API 能力指南

SDK

集成

Cookbook

参考

构建按 Token 计费的 AI 工具

你将构建什么

METERED vs PER_USE

前置条件

Step 1: OAuth 授权

Step 2: 调用 AI 模型（非流式）

Step 3: 流式响应

Step 4: 集成 OpenAI SDK

Step 5: Token 过期自动处理

Step 6: 错误处理与重试

可用模型

完整示例

下一步

PER_USE 计费教程

专家调用

Events API 参考

应用上架市场

​你将构建什么

​METERED vs PER_USE

​前置条件

​Step 1: OAuth 授权

​Step 2: 调用 AI 模型（非流式）

​Step 3: 流式响应

​Step 4: 集成 OpenAI SDK

​Step 5: Token 过期自动处理

​Step 6: 错误处理与重试

​可用模型

​完整示例

​下一步

PER_USE 计费教程

专家调用

Events API 参考

应用上架市场

你将构建什么

METERED vs PER_USE

前置条件

Step 1: OAuth 授权

Step 2: 调用 AI 模型（非流式）

Step 3: 流式响应

Step 4: 集成 OpenAI SDK

Step 5: Token 过期自动处理

Step 6: 错误处理与重试

可用模型

完整示例

下一步