Python FastAPI Full-Stack Integration

What You’ll Build

This tutorial walks you through building a FastAPI backend application using the dual-client architecture:

Profy (API Key) — Your app’s own API key for calling experts and AI models
ProfyApp (OAuth) — Users’ OAuth tokens for reporting billing events to the Profy platform

Prerequisites

Profy Developer Account

A registered Profy account with an App created in the Developer Console — you’ll need the Client ID, Client Secret, and API Key

Development Environment

Python 3.10+, pip, basic FastAPI and async/await experience

Step 1: Project Setup

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

Create a .env file for your credentials:

.env

# Profy client (API Key) — for calling experts and AI models
PROFY_API_KEY=sk-pro-your-api-key

# ProfyApp client (OAuth) — for user authorization and billing
PROFY_APP_ID=your_app_client_id
PROFY_APP_SECRET=your_app_client_secret
PROFY_CALLBACK_URL=http://localhost:8000/callback

Never commit .env to version control. Make sure .gitignore includes this file.

Step 2: Initialize Dual Clients

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"]

Each client serves a distinct purpose: client (Profy) holds your API Key for AI calls; app_client (ProfyApp) holds OAuth credentials for user authorization and billing.

Step 3: Token Storage

Manual token storage implementation. Use a database in production — here we use a dict to demonstrate the core pattern:

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

Production deployments must use persistent storage (database / Redis). The dict loses all tokens on process restart. See Token Management Best Practices.

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: Protected Routes

Use FastAPI dependency injection to automatically load and validate tokens:

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="Not logged in or token expired, please re-authorize")

Step 6: Report Billing Events

main.py (continued)

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": "Session expired, please re-authorize"},
        )
    except InsufficientBalanceError:
        return JSONResponse(
            status_code=402,
            content={"error": "Insufficient balance, please top up"},
        )

The on_refresh callback fires when the token is auto-refreshed. Because Profy’s refresh tokens are single-use (rotation), you must persist the new token in the callback.

Step 7: Call AI

Use the Profy (API Key) client to call experts or AI models — this is independent of the user’s OAuth token:

main.py (continued)

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

main.py (continued)

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": "Authentication failed, check your API Key or re-authorize"})


@app.exception_handler(InsufficientBalanceError)
async def handle_balance_error(request: Request, exc: InsufficientBalanceError):
    return JSONResponse(status_code=402, content={"error": "Insufficient balance, please top up"})


@app.exception_handler(ProfyApiError)
async def handle_profy_error(request: Request, exc: ProfyApiError):
    return JSONResponse(status_code=502, content={"error": f"Profy service error: {exc}"})

Contract limits worth coding against

The billing endpoint enforces these server-side. Validate before you call, so a typo doesn’t surface as a 400 in production:

Item	Value	What happens past it
`event` name length	100 characters	`400 invalid_request`
`idempotency_key` length	128 characters	`400 invalid_request`
Required fields	`event` + `idempotency_key`	`400 invalid_request` — the key is not optional server-side
Per-app per-user daily spend cap	500 credits by default	`429 daily_cap_exceeded`, with `today_spent` and `daily_limit` in the body
Duplicate `idempotency_key`	Returns the original charge	No second deduction — safe to retry
API key prefix	`sk-pro-`	A key without it is rejected as invalid
OAuth scope for reporting	`events:write`	Missing scope fails authorization

The SDK will generate an idempotency_key for you if you omit it — but a fresh UUID per attempt defeats the purpose. Derive the key from the unit of work (e.g. f"summarize:{document_id}"), so a retry after a timeout collapses into the same charge instead of double-billing your user.

Failures and fixes

