# OnlyFans Python SDK

> A typed Python client for the OnlyFans API. Call any endpoint from a script, a FastAPI service or a Django app, with sync and async clients and automatic retries.

- Requires Python 3.9+
- Install: `pip install onlyfans`, import: `from onlyfansapi import OnlyFansAPI`
- Source: https://github.com/onlyfansapi/onlyfansapi-python
- Package: `onlyfans` on PyPI: https://pypi.org/project/onlyfans/ (Apache 2.0)

## Make your first OnlyFans call in Python

Install the package with pip, add your API key, and list your connected accounts.

The OnlyFans Python SDK is on PyPI as onlyfans and needs Python 3.9 or newer. You install onlyfans but import onlyfansapi: the package name and the module name differ. httpx and Pydantic come with it, so there is nothing else to add.

```bash
pip install onlyfans
# or: uv add onlyfans
# or: poetry add onlyfans
```

Create an API key in the OnlyFansAPI dashboard and store it in the ONLYFANSAPI_API_KEY environment variable: export it in your shell, or keep it in .env and load it with python-dotenv. OnlyFansAPI() reads the variable automatically, so the key stays out of your code and your Git history. You can also pass it with OnlyFansAPI(api_key=...).

```bash
export ONLYFANSAPI_API_KEY=your_api_key

# Python does not read .env files on its own. To keep the key in .env,
# pip install python-dotenv and call load_dotenv() before creating the client.
```

List every OnlyFans account connected to your key. Each account has an acct_ ID, and you pass that ID to every other call: chats, fans, messages, media and statistics.

```python
from onlyfansapi import OnlyFansAPI

# Reads ONLYFANSAPI_API_KEY from the environment
client = OnlyFansAPI()

# GET /api/accounts: every OnlyFans account connected to your API key
accounts = client.accounts.list()

for account in accounts:
    print(f"{account.id} · @{account.onlyfans_username}")

# Output:
# acct_XXXXXXXXXXXXX · @sarah_k
```

## OnlyFans Python SDK code examples

Working calls for the jobs agencies automate first: tracking-link revenue, chats, messages, paid messages and media.

Fetch clicks, subscribers, spenders and lifetime revenue for a tracking link, then work out revenue per subscriber from the summary.

```python
from onlyfansapi import OnlyFansAPI

client = OnlyFansAPI()

# GET /api/{account}/tracking-links/{tracking_link_id}/stats
stats = client.tracking_links.get_stats("123456", account="acct_XXXXXXXXXXXXX")

summary = stats.data.summary if stats.data else None
if summary:
    subs = summary.subs_total or 0
    revenue = summary.revenue_total or 0.0
    arps = revenue / subs if subs else 0.0

    print(f"Clicks: {summary.clicks_total} · Subs: {subs}")
    print(f"Spenders: {summary.spenders_total}")
    print(f"ARPS: ${arps:.2f} · Lifetime: ${revenue:.2f}")

# Output:
# Clicks: 1240 · Subs: 300
# Spenders: 42
# ARPS: $12.40 · Lifetime: $3720.00
```

List the fans who paid after arriving through a tracking link. min_spend drops anyone under your threshold.

```python
from onlyfansapi import OnlyFansAPI

client = OnlyFansAPI()

# GET /api/{account}/tracking-links/{tracking_link_id}/spenders
spenders = client.tracking_links.list_spenders(
    "123456",
    account="acct_XXXXXXXXXXXXX",
    min_spend=10,
    limit=50,
)

for fan in spenders.data or []:
    total = fan.revenue.total if fan.revenue else 0.0
    print(f"@{fan.username} spent ${total:.2f}")

# Output:
# @sarah_k spent $189.00
```

Load unread chats for an account, with the fan, the unread count and the last message in each.

```python
from onlyfansapi import OnlyFansAPI

client = OnlyFansAPI()

# GET /api/{account}/chats
chats = client.chats.list("acct_XXXXXXXXXXXXX", filter="unread", limit="20")

for chat in chats.data or []:
    username = chat.fan.username if chat.fan else None
    text = chat.last_message.text if chat.last_message else ""
    print(f"@{username} · unread: {chat.unread_messages_count}")
    print(f"  {text}")

# Output:
# @sarah_k · unread: 3
#   Hey, are you around?
```

