FastSub API

# паблишер
Authorization: Bearer fsp_live_xxxxxxxxxxxxxxxx

# рекламодатель
Authorization: Bearer fsa_live_xxxxxxxxxxxxxxxx

# заголовка нет вообще
{ "detail": "missing authorization header" }

# не Bearer (Basic, Token, просто ключ без схемы…)
{ "detail": "invalid authorization scheme; expected Bearer" }

# ключ не найден или отозван
{ "detail": "invalid or revoked token" }

# аккаунт заблокирован администратором
{ "detail": "account disabled" }

# только Publisher API: старый ключ без привязки к боту
{ "detail": "token not bound to a bot — please regenerate via @FastSub_Robot" }

# HTTP 403 — владелец выключил бота в кабинете
{ "detail": "bot is currently disabled by its owner" }

# HTTP 404 — бот, к которому был привязан ключ, удалён
{ "detail": "associated bot not found" }

https://fastsub.org/api/v1

# Retry-After: 60
{ "detail": "Rate limit exceeded: 60 requests per 60s" }

# HTTP 404
{ "detail": "link_id not found: lnk_xxx" }

# HTTP 422 — не прошла валидация тела
{ "detail": [ {
    "type": "greater_than_equal",
    "loc": ["body", "count"],
    "msg": "Input should be greater than or equal to 1"
} ] }

# HTTP 402
{
  "ok": false,
  "error": {
    "code": "insufficient_funds",
    "message": "Недостаточно средств на балансе.",
    "details": {
      "required_rub": "1500.0000",
      "available_rub": "220.0000",
      "missing_rub": "1280.0000"
    }
  }
}

# 1. ваш бот
POST /request-op          → task_id + список заданий (у каждого link_id)

# 2. юзер жмёт кнопку и подписывается
                          → наш проверочный бот видит вступление
                          → выдача переходит в subscribed, начинается холд

# 3. вы спрашиваете результат — любым из двух способов
POST /check-subscription  → живой вопрос Telegram прямо сейчас
POST /check-task          → сохранённые статусы всех заданий разом

# 4. окно проверки отписок закрылось, юзер всё ещё подписан
                          → verified: выплата окончательна

{
  "ok": true,
  "task_id": "tsk_a1b2c3d4e5f6",
  "delivered": true,
  "tasks": [
    {
      "link_id": "lnk_a1b2c3d4e5f6",
      "type": "channel",
      "task_type": "subscribe",
      "button_name": "Подписаться",
      "title": "Example Channel",
      "theme_label": "Нейросети",
      // ссылок нет: сообщение уже у юзера
      "invite_link": null,
      "start_link": null,
      "reward_for_publisher": "1.5000"
    }
  ]
}

🔒 Доступ к боту

Подпишитесь на 2 спонсора ниже, чтобы продолжить.

▏После подписки вернитесь сюда и отправьте /start —
▏доступ откроется автоматически.

[ Подписаться: Example Channel ]
[ Подписаться: Нейросети каждый день ]

@dp.message(CommandStart())
async def start(m: Message):
    data = await api_post("/request-op", {"user_id": m.from_user.id})

    # веб-шаг: показываем ссылку и ждём возврата
    if data.get("reason") == "onboarding_required":
        return await m.answer("Остался шаг: " + data["onboarding_url"])

    # заданий нет — пускаем
    if not data["ok"]:
        return await grant_access(m)

    # блок ОП уже отправлен нами — проверяем, всё ли выполнено
    st = await api_post("/check-task", {"task_id": data["task_id"]})
    done = {"subscribed", "verified", "paid"}
    if all(i["status"] in done for i in st["items"]):
        await grant_access(m)

const data = await apiPost("/request-op", { user_id: userId });

if (data.reason === "onboarding_required")
  return send("Остался шаг: " + data.onboarding_url);

if (!data.ok) return grantAccess();

// сообщение со спонсорами уже отправлено нами
const st = await apiPost("/check-task", { task_id: data.task_id });
const done = ["subscribed", "verified", "paid"];
if (st.items.every(i => done.includes(i.status))) grantAccess();

{
  "event": "resource.verified",
  "link_id": "lnk_a1b2c3d4e5f6",
  "task_id": "tsk_a1b2c3d4e5f6",
  "user_id": 123456789,
  "campaign_resource_id": 412,
  "status": "verified",
  "reward_rub": "2.0000",
  "publisher_payout_rub": "1.5000",
  "platform_commission_rub": "0.5000",
  "issued_at": "2026-05-23T17:40:02Z",
  "subscribed_at": "2026-05-23T17:41:19Z",
  "verified_at": "2026-05-23T18:04:11Z",
  "unsubscribed_at": null,
  "expires_at": "2026-05-23T18:40:02Z",
  "timestamp": "2026-05-23T18:04:11.914Z"
}

