subscribeflow 1.0.19

Python SDK for SubscribeFlow API - Email subscription management

subscribeflow

Official Python SDK for the SubscribeFlow API.

Installation

Von GitHub (Empfohlen)

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

# Mit MCP-Server (für Claude Desktop/Code Integration)
pip install "subscribeflow[mcp] @ git+https://github.com/talent-factory/subscribe-flow.git#subdirectory=sdk/python"

In requirements.txt:

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

Mit MCP-Support:

subscribeflow[mcp] @ git+https://github.com/talent-factory/subscribe-flow.git#subdirectory=sdk/python

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

[project.dependencies]
subscribeflow = { git = "https://github.com/talent-factory/subscribe-flow.git", subdirectory = "sdk/python" }

# Oder mit MCP-Support
subscribeflow = { git = "https://github.com/talent-factory/subscribe-flow.git", subdirectory = "sdk/python", extras = ["mcp"] }

Lokale Entwicklung

Für parallele Entwicklung von SDK und Anwendung:

# Editable Install (Änderungen am SDK wirken sofort)
pip install -e /pfad/zu/subscribeflow/sdk/python

# Mit uv (schneller)
uv pip install -e /pfad/zu/subscribeflow/sdk/python

# Oder relativ vom Projekt
pip install -e ../subscribeflow/sdk/python

In pyproject.toml für lokale Entwicklung:

[tool.poetry.dependencies]
subscribeflow = { path = "../subscribeflow/sdk/python", develop = true }

Quick Start

import asyncio
from subscribeflow import SubscribeFlowClient

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

asyncio.run(main())

Usage

Subscribers

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

# Pagination
while result.next_cursor:
    result = await client.subscribers.list(cursor=result.next_cursor)
    for subscriber in result:
        print(subscriber.email)

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

# Update a subscriber
updated = await client.subscribers.update(
    "subscriber-id",
    metadata={"plan": "premium"},
)

# Delete a subscriber
await client.subscribers.delete("subscriber-id")

Tags

# Create a tag
tag = await client.tags.create(
    name="Product Updates",
    description="Get notified about new features",
)

# List tags
tags = await client.tags.list()
for tag in tags:
    print(f"{tag.name}: {tag.subscriber_count} subscribers")

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

# Delete a tag
await client.tags.delete("tag-id")

Templates

# 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",
)

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

# Get a template by slug
template = await client.templates.get("welcome-email")

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

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

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

Email Send

# Send a transactional email
result = await client.emails.send(
    template_slug="welcome-email",
    to="[e-mail]",
    variables={"company": "Acme Inc"},
    idempotency_key="unique-key-123",
)
print(f"Email queued: {result.id} (status: {result.status})")

Campaigns

from subscribeflow.models import TagFilter

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

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

# Preview recipient count
count = await client.campaigns.count_recipients(include_tags=["newsletter"])
print(f"Will be sent to {count} subscribers")

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

# Cancel a running campaign
await client.campaigns.cancel("campaign-id")

Email Triggers

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

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

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

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

Webhooks

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

# Store the signing secret securely!
print(f"Signing secret: {webhook.secret}")

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

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

# Test a webhook
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 signing secret
new_webhook = await client.webhooks.regenerate_secret("webhook-id")
print(f"New secret: {new_webhook.secret}")

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

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

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

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

Preference Center

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

# Access the preference center with the token
pref_center = client.preference_center(token_response.token)

# Get subscriber preferences and 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 to a tag
result = await pref_center.subscribe_tag("tag-id")

# Unsubscribe from a tag
result = await pref_center.unsubscribe_tag("tag-id")

# Export all data (DSGVO)
export = await pref_center.export_data()

# Delete account (DSGVO)
await pref_center.delete_account()

Error Handling

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

client = SubscribeFlowClient(
    # Required: Your API key
    api_key="sf_live_xxx",

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

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

Lokale API-Instanz

Für Entwicklung gegen eine lokale SubscribeFlow-Instanz:

client = SubscribeFlowClient(
    api_key="sf_dev_xxx",
    base_url="http://localhost:8000",  # SDK fügt /api/v1/... automatisch hinzu
)

MCP Server (Claude Integration)

Das SDK enthält einen MCP-Server für die Integration mit Claude Desktop und Claude Code.

Quick Start

Claude Desktop/Code Konfiguration (claude_desktop_config.json):

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

Nach der Konfiguration kannst du Claude bitten:

"Füge [e-mail] zum Newsletter hinzu"

Claude ruft automatisch die passenden Tools auf.

Vollständige Dokumentation: MCP Integration Guide

Without Context Manager

If you prefer not to use the async context manager, remember to close the client:

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 great with mypy and other type checkers:

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

• API Documentation
• Getting Started Guide
• Webhook Integration

License

MIT