Read the newest messages in one chat, newest first.

```python
from onlyfansapi import OnlyFansAPI

client = OnlyFansAPI()

# GET /api/{account}/chats/{chat_id}/messages
messages = client.chats.messages.list(
    "987654321",
    account="acct_XXXXXXXXXXXXX",
    limit="20",
    order="desc",
)

for message in messages.data or []:
    print(f"[{message.created_at}] {message.text}")

# Output:
# [2026-09-04T18:22:41+00:00] Just posted something new for you 💜
```

Send a text with a media attachment to a fan's chat and read back the new message ID.

```python
from onlyfansapi import OnlyFansAPI

client = OnlyFansAPI()

# POST /api/{account}/chats/{chat_id}/messages
sent = client.chats.messages.send(
    "987654321",
    account="acct_XXXXXXXXXXXXX",
    text="Hey! Thanks for subscribing 💜",
    media_files=["ofapi_media_XXXXXXXXXXXXX"],
)

if sent.data:
    print(f"Message sent #{sent.data.id}")

# Output:
# Message sent #6182934517
```

Upload a photo from disk with a Path, then send it as a paid message for $10. The upload returns an ofapi_media_ ID that works once, in one post or one message. Prices are in USD: 0 for free, or 3 to 200.

```python
from pathlib import Path

from onlyfansapi import OnlyFansAPI

client = OnlyFansAPI()
account = "acct_XXXXXXXXXXXXX"

# POST /api/{account}/media/upload
# The returned ID can be attached to one post or message.
upload = client.media.upload(account=account, file=Path("preview.jpg"))

# POST /api/{account}/chats/{chat_id}/messages
sent = client.chats.messages.send(
    "987654321",
    account=account,
    text="Something new, just for you 💜",
    media_files=[upload.prefixed_id],
    price=10,  # USD, 0 or 3–200
)

if sent.data:
    print(f"PPV sent #{sent.data.id}")
```

Stream a file from an OnlyFans CDN URL to disk through the API's proxy, chunk by chunk, so large videos never sit in memory.

```python
from onlyfansapi import OnlyFansAPI

client = OnlyFansAPI()

cdn_url = "https://cdn2.onlyfans.com/files/…/photo.jpg?Expires=…"

# GET /api/{account}/media/download/{cdn_url}
# Stream the bytes to disk instead of loading the whole file into memory.
with client.media.with_streaming_response.download(
    cdn_url,
    account="acct_XXXXXXXXXXXXX",
) as response:
    with open("photo.jpg", "wb") as file:
        for chunk in response.iter_bytes():
            file.write(chunk)
```

## Features

- **Sync and async**: OnlyFansAPI for scripts, Django and Flask. AsyncOnlyFansAPI for FastAPI and asyncio workers. Same methods, same arguments.
- **Typed models**: Keyword arguments with type hints, and Pydantic models for every response. Your editor autocompletes fields, and mypy or pyright catches typos.
- **Retries and timeouts**: Network errors, timeouts, 429 and 5xx responses are retried twice with backoff. Every request has a 60-second timeout you can change.

## Use the OnlyFans Python SDK with FastAPI

Open one async client when the app starts, inject it into your routes, and hand webhooks to a Celery worker.

Create one AsyncOnlyFansAPI in the lifespan handler so every request shares its connection pool, and the client closes cleanly on shutdown. A Depends alias injects it into any route.

```python
# main.py
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager
from typing import Annotated

from fastapi import Depends, FastAPI, Request
from onlyfansapi import AsyncOnlyFansAPI

from webhooks import router as webhook_router


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
    # One client for the whole app, closed on shutdown
    async with AsyncOnlyFansAPI() as client:
        app.state.onlyfans = client
        yield


app = FastAPI(lifespan=lifespan)
app.include_router(webhook_router)


def get_onlyfans(request: Request) -> AsyncOnlyFansAPI:
    client: AsyncOnlyFansAPI = request.app.state.onlyfans
    return client


OnlyFans = Annotated[AsyncOnlyFansAPI, Depends(get_onlyfans)]


@app.get("/fans/latest")
async def latest_fans(onlyfans: OnlyFans) -> list[str]:
    # GET /api/{account}/fans/latest
    fans = await onlyfans.fans.list_latest("acct_XXXXXXXXXXXXX", limit=10)
    users = (fans.data.users if fans.data else None) or []
    return [user.username for user in users if user.username]
```