import hmac, hashlib

def verify(body: bytes, signature: str, secret: str) -> bool:
    expected = hmac.new(
        secret.encode(), body, hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, signature)

import crypto from "node:crypto";

function verify(body, signature, secret) {
  const expected = crypto
    .createHmac("sha256", secret)
    .update(body).digest("hex");
  return crypto.timingSafeEqual(
    Buffer.from(expected), Buffer.from(signature));
}

from fastapi import FastAPI, Request, HTTPException

app = FastAPI()
SECRET = "whsec_…"
seen: set[str] = set()

@app.post("/fastsub/webhook")
async def hook(request: Request):
    body = await request.body()          # сырые байты!
    sig = request.headers.get("X-FastSub-Signature", "")
    if not verify(body, sig, SECRET):
        raise HTTPException(401)

    # идемпотентность: повтор придёт с тем же Delivery-Id
    delivery = request.headers.get("X-FastSub-Delivery-Id", "")
    if delivery in seen:
        return {"ok": True}
    seen.add(delivery)

    event = await request.json()
    if event.get("test"):
        return {"ok": True}
    if event["event"] == "resource.verified":
        ...  # начислить юзеру бонус
    return {"ok": True}       # любой 2xx = принято

Authorization: Bearer fsa_live_xxxxxxxxxxxxxxxx
Content-Type: application/json

базовая цена           1.00 ₽
countries: ["RU"]      × 1.5
min_age: min_18_plus   × 2.0
require_premium: true  × 1.3
─────────────────────────────
коэффициент            = 3.9
цена за подписчика     = 3.90 ₽
× 1000 подписчиков     = 3900.00 ₽

import httpx, uuid, time

key = str(uuid.uuid4())          # один на весь цикл повторов
payload = {"link": "@example", "name": "Запуск",
           "price_rub": "1.00", "quantity": 1000}

for attempt in range(5):
    r = httpx.post(
        "https://fastsub.org/api/v1/advertiser/orders",
        headers={"Authorization": "Bearer fsa_live_xxx",
                 "Idempotency-Key": key},
        json=payload, timeout=30,
    )
    if r.status_code == 409 and \
       r.json()["error"]["code"] == "idempotency_in_progress":
        time.sleep(2)
        continue
    break

def handle(r: httpx.Response):
    if r.is_success:
        return r.json()

    body = r.json()
    # 401/429 приходят в стиле FastAPI, остальное — конвертом
    if "error" not in body:
        raise RuntimeError(body.get("detail", r.text))

    code = body["error"]["code"]
    if code == "insufficient_funds":
        need = body["error"]["details"]["missing_rub"]
        raise RuntimeError(f"пополните баланс на {need} ₽")
    if code == "checker_not_admin":
        raise RuntimeError("добавьте нашего бота админом и повторите")
    raise RuntimeError(f"{code}: {body['error']['message']}")

pip install fastsub

# Асинхронный бот (aiogram):
from fastsub import FastSub

async with FastSub(api_key="fsp_live_...") as fs:
    answer = await fs.request_op(user_id=123456789)

# Синхронный бот (pyTelegramBotAPI, Flask и всё, что не на asyncio):
from fastsub import FastSubSync, telebot_gate

fs = FastSubSync(api_key="fsp_live_...")
gate = telebot_gate(fs, bot)

@bot.message_handler(commands=["start"])
def start(message):
    if not gate(message.from_user.id, message.chat.id):
        return                      # блок показан, дальше не пускаем
    bot.send_message(message.chat.id, "Доступ открыт")

// npm install fastsub — или скачайте fastsub.mjs, он без зависимостей
// Node 18+ / Bun / Deno
import { FastSub } from "fastsub";

const fs = new FastSub({ apiKey: "fsp_live_..." });
const answer = await fs.requestOp({ userId: 123456789 });

// На TypeScript положите рядом fastsub.d.ts — будет автодополнение:
//   /api/v1/sdk/js/types
const status = await fs.checkTask(answer.taskId);
if (status.allDone) console.log("пускаем");
else console.log("осталось:", status.remaining.length);

from fastsub import FastSub, FastSubError

