Build a Social Listening Agent for Threads with Claude

Threads launched in July 2023 and reached 200 million monthly active users faster than any social platform in history. It has no official API for third-party data access. For brands, marketers, and researchers who need to understand what is being said on the platform, web scraping is the only path.

TL;DR: Build a Threads social listening agent with 4 capabilities: brand mention monitor (Claude classifies sentiment, intent, and urgency per post), trending topic detector (engagement-ranked posts fed to Claude for narrative synthesis), content ideation engine (competitor posts → Claude generates post drafts in your brand voice), and share-of-voice tracker. Full weekly pipeline costs under $2 in API fees.

The Apify Threads Scraper handles the data collection. This guide shows you how to combine it with Claude to build a social listening agent that monitors your brand, detects trending conversations in your niche, and generates content recommendations — automatically.

What You Will Build

1. A brand mention monitor that classifies sentiment and flags replies requiring response
2. A trending topic detector that surfaces emerging conversations in your category
3. A content ideation engine that generates post ideas based on what is performing well
4. A competitive share-of-voice tracker

Setup

pip install apify-client anthropic python-dotenv

from apify_client import ApifyClient
import anthropic
import json
import os

apify = ApifyClient(os.environ["APIFY_TOKEN"])
claude = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

def fetch_threads_posts(
    query: str = None,
    usernames: list[str] = None,
    max_posts: int = 100,
) -> list[dict]:
    """Fetch posts from Threads by search query or user profiles."""
    
    run_input = {
        "maxPosts": max_posts,
        "includeReplies": True,
    }
    
    if query:
        run_input["searchQuery"] = query
    if usernames:
        run_input["usernames"] = usernames
    
    run = apify.actor("themineworks/threads-scraper").call(run_input=run_input)
    
    posts = list(apify.dataset(run["defaultDatasetId"]).iterate_items())
    return posts

Pattern 1: Brand Mention Monitor

Threads users discuss products candidly. This pipeline finds mentions, classifies them, and identifies which ones need a human response.

def monitor_brand_mentions(
    brand_names: list[str],
    response_threshold: int = 4,
) -> dict:
    """Monitor and classify brand mentions on Threads."""
    
    all_posts = []
    for brand in brand_names:
        posts = fetch_threads_posts(query=brand, max_posts=100)
        for p in posts:
            p["_queried_brand"] = brand
        all_posts.extend(posts)
    
    # Deduplicate by post ID
    seen_ids = set()
    unique_posts = []
    for post in all_posts:
        post_id = post.get("id") or post.get("postId")
        if post_id and post_id not in seen_ids:
            seen_ids.add(post_id)
            unique_posts.append(post)
    
    # Classify in batches
    classified = []
    batch_size = 10
    
    for i in range(0, len(unique_posts), batch_size):
        batch = unique_posts[i:i+batch_size]
        posts_text = "\n\n".join([
            f"POST {j+1}:\nUsername: {p.get('username', 'unknown')}\nContent: {p.get('text', '')[:500]}\nLikes: {p.get('likeCount', 0)}\nReplies: {p.get('replyCount', 0)}"
            for j, p in enumerate(batch)
        ])
        
        response = claude.messages.create(
            model="claude-haiku-4-5-20251001",
            max_tokens=1000,
            messages=[{
                "role": "user",
                "content": f"""Classify these Threads posts mentioning {brand_names}. Return a JSON array.

{posts_text}

For each post (in order), return:
{{
  "sentiment": "positive" | "negative" | "neutral" | "mixed",
  "intent": "complaint" | "praise" | "question" | "comparison" | "mention" | "thread",
  "urgency": 1-5,
  "response_recommended": true | false,
  "suggested_response_type": "acknowledge" | "support" | "clarify" | "thank" | "none",
  "key_issue": "one sentence if complaint or question, null otherwise"
}}

Return only a JSON array with {len(batch)} items."""
            }]
        )
        
        try:
            classifications = json.loads(response.content[0].text)
            for post, classification in zip(batch, classifications):
                classified.append({**post, **classification})
        except json.JSONDecodeError:
            classified.extend(batch)
    
    # Separate by priority
    needs_response = [p for p in classified if p.get("response_recommended") and p.get("urgency", 0) >= response_threshold]
    positive_mentions = [p for p in classified if p.get("sentiment") == "positive"]
    negative_mentions = [p for p in classified if p.get("sentiment") == "negative"]
    
    return {
        "total_mentions": len(classified),
        "needs_immediate_response": needs_response,
        "positive_count": len(positive_mentions),
        "negative_count": len(negative_mentions),
        "neutral_count": len([p for p in classified if p.get("sentiment") == "neutral"]),
        "all_classified": classified,
    }