OnlyFansAPI signs each webhook with an HMAC SHA256 of the raw body in the Signature header. Read await request.body() before parsing, compare with hmac.compare_digest, then skip events you have seen using X-OFAPI-Idempotency-Key and a Redis SET NX. Deliveries time out after 10 seconds, so queue the work and return 204. If queueing fails, the key is deleted so the retry gets through.

```python
# webhooks.py
import asyncio
import hashlib
import hmac
import os

from fastapi import APIRouter, HTTPException, Request, Response
from redis.asyncio import Redis

from tasks import process_event

router = APIRouter()
redis = Redis.from_url(os.environ["REDIS_URL"])
SECRET = os.environ["ONLYFANSAPI_WEBHOOK_SECRET"].encode()


@router.post("/webhooks/onlyfans", status_code=204)
async def onlyfans_webhook(request: Request) -> Response:
    # Sign the raw bytes, not JSON you parsed and dumped again
    body = await request.body()
    expected = hmac.new(SECRET, body, hashlib.sha256).hexdigest()
    signature = request.headers.get("Signature", "")
    if not hmac.compare_digest(expected.encode(), signature.encode()):
        raise HTTPException(status_code=401)

    # Retries reuse the same key, so each event is queued once.
    key = request.headers.get("X-OFAPI-Idempotency-Key")
    if key and not await redis.set(f"ofapi-webhook:{key}", 1, nx=True, ex=86400):
        return Response(status_code=204)

    try:
        # .delay() talks to the broker, so run it off the event loop
        await asyncio.to_thread(process_event.delay, await request.json())
    except Exception:
        # Queueing failed: forget the key so the retry is not skipped.
        if key:
            await redis.delete(f"ofapi-webhook:{key}")
        raise

    return Response(status_code=204)
```

The Celery task does the slow work after the response is sent. Each delivery has an event name, the account_id it belongs to and a payload. For fan events, payload["fan"] holds the fan's ID and username.

```python
# tasks.py, run with: celery -A tasks worker
import logging
import os
from typing import Any

from celery import Celery

app = Celery("onlyfans", broker=os.environ["REDIS_URL"])
log = logging.getLogger(__name__)


@app.task
def process_event(event: dict[str, Any]) -> None:
    # Every delivery has an event name, the account_id and a payload
    payload = event["payload"]

    if event["event"] == "subscriptions.new":
        log.info("New subscriber @%s", payload["fan"]["username"])
    elif event["event"] == "messages.received":
        log.info("New message on %s", event["account_id"])
```

## Use the OnlyFans Python SDK with Flask

A sync client, one route, and a signed webhook that queues jobs with RQ.

Create the client once at module level, not inside each view. It keeps a connection pool open, so later requests skip the TLS handshake.

```python
# app.py
from flask import Flask, jsonify
from flask.typing import ResponseReturnValue
from onlyfansapi import OnlyFansAPI

from webhooks import webhooks

app = Flask(__name__)
app.register_blueprint(webhooks)

# Create the client once per process; it keeps a connection pool open
onlyfans = OnlyFansAPI()


@app.get("/fans/latest")
def latest_fans() -> ResponseReturnValue:
    # GET /api/{account}/fans/latest
    fans = onlyfans.fans.list_latest("acct_XXXXXXXXXXXXX", limit=10)
    users = (fans.data.users if fans.data else None) or []
    return jsonify([user.username for user in users])
```

request.get_data() returns the raw body, which is what the signature covers. After the check, a Redis SET NX on the idempotency key stops duplicates, and RQ queues the event. A failed enqueue removes the key, so OnlyFansAPI's next attempt is processed.