# Ключ уже есть — просто клиент:
async with FastSub(api_key="fsp_live_...") as fs:

    # 1. Задания для юзера. Передавайте всё, что знаете о нём:
    #    откроются заказы с таргетингом — они дороже.
    answer = await fs.request_op(
        user_id=123456789,
        count=3,
        has_telegram_premium=True,
        has_username=True,
    )

    if answer.needs_web_step:
        # Один шаг на нашей странице, она сама вернёт юзера в бот.
        send(answer.onboarding_url)

    elif answer.has_tasks:
        for task in answer.tasks:
            print(task.title, task.link, task.reward_for_publisher)
    else:
        # Почему пусто — человеческим текстом, годится в логи.
        print(answer.explain)

    # 2. Проверить одно задание. check_subscription спрашивает Telegram
    #    сейчас, check_task читает сохранённый статус (дешевле).
    data = await fs.check_subscription("lnk_a1b2c3d4e5f6")
    if data["subscribed"]:
        grant_access()

    # 3. Баланс, настройки бота.
    me = await fs.me()
    print(me["balance_rub"], me["debt_rub"])

import { FastSub } from "fastsub";

const fs = new FastSub({ apiKey: "fsp_live_..." });

const answer = await fs.requestOp({
  userId: 123456789,
  count: 3,
  hasTelegramPremium: true,
});

if (answer.needsWebStep) send(answer.onboardingUrl);
else if (answer.hasTasks) for (const t of answer.tasks) console.log(t.title, t.link);
else console.log(answer.explain);

const data = await fs.checkSubscription("lnk_a1b2c3d4e5f6");
if (data.subscribed) grantAccess();

from aiogram import Dispatcher
from fastsub import FastSub, FastSubMiddleware, fastsub_check_router

fs = FastSub(api_key="fsp_live_...")
dp = Dispatcher()

# Спрашивает задания сам, рисует блок и кладёт ответ в data["fastsub"].
dp.message.middleware(FastSubMiddleware(fs))
dp.callback_query.middleware(FastSubMiddleware(fs))
dp.include_router(fastsub_check_router(fs))   # кнопка «Проверить»


@dp.message()
async def any_message(message, fastsub):
    # Уже всё сделано: если юзер не выполнил задания, сюда мы не дошли.
    # fastsub — это OpAnswer или None, если наш API был недоступен.
    await message.answer("Ваш обычный ответ")

# Где реклама не нужна — помечаем хендлер:
@dp.message(Command("help"), flags={"fastsub": "skip"})
async def help_cmd(message):
    await message.answer("Помощь без блока спонсоров")

# А где юзера надо остановить до выполнения — "gate":
@dp.message(Command("premium"), flags={"fastsub": "gate"})
async def premium(message):
    await message.answer("Доступно после заданий")

# Юзер выполнил два задания из трёх и просит ещё.
# keep вернёт те же три — включая выполненные. top_up добьёт до нужного числа:
answer = await fs.request_op(user_id=user_id, count=3, mode="top_up")

# «Пересобрать» — когда блок показываете вы сами и решаете, когда обновить:
answer = await fs.request_op(user_id=user_id, mode="fresh")

# «Обновлять после выполнения» — юзер возвращается за новой порцией:
answer = await fs.request_op(user_id=user_id, mode="rotate")

from fastapi import FastAPI
from fastsub import (
    ResourceReverted, ResourceUnsubscribed, ResourceVerified,
    WebhookEvents, fastsub_webhook_router,
)

app = FastAPI()

# Один раз: куда слать события. Секрет вернётся ровно один раз — сохраните.
result = await fs.configure_webhook(
    "https://your-bot.example/fastsub/webhook",
    events=["resource.verified", "resource.reverted"],
)
SECRET = result["secret"]

events = WebhookEvents()

@events.on("resource.verified")
async def paid(event: ResourceVerified):
    await credit(event.user_id, event.publisher_payout_rub)

@events.on("resource.unsubscribed", "resource.reverted")
async def taken_back(event: ResourceUnsubscribed | ResourceReverted):
    if event.money_was_taken:           # отписался внутри холда
        await debit(event.user_id, event.publisher_payout_rub)

# Подпись проверена, повторные доставки отсеяны, 2xx отвечен.
app.include_router(fastsub_webhook_router(SECRET, events))

from fastsub import FastSubAdvertiser

