构建按 Token 计费的 AI 工具

你将构建什么

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

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

METERED vs PER_USE

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

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

前置条件

已有 Profy 开发者账号，且创作者审核已通过
在 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 后，直接调用 Profy 的 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 是用户级别的凭证，Profy 会根据这个 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

Profy 的聊天端点兼容 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 时不要重试——用户余额不足不会因为重试而改变。向用户展示充值入口。

可用模型

Profy 平台管理员配置了哪些模型可用。你的 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 计费教程

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

Expert 调用

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

Events API 参考

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

应用上架市场

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

​构建按 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 计费教程

Expert 调用

Events API 参考

应用上架市场

构建按 Token 计费的 AI 工具

你将构建什么

METERED vs PER_USE

前置条件

Step 1: OAuth 授权

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

Step 3: 流式响应

Step 4: 集成 OpenAI SDK

Step 5: Token 过期自动处理

Step 6: 错误处理与重试

可用模型

完整示例

下一步