Building a Token-Metered AI Tool

What You’ll Build

An AI writing assistant: users enter a topic, and AI generates a draft article. Each call is billed based on actual token consumption from the user’s credits — you don’t need to manage API keys or build a billing system; the Profy platform handles everything. The end result:

Users authorize your App via OAuth
Your App calls Profy’s OpenAI-compatible endpoint to generate content
Profy automatically tracks token usage and deducts from the user’s credits
You (the creator) earn diamond revenue based on your revenue share ratio

METERED vs PER_USE

Dimension	METERED (Usage-Based)	PER_USE (Per-Call)
Billing Unit	Actual token consumption	Fixed price per call
Typical Scenarios	Conversations, article generation, translation	Report export, image processing
API Call	`POST /openapi/v1/events/chat`	`profy.reportEvent()`
Price Predictability	Variable based on usage	Completely fixed
Best For	AI scenarios with unpredictable output length	Well-defined single operations

This tutorial uses METERED mode. If your scenario involves fixed-price single operations, refer to the reportEvent() usage in the SDK Quick Start.

Prerequisites

Have a Profy developer account with approved creator verification
Created an App in Studio with billing type set to METERED
Obtained the App’s clientId and clientSecret
Configured an OAuth callback URL

Step 1: OAuth Authorization

Obtain the user’s Access Token via OAuth. All subsequent AI calls are authenticated and billed through this 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 — valid for 1 hour
// token.refreshToken — valid for 90 days, rotated on use

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()

For the complete OAuth flow (authorization page redirect, callback handling, Token storage), refer to the SDK Quick Start.

Step 2: Call AI Models (Non-Streaming)

With the Access Token in hand, call Profy’s OpenAI-compatible endpoint directly. The request format is identical to OpenAI’s /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: "You are a professional writing assistant." },
        { 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": "You are a professional writing assistant."},
                    {"role": "user", "content": prompt},
                ],
                "temperature": 0.7,
                "max_tokens": 2000,
            },
            timeout=60.0,
        )
        resp.raise_for_status()
        return resp.json()["choices"][0]["message"]["content"]

The Access Token is a user-level credential — Profy identifies the user through this Token and deducts from their credits. Never mix Tokens between different users.

Step 3: Streaming Responses

Set stream: true to receive SSE streaming responses, ideal for displaying AI output in real-time.

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: "You are a professional writing assistant." },
        { 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;
    }
  }
}

// Usage
for await (const text of chatStream(token.accessToken, "Write an article about 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": "You are a professional writing assistant."},
                    {"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

# Usage
async for text in chat_stream(access_token, "Write an article about AI"):
    print(text, end="", flush=True)

Step 4: OpenAI SDK Integration

Profy’s chat endpoint is OpenAI-compatible, so you can use the official OpenAI SDK directly — just change the baseURL and apiKey.

import OpenAI from "openai";

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

// Non-streaming
const completion = await openai.chat.completions.create({
  model: "deepseek-chat",
  messages: [
    { role: "system", content: "You are a professional writing assistant." },
    { role: "user", content: "Write a product description" },
  ],
  temperature: 0.7,
  max_tokens: 2000,
});

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

// Streaming
const stream = await openai.chat.completions.create({
  model: "deepseek-chat",
  messages: [{ role: "user", content: "Write a short essay" }],
  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",
)

# Non-streaming
completion = await client.chat.completions.create(
    model="deepseek-chat",
    messages=[
        {"role": "system", "content": "You are a professional writing assistant."},
        {"role": "user", "content": "Write a product description"},
    ],
    temperature=0.7,
    max_tokens=2000,
)

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

# Streaming
stream = await client.chat.completions.create(
    model="deepseek-chat",
    messages=[{"role": "user", "content": "Write a short essay"}],
    stream=True,
)

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

Using the OpenAI SDK gives you well-defined types, automatic retries, and robust streaming support. This approach is recommended for production environments.

Step 5: Automatic Token Expiry Handling

Access Tokens are valid for 1 hour. Wrap calls in an auto-refreshing function to avoid manual checks every time.

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: Error Handling and Retries

HTTP Status	Meaning	Handling
400	Invalid request parameters or model unavailable	Check the `model` name and request body format
401	Token expired or invalid	Refresh with Refresh Token and retry
402	Insufficient user credits	Prompt the user to top up — do not retry
502	Upstream model service error	Exponential backoff retry (max 3 attempts)

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("Insufficient user credits. Please top up and try again.");
      }

      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("Insufficient user credits. Please top up and try again.") 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

Do not retry on 402 — the user’s insufficient balance won’t change with retries. Show a top-up prompt instead.

Available Models

The Profy platform administrator configures which models are available. The models your App can use depend on the platform configuration. How to check:

API: GET /openapi/v1/meters returns the currently available Meter configuration
Studio: Check the “Billing Configuration” in your App settings page

Common models include deepseek-chat, deepseek-reasoner, qwen-plus, and more. The specific list of available models depends on the platform configuration.

Complete Example

An AI writing assistant backend integrating OAuth, Token refresh, streaming output, and error handling.

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) => {
    // Persist to database
  },
});

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: "Unauthorized" });

  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: "You are a professional writing assistant." },
        ...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: "Insufficient credits. Please top up." });
    }
    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": "Unauthorized"}, 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": "You are a professional writing assistant."},
                    *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")

Next Steps

PER_USE Billing Tutorial

Fixed-price per-use billing for deterministic operations like report exports

Expert Invocation

Invoke published AI Experts on the platform and receive SSE streaming responses

Events API Reference

Complete field documentation for the AI model call endpoint

Publish to Marketplace

After development, submit your App to the Profy Marketplace

Get Started

Authentication & Keys

API Guides

SDK

Integrations

Cookbook

Reference

Building a Token-Metered AI Tool

What You’ll Build

METERED vs PER_USE

Prerequisites

Step 1: OAuth Authorization

Step 2: Call AI Models (Non-Streaming)

Step 3: Streaming Responses

Step 4: OpenAI SDK Integration

Step 5: Automatic Token Expiry Handling

Step 6: Error Handling and Retries

Available Models

Complete Example

Next Steps

PER_USE Billing Tutorial

Expert Invocation

Events API Reference

Publish to Marketplace

​What You’ll Build

​METERED vs PER_USE

​Prerequisites

​Step 1: OAuth Authorization

​Step 2: Call AI Models (Non-Streaming)

​Step 3: Streaming Responses

​Step 4: OpenAI SDK Integration

​Step 5: Automatic Token Expiry Handling

​Step 6: Error Handling and Retries

​Available Models

​Complete Example

​Next Steps

PER_USE Billing Tutorial

Expert Invocation

Events API Reference

Publish to Marketplace

What You’ll Build

METERED vs PER_USE

Prerequisites

Step 1: OAuth Authorization

Step 2: Call AI Models (Non-Streaming)

Step 3: Streaming Responses

Step 4: OpenAI SDK Integration

Step 5: Automatic Token Expiry Handling

Step 6: Error Handling and Retries

Available Models

Complete Example

Next Steps