async with FastSubAdvertiser(api_key="fsa_live_...") as adv:
    # Сначала цена, потом деньги: quote ничего не списывает.
    quote = await adv.quote(quantity=1000, price_rub="1.50")
    print(quote["total_rub"], quote["coefficients"])

    # Повтор с тем же ключом вернёт созданный заказ, а не второй такой же.
    order = await adv.create_order(
        name="Мой канал", chat="@my_channel",
        quantity=1000, price_rub="1.50",
        idempotency_key="my-order-42",
    )

    card = await adv.order(order["id"])
    print(card["forecast"]["text"])       # «Около 9 ч при текущем темпе»
    print(card["retention"]["summary"])   # «Удержание: 1 дн — 92%, 7 дн — 84%»

    # Оплата целевого действия — за регистрацию, депозит, что угодно ваше.
    await adv.postback(link_id="lnk_...", event="registration", value_rub=25)

async for order in adv.iter_orders(status="active"):
    print(order["id"], order["progress"])

# Заказ на десять тысяч человек — это сотня страниц.
async for sub in adv.iter_subscribers(order_id):
    await crm.upsert(sub["user_id"])

# Наш сбой не должен останавливать вашего бота.
try:
    answer = await fs.request_op(user_id=user_id)
except FastSubError as e:
    print(e.status, e.detail)   # detail можно показать человеку
    answer = None               # и пропустить юзера дальше

import httpx

API = "https://fastsub.org/api/v1"
TOKEN = "fsp_live_xxx"
HEADERS = {"Authorization": f"Bearer {TOKEN}"}


def request_tasks(user_id: int, count: int = 3, **audience) -> dict:
    # Запросить задания. audience: has_telegram_premium, has_username, …
    r = httpx.post(f"{API}/request-op", headers=HEADERS,
                   json={"user_id": user_id, "count": count, **audience},
                   timeout=15)
    r.raise_for_status()
    return r.json()


def check_live(link_id: str) -> bool:
    # Живая проверка по кнопке «Проверить подписку»
    r = httpx.post(f"{API}/check-subscription", headers=HEADERS,
                   json={"link_id": link_id}, timeout=15)
    r.raise_for_status()
    return r.json()["subscribed"]


def check_all(task_id: str) -> list[dict]:
    # Пакетно: статусы всех ссылок одного задания
    r = httpx.post(f"{API}/check-task", headers=HEADERS,
                   json={"task_id": task_id}, timeout=15)
    r.raise_for_status()
    return r.json()["items"]

const API = "https://fastsub.org/api/v1";
const TOKEN = "fsp_live_xxx";
const HEADERS = {
  "Authorization": `Bearer ${TOKEN}`,
  "Content-Type": "application/json"
};

async function post(path, body) {
  const r = await fetch(`${API}${path}`, {
    method: "POST", headers: HEADERS, body: JSON.stringify(body)
  });
  const data = await r.json();
  if (!r.ok) throw new Error(data.detail || r.status);
  return data;
}

const requestTasks = (userId, count = 3, audience = {}) =>
  post("/request-op", { user_id: userId, count, ...audience });

const checkLive = async (linkId) =>
  (await post("/check-subscription", { link_id: linkId })).subscribed;

import asyncio
import httpx
from aiogram import Bot, Dispatcher, F
from aiogram.filters import CommandStart
from aiogram.types import (
    Message, CallbackQuery,
    InlineKeyboardMarkup, InlineKeyboardButton,
)

BOT_TOKEN = "123456:ABC..."
API = "https://fastsub.org/api/v1"
TOKEN = "fsp_live_xxx"
HEADERS = {"Authorization": f"Bearer {TOKEN}"}

bot = Bot(BOT_TOKEN)
dp = Dispatcher()


async def api_post(path: str, payload: dict) -> dict:
    async with httpx.AsyncClient(timeout=15) as c:
        r = await c.post(f"{API}{path}", headers=HEADERS, json=payload)
    r.raise_for_status()
    return r.json()


@dp.message(CommandStart())
async def start(message: Message):
    kb = InlineKeyboardMarkup(inline_keyboard=[[
        InlineKeyboardButton(text="Получить доступ",
                             callback_data="get_tasks")
    ]])
    await message.answer("Подпишитесь на спонсоров, чтобы продолжить:",
                         reply_markup=kb)


