subscribeflow

Official Python SDK for SubscribeFlow — async, fully typed, with built-in MCP server for Claude integration. Manage email subscriptions, campaigns, and GDPR-compliant preference centers.

Dashboard | Documentation | API Reference | MCP Integration

Installation

From PyPI (Recommended)

Install the SDK using pip, uv, or any PEP 517-compatible installer:

# Basic SDK
pip install subscribeflow

# With MCP server support (for Claude Desktop/Code integration)
pip install "subscribeflow[mcp]"

In requirements.txt:

subscribeflow

With MCP support:

subscribeflow[mcp]

In pyproject.toml (uv, Poetry, PDM):

[project]
dependencies = ["subscribeflow"]

# Or with MCP support
dependencies = ["subscribeflow[mcp]"]

From GitHub (Development Version)

To install the latest unreleased version directly from the repository:

pip install "subscribeflow @ git+https://github.com/talent-factory/subscribe-flow.git#subdirectory=sdk/python"

Local Development

If you are working on the SDK itself, use an editable install so your changes take effect immediately without reinstalling:

# Editable install (changes to SDK source apply immediately)
pip install -e /path/to/subscribeflow/sdk/python

# With uv (faster)
uv pip install -e /path/to/subscribeflow/sdk/python

Quick Start

The following example shows how to initialize the client and create your first subscriber. You need an API key, which you can generate from your SubscribeFlow admin dashboard.

The client uses an async context manager to manage the HTTP connection. This ensures the connection is properly closed when you are done.

import asyncio
from subscribeflow import SubscribeFlowClient

async def main():
    async with SubscribeFlowClient(api_key="sf_live_xxx") as client:
        # Create a subscriber with tags and metadata
        subscriber = await client.subscribers.create(
            email="user@example.com",
            tags=["newsletter", "product-updates"],
            metadata={"source": "website"},
        )
        print(f"Created subscriber: {subscriber.id}")

asyncio.run(main())

Usage

Subscribers

Subscribers are the core entity in SubscribeFlow. Each subscriber represents a person identified by their email address. You can list, create, update, and delete subscribers, as well as manage their tag subscriptions and metadata.

# List subscribers with optional filters
# Returns paginated results; iterate directly over the result object
result = await client.subscribers.list(
    limit=50,
    status="active",
)
for subscriber in result:
    print(subscriber.email)

# Use cursor-based pagination to iterate through all pages
while result.next_cursor:
    result = await client.subscribers.list(cursor=result.next_cursor)
    for subscriber in result:
        print(subscriber.email)

# Get a single subscriber by ID
subscriber = await client.subscribers.get("subscriber-id")

# Update subscriber metadata
# Only the fields you pass will be changed; everything else stays the same
updated = await client.subscribers.update(
    "subscriber-id",
    metadata={"plan": "premium"},
)

# Permanently delete a subscriber and all associated data
await client.subscribers.delete("subscriber-id")

Tags

Tags represent topics or categories that subscribers can opt into. They are the building block of SubscribeFlow's granular preference management. Unlike traditional mailing lists, subscribers can actively discover and subscribe to tags they are interested in.

# Create a new tag with a human-readable name and optional description
tag = await client.tags.create(
    name="Product Updates",
    description="Get notified about new features",
)

# List all tags in your organization, including subscriber counts
tags = await client.tags.list()
for tag in tags:
    print(f"{tag.name}: {tag.subscriber_count} subscribers")

# Update a tag's description or other properties
await client.tags.update(
    "tag-id",
    description="Updated description",
)

# Delete a tag (subscribers will be automatically unsubscribed)
await client.tags.delete("tag-id")

Templates

Templates define the content and layout of your emails. SubscribeFlow uses MJML for responsive email rendering and supports Mustache-style variables for dynamic content.

