Workflow recipes

deadline = time.time() + 60 * 60  # stop waiting after an hour (rare)
while True:
    agent = requests.get(
        f'https://api.virlo.ai/v1/agents/{agent_id}',
        headers={'Authorization': 'Bearer YOUR_API_KEY'}
    ).json()['data']

    if agent['latest_run']['status'] == 'failed':
        raise Exception('The run failed.')
    # finalized: videos collected and the AI report written
    #   (latest_run.status "completed" alone only means videos are collected).
    # last_run_at: set about a minute later, when creator scoring is done.
    if agent['finalized'] and agent['last_run_at']:
        break
    if time.time() > deadline:
        # Check again later. Creating a new agent would charge you again.
        raise TimeoutError('Still running after an hour.')

    print(f"{agent['stage']}: {agent['progress_pct']}%")  # e.g. "collecting: 40%"
    time.sleep(30)

# analysis is only the headline. The full report (themes, tactics, posting times,
# top 10 videos) is in analysis_data, also at GET /v1/agents/:id/analysis/latest.
report = agent['analysis_data'] or {}
print(agent['analysis'])  # the main takeaway, in one or two sentences
for theme in report.get('themes', []):
    print(f"- {theme['name']}: {theme['why_it_works']}")

# Example output, from a latte art agent:
# The most impactful insight for creators is the overwhelming success of 'progress journey' content, ...
# - Progress Journey Latte Art: Documenting daily or incremental improvements creates a relatable narrative, ...

# Free, including the filters (min_views, start_date, end_date, platforms, region).

# The 50 most-viewed videos
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 ads (needs meta_ads_enabled at create time)
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']

# Standout creators ("creator outliers"), best Virality Score first.
# 35+ exceptional, 25-35 very strong, 18-25 strong, 10-18 promising
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']

for creator in outliers[:5]:
    print(f"{creator['platform']}: {creator['follower_count']:,} followers, "
          f"{creator['avg_views']:,.0f} avg views, score {creator['weighted_score']}")
    print(f"  {creator['creator_url']}")

# Example output:
# instagram: 1,791 followers, 475,807 avg views, score 41.81
#   https://www.instagram.com/...

for creator in outliers[:3]:
    handle = creator['creator_url'].rstrip('/').split('/')[-1].lstrip('@')
    # On YouTube this is often a channel ID (/channel/UC...), not a handle.
    print(creator['platform'], handle)  # tiktok outside.kev

# The most-viewed of the recent videos the lookup read
top_video = max(creator['videos'], key=lambda v: v['views'])

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

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

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

analysis = result['result']['analysis']
print(f"Outlier score: {analysis['outlier_score']}")    # views / the creator's median views
print(f"Percentile: {analysis['percentile']}")          # % of their recent videos it beat
print(f"Label: {analysis['performance_label']}")        # average | above_average | viral | mega_viral

# Both results are saved. Reopen either for free with GET /v1/satellite/runs/{job_id},
# so keep both IDs. A new paid lookup of the same creator gets its own ID; old runs stay.

import requests

API = 'https://api.virlo.ai/v1'
headers = {'Authorization': 'Bearer YOUR_API_KEY'}

digest = requests.get(f'{API}/topic-trends/today', headers=headers).json()['data']

for group in digest:
    print(group['local_date'])  # the list's date. Don't use 'title', which can be a day off.
    for item in group['trends']:
        print(f"  #{item['ranking']}: {item['trend']['name']}")

trend_name = digest[0]['trends'][0]['trend']['name']  # or pick one from the list
intent = f'Find what is working in short videos about {trend_name}'

# Keyword ideas (free)
suggestion = requests.post(
    f'{API}/agents/suggest-keywords',
    headers=headers,
    json={'intent': intent}
).json()['data']
# Before you pay for a run, check that suggestion['quality']['passes'] is True.

# One-time agent ($0.50). Optional: skip it to start at $0.75, but keep the keyword ideas.
agent = requests.post(
    f'{API}/agents',
    headers=headers,
    json={
        'is_recurring': False,
        'intent': intent,
        'keywords': suggestion['keywords'],
        'exclude_keywords': suggestion['exclude_keywords'],  # words that point to other topics
    }
).json()['data']
agent_id = agent['id']

# Your balance must cover the first run ($0.50), or this returns a 402 error.
monitor = requests.post(
    f'{API}/agents',
    headers=headers,
    json={
        'is_recurring': True,
        'cadence': 'weekly',  # same day and time each week; next_run_at says when
        'name': f'Weekly watch: {trend_name}',
        'intent': intent,
        'keywords': suggestion['keywords'],
        'exclude_keywords': suggestion['exclude_keywords'],
    }
).json()['data']

print(f"Agent ID: {monitor['id']}")
print(f"Next scheduled run: {monitor['next_run_at']}")  # the first run has already started

import requests
from datetime import date, timedelta

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

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']

# Generic tags like #fyp lead. Hashtags come back lowercase, without the #.
for tag in hashtags[:5]:
    print(f"#{tag['hashtag']}: {tag['count']:,} videos, {tag['total_views']:,} views")

perf = requests.get(
    'https://api.virlo.ai/v1/hashtags/fyp/performance',  # no # in the path
    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")

# YouTube and Instagram: /v1/youtube/hashtags and /v1/instagram/hashtags
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']

for tag in tiktok[:3]:
    print(f"#{tag['hashtag']}: {tag['count']:,} videos, {tag['total_views']:,} views")

# Example output (videos posted Aug 25 to Sep 23, 2026):
# #fyp: 31,049 videos, 5,162,705,739 views
# #viral: 9,971 videos, 1,693,054,907 views
# #funny: 4,276 videos, 1,242,478,023 views

curl https://api.virlo.ai/v1/raw/tiktok/creator/creatorhandle   -H "Authorization: Bearer YOUR_API_KEY"

curl https://api.virlo.ai/v1/raw/tiktok/creator/creatorhandle/videos   -H "Authorization: Bearer YOUR_API_KEY"

import time, requests

API = 'https://api.virlo.ai/v1/satellite/place'
headers = {'Authorization': 'Bearer YOUR_API_KEY'}

job = requests.post(API, headers=headers, json={'place_id': 'YOUR_PLACE_ID', 'max_videos': 50}).json()['data']

while True:
    state = requests.get(f"{API}/status/{job['job_id']}", headers=headers).json()['data']
    if state.get('finalized'):
        break
    time.sleep(5)

change = state['result']['change']
print('tags now', change['video_count'], 'change', change['video_count_change'])
print('posts not seen before', change['new_since_previous'])