amplitude

Original source / Raw SKILL.md

---
name: amplitude
description: "Amplitude product analytics — track events, analyze user behavior, run cohort analysis, manage user properties, and query funnel/retention data via the Amplitude API. Understand product usage, measure feature adoption, and analyze user journeys. Built for AI agents — Python stdlib only, zero dependencies. Use for product analytics, user behavior tracking, funnel analysis, retention analysis, and cohort segmentation."
homepage: https://www.agxntsix.ai
license: MIT
compatibility: Python 3.10+ (stdlib only — no dependencies)
metadata: {"openclaw": {"emoji": "📉", "requires": {"env": ["AMPLITUDE_API_KEY", "AMPLITUDE_SECRET_KEY"]}, "primaryEnv": "AMPLITUDE_API_KEY", "homepage": "https://www.agxntsix.ai"}}
---

# 📉 Amplitude

Amplitude product analytics — track events, analyze user behavior, run cohort analysis, manage user properties, and query funnel/retention data via the Amplitude API.

## Features

- **Event tracking** — log user events with properties
- **User analytics** — active users, sessions, engagement
- **Funnel analysis** — conversion through event sequences
- **Retention analysis** — user return rates over time
- **Cohort management** — create and manage user cohorts
- **User properties** — set and query user attributes
- **Revenue analytics** — LTV, ARPU, revenue tracking
- **Segmentation** — query by properties and events
- **Event segmentation** — event counts and breakdowns
- **Dashboard export** — export chart data

## Requirements

| Variable | Required | Description |
|----------|----------|-------------|
| `AMPLITUDE_API_KEY` | ✅ | API key/token for Amplitude |
| `AMPLITUDE_SECRET_KEY` | ✅ | Amplitude secret key for Export/Dashboard APIs |

## Quick Start

```bash
# Track an event
python3 {baseDir}/scripts/amplitude.py track '{"user_id":"user123","event_type":"purchase","event_properties":{"amount":29.99}}'
```

```bash
# Track batch events
python3 {baseDir}/scripts/amplitude.py track-batch events.json
```

```bash
# Set user properties
python3 {baseDir}/scripts/amplitude.py identify '{"user_id":"user123","user_properties":{"plan":"pro","company":"Acme"}}'
```

```bash
# Get active user counts
python3 {baseDir}/scripts/amplitude.py active-users --start 2026-01-01 --end 2026-02-01
```



## Commands

### `track`
Track an event.
```bash
python3 {baseDir}/scripts/amplitude.py track '{"user_id":"user123","event_type":"purchase","event_properties":{"amount":29.99}}'
```

### `track-batch`
Track batch events.
```bash
python3 {baseDir}/scripts/amplitude.py track-batch events.json
```

### `identify`
Set user properties.
```bash
python3 {baseDir}/scripts/amplitude.py identify '{"user_id":"user123","user_properties":{"plan":"pro","company":"Acme"}}'
```

### `active-users`
Get active user counts.
```bash
python3 {baseDir}/scripts/amplitude.py active-users --start 2026-01-01 --end 2026-02-01
```

### `events`
Get event data.
```bash
python3 {baseDir}/scripts/amplitude.py events --start 2026-01-01 --end 2026-02-01 --event purchase
```

### `funnel`
Run funnel analysis.
```bash
python3 {baseDir}/scripts/amplitude.py funnel '{"events":[{"event_type":"page_view"},{"event_type":"signup"},{"event_type":"purchase"}]}' --start 2026-01-01 --end 2026-02-01
```

### `retention`
Retention analysis.
```bash
python3 {baseDir}/scripts/amplitude.py retention --start 2026-01-01 --end 2026-02-01
```

### `cohorts`
List cohorts.
```bash
python3 {baseDir}/scripts/amplitude.py cohorts
```

### `cohort-get`
Get cohort details.
```bash
python3 {baseDir}/scripts/amplitude.py cohort-get abc123
```

### `revenue`
Revenue analysis.
```bash
python3 {baseDir}/scripts/amplitude.py revenue --start 2026-01-01 --end 2026-02-01
```

### `user-search`
Search for a user.
```bash
python3 {baseDir}/scripts/amplitude.py user-search "[email protected]"
```

### `user-activity`
Get user activity.
```bash
python3 {baseDir}/scripts/amplitude.py user-activity user123
```

### `segments`
Event segmentation query.
```bash
python3 {baseDir}/scripts/amplitude.py segments --event purchase --group-by platform --start 2026-01-01 --end 2026-02-01
```


## Output Format

All commands output JSON by default. Add `--human` for readable formatted output.

```bash
# JSON (default, for programmatic use)
python3 {baseDir}/scripts/amplitude.py track --limit 5

# Human-readable
python3 {baseDir}/scripts/amplitude.py track --limit 5 --human
```

## Script Reference

