CycleTask Python SDK

The official Python SDK for the CycleTask Mail Forwarding API.

Installation

pip install cycletask

Quick Start

from cycletask import CycleTask

client = CycleTask(api_key="ct_live_your_api_key")

# List your domains
domains = client.domains.list()
print(domains)

# Create a forwarding alias
alias = client.aliases.create(
    alias="hello",
    domain_id="your_domain_id",
    destinations=["me@gmail.com"],
)
print(alias)

Async Support

import asyncio
from cycletask import AsyncCycleTask

async def main():
    client = AsyncCycleTask(api_key="ct_live_your_api_key")

    domains = await client.domains.list()
    print(domains)

    await client.close()

asyncio.run(main())

Configuration

client = CycleTask(
    api_key="ct_live_your_api_key",
    base_url="https://api.task-cycle.com/api",  # default
    timeout=30.0,                                # seconds, default
    max_retries=3,                               # default
)

Parameter	Type	Default	Description
api_key	str	None	Your API key. Required for authenticated endpoints.
base_url	str	https://api.task-cycle.com/api	API base URL.
timeout	float	30.0	Request timeout in seconds.
max_retries	int	3	Max retries on 429/5xx with exponential backoff.

API Reference

Domains

Manage your email domains.

# List all domains
domains = client.domains.list()

# Create a domain
domain = client.domains.create(domain="example.com")

# Get domain details
domain = client.domains.get("domain_id")

# Delete a domain
client.domains.delete("domain_id")

# Trigger DNS verification
result = client.domains.verify("domain_id")

# Get required DNS records
records = client.domains.get_dns_records("domain_id")

Aliases

Manage email forwarding aliases.

# List aliases (with pagination and filtering)
aliases = client.aliases.list(
    page=1,
    limit=10,
    search="hello",
    domain="domain_id",
    status="active",
    sort="created_at:desc",
)

# Create an alias
alias = client.aliases.create(
    alias="hello",
    domain_id="domain_id",
    destinations=["me@gmail.com", "backup@gmail.com"],
    description="My forwarding alias",
    privacy_mode=False,
)

# Update an alias
client.aliases.update("alias_id", is_active=False)
client.aliases.update("alias_id", destinations=["new@gmail.com"])

# Delete an alias
client.aliases.delete("alias_id")

# Get forwarding logs for an alias
logs = client.aliases.get_logs("alias_id", page=1, limit=50)

Stats

Access dashboard analytics and email logs.

# Dashboard overview
overview = client.stats.overview()

# Chart data (email volume over time)
chart = client.stats.chart(days=30)

# Global email logs
logs = client.stats.logs(
    page=1,
    limit=20,
    status="delivered",
    search="sender@example.com",
)

Status (Public)

Check service health. These endpoints do not require authentication.

# Can create a client without an API key for status checks
client = CycleTask()

# Service health
status = client.status.get()

# 90-day uptime history
uptime = client.status.uptime()

# Incident reports
incidents = client.status.incidents()

Error Handling

The SDK raises typed exceptions for different error conditions:

from cycletask import (
    CycleTask,
    CycleTaskError,
    AuthenticationError,
    NotFoundError,
    ValidationError,
    RateLimitError,
)

client = CycleTask(api_key="ct_live_your_api_key")

try:
    domain = client.domains.get("nonexistent_id")
except AuthenticationError as e:
    print(f"Bad API key: {e.message}")
except NotFoundError as e:
    print(f"Not found: {e.message}")
except ValidationError as e:
    print(f"Invalid input: {e.message}")
    print(f"Field errors: {e.errors}")
except RateLimitError as e:
    print(f"Rate limited. Retry after: {e.retry_after}s")
except CycleTaskError as e:
    print(f"API error {e.status_code}: {e.message}")

Exception Hierarchy

Exception	HTTP Status	Description
CycleTaskError	—	Base exception for all errors
AuthenticationError	401	Invalid or missing API key
PermissionError	403	Insufficient permissions
NotFoundError	404	Resource not found
ValidationError	400, 422	Request validation failed
ConflictError	409	Resource conflict
RateLimitError	429	Rate limit exceeded
InternalServerError	5xx	Server-side error
ConnectionError	—	Network connection failure
TimeoutError	—	Request timed out

Context Manager

Both sync and async clients support context managers for automatic cleanup:

# Sync
with CycleTask(api_key="ct_live_your_api_key") as client:
    domains = client.domains.list()

# Async
async with AsyncCycleTask(api_key="ct_live_your_api_key") as client:
    domains = await client.domains.list()

Automatic Retries

The SDK automatically retries requests that fail with:

• 429 (Rate Limited) — respects the Retry-After header
• 500, 502, 503, 504 (Server Errors)

Retries use exponential backoff starting at 0.5s, doubling each attempt, capped at 30s. The default maximum is 3 retry attempts.

Type Hints

The SDK ships with complete type annotations and a py.typed marker for PEP 561 compliance. All response types are available as TypedDict definitions:

from cycletask import Domain, Alias, StatsOverview

Requirements

• Python 3.8+
• httpx >= 0.24.0

License

MIT — see LICENSE for details.