```python
# webhooks.py
import hashlib
import hmac
import os

from flask import Blueprint, abort, request
from flask.typing import ResponseReturnValue
from redis import Redis
from rq import Queue

from jobs import process_event

webhooks = Blueprint("webhooks", __name__)
redis = Redis.from_url(os.environ["REDIS_URL"])
queue = Queue("onlyfans", connection=redis)
SECRET = os.environ["ONLYFANSAPI_WEBHOOK_SECRET"].encode()


@webhooks.post("/webhooks/onlyfans")
def onlyfans_webhook() -> ResponseReturnValue:
    # get_data() is the raw body, which is what the signature covers
    expected = hmac.new(SECRET, request.get_data(), hashlib.sha256).hexdigest()
    signature = request.headers.get("Signature", "")
    if not hmac.compare_digest(expected.encode(), signature.encode()):
        abort(401)

    # Retries reuse the same key, so each event is queued once.
    key = request.headers.get("X-OFAPI-Idempotency-Key")
    if key and not redis.set(f"ofapi-webhook:{key}", 1, nx=True, ex=86400):
        return "", 204

    try:
        queue.enqueue(process_event, request.get_json())
    except Exception:
        # Queueing failed: forget the key so the retry is not skipped.
        if key:
            redis.delete(f"ofapi-webhook:{key}")
        raise

    return "", 204
```

An RQ job that welcomes new subscribers. OnlyFans uses the fan's user ID as the chat ID, so payload["fan"]["id"] is all you need to message them. Run it with rq worker onlyfans.

```python
# jobs.py, run with: rq worker onlyfans
from typing import Any

from onlyfansapi import OnlyFansAPI


def process_event(event: dict[str, Any]) -> None:
    if event["event"] != "subscriptions.new":
        return

    # Welcome each new subscriber from the account that received the event
    fan = event["payload"]["fan"]
    client = OnlyFansAPI()
    client.chats.messages.send(
        str(fan["id"]),
        account=event["account_id"],
        text=f"Welcome, {fan['name']}! Thanks for subscribing 💜",
    )
```

## Use the OnlyFans Python SDK with Django

Keep the key in settings, share one client across views, and process webhooks with Django's built-in tasks.

Read the API key and signing secret from the environment in settings.py. Django 6 ships a Tasks framework; django-tasks-db stores queued tasks in your database and runs them with manage.py db_worker.

```python
# settings.py
import os

INSTALLED_APPS = [
    # ...
    "django_tasks_db",
    "onlyfans",
]

ONLYFANSAPI_API_KEY = os.environ["ONLYFANSAPI_API_KEY"]
ONLYFANSAPI_WEBHOOK_SECRET = os.environ["ONLYFANSAPI_WEBHOOK_SECRET"]

# Shared store for the webhook duplicate check
CACHES = {
    "default": {
        "BACKEND": "django.core.cache.backends.redis.RedisCache",
        "LOCATION": os.environ["REDIS_URL"],
    }
}

# pip install django-tasks-db, then run: python manage.py db_worker
TASKS = {
    "default": {
        "BACKEND": "django_tasks_db.DatabaseBackend",
        "QUEUES": ["default"],
    }
}
```

get_client() builds the client on first use and returns the same instance after that, so views, tasks and management commands share one connection pool.

```python
# onlyfans/client.py
from functools import lru_cache

from django.conf import settings
from onlyfansapi import OnlyFansAPI


@lru_cache(maxsize=1)
def get_client() -> OnlyFansAPI:
    # Built on first use, then shared by every view and task in the process
    return OnlyFansAPI(api_key=settings.ONLYFANSAPI_API_KEY)
```

A plain function view that returns the latest fans as JSON.

```python
# onlyfans/views.py
from django.http import HttpRequest, JsonResponse

from .client import get_client


def latest_fans(request: HttpRequest) -> JsonResponse:
    # GET /api/{account}/fans/latest
    fans = get_client().fans.list_latest("acct_XXXXXXXXXXXXX", limit=10)
    users = (fans.data.users if fans.data else None) or []
    return JsonResponse({"fans": [user.username for user in users]})
```

request.body holds the raw bytes for the signature check. cache.add only succeeds the first time it sees a key, so a retried delivery is answered with 204 and not queued again. Mark the view csrf_exempt: OnlyFansAPI cannot send a CSRF token.

