Skip to content

Python SDK Reference

The official Python SDK for the SubscribeFlow API. Fully async, fully typed.

Installation

pip install subscribeflow

Or in pyproject.toml (uv, Poetry, PDM):

[project.dependencies]
subscribeflow = ">=0.1.0"

Client initialization

The client uses an async context manager to manage the HTTP connection:

import asyncio
from subscribeflow import SubscribeFlowClient

async def main():
    async with SubscribeFlowClient(
        api_key="sf_live_...",
        base_url="https://api.subscribeflow.net",  # default
        timeout=30.0,                               # seconds, default: 30
    ) as client:
        subscribers = await client.subscribers.list(limit=10)

asyncio.run(main())

You can also manage the lifecycle manually:

client = SubscribeFlowClient(api_key="sf_live_...")
try:
    subscriber = await client.subscribers.get("subscriber-id")
finally:
    await client.close()

Resources

subscribers

create(email, tags=None, metadata=None)

Create a new subscriber.

subscriber = await client.subscribers.create(
    email="alice@example.com",
    tags=["newsletter"],
    metadata={"source": "website"},
)

get(subscriber_id)

Get a subscriber by ID.

subscriber = await client.subscribers.get("subscriber-id")

get_by_email(email)

Get a subscriber by email address.

subscriber = await client.subscribers.get_by_email("alice@example.com")

list(limit=50, cursor=None, status=None)

List subscribers with cursor-based pagination.

result = await client.subscribers.list(limit=50, status="active")
for subscriber in result:
    print(subscriber.email)

update(subscriber_id, **kwargs)

Update subscriber fields (email, status, metadata).

updated = await client.subscribers.update(
    "subscriber-id",
    metadata={"plan": "professional"},
)

delete(subscriber_id)

Permanently delete a subscriber and all associated data.

await client.subscribers.delete("subscriber-id")

add_tags(subscriber_id, tags)

Add tags to a subscriber.

await client.subscribers.add_tags("subscriber-id", tags=["beta-testers"])

remove_tag(subscriber_id, tag_slug)

Remove a single tag from a subscriber.

await client.subscribers.remove_tag("subscriber-id", "beta-testers")

generate_token(subscriber_id)

Generate a preference center token.

token_response = await client.subscribers.generate_token("subscriber-id")
print(token_response.token)

tags

create(name, description=None, category=None, is_public=True)

Create a new tag.

tag = await client.tags.create(
    name="Product Updates",
    description="New features and improvements",
    category="product",
    is_public=True,
)

get(tag_id)

Get a tag by ID.

tag = await client.tags.get("tag-id")

get_by_name(name)

Get a tag by name.

tag = await client.tags.get_by_name("Product Updates")

list(category=None)

List all tags, optionally filtered by category.

tags = await client.tags.list(category="product")

update(tag_id, **kwargs)

Update tag fields (name, description, category, is_public).

await client.tags.update("tag-id", description="Updated description")

delete(tag_id)

Delete a tag and remove all subscriber associations.

await client.tags.delete("tag-id")

templates

create(name, subject, mjml_content, category=None)

Create an email template.

template = await client.templates.create(
    name="Welcome Email",
    subject="Welcome to {{company}}!",
    mjml_content="<mjml><mj-body>...</mj-body></mjml>",
    category="transactional",
)

get(template_id)

Get a template by ID or slug.

template = await client.templates.get("template-id")

list(category=None)

List templates, optionally filtered by category.

templates = await client.templates.list(category="transactional")

update(template_id, **kwargs)

Update template fields.

await client.templates.update("template-id", subject="New Subject")

delete(template_id)

Delete a template.

await client.templates.delete("template-id")

preview(template_id, variables=None)

Render a template with variables and return the HTML output.

preview = await client.templates.preview(
    "template-id",
    variables={"company": "Acme Inc"},
)
print(preview.html)

campaigns

create(name, template_id, tag_filter=None)

Create a campaign draft.

campaign = await client.campaigns.create(
    name="March Newsletter",
    template_id="template-uuid",
    tag_filter={"include_tags": ["newsletter"], "match": "any"},
)

get(campaign_id)

Get a campaign by ID.

campaign = await client.campaigns.get("campaign-id")

list(status=None)

List campaigns, optionally filtered by status.

campaigns = await client.campaigns.list(status="draft")

update(campaign_id, **kwargs)

Update campaign fields.

await client.campaigns.update("campaign-id", name="Updated Name")

send(campaign_id)

Send a campaign to all matching subscribers.

result = await client.campaigns.send("campaign-id")
print(f"Sending to {result.total_recipients} recipients")

cancel(campaign_id)

Cancel a running campaign.

await client.campaigns.cancel("campaign-id")

duplicate(campaign_id)

Duplicate a campaign as a new draft.

new_campaign = await client.campaigns.duplicate("campaign-id")

retry(campaign_id)

Retry a failed campaign.

await client.campaigns.retry("campaign-id")

count_recipients(include_tags=None, match=None)

Preview how many subscribers match a tag filter.

count = await client.campaigns.count_recipients(include_tags=["newsletter"])

