OnlyFans API Python SDK

# Install the requests library
pip install requests

# ofapi.py — shared client config
import requests
import os

BASE_URL = "http://157.180.79.226:4024/api/v1/onlyfans"

HEADERS = {
    "X-API-Key": os.environ["OFAPI_KEY"],
    "Content-Type": "application/json",
}

def ofapi(path, method="GET", **kwargs):
    url = f"{BASE_URL}{path}"
    resp = requests.request(method, url, headers=HEADERS, **kwargs)
    resp.raise_for_status()
    return resp.json()

from ofapi import ofapi

def get_revenue_stats(creator_id):
    data = ofapi(f"/models/{creator_id}/statistics/overview")
    return data

stats = get_revenue_stats("creator_abc123")
print(f"Total earnings: ${stats['totalEarned']}")
print(f"Subscriptions:  ${stats['subscriptions']}")
print(f"Tips:           ${stats['tips']}")
print(f"PPV:            ${stats['ppv_messages'] + stats['ppv_posts']}")

from ofapi import ofapi

def get_top_fans(creator_id, limit=10):
    data = ofapi(
        f"/models/{creator_id}/stats/fans/top",
        params={"limit": limit}
    )
    return data["fans"]

fans = get_top_fans("creator_abc123", limit=5)
for i, fan in enumerate(fans, 1):
    print(f"#{i} {fan['username']} — ${fan['total_spent']}")

from ofapi import ofapi
from datetime import datetime

def get_payouts(creator_id, page=1, limit=20):
    data = ofapi(
        f"/models/{creator_id}/payouts/transactions",
        params={"page": page, "limit": limit}
    )
    return data

result = get_payouts("creator_abc123")
now = datetime.now()

# Sum payouts for the current month
this_month = [
    t for t in result["transactions"]
    if datetime.fromisoformat(t["date"].replace("Z", "+00:00")).month == now.month
]
total = sum(t["amount"] for t in this_month)
print(f"Paid out this month: ${total}")

from ofapi import ofapi

def get_top_subscribers(creator_id, limit=20):
    data = ofapi(
        f"/models/{creator_id}/subscribers/top",
        params={"limit": limit}
    )
    return data["subscribers"]

subs = get_top_subscribers("creator_abc123")

# Flag high-value subs with no recent activity
from datetime import datetime, timezone, timedelta
cutoff = datetime.now(timezone.utc) - timedelta(days=14)

at_risk = [
    s for s in subs
    if s["total_spent"] > 200
    and datetime.fromisoformat(
        s["last_purchase_at"].replace("Z", "+00:00")
    ) < cutoff
]
print(f"{len(at_risk)} high-value subs at churn risk")

from ofapi import ofapi

def send_message(creator_id, fan_id, text, ppv_price=None, media_ids=None):
    body = {
        "creator_id": creator_id,
        "fan_id": fan_id,
        "text": text,
    }
    if ppv_price is not None:
        body["ppv_price"] = ppv_price
    if media_ids:
        body["media_ids"] = media_ids

    result = ofapi("/messages/send", method="POST", json=body)
    return result

# Send a PPV re-engagement message
result = send_message(
    creator_id="creator_abc123",
    fan_id="fan_001",
    text="Hey! Made something special just for you 🔥",
    ppv_price=12,
    media_ids=["media_xyz789"],
)
print(f"Message sent: {result['message_id']}")

import requests
from ofapi import ofapi

try:
    stats = ofapi(f"/models/creator_abc123/statistics/overview")
except requests.exceptions.HTTPError as e:
    status = e.response.status_code
    if status == 401:
        print("Invalid API key — check OFAPI_KEY env var")
    elif status == 403:
        print("Creator not linked to your account")
    elif status == 429:
        retry_after = e.response.headers.get("Retry-After", "5")
        print(f"Rate limited — retry after {retry_after}s")
    else:
        print(f"API error {status}: {e.response.text}")
except requests.exceptions.ConnectionError:
    print("Network error — API unreachable")