How to Build an Automated Threat Intelligence Pipeline with Claude + Xpoz

Attackers often announce what they've done before your tools pick it up. A combo list goes up on Twitter/X with your domain in the post. A Reddit thread asks whether a ransomware group's leak-site claim about one of your suppliers is real. An "official support" account copying your logo starts collecting DMs on Instagram. If someone on your team only finds these posts by chance while scrolling, they'll find them late.

Introduction

A threat intelligence pipeline is a scheduled job with four parts. It collects public posts that match your watchlist. Claude triages each post into a structured verdict. High-severity hits get enriched with extra context. Alerts are then routed to Slack, your SIEM or a ticket queue. This tutorial builds that pipeline end to end, using Xpoz for social data and Claude for judgment.

Earlier Xpoz security articles covered ad-hoc hunting with chat prompts. This one is a build guide for an unattended pipeline. It covers five threat classes:

The main design choice is to run collection as deterministic code and use Claude only for judgment. Code is cheap, repeatable and easy to audit. Claude does the part that code can't: reading a messy post and deciding whether it matters.

What You'll Build

The pipeline runs every 15–60 minutes and has seven stages. Each stage is a few dozen lines of Python, and SQLite holds all the state.

1. Watchlist (YAML): brand names, domains, executives, vendor stack, suppliers, and ransomware groups your CTI team tracks.
2. Collect: Xpoz keyword searches per threat class, sorted by latest, from the last run date.
3. Normalize and dedupe: key on platform + post_id, store seen IDs.
4. Triage: Claude returns category, severity, confidence, indicators, claim_verified, rationale and action as strict JSON.
5. Enrich (high severity only): replies, quotes, author profile and volume spikes.
6. Route: Slack for medium and above, SIEM for everything, and a ticket for high and critical.
7. Feedback: promote new indicators into the watchlist and Xpoz tracked items.

What it costs to run

Xpoz charges per query, not per row returned: 2 credits for an X or Reddit query, 5 for TikTok and 12 for Instagram (pricing). Here is a typical configuration of 4 X queries and 2 Reddit queries per run:

Add a daily Instagram/TikTok impersonation sweep on top of that. One Instagram query plus two TikTok queries costs 22 credits a day. On the Claude side, claude-haiku-5-5 starts at $0.10/$0.50 per million tokens, which keeps bulk triage of a few hundred short posts a day cheap.

Prerequisites

• An Xpoz account and API key from xpoz.ai/settings. The free tier gives you 500 credits and needs no credit card. Caveat: the pipeline's default (non-CSV) SDK calls go to live REST routes in xpoz 0.10.0, and the SDK README says live routes return HTTP 403 for free-tier accounts, so scheduled runs need a paid plan.
• An Anthropic API key.
• Python 3.10+ with pip install xpoz anthropic pyyaml requests.
• A Slack incoming webhook URL. Optionally, a SIEM HTTP event collector and a Jira or ServiceNow API token.

You don't need any Twitter, Reddit, TikTok or Instagram developer keys. Xpoz handles platform access and searches 3B+ indexed posts across Twitter, Instagram, TikTok, and Reddit.

Step 1: Define Your Watchlist

The watchlist is the only file your analysts should need to edit. Keep it in version control so every change to detection scope can be reviewed.

# watchlist.yaml
brand: ["acme", "acme.com"]
executives: ["Jane Doe"]
stack: ["fortinet", "ivanti", "citrix", "palo alto"]
suppliers: ["acme supplier"]
ransomware_groups: ["akira", "qilin", "lockbit", "play ransomware"]
lookalikes: ["acme-login", "acme-support", "acme-verify", "acme-wallet"]
tracked_cves: []   # filled by the feedback loop

• Use the names your customers and attackers actually type, not just your legal entity name.
• Treat the ransomware group names as examples. Replace them with the groups your CTI team tracks.
• Add executives sparingly. Common names produce noise.

Step 2: Connect Xpoz

Connect Xpoz in two ways: interactively for hunting and tuning queries, and through the SDK for the pipeline itself. Both use the same remote MCP server, https://mcp.xpoz.ai/mcp.