Pattern 2: Trending Topic Detector

What is your industry talking about on Threads right now? This pattern fetches posts in your niche and asks Claude to identify emerging threads of conversation — before they peak.

def detect_trending_topics(
    category_keywords: list[str],
    your_niche: str,
) -> str:
    """Identify trending conversations in a category."""
    
    all_posts = []
    for keyword in category_keywords[:5]:  # Limit API calls
        posts = fetch_threads_posts(query=keyword, max_posts=50)
        all_posts.extend(posts)
    
    # Sort by engagement
    all_posts.sort(key=lambda x: x.get("likeCount", 0) + x.get("replyCount", 0) * 3, reverse=True)
    
    # Build high-engagement digest
    top_posts = all_posts[:40]
    posts_digest = "\n\n".join([
        f"- [{p.get('likeCount', 0)} likes, {p.get('replyCount', 0)} replies] @{p.get('username', '?')}: {p.get('text', '')[:200]}"
        for p in top_posts
    ])
    
    response = claude.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1500,
        messages=[{
            "role": "user",
            "content": f"""You are a social media trend analyst for the {your_niche} space.

TOP THREADS POSTS BY ENGAGEMENT IN THIS CATEGORY:
{posts_digest}

Identify and analyze trending topics. For each trend:
1. Topic name and description
2. Why it is gaining traction right now
3. The sentiment and narrative around it (positive, skeptical, contested)
4. Opportunity for a brand in this space to participate
5. Recommended angle: what kind of post would add value to this conversation

Identify 4-6 distinct trends. Focus on emerging conversations, not just the most obvious topics. Look for opinion threads, debates, and questions that are generating discussion.

Be specific — name the actual narratives and arguments happening, not vague categories."""
        }]
    )
    
    return response.content[0].text

Pattern 3: Content Ideation Engine

The best content strategy is observational: what is your audience already talking about, and how can you add to it? This pattern analyzes high-performing posts in your niche and generates tailored content ideas.

