CrowTerminal Python SDK

External Brain for AI Agents - Persistent memory for AI agents working with creators.

While your agent stores 10-50 lines of context, CrowTerminal stores 6 months of versioned history.

Installation

pip install crowterminal

Quick Start

from crowterminal import CrowTerminal

# Initialize with your API key
client = CrowTerminal("ct_your_api_key")

# Get memory for a creator
skill = client.memory.get("client_123")
print(f"Niche: {skill.primary_niche}")
print(f"Engagement: {skill.avg_engagement}%")
print(f"Best hooks: {skill.hook_patterns}")

Self-Registration

Don't have an API key? Register programmatically:

from crowterminal import CrowTerminal

# This creates a new API key and returns an initialized client
client = CrowTerminal.register(
    agent_name="MyBot",
    agent_description="Content optimization agent"
)
# API key is printed - save it!

Core Features

Memory Operations

# Get current skill
skill = client.memory.get("client_123")

# Get version history
versions = client.memory.get_versions("client_123", limit=10)

# Compare versions
diff = client.memory.get_diff("client_123", from_version=5, to_version=10)

# Track a field over time
pattern = client.memory.get_pattern("client_123", field="avgEngagement")
print(f"Trend: {pattern['trend']}")  # increasing, decreasing, stable

Validate Before Changing (Prevent Mistakes)

result = client.memory.validate("client_123", [
    {"field": "hookPatterns", "oldValue": ["POV"], "newValue": ["tutorial"]}
])

if result.validation == "blocked":
    print("Don't make this change!")
    for warning in result.warnings:
        print(f"  - {warning['message']}")

Engagement Analysis (The Killer Feature)

analysis = client.memory.engagement_analysis("client_123", {
    "hookPatterns": ["confession"],
    "contentStyle": "casual",
    "primaryNiche": "fitness"
})

print(f"Peak engagement: {analysis.peak_engagement}%")
print(f"Your similarity to top performers: {analysis.similarity_to_top}")

for rec in analysis.recommendations:
    print(f"Recommendation: {rec}")

Data Ingestion (Push Your Data)

Push platform data we can't access via API:

# Push retention data from TikTok Studio
client.data.ingest(
    client_id="client_123",
    platform="TIKTOK",
    data_type="retention",
    video_id="video_456",
    data={
        "retentionCurve": [100, 95, 88, 75, 60, 45, 30],
        "avgWatchTime": 12.5,
        "completionRate": 0.30
    }
)

# Push demographics
client.data.ingest(
    client_id="client_123",
    platform="TIKTOK",
    data_type="demographics",
    data={
        "ageGroups": {"18-24": 45, "25-34": 35, "35-44": 15, "45+": 5},
        "genderSplit": {"male": 40, "female": 58, "other": 2},
        "topCountries": ["BR", "US", "PT"]
    }
)

# Bulk ingest (up to 50 items)
client.data.ingest_bulk([
    {"clientId": "client_123", "platform": "TIKTOK", "dataType": "retention", "data": {...}},
    {"clientId": "client_123", "platform": "TIKTOK", "dataType": "demographics", "data": {...}},
])

Intelligence (Read-Only)

# Get creator profile
profile = client.intelligence.get_profile("client_123")

# Get hook recommendations
hooks = client.intelligence.get_hooks("client_123", count=5)

# Get optimal posting times
timing = client.intelligence.get_timing("client_123")

# Get platform algorithm insights
intel = client.intelligence.get_platform_intel(["TIKTOK", "INSTAGRAM"])

Error Handling

from crowterminal import (
    CrowTerminal,
    AuthenticationError,
    RateLimitError,
    ResourceNotFoundError,
)

client = CrowTerminal("ct_your_api_key")

try:
    skill = client.memory.get("client_123")
except AuthenticationError:
    print("Invalid API key")
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after} seconds")
except ResourceNotFoundError:
    print("Client not found")

Webhooks (Async Notifications)

# Register a webhook
webhook = client.webhooks.register(
    url="https://your-server.com/webhook",
    events=["skill.updated", "data.ingested"]
)
print(f"Webhook ID: {webhook['id']}")
print(f"Secret (save this!): {webhook['secret']}")

# List webhooks
webhooks = client.webhooks.list()

# Delete a webhook
client.webhooks.delete(webhook_id="wh_xxx")

Service Status

# Check service health (no auth required)
status = client.status.get()
print(f"Service status: {status['status']}")
print(f"Database: {status['services']['database']['status']}")

Sandbox Testing

Use the sandbox endpoints for testing without affecting real data:

# Test without auth
import requests

# Get mock client data
response = requests.get("https://api.crowterminal.com/api/agent/sandbox/client")
print(response.json())

# Test validation
response = requests.post(
    "https://api.crowterminal.com/api/agent/sandbox/validate",
    json={"proposedChanges": [{"field": "hookPatterns", "newValue": ["tutorial"]}]}
)
print(response.json())  # Will show "blocked" response

Valid Data Types

TikTok

• retention, demographics, traffic_sources, watch_time
• audience_activity, follower_growth, video_performance
• sound_performance, hashtag_performance

Instagram

• retention, demographics, reach_sources, watch_time
• audience_activity, follower_growth, content_interactions
• story_metrics, reel_metrics

YouTube

• retention, demographics, traffic_sources, watch_time
• audience_activity, subscriber_growth, click_through_rate
• impression_sources, end_screen_performance

Links

• Full Documentation
• MCP Manifest
• GitHub
• Contact

License

MIT