• claude.ai or Claude Desktop: add a custom connector named "Xpoz" with the URL above and sign in with Google. Use this to test queries in chat before you put them in code.
• Claude Code: run this from your system terminal, then run /mcp inside claude to confirm the connection:

claude mcp add --transport http --scope user xpoz-mcp https://mcp.xpoz.ai/mcp \
  --header "Authorization: Bearer YOUR_XPOZ_API_KEY"

• Pipeline: the xpoz Python SDK reads XPOZ_API_KEY from the environment. Collection stays plain code, so no model tokens are spent fetching posts.

Step 3: Collect Signals Across X, Reddit, TikTok and Instagram

Collection runs one keyword query per threat class per platform. Each query is sorted by latest and limited to posts since the last run. Xpoz queries support uppercase AND/OR/NOT, quoted phrases and parentheses, with a 250-character limit. A bare space means OR, so be explicit.

TikTok search covers video transcripts as well as descriptions. Instagram search covers captions and video subtitles. Run searchInstagramUsers, searchTiktokUsers and searchTwitterUsers with names like "acme support" or "acme official" to find lookalike accounts.

import sqlite3, datetime as dt
from xpoz import XpozClient

xpoz = XpozClient()  # reads XPOZ_API_KEY
db = sqlite3.connect("tip.db")
db.execute("CREATE TABLE IF NOT EXISTS seen (key TEXT PRIMARY KEY, first_seen TEXT)")

X_QUERIES = {
    "credentials": '("combo list" OR combolist OR "stealer logs" OR "credential dump") AND (acme OR "acme.com")',
    "cve": '("zero-day" OR 0day OR "actively exploited" OR "CVE-2026") AND (fortinet OR ivanti OR citrix OR "palo alto")',
    "ransomware": '(akira OR qilin OR lockbit OR "play ransomware") AND (acme OR "acme.com")',
    "phishing": '"acme-login" OR "acme-support" OR "acme-verify" OR "acme-wallet"',
}

def is_new(platform, post_id):
    cur = db.execute("INSERT OR IGNORE INTO seen VALUES (?, ?)",
                     (f"{platform}:{post_id}", dt.datetime.utcnow().isoformat()))
    db.commit()
    return cur.rowcount == 1

def collect(since):
    hits = []
    for cls, q in X_QUERIES.items():
        res = xpoz.twitter.search_posts(q, sort_by="latest", start_date=since)
        for p in res.data:
            if is_new("twitter", p.id):
                hits.append({"platform": "twitter", "id": p.id, "class": cls,
                             "author": p.author_username, "text": p.text})
    return hits

Caveat: SDK keyword arguments mirror the MCP parameters in snake_case. Check the exact names against the xpoz SDK README before deploying. For backfills, use paging or CSV responses: they return an operationId, and you poll checkOperationStatus about every 5 seconds until it reports success.

Step 4: Triage With Claude Using a Strict JSON Schema

Triage turns each raw post into a structured verdict your router can act on without parsing free text. Use claude-haiku-5-5 at low effort for bulk classification. Escalate anything at severity 4 or above to claude-sonnet-5-5 for a second opinion.

The claim_verified field matters most. In test queries, Reddit posts relayed a ransomware group's victim claims while also noting "no proof of any breach yet." Social posts often assert a breach with nothing behind it. Your alert should say so.

import json, anthropic
claude = anthropic.Anthropic()

TRIAGE_SCHEMA = {
  "type": "object",
  "properties": {
    "category": {"type": "string", "enum": ["credentials", "cve", "ransomware", "phishing", "impersonation", "noise"]},
    "severity": {"type": "integer", "minimum": 1, "maximum": 5},
    "confidence": {"type": "number"},
    "indicators": {"type": "array", "items": {"type": "string"}},
    "claim_verified": {"type": "boolean"},
    "rationale": {"type": "string"},
    "action": {"type": "string"}
  },
  "required": ["category", "severity", "confidence", "indicators", "claim_verified", "rationale", "action"],
  "additionalProperties": False
}