@dp.callback_query(F.data == "get_tasks")
async def get_tasks(cb: CallbackQuery):
    u = cb.from_user
    data = await api_post("/request-op", {
        "user_id": u.id,
        "count": 3,
        # без этих полей часть кампаний вам просто не выдадут
        "has_telegram_premium": bool(u.is_premium),
        "has_username": bool(u.username),
    })

    if not data["ok"]:
        # ВАЖНО: ветвимся по reason, а не просто «не ok — пускаем»
        if data.get("reason") == "onboarding_required":
            kb = InlineKeyboardMarkup(inline_keyboard=[[
                InlineKeyboardButton(text="Пройти регистрацию",
                                     url=data["onboarding_url"])
            ]])
            await cb.message.answer("Остался один шаг:",
                                    reply_markup=kb)
        else:
            await grant_access(cb.message)   # no_tasks — пускаем
        await cb.answer()
        return

    if data.get("delivered"):
        # режим «под ключ»: блок ОП мы уже отправили сами
        await cb.answer()
        return

    for task in data["tasks"]:
        link = task["invite_link"] or task["start_link"]
        action = "Забустить" if task["task_type"] == "boost" \
            else ("Запустить" if task["task_type"] == "start_bot"
                  else "Подписаться")
        kb = InlineKeyboardMarkup(inline_keyboard=[
            [InlineKeyboardButton(text=action, url=link)],
            [InlineKeyboardButton(text="Проверить",
                callback_data=f"check:{task['link_id']}")],
        ])
        await cb.message.answer(task["title"], reply_markup=kb)

    # task_id пригодится, чтобы проверить всё разом
    await cb.message.answer(
        "Когда закончите — нажмите «Готово»",
        reply_markup=InlineKeyboardMarkup(inline_keyboard=[[
            InlineKeyboardButton(text="Готово",
                callback_data=f"done:{data['task_id']}")
        ]]))
    await cb.answer()


@dp.callback_query(F.data.startswith("check:"))
async def check_one(cb: CallbackQuery):
    link_id = cb.data.split(":", 1)[1]
    data = await api_post("/check-subscription", {"link_id": link_id})

    if data["subscribed"]:
        await cb.answer("Засчитано!", show_alert=True)
    elif data["reason"] == "unsupported_task_type":
        await cb.answer("Запуск бота засчитается автоматически",
                       show_alert=True)
    elif data["reason"] == "check_unavailable":
        await cb.answer("Проверка недоступна, попробуйте через минуту",
                       show_alert=True)
    else:
        await cb.answer("Не вижу подписку. Подпишитесь и нажмите ещё раз",
                       show_alert=True)


@dp.callback_query(F.data.startswith("done:"))
async def check_all(cb: CallbackQuery):
    task_id = cb.data.split(":", 1)[1]
    data = await api_post("/check-task", {"task_id": task_id})
    done = {"subscribed", "verified", "paid"}
    left = [i for i in data["items"] if i["status"] not in done]

    if left:
        titles = ", ".join(i["title"] or "—" for i in left)
        await cb.answer(f"Осталось: {titles}", show_alert=True)
        return
    await grant_access(cb.message)
    await cb.answer()


async def grant_access(message: Message):
    await message.answer("Доступ открыт. Держите контент 🎁")


async def main():
    await dp.start_polling(bot)


if __name__ == "__main__":
    asyncio.run(main())

import httpx
from aiogram import Bot, Dispatcher
from aiogram.filters import CommandObject, CommandStart
from aiogram.types import Message

FASTSUB = "https://fastsub.org/api/v1/advertiser/confirm-start"
KEY = "fsa_live_xxx"

dp = Dispatcher()


@dp.message(CommandStart())
async def start(message: Message, command: CommandObject):
    payload = (command.args or "").strip()

    if payload.startswith("fastsub_"):
        try:
            async with httpx.AsyncClient(timeout=10) as c:
                r = await c.post(
                    FASTSUB,
                    headers={"Authorization": f"Bearer {KEY}"},
                    json={"start_param": payload,
                          "user_id": message.from_user.id},
                )
            # 409 = ссылку выдали другому человеку; 4xx не ретраим
            if r.status_code >= 500:
                ...  # положить в очередь на повтор
        except httpx.HTTPError:
            ...      # сеть моргнула — тоже в очередь

    # пользователю отвечаем в любом случае
    await message.answer("Добро пожаловать!")

import httpx
from telegram import Update
from telegram.ext import ContextTypes

FASTSUB = "https://fastsub.org/api/v1/advertiser/confirm-start"
KEY = "fsa_live_xxx"


async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    payload = context.args[0] if context.args else ""

    if payload.startswith("fastsub_"):
        async with httpx.AsyncClient(timeout=10) as c:
            await c.post(
                FASTSUB,
                headers={"Authorization": f"Bearer {KEY}"},
                json={"start_param": payload,
                      "user_id": update.effective_user.id},
            )

    await update.message.reply_text("Добро пожаловать!")