# Create a new email template with MJML content
# Variables like {{company}} will be replaced when sending
template = await client.templates.create(
    name="Welcome Email",
    subject="Welcome to {{company}}!",
    mjml_content="<mjml><mj-body>...</mj-body></mjml>",
    category="transactional",
)

# List templates, optionally filtered by category
templates = await client.templates.list(category="transactional")

# Look up a template by its slug (useful for send operations)
template = await client.templates.get("welcome-email")

# Preview how a template will look with specific variable values
preview = await client.templates.preview(
    "template-id",
    variables={"company": "Acme Inc"},
)
print(preview.html)

# Update a template's subject or content
await client.templates.update("template-id", subject="New Subject")

# Delete a template
await client.templates.delete("template-id")

Email Send

Send individual transactional emails using a template. Each send requires a template slug and recipient. The optional idempotency_key prevents duplicate sends if the same request is retried.

# Send a transactional email to a single recipient
result = await client.emails.send(
    template_slug="welcome-email",
    to="user@example.com",
    variables={"company": "Acme Inc"},
    idempotency_key="unique-key-123",  # prevents duplicate sends on retry
)
print(f"Email queued: {result.id} (status: {result.status})")

Campaigns

Campaigns let you send emails to groups of subscribers based on tag filters. Create a draft, preview the recipient count, and then send it. Running campaigns can be cancelled.

from subscribeflow.models import TagFilter

# Create a campaign draft targeting subscribers with specific tags
campaign = await client.campaigns.create(
    name="February Newsletter",
    template_id="template-uuid",
    tag_filter={"include_tags": ["newsletter"], "match": "any"},
)

# List campaigns filtered by status
campaigns = await client.campaigns.list(status="draft")

# Preview how many subscribers will receive this campaign
count = await client.campaigns.count_recipients(include_tags=["newsletter"])
print(f"Will be sent to {count} subscribers")

# Send the campaign (moves from draft to sending)
result = await client.campaigns.send("campaign-id")
print(f"Sending to {result.total_recipients} recipients")

# Cancel a running campaign (emails already sent cannot be recalled)
await client.campaigns.cancel("campaign-id")

Email Triggers

Triggers automatically send emails in response to events. For example, you can send a welcome email whenever a new subscriber is created. Triggers can be activated or deactivated without deleting them.

# Create a trigger that fires when a subscriber is created
trigger = await client.triggers.create(
    event_type="subscriber.created",
    template_id="welcome-template-uuid",
    description="Send welcome email on signup",
)

# List all triggers
triggers = await client.triggers.list()

# Deactivate a trigger without deleting it
await client.triggers.update("trigger-id", is_active=False)

# Permanently delete a trigger
await client.triggers.delete("trigger-id")

Webhooks

Webhooks let your application receive real-time notifications when events occur in SubscribeFlow. Each webhook endpoint receives signed HTTP POST requests that you can verify using the signing secret.

# Register a new webhook endpoint for specific event types
webhook = await client.webhooks.create(
    url="https://your-app.com/webhooks/subscribeflow",
    events=["subscriber.created", "tag.subscribed"],
    description="Main webhook endpoint",
)

# Important: the signing secret is only returned on creation — store it securely
print(f"Signing secret: {webhook.secret}")

# List all registered webhook endpoints
webhooks = await client.webhooks.list()

# Update the events a webhook listens to
await client.webhooks.update(
    "webhook-id",
    events=["subscriber.created", "subscriber.deleted"],
)

# Test a webhook by sending a sample payload to your endpoint
result = await client.webhooks.test("webhook-id")
if result.success:
    print(f"Webhook working! Response time: {result.response_time_ms}ms")
else:
    print(f"Webhook failed: {result.error}")

# Rotate the signing secret (invalidates the old one immediately)
new_webhook = await client.webhooks.regenerate_secret("webhook-id")
print(f"New secret: {new_webhook.secret}")

# View delivery history to debug failed deliveries
deliveries = await client.webhooks.list_deliveries("webhook-id")
for d in deliveries:
    print(f"{d.event_type}: {d.status}")

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