SYSTEM = ("You are a defensive SOC analyst triaging PUBLIC social media posts about our organization. "
          "Extract indicators (CVE ids, domains, handles). Set claim_verified=false when a post asserts "
          "a breach or leak without evidence such as samples, a file tree, or a credible reporter.")

def triage(post, model="claude-haiku-5-5", effort="low"):
    resp = claude.messages.create(
        model=model, max_tokens=1024, system=SYSTEM,
        messages=[{"role": "user", "content": json.dumps(post)}],
        output_config={"effort": effort,
                       "format": {"type": "json_schema", "schema": TRIAGE_SCHEMA}},
    )
    if resp.stop_reason == "refusal":
        return {"category": "review", "severity": 3, "rationale": "model declined; route to analyst"}
    return json.loads(resp.content[0].text)

Two operational details:

• Refusals: Sonnet 5.5 and Opus 5.5 run safety classifiers that include a cyber category. A declined request returns HTTP 200 with stop_reason: "refusal", so check that field before parsing. Frame the prompt as defensive triage of public posts (refusal docs).
• Forced tool choice: tool_choice of any or tool returns a 400 on Opus 5.5 and Sonnet 5.5. Structured outputs, as used above, avoid that problem.

Step 5: Enrich High-Severity Hits

Enrichment runs only on severity 4–5 verdicts, so its cost scales with real incidents rather than with noise. It answers three questions: is there technical substance, is the author credible, and is volume spiking?

Enrichment benefits from an agent that picks its own next tool call. Claude Code in headless mode with the Xpoz MCP server connected handles this well:

claude -p "Enrich this triaged threat post using Xpoz: pull quotes and replies, \
the author profile, and a countTweets spike check. Post: $POST_JSON" \
  --output-format json --json-schema "$(cat enrich_schema.json)"

The structured result appears in .structured_output. For reproducible CI runs, add --bare and --mcp-config so no local settings leak in (headless docs).

Step 6: Route Alerts to Slack, SIEM and Ticketing

Xpoz returns data and has no built-in alert delivery, so routing lives in your own code. That also means you control the thresholds and destinations.

import requests, os

def route(post, v):
    flag = "VERIFIED" if v["claim_verified"] else "UNVERIFIED CLAIM"
    requests.post(os.environ["SIEM_URL"], json={**post, **v}, timeout=10)
    if v["severity"] >= 3:
        requests.post(os.environ["SLACK_WEBHOOK"], json={"text":
            f"[sev {v['severity']}] [{flag}] {v['category']}: {v['rationale']}\n"
            f"Indicators: {', '.join(v['indicators'])}\nAction: {v['action']}"}, timeout=10)

Use platform:post_id as the dedupe key in every destination. A retried run should then update existing records rather than create duplicate tickets.

Step 7: Schedule It and Close the Loop

Schedule the job with cron, a CI scheduler or a container job. Store the last successful run date so each run's startDate picks up where the previous run stopped.

*/30 * * * * cd /opt/tip && python run.py >> tip.log 2>&1

The feedback loop is what keeps the pipeline current. When triage extracts a new indicator, such as a fresh CVE id or a new lookalike handle, append it to watchlist.yaml and register it with addTrackedItems:

{"items": [{"phrase": "CVE-2026-XXXXX", "type": "keyword", "platform": "twitter"}]}

Tracked-item limits depend on your plan: Free 1, Pro 3, Max 10, Enterprise unlimited. The call returns an error once you exceed your limit. Quote full CVE ids in queries, because tokenization of hyphenated ids is unverified.

How Xpoz Addresses This

Social media threat monitoring usually stalls on data access. Each platform has its own API, keys, rate limits and data schema. Xpoz replaces those with one read-only data layer and the same query syntax across all four platforms.