def generate_content_ideas(
    competitor_usernames: list[str],
    niche_keywords: list[str],
    your_brand_voice: str,
    num_ideas: int = 10,
) -> list[dict]:
    """Generate content ideas based on what's performing in your niche."""
    
    # Fetch competitor posts + search results
    competitor_posts = []
    if competitor_usernames:
        competitor_data = fetch_threads_posts(usernames=competitor_usernames, max_posts=50)
        competitor_posts.extend(competitor_data)
    
    niche_posts = []
    for keyword in niche_keywords[:3]:
        posts = fetch_threads_posts(query=keyword, max_posts=40)
        niche_posts.extend(posts)
    
    # Filter to high-engagement posts
    def engagement_score(post):
        return post.get("likeCount", 0) + post.get("replyCount", 0) * 3
    
    top_competitor = sorted(competitor_posts, key=engagement_score, reverse=True)[:15]
    top_niche = sorted(niche_posts, key=engagement_score, reverse=True)[:20]
    
    # Build analysis context
    competitor_digest = "\n".join([
        f"@{p.get('username')} ({p.get('likeCount', 0)}L, {p.get('replyCount', 0)}R): {p.get('text', '')[:250]}"
        for p in top_competitor
    ])
    
    niche_digest = "\n".join([
        f"({p.get('likeCount', 0)}L, {p.get('replyCount', 0)}R): {p.get('text', '')[:250]}"
        for p in top_niche
    ])
    
    response = claude.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=2500,
        messages=[{
            "role": "user",
            "content": f"""You are a content strategist. Generate {num_ideas} Threads post ideas based on what is performing in this niche.

BRAND VOICE:
{your_brand_voice}

TOP COMPETITOR POSTS (by engagement):
{competitor_digest}

TOP NICHE POSTS (by engagement):
{niche_digest}

Generate {num_ideas} original content ideas. For each idea return a JSON object:
{{
  "hook": "the opening line of the post (under 80 chars)",
  "content_type": "opinion" | "how_to" | "observation" | "question" | "list" | "story",
  "full_draft": "complete Threads post (150-300 chars max)",
  "why_it_works": "one sentence explaining the engagement mechanism",
  "best_time_to_post": "morning" | "afternoon" | "evening"
}}

Return a JSON array of {num_ideas} ideas. Make the drafts immediately usable — not templates, actual posts. Match the brand voice exactly."""
        }]
    )
    
    try:
        ideas = json.loads(response.content[0].text)
        return ideas
    except json.JSONDecodeError:
        return [{"hook": "Parse error", "full_draft": response.content[0].text}]

Pattern 4: Share-of-Voice Tracker

How much of your category’s Threads conversation mentions your brand versus competitors?

from collections import Counter

def track_share_of_voice(
    brands: dict[str, list[str]],
    category_keywords: list[str],
) -> str:
    """
    brands: {"Your Brand": ["yourbrand", "your brand"], "Competitor A": ["competitora"]}
    """
    
    # Fetch category posts
    all_posts = []
    for keyword in category_keywords[:5]:
        posts = fetch_threads_posts(query=keyword, max_posts=100)
        all_posts.extend(posts)
    
    # Deduplicate
    seen = set()
    unique_posts = []
    for post in all_posts:
        pid = post.get("id") or post.get("postId") or str(post.get("text", ""))[:50]
        if pid not in seen:
            seen.add(pid)
            unique_posts.append(post)
    
    # Count brand mentions
    brand_counts = {brand: 0 for brand in brands}
    brand_posts = {brand: [] for brand in brands}
    
    for post in unique_posts:
        text = (post.get("text", "") or "").lower()
        for brand, terms in brands.items():
            if any(term.lower() in text for term in terms):
                brand_counts[brand] += 1
                brand_posts[brand].append(post)
    
    total_mentions = sum(brand_counts.values())
    
    sov_data = {
        brand: {
            "mention_count": count,
            "share_of_voice": round(count / total_mentions * 100, 1) if total_mentions > 0 else 0,
            "sample_posts": [p.get("text", "")[:200] for p in brand_posts[brand][:3]],
        }
        for brand, count in brand_counts.items()
    }
    
    response = claude.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=800,
        messages=[{
            "role": "user",
            "content": f"""Analyze this Threads share-of-voice data for {', '.join(brands.keys())}.

SHARE OF VOICE DATA:
{json.dumps(sov_data, indent=2)}

TOTAL POSTS ANALYZED: {len(unique_posts)}
TOTAL BRAND MENTIONS: {total_mentions}

Write a 200-word SOV analysis:
1. Which brand owns the most conversation
2. Quality of mentions (are high-SOV brands being mentioned positively or negatively)
3. Gap analysis for the lowest-SOV brand
4. One actionable recommendation to improve SOV"""
        }]
    )
    
    return response.content[0].text