```python
# onlyfans/webhooks.py
import hashlib
import hmac
import json

from django.conf import settings
from django.core.cache import cache
from django.http import HttpRequest, HttpResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST

from .tasks import process_event


@csrf_exempt
@require_POST
def onlyfans_webhook(request: HttpRequest) -> HttpResponse:
    secret = settings.ONLYFANSAPI_WEBHOOK_SECRET.encode()
    expected = hmac.new(secret, request.body, hashlib.sha256).hexdigest()
    signature = request.headers.get("Signature", "")
    if not hmac.compare_digest(expected.encode(), signature.encode()):
        return HttpResponse(status=401)

    # Retries reuse the same key, so each event is queued once.
    key = request.headers.get("X-OFAPI-Idempotency-Key")
    if key and not cache.add(f"ofapi-webhook:{key}", True, timeout=86400):
        return HttpResponse(status=204)

    try:
        process_event.enqueue(json.loads(request.body))
    except Exception:
        # Queueing failed: forget the key so the retry is not skipped.
        if key:
            cache.delete(f"ofapi-webhook:{key}")
        raise

    return HttpResponse(status=204)
```

A @task function from django.tasks. The view calls process_event.enqueue(event) and returns at once, and db_worker runs the task in the background.

```python
# onlyfans/tasks.py
import logging
from typing import Any

from django.tasks import task

log = logging.getLogger(__name__)


@task
def process_event(event: dict[str, Any]) -> None:
    payload = event["payload"]

    if event["event"] == "subscriptions.new":
        log.info("New subscriber @%s on %s", payload["fan"]["username"], event["account_id"])
```

## Call the OnlyFans API concurrently with asyncio

The async client runs many requests at once, which helps most when an agency manages dozens of accounts.

Use AsyncOnlyFansAPI as an async context manager so its connections close when you are done. Every method matches the sync client, with await in front.

```python
import asyncio

from onlyfansapi import AsyncOnlyFansAPI


async def main() -> None:
    async with AsyncOnlyFansAPI() as client:
        # Same methods as the sync client, awaited
        fans = await client.fans.list_latest("acct_XXXXXXXXXXXXX", limit=10)
        for fan in (fans.data.users if fans.data else None) or []:
            print(f"@{fan.username}")


asyncio.run(main())
```

Count unread chats on every connected account at the same time with asyncio.gather. Each count follows next_page until the last page, and a semaphore keeps five requests in flight, so you stay inside your rate limit.

```python
import asyncio

from onlyfansapi import AsyncOnlyFansAPI

client = AsyncOnlyFansAPI()
limit = asyncio.Semaphore(5)  # at most 5 requests in flight


async def unread_chats(account_id: str) -> int:
    total, offset = 0, 0
    while True:
        async with limit:
            page = await client.chats.list(account_id, filter="unread", limit="100", offset=str(offset))
        total += len(page.data or [])

        # Chats have no has_more flag: follow next_page until it is empty
        if not (page.api_pagination and page.api_pagination.next_page):
            return total
        offset += 100


async def main() -> None:
    accounts = await client.accounts.list()
    ids = [a.id for a in accounts if a.id]

    # Every connected account, counted concurrently
    counts = await asyncio.gather(*(unread_chats(i) for i in ids))
    for account_id, count in zip(ids, counts):
        print(f"{account_id}: {count} unread chats")


asyncio.run(main())
```

Install onlyfans[aiohttp] and pass DefaultAioHttpClient() to use aiohttp instead of httpx, for better performance under heavy concurrency.

```python
# pip install "onlyfans[aiohttp]"
import asyncio

from onlyfansapi import AsyncOnlyFansAPI, DefaultAioHttpClient


async def main() -> None:
    # aiohttp instead of httpx for high-concurrency workers
    async with AsyncOnlyFansAPI(http_client=DefaultAioHttpClient()) as client:
        whoami = await client.whoami.retrieve()
        print(whoami.team.name if whoami.team else None)


asyncio.run(main())
```

## Load every fan, chat or post page by page

List endpoints return one page at a time. Loop until has_more is false and move the offset by the full page size each time.

OnlyFans filters some entries on its side, so a page can hold 19 fans when you asked for 20 and still not be the last one. Do not stop on a short page. has_more is the reliable end signal, and moving the offset by the full limit skips nobody.

```python
from onlyfansapi import OnlyFansAPI

client = OnlyFansAPI()

account = "acct_XXXXXXXXXXXXX"
limit = 20
offset = 0
usernames: list[str] = []

while True:
    # GET /api/{account}/fans/all
    page = client.fans.list_all(account, limit=limit, offset=offset)
    if page.data is None:
        break

    for fan in page.data.list or []:
        if fan.username:
            usernames.append(fan.username)

    # A page can hold fewer than limit fans and still not be the last one.
    # Trust has_more, and move the offset by the full limit.
    if not page.data.has_more:
        break
    offset += limit

print(f"{len(usernames)} fans loaded")
```

