AI Agent for Social Media Analytics: Monitoring Mentions, Sentiment, and Trends

Why Social Media Analytics Needs an Agent

Social media monitoring tools collect data. They count mentions, track hashtags, and chart volume over time. But they rarely answer the questions that actually matter: "Why did mentions spike on Tuesday?" "Is this negative sentiment about our product or our competitor's?" "Should we respond to this emerging thread?" An AI agent goes beyond counting — it reads, interprets, and recommends.

The agent we build here integrates with social media APIs, processes mentions through sentiment analysis, detects anomalies and trends, and generates alerts when attention is needed.

Mention Collection Tool

This tool fetches recent mentions from a social media platform. The example uses a generic API pattern that applies to Twitter/X, Reddit, or any platform with a search endpoint:

flowchart LR
    INPUT(["User intent"])
    PARSE["Parse plus<br/>classify"]
    PLAN["Plan and tool<br/>selection"]
    AGENT["Agent loop<br/>LLM plus tools"]
    GUARD{"Guardrails<br/>and policy"}
    EXEC["Execute and<br/>verify result"]
    OBS[("Trace and metrics")]
    OUT(["Outcome plus<br/>next action"])
    INPUT --> PARSE --> PLAN --> AGENT --> GUARD
    GUARD -->|Pass| EXEC --> OUT
    GUARD -->|Fail| AGENT
    AGENT --> OBS
    style AGENT fill:#4f46e5,stroke:#4338ca,color:#fff
    style GUARD fill:#f59e0b,stroke:#d97706,color:#1f2937
    style OBS fill:#ede9fe,stroke:#7c3aed,color:#1e1b4b
    style OUT fill:#059669,stroke:#047857,color:#fff

import httpx
from datetime import datetime, timedelta
from agents import Agent, Runner, function_tool

_mentions_store: list[dict] = []

@function_tool
async def fetch_mentions(
    query: str, platform: str = "twitter", hours_back: int = 24
) -> str:
    """Fetch recent social media mentions matching a search query."""
    since = datetime.utcnow() - timedelta(hours=hours_back)

    # Platform-specific API calls
    headers = {"Authorization": "Bearer YOUR_TOKEN"}
    async with httpx.AsyncClient() as client:
        if platform == "twitter":
            resp = await client.get(
                "https://api.twitter.com/2/tweets/search/recent",
                headers=headers,
                params={
                    "query": query,
                    "max_results": 50,
                    "tweet.fields": "created_at,public_metrics,lang",
                    "start_time": since.strftime("%Y-%m-%dT%H:%M:%SZ"),
                },
            )
        elif platform == "reddit":
            resp = await client.get(
                "https://oauth.reddit.com/search",
                headers=headers,
                params={"q": query, "sort": "new", "limit": 50, "t": "day"},
            )
        else:
            return f"Unsupported platform: {platform}"

        if resp.status_code != 200:
            return f"API error: {resp.status_code}"

    data = resp.json()
    mentions = data.get("data", [])

    for m in mentions:
        _mentions_store.append({
            "text": m.get("text", m.get("title", "")),
            "created_at": m.get("created_at", ""),
            "metrics": m.get("public_metrics", {}),
            "platform": platform,
        })

    return f"Collected {len(mentions)} mentions. Total stored: {len(_mentions_store)}"

Sentiment Tracking Tool

Process collected mentions through sentiment classification and track the distribution:

@function_tool
def analyze_mention_sentiment(batch_size: int = 30) -> str:
    """Analyze sentiment of collected mentions. Returns mentions
    grouped for LLM-based sentiment classification."""
    if not _mentions_store:
        return "No mentions collected. Call fetch_mentions first."

    batch = _mentions_store[:batch_size]
    lines = [f"Classify sentiment for {len(batch)} mentions:"]
    for i, m in enumerate(batch):
        text = m["text"][:200]
        lines.append(f"  [{i}] ({m['platform']}) {text}")

    lines.append(
        "\nFor each mention, classify as: positive, negative, neutral, or mixed."
        "\nAlso identify the primary topic (product, service, pricing, competitor, other)."
        "\nReturn a summary with counts and the 3 most concerning negative mentions."
    )
    return "\n".join(lines)

Anomaly Detection Tool

Detecting spikes or drops in mention volume or sentiment shifts is critical for crisis management:

from collections import Counter

