zetsubou-sdk 1.1.0

Python SDK for the Zetsubou.life API v2

Zetsubou.life Python SDK

A comprehensive Python SDK for the Zetsubou.life API v2, providing easy access to AI-powered tools, encrypted file storage, chat capabilities, and webhooks.

Features

• 🛠️ 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

Installation

pip install zetsubou-sdk

Quick Start

from 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")

Core Features

🛠️ Tool Execution

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)

📋 Job Management

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)

🔒 File Storage (VFS)

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)

💬 Chat Integration

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_id=conversation.id,
    content="Hello! Can you help me process some images?"
)

# Get conversation history
messages = client.chat.get_messages(conversation.id, limit=50)
for msg in messages:
    print(f"{msg.role}: {msg.content}")

# Delete conversation
client.chat.delete_conversation(conversation.id)

🪝 Webhooks

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)

📊 Account Management

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)

🎨 NFT Generation

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}")

🔷 GraphQL API

Execute GraphQL queries and mutations:

# 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
            }
        }
    }
''')

Available Tools

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.

Basic Tools (Free Tier)

• 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 Tools (Creator Tier)

• 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

Advanced Tools (Pro Tier)

• 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.

Error Handling

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}")

Configuration

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 automatically

Examples

Check 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

API Reference

Full API documentation is available at docs.zetsubou.life/sdk/python.

Requirements

• Python 3.8 or higher
• requests >= 2.25.0
• urllib3 >= 1.26.0

Contributing

We welcome contributions! Please see our Contributing Guide for details.

License

This project is licensed under the MIT License - see the LICENSE file for details.

Support

• 📧 Email: [e-mail]
• 📖 Documentation: docs.zetsubou.life
• 🐛 Issues: GitHub Issues

Changelog

See CHANGELOG.md for a list of changes and version history.