• One query language, four platforms. The same boolean syntax works on X, Reddit, TikTok and Instagram.
• Transcript and subtitle search. Impersonation scams often say the brand name in a video without ever typing it. TikTok transcripts and Instagram subtitles are searchable.
• Field selection. A fields array keeps responses small, which keeps triage prompts small and cheap.
• Per-query pricing. Pulling more rows costs no extra credits, so you can cast a wide net and let Claude filter.
• Same server for humans and code. Analysts hunt in claude.ai, and the pipeline calls the same data through the SDK.

Practical Examples

• Supplier ransomware claim: a Reddit post in r/cybersecurity relays a leak-site claim naming your logistics supplier. Triage returns ransomware, severity 4, claim_verified=false. Enrichment finds no samples in the thread, and countTweets shows no spike. A Slack alert and a ticket go to vendor risk, both labeled "UNVERIFIED CLAIM", with a request to check with the supplier.
• Edge device zero-day: a burst of X posts mentions "actively exploited" alongside a firewall vendor in your stack. The second-pass model extracts a CVE id, the feedback loop tracks it, and the next runs watch for "PoC" or "exploit released" posts mentioning that id.
• Fake support account: searchInstagramUsers returns "acme.support.help" with a copied bio. Triage classes it as impersonation, severity 3, and the brand team files a takedown request.

Tuning: False Positives, Unverified Claims and Cost

Most early alerts will be noise. Tune in this order:

1. Tighten queries before tightening prompts. Add NOT terms for recurring noise, such as a product that shares your brand name.
2. Keep severity 1–2 out of Slack. Send it only to the SIEM, so analysts can query it later without getting paged.
3. Never drop unverified claims. Label them instead. An unverified ransomware claim about you is still something leadership needs to hear from your team first.
4. Watch spend. Run getCreditsUsageHistory with a 7d window weekly, and lengthen the cadence for quiet threat classes.

Limit: Xpoz covers public posts on Twitter/X, Instagram, TikTok and Reddit only. Telegram, Discord and dark-web forums need separate sources.

FAQ

Q: What is a threat intelligence pipeline?

A: A threat intelligence pipeline is an automated workflow that collects external signals, scores them and routes the important ones to responders. In this build, Xpoz collects public social posts matching a watchlist. Claude classifies each post by category, severity and evidence quality. Your own code sends the results to Slack, a SIEM or a ticket system on a schedule.

Q: How do I keep Claude from treating rumors as confirmed breaches?

A: Make evidence a required field. The triage schema includes claim_verified, and the system prompt tells Claude to set it to false when a post asserts a breach without samples, file trees or a credible reporter. Every alert carries that label, so responders can see straight away whether they are acting on a claim or on evidence.

Q: Does Xpoz send alerts by itself?

A: No. Xpoz is a read-only data layer that returns posts, profiles and counts. Tracked items register keywords or users for collection, but delivery to Slack, a SIEM or ticketing is handled by the pipeline code in this tutorial or by your automation platform.

Q: How many credits does automated threat intelligence use?

A: X and Reddit queries cost 2 credits each, TikTok 5 and Instagram 12, with no charge per row. Six X and Reddit queries every two hours use about 4,300 credits a month, which fits the Pro plan's 7,500. The free 500 credits are enough to prototype the queries and triage prompts.

Key Takeaways

• Run collection as deterministic code and use Claude only for judgment. This keeps the threat intelligence pipeline cheap, auditable and repeatable.
• Cover five classes: leaked credentials monitoring, CVE chatter, ransomware mention alerts, phishing lookalikes and brand impersonation detection.
• Force triage into a strict JSON schema with a claim_verified flag, and check stop_reason before parsing.
• Enrich only severity 4–5 hits, and use countTweets to separate real spikes from one-off posts.
• Feed new indicators back into the watchlist and Xpoz tracked items so coverage keeps up with the threat.

Conclusion

You now have a Claude threat intelligence workflow made of a YAML watchlist, Xpoz collection across four platforms, schema-bound Claude triage, selective enrichment and routing you control. Start with a single threat class, such as leaked credentials, run it for a week, tune out the noise, and then add the next class. Create an account at xpoz.ai to start free with 500 credits, compare plans on the pricing page, and find setup help at help.xpoz.ai.