Symptom	HTTP	Cause	Fix
`invalid_request` on a valid-looking call	400	Missing `idempotency_key`, or a field past its length cap	Send both fields; keep event ≤ 100 and key ≤ 128 chars
`AuthenticationError` on AI calls but OAuth works	401	Wrong client — AI calls use the API key, not the user token	Use `client` (Profy) for AI, `app_client` (ProfyApp) for billing
`AuthenticationError` on billing but AI works	401	The user’s token expired and refresh failed	Re-run the OAuth flow for that user
Tokens vanish after deploy	—	The `dict` store is per-process	Move to a database or Redis before running multiple workers
Refresh works once, then always fails	—	Refresh tokens are single-use (rotation), and you didn’t persist the rotated one	Persist inside the `on_refresh` callback
`daily_cap_exceeded`	429	This user hit the per-app daily cap (500 credits default)	Read `today_spent` / `daily_limit` from the body and tell the user
`InsufficientBalanceError`	402	The user has no credits	Prompt a top-up; don’t retry blindly
A user got billed twice for one action	—	A new idempotency key on each retry	Derive the key from the work unit, not from `uuid4()`
Everything 502s after a Profy incident	502	Upstream failure surfaced through `ProfyApiError`	Fail the request; never mark the work as billed

Complete Project Structure

profy-fastapi-demo/
├── main.py               # FastAPI application entry point
├── config.py              # Dual client initialization
├── token_store.py         # Token storage
├── dependencies.py        # Route dependencies (token validation)
├── .env                   # Environment variables
└── 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

Start the development server:

uvicorn main:app --reload --port 8000

Deployment Tips

Multi-Worker Deployment

In production, use gunicorn main:app -k uvicorn.workers.UvicornWorker -w 4. With multiple workers, the dict token store isn’t shared — you must use a database or Redis.

Environment Variables

In production, inject credentials via Kubernetes Secrets or cloud platform environment variables instead of .env files. Set secure=True on cookies.

Database Migrations

Use SQLAlchemy + Alembic to manage the token table. Don’t use create_all in production.

Reverse Proxy

Run behind Nginx / Caddy for HTTPS termination. Ensure X-Forwarded-Proto is correctly forwarded.

Next Steps

Next.js Full-Stack Integration

TypeScript version of the dual-client full-stack app

Per-Use Billing

ProfyApp reportEvent per-call billing

Token Management Best Practices

Concurrent refresh, secure storage, degradation strategies

Full SDK Guide

Complete bilingual API reference

Get Started

Authentication & Keys

API Guides

SDK

Integrations

Cookbook

Reference

Python FastAPI Full-Stack Integration

What You’ll Build

Prerequisites

Profy Developer Account

Development Environment

Step 1: Project Setup

Step 2: Initialize Dual Clients

Step 3: Token Storage

Step 5: Protected Routes

Step 6: Report Billing Events

Step 7: Call AI

Step 8: Error Handling Middleware

Contract limits worth coding against

Failures and fixes

Complete Project Structure

Deployment Tips

Multi-Worker Deployment

Environment Variables

Database Migrations

Reverse Proxy

Next Steps

Next.js Full-Stack Integration

Per-Use Billing

Token Management Best Practices

Full SDK Guide

​What You’ll Build

​Prerequisites

Profy Developer Account

Development Environment

​Step 1: Project Setup

​Step 2: Initialize Dual Clients

​Step 3: Token Storage

​Step 4: OAuth Login Flow

​Step 5: Protected Routes

​Step 6: Report Billing Events

​Step 7: Call AI

​Step 8: Error Handling Middleware

​Contract limits worth coding against

​Failures and fixes

​Complete Project Structure

​Deployment Tips

Multi-Worker Deployment

Environment Variables

Database Migrations

Reverse Proxy

​Next Steps

Next.js Full-Stack Integration

Per-Use Billing

Token Management Best Practices

Full SDK Guide

What You’ll Build

Prerequisites

Step 1: Project Setup

Step 2: Initialize Dual Clients

Step 3: Token Storage

Step 4: OAuth Login Flow

Step 5: Protected Routes

Step 6: Report Billing Events

Step 7: Call AI

Step 8: Error Handling Middleware

Contract limits worth coding against

Failures and fixes

Complete Project Structure

Deployment Tips

Next Steps