> ## Documentation Index
> Fetch the complete documentation index at: https://docs.profy.cn/llms.txt
> Use this file to discover all available pages before exploring further.

# 构建按 Token 计费的 AI 工具

> 使用 Profy Events API 的 METERED 模式调用 AI 模型，按实际 token 消耗自动扣费，无需自建计费系统。

# 构建按 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 场景                 | 明确单次操作                |

<Tip>
  本教程使用 METERED 模式。如果你的场景是固定价格的单次操作，参考 [SDK 快速开始](/zh/sdk/quickstart) 中的 `reportEvent()` 用法。
</Tip>

## 前置条件

1. 已有 Profy 开发者账号，且创作者审核已通过
2. 在 [Studio](https://app.profy.cn/studio) 创建了一个 App，计费类型选择 **METERED**
3. 获取 App 的 `clientId` 和 `clientSecret`
4. 配置了 OAuth 回调地址

## Step 1: OAuth 授权

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

<CodeGroup>
  ```typescript TypeScript theme={null}
  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 天有效，使用后轮换
  ```

  ```python Python theme={null}
  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()
  ```
</CodeGroup>

<Note>
  OAuth 完整流程（授权页跳转、回调处理、Token 存储）参考 [SDK 快速开始](/zh/sdk/quickstart)。
</Note>

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

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

<CodeGroup>
  ```typescript TypeScript theme={null}
  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;
  }
  ```

  ```python Python theme={null}
  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"]
  ```
</CodeGroup>

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

## Step 3: 流式响应

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

<CodeGroup>
  ```typescript TypeScript theme={null}
  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);
  }
  ```

  ```python Python theme={null}
  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)
  ```
</CodeGroup>

## Step 4: 集成 OpenAI SDK

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

<CodeGroup>
  ```typescript TypeScript theme={null}
  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);
  }
  ```

  ```python Python theme={null}
  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)
  ```
</CodeGroup>

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

## Step 5: Token 过期自动处理

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

<CodeGroup>
  ```typescript TypeScript theme={null}
  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();
    }
  }
  ```

  ```python Python theme={null}
  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()
  ```
</CodeGroup>

## Step 6: 错误处理与重试

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

<CodeGroup>
  ```typescript TypeScript theme={null}
  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;
      }
    }
  }
  ```

  ```python Python theme={null}
  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
  ```
</CodeGroup>

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

## 可用模型

Profy 平台管理员配置了哪些模型可用。你的 App 可以调用的模型取决于平台配置。

查询方式：

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

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

## 完整示例

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

<CodeGroup>
  ```typescript server.ts (Express) theme={null}
  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);
  ```

  ```python server.py (FastAPI) theme={null}
  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")
  ```
</CodeGroup>

## 下一步

<CardGroup cols={2}>
  <Card title="PER_USE 计费教程" icon="coins" href="/zh/sdk/cookbook/per-use-billing">
    固定价格按次扣费，适合导出报告等确定性操作
  </Card>

  <Card title="Expert 调用" icon="robot" href="/zh/sdk/cookbook/expert-invoke">
    调用平台上已发布的 AI Expert，获取 SSE 流式回复
  </Card>

  <Card title="Events API 参考" icon="code" href="/zh/api/post-chat">
    AI 模型调用端点完整字段说明
  </Card>

  <Card title="应用上架市场" icon="store" href="/zh/documentation/integration-quickstart">
    完成开发后，将你的 App 提交到 Profy 市场
  </Card>
</CardGroup>