A comprehensive Python SDK for the Zetsubou.life API v2, providing easy access to AI-powered tools, encrypted file storage, chat capabilities, and webhooks.
- 🛠️ AI Tools: Execute 15+ AI-powered tools for image and video processing (dynamically loaded from API)
- 🔒 Encrypted Storage: Zero-knowledge file storage with client-side encryption
- 💬 Chat API: Conversational AI with multiple models (Llama 3.2, Qwen 2.5 VL, etc.)
- 📊 Job Management: Track, monitor, and download job results
- 🪝 Webhooks: Real-time event notifications for jobs, files, and storage
- 🔑 API Keys: Secure authentication with scope-based permissions
- 📈 Usage Tracking: Monitor storage, API usage, and job statistics
- ⚡ S3 Storage: High-performance cloud storage backend
- 🎨 NFT Generation: Create and manage NFT projects, layers, and collections
- 🔷 GraphQL API: Flexible query-based interface for fetching multiple resources
pip install zetsubou-sdkfrom zetsubou import ZetsubouClient
# Initialize the client
client = ZetsubouClient(api_key="ztb_live_your_api_key_here")
# List available tools
tools = client.tools.list()
print(f"Available tools: {[tool.name for tool in tools]}")
# Get account info
account = client.account.get_account()
print(f"Tier: {account.tier}, Username: {account.username}")
# Check storage quota
quota = client.account.get_storage_quota()
print(f"Storage: {quota.used_bytes / 1024 / 1024:.2f} MB / {quota.quota_bytes / 1024 / 1024:.2f} MB")Execute AI-powered tools with simple Python calls:
# Get tool metadata
tool = client.tools.get('remove_bg')
print(f"Tool: {tool.name} (Tier: {tool.required_tier})")
print(f"Supports batch: {tool.supports_batch}")
print(f"Supports audio: {tool.supports_audio}")
# Execute a tool (note: actual file execution requires multipart upload)
# This is for demonstration - see examples/tool_execution.py for working code
job = client.tools.execute(
tool_id="remove_bg",
files=[open("image.jpg", "rb")],
options={"model_name": "isnet-general-use"}
)
# Wait for completion
completed_job = client.jobs.wait_for_completion(job.id)
print(f"Job completed! Status: {completed_job.status}")
# Download results
results_zip = client.jobs.download_results(job.id)
with open("output.zip", "wb") as f:
f.write(results_zip)Track and manage asynchronous jobs:
# List recent jobs
jobs = client.jobs.list(limit=10, status="completed")
for job in jobs:
print(f"Job {job.id}: {job.tool_id} - {job.status}")
# Get specific job
job = client.jobs.get(job_id="your-job-id")
print(f"Status: {job.status}, Progress: {job.progress}%")
# Cancel a running job
success = client.jobs.cancel(job_id)
# Retry a failed job
retried_job = client.jobs.retry(job_id)
# Delete job and free storage
client.jobs.delete(job_id)Manage encrypted files with the Virtual File System:
# Upload files
with open("document.pdf", "rb") as f:
node = client.vfs.upload_file(file=f, parent_id=None, encrypt=True)
print(f"Uploaded: {node.name} ({node.size_bytes} bytes)")
# List files
files = client.vfs.list_nodes(node_type="file", limit=50)
for file in files:
print(f"{file.name} - {file.mime_type} - {file.size_bytes} bytes")
# Download files
content = client.vfs.download_file(node.id)
with open("downloaded.pdf", "wb") as f:
f.write(content)
# Create folders
folder = client.vfs.create_folder("My Projects", parent_id=None)
# Update node metadata
updated = client.vfs.update_node(node.id, name="New Name")
# Delete node
client.vfs.delete_node(node.id)
# Search files
images = client.vfs.get_images(limit=100)
videos = client.vfs.get_videos(limit=100)
# Delete workspace (deletes all associated folders)
success = client.vfs.delete_workspace(workspace_id="workspace-uuid")
if success:
print("Workspace deleted successfully")Create and manage AI conversations:
# List conversations
conversations = client.chat.list_conversations(limit=10)
# Create a conversation
conversation = client.chat.create_conversation(
title="AI Assistant",
model="llama3.2",
system_prompt="You are a helpful AI assistant."
)
# Send messages
message = client.chat.send_message(
conversation_uuid=conversation.uuid,
content="Hello! Can you help me process some images?"
)
# Get conversation history
messages = client.chat.get_messages(conversation.uuid, limit=50)
for msg in messages:
print(f"{msg.role}: {msg.content}")
# Delete conversation
client.chat.delete_conversation(conversation.uuid)
# Export conversation in various formats
# JSON export
json_data = client.chat.export_conversation(conversation.uuid, format="json")
# Markdown export
md_content = client.chat.export_conversation(conversation.uuid, format="md")
# HTML export (styled)
html_content = client.chat.export_conversation(conversation.uuid, format="html", output_path="conversation.html")
# PDF export
pdf_bytes = client.chat.export_conversation(conversation.uuid, format="pdf", output_path="conversation.pdf")Set up real-time event notifications:
# Create a webhook
webhook = client.webhooks.create(
url="https://your-app.com/webhooks/jobs",
events=["job.completed", "job.failed"],
secret="your_webhook_secret"
)
# List webhooks
webhooks = client.webhooks.list()
# Update webhook
updated = client.webhooks.update(
webhook_id=webhook.id,
enabled=True
)
# Test a webhook
client.webhooks.test(webhook.id)
# Delete webhook
client.webhooks.delete(webhook.id)Monitor usage and manage API keys:
# Get account information
account = client.account.get_account()
print(f"Tier: {account.tier}")
print(f"Username: {account.username}")
print(f"Email: {account.email}")
# Check storage usage
quota = client.account.get_storage_quota()
print(f"Used: {quota.used_bytes / 1024 / 1024:.2f} MB")
print(f"Limit: {quota.quota_bytes / 1024 / 1024:.2f} MB")
print(f"Usage: {quota.usage_percent}%")
print(f"Files: {quota.file_count}, Folders: {quota.folder_count}")
# Get storage breakdown
for category, data in quota.breakdown.items():
print(f"{category}: {data['bytes'] / 1024 / 1024:.2f} MB ({data['count']} files)")
# Get usage statistics
usage = client.account.get_usage_stats(period="30d")
# List API keys
api_keys = client.account.list_api_keys()
for key in api_keys:
print(f"{key['name']}: {key['scopes']}")
# Create API key
api_key = client.account.create_api_key(
name="My App Key",
scopes=["tools:execute", "files:read", "files:write"],
expires_at="2025-12-31T23:59:59Z",
bypass_drive_lock=False
)
print(f"New API key: {api_key['key']}")
# Delete API key
client.account.delete_api_key(key_id)Create and manage NFT projects, layers, and generations:
# Get NFT limits
limits = client.nft.get_limits()
print(f"Tier: {limits['tier']}, Max projects: {limits['limits']['max_projects']}")
# List projects
projects = client.nft.list_projects(include_archived=False)
for project in projects:
print(f"{project.name} - {project.layer_count} layers")
# Create project
project = client.nft.create_project(
name="Cool Apes Club",
collection_config={
"network": "solana",
"name": "Cool Apes Club",
"symbol": "CAC",
"description": "777 unique apes",
"seller_fee_basis_points": 500
},
generation_config={
"format": {"width": 2000, "height": 2000}
}
)
# Create layer
layer = client.nft.create_layer(
project_id=project.id,
name="Background",
order_index=0,
is_required=True
)
# Start generation
generation = client.nft.create_generation(
project_id=project.id,
total_pieces=777
)
# Poll for completion
import time
while generation.status != 'completed':
if generation.status == 'failed':
print(f"Generation failed: {generation.error_message}")
break
time.sleep(5)
generation = client.nft.get_generation(generation.id)
print(f"Status: {generation.status}")Get wallet information with dual-currency support:
# Get wallet info (SOL and USDC balances)
wallet = client.account.get_wallet_info()
print(f"Deposit Address: {wallet['deposit_address']}")
print(f"USDC Balance: ${wallet['balance']:.2f}")
print(f"SOL Balance: {wallet['sol_balance']:.4f} SOL")
print(f"Credits Remaining: {wallet['credits_remaining']}")
print(f"Pay-as-you-go: {'Enabled' if wallet['paug_enabled'] else 'Disabled'}")Execute GraphQL queries and mutations:
# Create shared folder shortcut via GraphQL
result = client.graphql.create_shared_folder_shortcut(
folder_id="shared-folder-uuid",
name="My Shortcut",
parent_id="parent-folder-uuid" # Optional
)
if result['data']['createSharedFolderShortcut']['success']:
shortcut = result['data']['createSharedFolderShortcut']['shortcut']
print(f"Shortcut created: {shortcut['name']} at {shortcut['path']}")
# Simple query
result = client.graphql.query('''
query {
viewer {
username
tier
}
jobs(limit: 5) {
jobs {
job_id
tool
status
}
}
}
''')
print(f"User: {result['data']['viewer']['username']}")
# Query with variables
result = client.graphql.query(
query='''
query GetJob($jobId: String!) {
jobs(job_id: $jobId) {
jobs {
job_id
status
}
}
}
''',
variables={'jobId': 'your-job-id'}
)
# Mutation
result = client.graphql.mutate('''
mutation {
createNftProject(
name: "My Project"
collectionConfig: {
network: "solana"
name: "My Project"
symbol: "MP"
}
) {
success
project {
id
name
}
}
}
''')The SDK provides access to 15+ AI-powered tools across 3 tiers. Tools are dynamically loaded from the API, so new tools become available automatically without SDK updates.
- Remove Background (
remove_bg): AI-powered background removal with 15 models - Polar Effect (
polar_effect): Create polaroid-style effects - P2Rotatooor (
p2rotatooor): Advanced polar effects with Node.js - The Process (
the_process): Image enhancement and processing - Batch Resize (
batch_resize): Resize multiple images at once
- Video Process (
video_process): Apply effects to videos - Video Background Remove (
video_bgremove): Remove backgrounds from videos - Video Batch Resize (
video_batch_resize): Resize multiple videos - Extract Frames (
extract_frames): Extract frames from videos - Clip Maker (
clip_maker): AI music video generator with audio support - PSD Layer Extractor (
psd_extractor): Extract individual layers from Photoshop PSD files
- Datamosher (
datamosher): Glitch art video effects with audio - Datamelter (
datamelter): Melting video effects with audio - Background-Foreground Matcher (
bgfg_matcher): Match and composite images - Batch Bloomer (
batch_bloomer): Apply bloom effects to multiple images/videos
Note: Tool availability is determined by your account tier and API key scopes. Use
client.tools.list()to see all tools available to your account.
Comprehensive error handling with custom exceptions:
from zetsubou import (
ZetsubouClient,
ZetsubouError,
AuthenticationError,
RateLimitError,
ValidationError,
NotFoundError,
ServerError
)
try:
job = client.tools.execute("invalid_tool", files=["test.jpg"])
except AuthenticationError as e:
print(f"Authentication failed: {e.message}")
print(f"Error code: {e.code}")
except ValidationError as e:
print(f"Validation error: {e.message}")
except RateLimitError as e:
print(f"Rate limited. Retry after {e.retry_after} seconds")
except NotFoundError as e:
print(f"Resource not found: {e.message}")
except ServerError as e:
print(f"Server error: {e.message}")
except ZetsubouError as e:
print(f"SDK error: {e.message}")Configure the client with various options:
client = ZetsubouClient(
api_key="ztb_live_your_key",
base_url="https://zetsubou.life", # or custom endpoint
timeout=60, # Request timeout in seconds (default: 30)
retry_attempts=5 # Number of retry attempts (default: 3)
)
# Use context manager for automatic cleanup
with ZetsubouClient(api_key="your_key") as client:
tools = client.tools.list()
# client.close() called automaticallyCheck out the examples directory for complete working examples:
- basic_usage.py - Get started with account info and tool listing
- tool_execution.py - Execute tools and manage jobs
- file_management.py - Upload, download, and manage VFS files
- storage_monitor.py - Monitor storage usage and quota
Full API documentation is available at docs.zetsubou.life/sdk/python.
- Python 3.8 or higher
requests>= 2.25.0urllib3>= 1.26.0
We welcome contributions! Please see our Contributing Guide for details.
This project is licensed under the MIT License - see the LICENSE file for details.
- 📧 Email: support@zetsubou.life
- 📖 Documentation: docs.zetsubou.life
- 🐛 Issues: GitHub Issues
See CHANGELOG.md for a list of changes and version history.