Workflow Recipes

# Step 2: Poll until everything (scrape + AI analysis) is done
while True:
    result = requests.get(
        f'https://api.virlo.ai/v1/agents/{agent_id}',
        headers={'Authorization': 'Bearer YOUR_API_KEY'}
    ).json()['data']

    if result.get('finalized'):
        break
    if result['latest_run']['status'] == 'failed':
        raise Exception('Search failed')
    # latest_run 'completed' or 'partial_failure' with finalized=False just
    # means secondary AI jobs are still running — keep polling.

    time.sleep(60)

# Read the AI intelligence report (now guaranteed populated)
print(result['analysis'])

# Top videos sorted by views. Read-time filters (min_views, start_date, etc.)
# are free — slice the collection as many ways as you like.
videos = requests.get(
    f'https://api.virlo.ai/v1/agents/{agent_id}/videos',
    headers={'Authorization': 'Bearer YOUR_API_KEY'},
    params={'limit': 50, 'order_by': 'views', 'sort': 'desc'}
).json()['data']['videos']

# Meta ad campaigns
ads = requests.get(
    f'https://api.virlo.ai/v1/agents/{agent_id}/ads',
    headers={'Authorization': 'Bearer YOUR_API_KEY'},
    params={'limit': 50}
).json()['data']['ads']

# Rising creators outperforming their follower count.
# Sort by weighted_score (ln(outlier_ratio) x ln(followers)) — it balances
# outperformance against audience size and is what Virlo's own UI uses.
outliers = requests.get(
    f'https://api.virlo.ai/v1/agents/{agent_id}/creators/outliers',
    headers={'Authorization': 'Bearer YOUR_API_KEY'},
    params={'limit': 20, 'order_by': 'weighted_score', 'sort': 'desc'}
).json()['data']['outliers']

# Get the top-performing video URL from the creator's videos
top_video = max(creator['videos'], key=lambda v: v['views'])

# Start video outlier analysis
response = requests.post(
    'https://api.virlo.ai/v1/satellite/video-outlier',
    headers={'Authorization': 'Bearer YOUR_API_KEY'},
    json={'url': top_video['url'], 'platform': 'tiktok'}
)
job_id = response.json()['data']['job_id']

# Poll until complete
while True:
    result = requests.get(
        f'https://api.virlo.ai/v1/satellite/video-outlier/status/{job_id}',
        headers={'Authorization': 'Bearer YOUR_API_KEY'}
    ).json()['data']

    if result['status'] == 'completed':
        break
    time.sleep(10)

analysis = result['result']['analysis']
print(f"Outlier score: {analysis['outlier_score']}")       # vs the creator's own baseline
print(f"Percentile: {analysis['percentile']}")             # 0-100 within their catalogue
print(f"Label: {analysis['performance_label']}")           # average | above_average | viral | mega_viral

import requests

trends = requests.get(
    'https://api.virlo.ai/v1/trends/digest',
    headers={'Authorization': 'Bearer YOUR_API_KEY'}
).json()['data']

for group in trends:
    print(f"\n{group['title']}")
    for item in group['trends']:
        print(f"  #{item['ranking']}: {item['trend']['name']}")

# (Legacy: POST /v1/comet.)
response = requests.post(
    'https://api.virlo.ai/v1/agents',
    headers={'Authorization': 'Bearer YOUR_API_KEY'},
    json={
        'is_recurring': True,
        'intent': 'Monitor Devil Wears Prada 2 buzz for a merch brand',
        'name': 'Trend Monitoring - Devil Wears Prada 2',
        'keywords': ['Devil Wears Prada 2', 'Devil Wears Prada sequel', 'Prada 2 merch'],
        'platforms': ['youtube', 'tiktok', 'instagram'],
        'cadence': 'weekly',
        'meta_ads_enabled': True
    }
)
agent = response.json()['data']
print(f"Next run: {agent.get('next_run_at', 'scheduled')}")

import requests
from datetime import date, timedelta

headers = {'Authorization': 'Bearer YOUR_API_KEY'}
end = date.today().isoformat()
start = (date.today() - timedelta(days=30)).isoformat()

# Top hashtags by views ($0.05)
hashtags = requests.get(
    'https://api.virlo.ai/v1/hashtags',
    headers=headers,
    params={'start_date': start, 'end_date': end, 'limit': 20, 'order_by': 'views', 'sort': 'desc'}
).json()['data']

# Drill into a specific hashtag ($0.05)
perf = requests.get(
    f'https://api.virlo.ai/v1/hashtags/fyp/performance',
    headers=headers,
    params={'start_date': start, 'end_date': end}
).json()['data']

print(f"#fyp: {perf['video_count']} videos, {perf['total_views']:,} views, avg {perf['avg_views']:,.0f} views/video")

# Platform-specific hashtags ($0.05 each)
tiktok = requests.get(
    'https://api.virlo.ai/v1/tiktok/hashtags',
    headers=headers,
    params={'start_date': start, 'end_date': end, 'limit': 10, 'order_by': 'views', 'sort': 'desc'}
).json()['data']