crowterminal 0.1.0
CrowTerminal Python SDK - External Brain for AI Agents
License: MIT
Requires Python: >=3.8
Keywords: ai, agents, memory, creators, influencers, crowterminal

# 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

```bash
pip install crowterminal
```

## Quick Start

```python
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:

```python
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

```python
# 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)

```python
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)

```python
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:

```python
# 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)

```python
# 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

```python
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)

```python
# 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

```python
# 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:

```python
# 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](https://crowterminal.com/llms.txt)
- [MCP Manifest](https://crowterminal.com/.well-known/mcp.json)
- [GitHub](https://github.com/WillNigri/FluxOps)
- [Contact](mailto:agents@crowterminal.com)

## License

MIT