# Retry a specific failed delivery
await client.webhooks.retry_delivery("webhook-id", "delivery-id")

# Remove a webhook endpoint
await client.webhooks.delete("webhook-id")

Preference Center

The Preference Center allows subscribers to manage their own email preferences. Generate a secure token for a subscriber, then use it to access their preferences. This powers the self-service UI where subscribers can subscribe to new tags, unsubscribe, export their data, or delete their account (GDPR compliance).

# Generate a time-limited preference center token for a subscriber
token_response = await client.subscribers.generate_token("subscriber-id")

# Create a preference center client using the token
pref_center = client.preference_center(token_response.token)

# Retrieve the subscriber's current preferences and all available tags
info = await pref_center.get_preferences()
print(f"Subscribed to: {len(info.subscribed_tags)} tags")
print(f"Available: {len(info.available_tags)} tags")

# Subscribe or unsubscribe from individual tags
result = await pref_center.subscribe_tag("tag-id")
result = await pref_center.unsubscribe_tag("tag-id")

# Export all subscriber data as JSON (GDPR Art. 20 — Right to Data Portability)
export = await pref_center.export_data()

# Permanently delete the subscriber account (GDPR Art. 17 — Right to Erasure)
await pref_center.delete_account()

Error Handling

All API errors are raised as typed exceptions that inherit from SubscribeFlowError. You can catch specific error types to handle different failure modes. Each exception includes the HTTP status code, a machine-readable error type, and a human-readable description.

from subscribeflow import (
    SubscribeFlowClient,
    SubscribeFlowError,
    AuthenticationError,
    AuthorizationError,
    NotFoundError,
    ValidationError,
    RateLimitError,
)

try:
    subscriber = await client.subscribers.get("non-existent-id")
except NotFoundError as e:
    print(f"Subscriber 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 RateLimitError as e:
    print("Rate limit exceeded, please retry later")
except AuthenticationError as e:
    print("Invalid API key")
except AuthorizationError as e:
    print("Insufficient permissions")
except SubscribeFlowError as e:
    print(f"API error ({e.status}): {e.detail}")

Configuration

The client accepts configuration options when initialized. Only the api_key is required.

client = SubscribeFlowClient(
    # Required: Your API key (starts with sf_live_ or sf_test_)
    api_key="sf_live_xxx",

    # Optional: API base URL (default: https://api.subscribeflow.net)
    # The SDK appends /api/v1/... automatically
    base_url="https://api.subscribeflow.net",

    # Optional: Request timeout in seconds (default: 30)
    timeout=30.0,
)

Local API Instance

When developing against a local SubscribeFlow backend, point the client to your local server:

client = SubscribeFlowClient(
    api_key="sf_dev_xxx",
    base_url="http://localhost:8000",  # SDK appends /api/v1/... automatically
)

MCP Server (Claude Integration)

The SDK includes an MCP (Model Context Protocol) server for integration with Claude Desktop and Claude Code. This allows Claude to interact with your SubscribeFlow account directly using natural language.

Quick Start

Add the following to your Claude Desktop/Code configuration (claude_desktop_config.json):

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

After configuration, you can ask Claude things like:

"Add user@example.com to the newsletter"

Claude will automatically call the appropriate SubscribeFlow tools.

Full documentation: MCP Integration Guide

Without Context Manager

If you prefer not to use the async context manager, remember to close the client manually when you are done:

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

Type Hints

This SDK is fully typed and works with mypy, pyright, and other type checkers out of the box:

from subscribeflow import SubscribeFlowClient, Subscriber, Tag

async def get_active_subscribers(client: SubscribeFlowClient) -> list[Subscriber]:
    result = await client.subscribers.list(status="active")
    return result.items

Links

• Dashboard
• Documentation
• API Reference
• MCP Integration
• Feedback

License

MIT