@function_tool
def detect_anomalies() -> str:
    """Detect unusual patterns in mention volume and content."""
    if len(_mentions_store) < 10:
        return "Need at least 10 mentions for anomaly detection."

    # Volume analysis by hour
    hourly_counts: dict[str, int] = Counter()
    for m in _mentions_store:
        created = m.get("created_at", "")
        if created:
            hour = created[:13]  # "2026-03-17T14"
            hourly_counts[hour] = hourly_counts.get(hour, 0) + 1

    if hourly_counts:
        avg_volume = sum(hourly_counts.values()) / len(hourly_counts)
        spikes = {h: c for h, c in hourly_counts.items() if c > avg_volume * 2}
    else:
        avg_volume = 0
        spikes = {}

    # Content clustering — find repeated phrases
    all_text = " ".join(m["text"].lower() for m in _mentions_store)
    words = all_text.split()
    bigrams = [f"{words[i]} {words[i+1]}" for i in range(len(words) - 1)]
    common_phrases = Counter(bigrams).most_common(10)

    report = f"Anomaly Detection Report:\n"
    report += f"  Average hourly volume: {avg_volume:.1f}\n"

    if spikes:
        report += f"  VOLUME SPIKES detected:\n"
        for hour, count in sorted(spikes.items()):
            report += f"    {hour}: {count} mentions ({count/avg_volume:.1f}x average)\n"
    else:
        report += "  No volume spikes detected.\n"

    report += f"\n  Top phrases:\n"
    for phrase, count in common_phrases:
        report += f"    '{phrase}': {count} occurrences\n"

    return report

Alert Generation Tool

When the agent detects something actionable, it generates a structured alert:

_alerts: list[dict] = []

@function_tool
def create_alert(severity: str, title: str, details: str, recommended_action: str) -> str:
    """Create an alert for the social media team.
    Severity: critical, warning, info."""
    alert = {
        "severity": severity,
        "title": title,
        "details": details,
        "action": recommended_action,
        "timestamp": datetime.utcnow().isoformat(),
    }
    _alerts.append(alert)
    return f"Alert created [{severity.upper()}]: {title}"

@function_tool
def get_all_alerts() -> str:
    """Return all generated alerts."""
    if not _alerts:
        return "No alerts generated."
    lines = []
    for a in _alerts:
        lines.append(
            f"[{a['severity'].upper()}] {a['title']}\n"
            f"  Details: {a['details']}\n"
            f"  Action: {a['action']}\n"
            f"  Time: {a['timestamp']}"
        )
    return "\n\n".join(lines)

Assembling the Social Media Agent

social_agent = Agent(
    name="Social Media Analyst",
    instructions="""You are a social media analytics agent. Your workflow:
1. Use fetch_mentions to collect recent mentions for the target brand.
2. Call analyze_mention_sentiment to classify sentiment.
3. Run detect_anomalies to find unusual patterns.
4. For any concerning findings, create_alert with appropriate severity:
   - critical: sudden negative sentiment spike, potential PR crisis
   - warning: gradual sentiment decline, competitor mentions increasing
   - info: positive trend, viral content opportunity
5. Compile a report: Mention Volume, Sentiment Breakdown, Anomalies,
   Active Alerts, Recommended Actions.
6. Call get_all_alerts at the end to include the alert summary.""",
    tools=[
        fetch_mentions, analyze_mention_sentiment, detect_anomalies,
        create_alert, get_all_alerts,
    ],
)

Running a Monitoring Session

import asyncio

async def main():
    result = await Runner.run(
        social_agent,
        "Monitor social media mentions of 'CallSphere' across Twitter "
        "and Reddit for the last 24 hours. Flag any negative sentiment "
        "spikes and identify the top discussion topics.",
    )
    print(result.final_output)

asyncio.run(main())

FAQ

How do I make monitoring continuous rather than on-demand?

Wrap the agent execution in a scheduled loop using APScheduler or a cron job. Store results in a database between runs and have the agent compare current sentiment against the previous run to detect shifts.

Which social media APIs are best for this?

Twitter/X API v2 is the most structured for search and metrics. Reddit's API is free and provides rich text data. For broader coverage, third-party aggregators like Brandwatch or Mention provide unified access to multiple platforms through a single API.

How do I avoid false positive alerts?

Set thresholds based on your baseline. If your brand normally gets 50 mentions per hour, a spike to 75 is not alarming — but 200 is. Calibrate the anomaly detection multiplier (currently 2x) based on your historical data patterns.

---

#SocialMedia #SentimentAnalysis #Monitoring #Analytics #AIAgents #AgenticAI #LearnAI #AIEngineering