Write every active fan to a CSV with Python's csv module, for a CRM import or a spreadsheet. The loop follows has_more, so large accounts export completely.

```python
import csv

from onlyfansapi import OnlyFansAPI

client = OnlyFansAPI()

with open("fans.csv", "w", newline="") as file:
    writer = csv.writer(file)
    writer.writerow(["id", "username", "name"])

    offset = 0
    has_more = True
    while has_more:
        # GET /api/{account}/fans/active
        page = client.fans.list_active("acct_XXXXXXXXXXXXX", limit=20, offset=offset)
        data = page.data
        for fan in (data.list if data else None) or []:
            writer.writerow([fan.id, fan.username, fan.name])

        has_more = bool(data and data.has_more)
        offset += 20
```

## Handle errors, retries and timeouts

Every failed request raises a typed exception, and short outages are retried before your code sees them.

Each HTTP status has its own exception class, and all of them inherit from onlyfansapi.APIError. Catch the ones you can act on, then APIStatusError for the rest; it carries status_code and the full response.

```python
import onlyfansapi
from onlyfansapi import OnlyFansAPI

client = OnlyFansAPI()

try:
    chats = client.chats.list("acct_XXXXXXXXXXXXX", limit="10")
except onlyfansapi.AuthenticationError:
    ...  # 401: the API key is missing, revoked or mistyped
except onlyfansapi.NotFoundError:
    ...  # 404: wrong account ID, or the account was disconnected
except onlyfansapi.RateLimitError:
    ...  # 429: still limited after the SDK's own retries, so back off
except onlyfansapi.APITimeoutError:
    ...  # no answer within the timeout, even after retries
except onlyfansapi.APIConnectionError as e:
    print(e.__cause__)  # DNS, TLS or network failure from httpx
except onlyfansapi.APIStatusError as e:
    print(e.status_code, e.response.text)  # any other 4xx or 5xx
```

The SDK retries connection errors, timeouts, and 408, 409, 429 and 5xx responses twice with exponential backoff, and follows Retry-After. Raise max_retries for background jobs, or use with_options(max_retries=0) while a user waits.

```python
from onlyfansapi import OnlyFansAPI

# Connection errors, timeouts, 408, 409, 429 and 5xx responses are retried
# twice by default, with exponential backoff that respects Retry-After.
client = OnlyFansAPI(max_retries=4)

# Or switch retries off for one call, e.g. while a user waits on the page
chats = client.with_options(max_retries=0).chats.list("acct_XXXXXXXXXXXXX", limit="10")
```

Requests time out after 60 seconds, with 5 seconds to connect. Pass a float or an httpx.Timeout to the client, or with_options(timeout=...) for one call. A timed-out request is retried, then raises APITimeoutError.

```python
import httpx
from onlyfansapi import OnlyFansAPI

# The default is 60 seconds per request, with 5 seconds to connect
client = OnlyFansAPI(timeout=httpx.Timeout(30.0, connect=5.0))

# Shorter timeout for one call; on expiry the SDK retries, then raises APITimeoutError
chats = client.with_options(timeout=10.0).chats.list("acct_XXXXXXXXXXXXX", limit="10")
```

New endpoints can ship before the SDK has a method for them. client.get() and client.post() call any path with the same authentication, retries and timeouts, and extra_query, extra_body and extra_headers add fields to a typed call.

```python
import httpx
from onlyfansapi import OnlyFansAPI

client = OnlyFansAPI()

# Call an endpoint the SDK has no method for yet.
# Authentication, retries and timeouts still apply.
response = client.get(
    "/api/acct_XXXXXXXXXXXXX/some-new-endpoint",
    cast_to=httpx.Response,
    options={"params": {"limit": 10}},
)
print(response.json())

# Or add a parameter the SDK does not know about to a typed call
chats = client.chats.list(
    "acct_XXXXXXXXXXXXX",
    extra_query={"new_filter": "value"},
)
```