emails

send(template_slug, to, variables=None, idempotency_key=None)

Send a transactional email to a single recipient.

result = await client.emails.send(
    template_slug="welcome-email",
    to="alice@example.com",
    variables={"company": "Acme Inc"},
    idempotency_key="welcome-alice-2024",
)

webhooks

create(url, events, description=None)

Create a webhook endpoint. Returns the signing secret.

webhook = await client.webhooks.create(
    url="https://your-app.com/webhooks/subscribeflow",
    events=["subscriber.created", "tag.subscribed"],
)
print(f"Secret: {webhook.secret}")

get(webhook_id)

Get a webhook by ID.

webhook = await client.webhooks.get("webhook-id")

list()

List all webhook endpoints.

webhooks = await client.webhooks.list()

update(webhook_id, **kwargs)

Update webhook fields (url, events, description, is_active).

await client.webhooks.update("webhook-id", events=["subscriber.created"])

delete(webhook_id)

Delete a webhook endpoint.

await client.webhooks.delete("webhook-id")

test(webhook_id)

Send a test event to a webhook endpoint.

result = await client.webhooks.test("webhook-id")
print(f"Success: {result.success}")

list_deliveries(webhook_id)

List delivery history for a webhook.

deliveries = await client.webhooks.list_deliveries("webhook-id")

retry_delivery(webhook_id, delivery_id)

Retry a failed delivery.

await client.webhooks.retry_delivery("webhook-id", "delivery-id")

get_stats(webhook_id)

Get delivery statistics for a webhook.

stats = await client.webhooks.get_stats("webhook-id")
print(f"Success rate: {stats.success_rate}%")

regenerate_secret(webhook_id)

Rotate the signing secret.

new_webhook = await client.webhooks.regenerate_secret("webhook-id")
print(f"New secret: {new_webhook.secret}")

triggers

create(event_type, template_id, description=None)

Create an event-based email trigger.

trigger = await client.triggers.create(
    event_type="subscriber.created",
    template_id="welcome-template-uuid",
    description="Send welcome email on signup",
)

get(trigger_id)

Get a trigger by ID.

trigger = await client.triggers.get("trigger-id")

list()

List all triggers.

triggers = await client.triggers.list()

update(trigger_id, **kwargs)

Update trigger fields.

await client.triggers.update("trigger-id", is_active=False)

delete(trigger_id)

Delete a trigger.

await client.triggers.delete("trigger-id")

billing

get_subscription()

Get the current billing subscription.

subscription = await client.billing.get_subscription()

create_checkout_session(plan)

Create a Stripe Checkout session for plan upgrade.

session = await client.billing.create_checkout_session("starter")
print(f"Checkout URL: {session.url}")

create_portal_session()

Create a Stripe Billing Portal session.

session = await client.billing.create_portal_session()
print(f"Portal URL: {session.url}")

sync_subscription()

Synchronize the subscription state with Stripe.

await client.billing.sync_subscription()

preference_center

Access preference center operations using a subscriber token.

pref_center = client.preference_center(token)

get_preferences()

Get the subscriber's preferences and available tags.

info = await pref_center.get_preferences()

subscribe_tag(tag_id)

Subscribe to a tag.

await pref_center.subscribe_tag("tag-id")

unsubscribe_tag(tag_id)

Unsubscribe from a tag.

await pref_center.unsubscribe_tag("tag-id")

export_data()

Export all subscriber data as JSON (DSGVO Art. 20).

export = await pref_center.export_data()

delete_account()

Permanently delete the subscriber account (DSGVO Art. 17).

await pref_center.delete_account()

Error handling

All errors extend SubscribeFlowError:

from subscribeflow import (
    SubscribeFlowError,      # Base error (has .status, .detail, .type)
    AuthenticationError,      # 401 -- invalid or missing API key
    NotFoundError,            # 404 -- resource not found
    ValidationError,          # 422 -- request validation failed
    RateLimitError,           # 429 -- rate limit exceeded
    LimitExceededError,       # 402 -- plan limit exceeded
)

Example:

from subscribeflow import NotFoundError, ValidationError, LimitExceededError

try:
    subscriber = await client.subscribers.get("non-existent-id")
except NotFoundError as e:
    print(f"Not found: {e.detail}")
except ValidationError as e:
    print(f"Validation failed: {e.detail}")
    for error in e.errors:
        print(f"  - {error['loc']}: {error['msg']}")
except LimitExceededError as e:
    print(f"Plan limit exceeded: {e.detail}")
except SubscribeFlowError as e:
    print(f"API error ({e.status}): {e.detail}")

MCP Server

The Python SDK includes an MCP server for Claude Desktop and Claude Code integration. See the MCP integration guide for setup instructions.

{
  "mcpServers": {
    "subscribeflow": {
      "command": "uv",
      "args": [
        "--directory", "/path/to/subscribe-flow/sdk/python",
        "run", "subscribeflow-mcp"
      ],
      "env": {
        "SUBSCRIBEFLOW_API_KEY": "sf_live_..."
      }
    }
  }
}