| Script | Description |
|--------|-------------|
| `{baseDir}/scripts/amplitude.py` | Main CLI — all Amplitude operations |

## Data Policy

This skill **never stores data locally**. All requests go directly to the Amplitude API and results are returned to stdout. Your data stays on Amplitude servers.

## Credits
---
Built by [M. Abidi](https://www.linkedin.com/in/mohammad-ali-abidi) | [agxntsix.ai](https://www.agxntsix.ai)
[YouTube](https://youtube.com/@aiwithabidi) | [GitHub](https://github.com/aiwithabidi)
Part of the **AgxntSix Skill Suite** for OpenClaw agents.

📅 **Need help setting up OpenClaw for your business?** [Book a free consultation](https://cal.com/agxntsix/abidi-openclaw)


---

## Skill Companion Files

> Additional files collected from the skill directory layout.

### _meta.json

```json
{
  "owner": "aiwithabidi",
  "slug": "amplitude",
  "displayName": "Amplitude",
  "latest": {
    "version": "1.0.0",
    "publishedAt": 1771452314623,
    "commit": "https://github.com/openclaw/skills/commit/af76ae3fa82dca9e1f77dbc40e3a011878a58551"
  },
  "history": []
}

```

### scripts/amplitude.py

```python
#!/usr/bin/env python3
"""Amplitude CLI — comprehensive API integration for AI agents.

Full CRUD operations, search, reporting, and automation.
Zero dependencies beyond Python stdlib.
"""

import argparse
import json
import os
import sys
import urllib.request
import urllib.error
import urllib.parse
from datetime import datetime, timezone

API_BASE = "https://amplitude.com/api/2"


def get_token():
    """Get API token from environment."""
    token = os.environ.get("AMPLITUDE_API_KEY", "")
    if not token:
        env_path = os.path.join(
            os.environ.get("WORKSPACE", os.path.expanduser("~/.openclaw/workspace")),
            ".env"
        )
        if os.path.exists(env_path):
            with open(env_path) as f:
                for line in f:
                    line = line.strip()
                    if line.startswith("AMPLITUDE_API_KEY="):
                        token = line.split("=", 1)[1].strip().strip('"').strip("'")
                        break
    if not token:
        print(f"Error: AMPLITUDE_API_KEY not set", file=sys.stderr)
        sys.exit(1)
    return token


def api(method, path, data=None, params=None):
    """Make an API request."""
    token = get_token()
    url = f"{API_BASE}{path}"
    if params:
        qs = urllib.parse.urlencode({k: v for k, v in params.items() if v is not None})
        if qs:
            url = f"{url}?{qs}"
    body = json.dumps(data).encode() if data else None
    req = urllib.request.Request(url, data=body, method=method)
    req.add_header("Authorization", f"Bearer {token}")
    req.add_header("Content-Type", "application/json")
    req.add_header("Accept", "application/json")
    try:
        resp = urllib.request.urlopen(req, timeout=30)
        raw = resp.read().decode()
        return json.loads(raw) if raw.strip() else {"ok": True}
    except urllib.error.HTTPError as e:
        err_body = e.read().decode()
        print(json.dumps({"error": True, "code": e.code, "message": err_body}), file=sys.stderr)
        sys.exit(1)


def output(data, human=False):
    """Output data as JSON or human-readable."""
    if human and isinstance(data, list):
        for item in data:
            if isinstance(item, dict):
                for k, v in item.items():
                    print(f"  {k}: {v}")
                print()
            else:
                print(item)
    elif human and isinstance(data, dict):
        for k, v in data.items():
            print(f"  {k}: {v}")
    else:
        print(json.dumps(data, indent=2, default=str))


def cmd_track(args):
    """Track an event."""
    params = {}
    if hasattr(args, 'limit') and args.limit:
        params["limit"] = args.limit
    if hasattr(args, 'id') and args.id:
        data = api("POST", f"/track/{args.id}")
    else:
        data = api("POST", "/track", params=params)
    output(data, getattr(args, 'human', False))

def cmd_track_batch(args):
    """Track batch events."""
    params = {}
    if hasattr(args, 'limit') and args.limit:
        params["limit"] = args.limit
    if hasattr(args, 'id') and args.id:
        data = api("POST", f"/track/{args.id}")
    else:
        data = api("POST", "/track/batch", params=params)
    output(data, getattr(args, 'human', False))

def cmd_identify(args):
    """Set user properties."""
    params = {}
    if hasattr(args, 'limit') and args.limit:
        params["limit"] = args.limit
    if hasattr(args, 'id') and args.id:
        data = api("GET", f"/identify/{args.id}")
    else:
        data = api("GET", "/identify", params=params)
    output(data, getattr(args, 'human', False))

def cmd_active_users(args):
    """Get active user counts."""
    params = {}
    if hasattr(args, 'limit') and args.limit:
        params["limit"] = args.limit
    if hasattr(args, 'id') and args.id:
        data = api("GET", f"/active/{args.id}")
    else:
        data = api("GET", "/active/users", params=params)
    output(data, getattr(args, 'human', False))

def cmd_events(args):
    """Get event data."""
    params = {}
    if hasattr(args, 'limit') and args.limit:
        params["limit"] = args.limit
    if hasattr(args, 'id') and args.id:
        data = api("GET", f"/events/{args.id}")
    else:
        data = api("GET", "/events", params=params)
    output(data, getattr(args, 'human', False))

def cmd_funnel(args):
    """Run funnel analysis."""
    params = {}
    if hasattr(args, 'limit') and args.limit:
        params["limit"] = args.limit
    if hasattr(args, 'id') and args.id:
        data = api("GET", f"/funnel/{args.id}")
    else:
        data = api("GET", "/funnel", params=params)
    output(data, getattr(args, 'human', False))

def cmd_retention(args):
    """Retention analysis."""
    params = {}
    if hasattr(args, 'limit') and args.limit:
        params["limit"] = args.limit
    if hasattr(args, 'id') and args.id:
        data = api("GET", f"/retention/{args.id}")
    else:
        data = api("GET", "/retention", params=params)
    output(data, getattr(args, 'human', False))

def cmd_cohorts(args):
    """List cohorts."""
    params = {}
    if hasattr(args, 'limit') and args.limit:
        params["limit"] = args.limit
    if hasattr(args, 'id') and args.id:
        data = api("GET", f"/cohorts/{args.id}")
    else:
        data = api("GET", "/cohorts", params=params)
    output(data, getattr(args, 'human', False))

def cmd_cohort_get(args):
    """Get cohort details."""
    params = {}
    if hasattr(args, 'limit') and args.limit:
        params["limit"] = args.limit
    if hasattr(args, 'id') and args.id:
        data = api("GET", f"/cohort/{args.id}")
    else:
        data = api("GET", "/cohort/get", params=params)
    output(data, getattr(args, 'human', False))

def cmd_revenue(args):
    """Revenue analysis."""
    params = {}
    if hasattr(args, 'limit') and args.limit:
        params["limit"] = args.limit
    if hasattr(args, 'id') and args.id:
        data = api("GET", f"/revenue/{args.id}")
    else:
        data = api("GET", "/revenue", params=params)
    output(data, getattr(args, 'human', False))

def cmd_user_search(args):
    """Search for a user."""
    params = {}
    if hasattr(args, 'limit') and args.limit:
        params["limit"] = args.limit
    if hasattr(args, 'id') and args.id:
        data = api("GET", f"/user/{args.id}")
    else:
        data = api("GET", "/user/search", params=params)
    output(data, getattr(args, 'human', False))

def cmd_user_activity(args):
    """Get user activity."""
    params = {}
    if hasattr(args, 'limit') and args.limit:
        params["limit"] = args.limit
    if hasattr(args, 'id') and args.id:
        data = api("GET", f"/user/{args.id}")
    else:
        data = api("GET", "/user/activity", params=params)
    output(data, getattr(args, 'human', False))

def cmd_segments(args):
    """Event segmentation query."""
    params = {}
    if hasattr(args, 'limit') and args.limit:
        params["limit"] = args.limit
    if hasattr(args, 'id') and args.id:
        data = api("GET", f"/segments/{args.id}")
    else:
        data = api("GET", "/segments", params=params)
    output(data, getattr(args, 'human', False))


COMMANDS = {
    "track": cmd_track,
    "track-batch": cmd_track_batch,
    "identify": cmd_identify,
    "active-users": cmd_active_users,
    "events": cmd_events,
    "funnel": cmd_funnel,
    "retention": cmd_retention,
    "cohorts": cmd_cohorts,
    "cohort-get": cmd_cohort_get,
    "revenue": cmd_revenue,
    "user-search": cmd_user_search,
    "user-activity": cmd_user_activity,
    "segments": cmd_segments,
}


def main():
    parser = argparse.ArgumentParser(
        description="Amplitude CLI — AI agent integration",
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument("command", choices=list(COMMANDS.keys()), help="Command to run")
    parser.add_argument("args", nargs="*", help="Command arguments")
    parser.add_argument("--human", action="store_true", help="Human-readable output")
    parser.add_argument("--limit", type=int, help="Limit results")
    parser.add_argument("--id", help="Resource ID")
    parser.add_argument("--from", dest="from_date", help="Start date")
    parser.add_argument("--to", dest="to_date", help="End date")
    parser.add_argument("--status", help="Filter by status")
    parser.add_argument("--sort", help="Sort field")
    parser.add_argument("--query", help="Search query")

    parsed = parser.parse_args()
    cmd_func = COMMANDS[parsed.command]
    cmd_func(parsed)


if __name__ == "__main__":
    main()

```