## Track credits and rate limits from Python

Every response reports what the call cost and how much room you have left, so you can slow down before you hit a limit.

The API's _meta block is api_meta in Python, with api_credits, api_rate_limits and api_cache inside. It shows credits used and left, requests left this minute and today, and whether the answer came from cache. Cached responses cost 0 credits but still count toward your rate limit.

```python
from onlyfansapi import OnlyFansAPI

client = OnlyFansAPI()

fans = client.fans.list_all("acct_XXXXXXXXXXXXX", limit=20)

# Every response carries _meta with credits, rate limits and cache state.
# The SDK exposes it as api_meta, with api_credits, api_rate_limits and api_cache.
meta = fans.api_meta
if meta and meta.api_credits and meta.api_rate_limits and meta.api_cache:
    print(f"Credits used: {meta.api_credits.used}")
    print(f"Credits left: {meta.api_credits.balance}")
    print(f"Requests left this minute: {meta.api_rate_limits.remaining_minute}")
    print(f"From cache: {'yes' if meta.api_cache.is_cached else 'no'}")

# Output:
# Credits used: 1
# Credits left: 9876
# Requests left this minute: 998
# From cache: no
```

## What Python teams build with the OnlyFans API

- **Analytics Dashboards**: Feed earnings, tracking-link revenue and fan counts into Streamlit, Metabase or a Django admin page.
- **Data Pipelines**: Load fans, transactions and statistics into pandas, Postgres or BigQuery on a schedule with Airflow, Prefect or cron.
- **AI Chat Assistants**: Read chat history, draft replies with an LLM, and send them back through the same client.
- **Multi-account tools**: Run one async worker across every creator account on your key, each addressed by its acct_ ID.
- **Media Workflows**: Upload from local files or object storage, attach media to posts and paid messages, and stream downloads to disk.
- **Custom integrations**: Connect OnlyFans data to a CRM, Slack or your own backend through webhooks and the typed client.

## FAQ

### How do I install the OnlyFans Python SDK?

Run pip install onlyfans, or uv add onlyfans or poetry add onlyfans. Then import it with from onlyfansapi import OnlyFansAPI. The package is called onlyfans on PyPI, but the module you import is onlyfansapi.

### Which Python versions are supported?

Python 3.9 and newer. We tested the examples on this page on Python 3.9 and 3.14.

### Does the SDK support async?

Yes. AsyncOnlyFansAPI has the same methods as OnlyFansAPI, awaited. It uses httpx by default; install onlyfans[aiohttp] to switch to aiohttp.

### Does it work with FastAPI, Flask and Django?

Yes. Use the async client in FastAPI and the sync client in Flask and Django. The sections above show the setup for each, with a signed webhook route.

### How do I verify webhook signatures in Python?

Compute hmac.new(secret.encode(), raw_body, hashlib.sha256).hexdigest() and compare it with the Signature header using hmac.compare_digest. Hash the raw request body, not JSON you parsed and dumped again, or the signature will not match.

### What happens when a request fails?

The SDK raises a subclass of onlyfansapi.APIError, such as AuthenticationError, NotFoundError, RateLimitError or APITimeoutError. Connection errors, timeouts, and 408, 409, 429 and 5xx responses are first retried twice with exponential backoff.

### Are responses typed?

Yes. Every response is a Pydantic model and every argument has a type hint, so editors autocomplete fields and mypy or pyright checks your calls. Most fields are Optional, so check for None before you use them.

### Can I call an endpoint that is not in the SDK yet?

Yes. client.get() and client.post() call any path with the same authentication and retries, and extra_query, extra_body and extra_headers add fields to a typed call.

### Is the OnlyFans Python SDK free?

Yes. The SDK is open source under the Apache 2.0 license and free to install. The API calls it makes use credits from your OnlyFansAPI plan.

## Links

- Get your API key: https://app.onlyfansapi.com/register
- Python SDK docs: https://github.com/onlyfansapi/onlyfansapi-python/blob/main/README.md
- Source on GitHub: https://github.com/onlyfansapi/onlyfansapi-python
- API reference: https://docs.onlyfansapi.com/introduction
- Webhook signatures: https://docs.onlyfansapi.com/webhooks/protecting-your-webhooks
- All SDKs: https://onlyfansapi.com/sdk