# Full weekly social listening run
def run_weekly_social_listening(
    brand_names: list[str],
    competitor_usernames: list[str],
    category_keywords: list[str],
    your_niche: str,
    brand_voice: str,
):
    print("Running weekly Threads social listening...")
    
    # 1. Brand mentions
    print("Checking brand mentions...")
    mentions = monitor_brand_mentions(brand_names)
    print(f"  {mentions['total_mentions']} mentions found")
    print(f"  {len(mentions['needs_immediate_response'])} need response")
    
    # 2. Trending topics
    print("Detecting trending topics...")
    trends = detect_trending_topics(category_keywords, your_niche)
    
    # 3. Content ideas
    print("Generating content ideas...")
    ideas = generate_content_ideas(competitor_usernames, category_keywords, brand_voice, num_ideas=10)
    
    # Save report
    with open("weekly_social_report.json", "w") as f:
        json.dump({
            "mention_summary": {
                "total": mentions["total_mentions"],
                "positive": mentions["positive_count"],
                "negative": mentions["negative_count"],
                "needs_response": [
                    {"text": p.get("text", "")[:200], "url": p.get("url"), "urgency": p.get("urgency")}
                    for p in mentions["needs_immediate_response"]
                ],
            },
            "trends": trends,
            "content_ideas": ideas,
        }, f, indent=2)
    
    print("Report saved to weekly_social_report.json")

Threads vs. Other Platforms for Social Listening

Threads is worth monitoring specifically because it skews toward authentic, real-time opinion. The demographics lean toward tech-literate users and creators who have strong opinions and express them directly. Unlike LinkedIn (performative) or Twitter/X (high noise-to-signal), Threads conversations tend to be genuine reactions.

For brands in the developer tools, SaaS, and consumer tech spaces, Threads is where the actual opinion-formers talk. A single viral negative thread from a respected developer can reach tens of thousands of relevant people. Monitoring it weekly costs under $2 in API fees. Not monitoring it is a choice that gets made by default — and occasionally gets expensive.

Frequently Asked Questions

How do you monitor brand mentions on Threads without an official read API?

Web scraping is the only method — Meta has not released a public read API for Threads as of 2025. Use the Apify Threads Scraper to search for your brand name and product names, which returns posts matching the query along with engagement metrics. Run collections daily for high-stakes monitoring or weekly for standard brand tracking. The scraper handles session management and proxy rotation; you only need to define search queries and process the results.

How does Claude improve brand mention classification over keyword-based approaches?

Keyword classifiers produce high false-positive rates on Threads because short posts lack context. “I hate that I love [product]” is positive; keyword classifiers flag “hate” as negative. Claude processes the full post including capitalization, punctuation, and conversational tone to infer intent accurately. In brand monitoring tests, Claude classification achieves 85-90% accuracy on Threads posts versus 60-70% for keyword-based VADER sentiment — a 20-30 point improvement that makes the difference between actionable signal and noise.

What data does the Threads Scraper return for each post?

Each post returns: text content, username, like count, reply count, repost/rethread count, timestamp, post URL, and optionally reply threads. Profile data (follower count, bio) requires a separate profile lookup. The scraper does not return impression counts or reach metrics — those are only available to the post author in the Threads app. For brand monitoring, like count + reply count provides a sufficient engagement proxy for prioritizing which mentions to review.

How do you detect trending topics in a specific niche on Threads?

Collect the top 40-50 posts by engagement (likes + 3x replies as engagement score) for your category keywords over a 7-day window. Feed them to Claude with a prompt asking it to identify 4-6 distinct conversation threads — not just topic categories, but the actual narratives and arguments being made. Claude identifies which conversations are gaining momentum and what positions people are taking. This is more useful than keyword frequency counts because it surfaces the narrative layer, not just the topic layer.

What is the typical weekly cost of running a Threads social listening agent?

A full pipeline — brand mention monitoring (3 brand terms), trending topic detection (5 category keywords), content ideation (10 competitor profiles), and share-of-voice tracking — collecting ~2,000 posts/week costs approximately $1-2/week. Apify Threads Scraper at PPE rates: ~$0.50-0.80 for 2,000 posts. Claude Haiku for post classification at ~150 tokens each: ~$0.05. Claude Sonnet for trend synthesis and content ideation: ~$0.30-0.50. Total: under $2/week for continuous Threads intelligence.