Skip to content

Build with the API / A lookup tool for LLM agents

A lookup tool for LLM agents

A tool-calling function `threatcluster_lookup(query)` for LLM agents.

What you get

Wraps the unified search endpoint as a plain Python function that returns a compact, JSON-serialisable answer: a one-line summary, the matching clusters with scores, the matching entities and dark-web groups, and a `citations` list of ThreatCluster URLs the model can quote. No LLM client is needed to run it; see TOOL_SCHEMA for registration.

Endpoints
/search
Credits per run
5
Key
Works on the free key

Step by step

  1. 1. Get a key

    Sign in, open Settings, then API & Feeds, and select Generate API key. The key is shown once, so copy it. Every account has one and the free key covers this recipe.

    The ThreatCluster API page with the Get a free key button and the quick start, which begins with creating a key
    The API page. The key itself is created in Settings, under API & Feeds.

  2. 2. Download the script and set the key

    The script is one file with one dependency. It reads the key from the environment, so the key never goes into the file.

    curl -sO https://raw.githubusercontent.com/Jam0k/Cyber-Threat-Intelligence-API/main/examples/recipes/agent-tool.py
pip install requests
export THREATCLUSTER_API_KEY=tc_live_your_key_here

  3. 3. Run it

    Run from the command line, the script prints the JSON a model would receive from the tool: a one-line answer, the matching incidents, and a list of citation links.

    python3 agent-tool.py "citrix netscaler"
    A terminal running the tool with the query citrix netscaler and printing JSON with an answer line and the first incident, including its title, score of 80 and summary
    Run on 29 September 2026. Five incidents matched. The answer line names the highest-scoring one.

  4. 4. Follow a citation

    Every incident in the result carries its link, so the model can cite what it used and a person can check it.

    The ThreatCluster incident page for the Citrix NetScaler zero-days confirmed by CISA, with key points, the summary and 13 extracted entities
    The first citation from the run above.

  5. 5. Register it with a model

    Import the function and its schema from the file. The same schema works for OpenAI and for Claude, which names the field input_schema.

    # OpenAI
tools=[{"type": "function", "function": TOOL_SCHEMA}]

# Claude
tools=[{"name": TOOL_SCHEMA["name"], "description": TOOL_SCHEMA["description"],
        "input_schema": TOOL_SCHEMA["parameters"]}]

# either: on a tool call, return this as the tool result
json.dumps(threatcluster_lookup(**args))

The code

agent-tool.py, 133 lines. Python 3, needs requests.

#!/usr/bin/env python3
"""agent-tool: a tool-calling function `threatcluster_lookup(query)` for LLM agents.

What it does
  Wraps the unified search endpoint as a plain Python function that returns a compact,
  JSON-serialisable answer: a one-line summary, the matching clusters with scores, the
  matching entities and dark-web groups, and a `citations` list of ThreatCluster URLs the
  model can quote. No LLM client is needed to run it; see TOOL_SCHEMA for registration.

Endpoints
  GET /search?q=<query>&limit=<n>   (5 credits)

Cost: 5 credits per call. Works on the free tier (7-day window).

Usage
  python3 agent-tool.py "lockbit"            # prints the JSON the model would receive

Registering it as a tool (three lines, either vendor):
  OpenAI:  tools=[{"type": "function", "function": TOOL_SCHEMA}]; on a tool call run threatcluster_lookup(**args)
  Claude:  tools=[{"name": TOOL_SCHEMA["name"], "description": TOOL_SCHEMA["description"], "input_schema": TOOL_SCHEMA["parameters"]}]
  Either:  return json.dumps(threatcluster_lookup(**args)) as the tool result; the model cites the `citations` URLs.
"""
import json
import os
import sys
import time

import requests

API_BASE = os.environ.get("THREATCLUSTER_API_BASE", "https://threatcluster.io/api/public/v1").rstrip("/")
API_KEY = os.environ.get("THREATCLUSTER_API_KEY", "")
SITE = os.environ.get("THREATCLUSTER_SITE", "https://threatcluster.io")  # where the linked pages live
PACE_SECONDS = 2.1  # free keys get 30 requests/min; ~28/min keeps loops clear of 429s
_credits = {"used": 0, "remaining": None}


def _auth_headers():
    # A tc_... key is sent as X-API-Key; anything else is a short-lived bearer (e.g. from `tc login`).
    if not API_KEY:
        sys.exit("THREATCLUSTER_API_KEY is not set. Free keys: https://threatcluster.io/about/api")
    if API_KEY.startswith("tc_"):
        return {"X-API-Key": API_KEY}
    return {"Authorization": "Bearer " + API_KEY}


def api_get(path, _allow=(), **params):
    """GET one endpoint. Exits non-zero with the API's own error body, unless the
    status is listed in _allow (per-item lookups where a 404 is itself an answer)."""
    r = requests.get(API_BASE + path, headers=_auth_headers(), params=params, timeout=60)
    if "X-Request-Cost" in r.headers:
        _credits["used"] += int(r.headers["X-Request-Cost"])
    if "X-RateLimit-Remaining" in r.headers:
        _credits["remaining"] = r.headers["X-RateLimit-Remaining"]
    if r.status_code >= 400 and r.status_code not in _allow:
        try:
            body = json.dumps(r.json())
        except ValueError:
            body = r.text[:500]
        sys.exit("HTTP %d from %s: %s" % (r.status_code, path, body))
    return r


def cluster_url(cluster):
    # The free payload carries `slug` (canonical page) and `cluster_id`; the last 8 hex
    # chars of the id also resolve, via a 301 to the slug, when slug is missing.
    ident = cluster.get("slug") or cluster.get("short_id") or cluster.get("cluster_id", "").replace("-", "")[-8:]
    return "%s/cluster/%s" % (SITE, ident)


def report_credits():
    # Stderr, so it never pollutes piped output (brief.md, blocklists, JSON).
    # Keys without a daily cap send no remaining-credits header.
    left = ", remaining today: %s" % _credits["remaining"] if _credits["remaining"] is not None else ""
    sys.stderr.write("[threatcluster] credits used this run: %d%s\n" % (_credits["used"], left))

TOOL_SCHEMA = {
    "name": "threatcluster_lookup",
    "description": ("Look up a threat actor, malware family, CVE, vendor, product or campaign in ThreatCluster's "
                    "incident corpus (last 7 days on a free key). Returns matching incident clusters with threat "
                    "scores, related entities, dark-web groups, and citation URLs."),
    "parameters": {
        "type": "object",
        "properties": {
            "query": {"type": "string", "description": "What to look up, e.g. 'LockBit', 'CVE-2026-83549', 'Ivanti'"},
            "limit": {"type": "integer", "description": "Max clusters to return (1-10)", "default": 5},
        },
        "required": ["query"],
    },
}


def threatcluster_lookup(query, limit=5):
    data = api_get("/search", q=query, limit=max(1, min(int(limit), 10))).json()
    clusters = []
    for c in data.get("clusters", []):
        clusters.append({
            "title": c.get("ai_title") or c.get("title"),
            "threat_score": c.get("threat_score"),
            "urgency": c.get("urgency_level"),
            "latest_activity": (c.get("date_range_latest") or c.get("updated_at") or "")[:10],
            "summary": c.get("ai_summary"),
            "url": cluster_url(c),
        })
    entities = [{"type": e.get("entity_type"), "value": e.get("entity_value"),
                 "clusters": e.get("cluster_count"),
                 "url": "%s/entities/%s/%s" % (SITE, (e.get("entity_type") or "").replace("_", "-"), e.get("entity_value"))}
                for e in data.get("entities", [])]
    darkweb = [{"type": d.get("type"), "name": d.get("name"), "active": d.get("active"),
                "victim_count": d.get("victim_count")} for d in data.get("darkweb", [])]

    if clusters:
        top = clusters[0]
        answer = ("%d incident cluster(s) in the last %s days match %r; highest-scoring: %r (score %s, %s)."
                  % (len(clusters), data.get("days") or data.get("lookback_days") or 7, query,
                     top["title"], top["threat_score"], top["latest_activity"]))
    else:
        answer = "No incident clusters in the last %s days match %r." % (data.get("days") or 7, query)
    return {
        "query": query,
        "window_days": data.get("days") or data.get("lookback_days"),
        "answer": answer,
        "clusters": clusters,
        "entities": entities,
        "darkweb": darkweb,
        "citations": [c["url"] for c in clusters] + [e["url"] for e in entities[:3]],
    }


if __name__ == "__main__":
    if len(sys.argv) < 2:
        sys.exit("usage: agent-tool.py <query>")
    print(json.dumps(threatcluster_lookup(" ".join(sys.argv[1:])), indent=2))
    report_credits()

The shell one-liner, agent-tool.sh:

#!/usr/bin/env bash
# agent-tool.sh: the raw call behind threatcluster_lookup(query): unified search, compact JSON with citations. 5 credits. Free tier OK.
# usage: ./agent-tool.sh lockbit
set -euo pipefail
: "${THREATCLUSTER_API_KEY:?set THREATCLUSTER_API_KEY (free keys: https://threatcluster.io/about/api)}"
BASE="${THREATCLUSTER_API_BASE:-https://threatcluster.io/api/public/v1}"
SITE="${THREATCLUSTER_SITE:-https://threatcluster.io}"
# A tc_... key goes in X-API-Key; anything else is a bearer (e.g. from `tc login`).
if [[ "$THREATCLUSTER_API_KEY" == tc_* ]]; then AUTH="X-API-Key: $THREATCLUSTER_API_KEY"; else AUTH="Authorization: Bearer $THREATCLUSTER_API_KEY"; fi
# tc_get <path> [curl -G args...]: prints the body; on HTTP >= 400 prints the API error to stderr and fails.
tc_get() { local path=$1; shift; local out code
  out=$(curl -sS -G -H "$AUTH" -w $'\n%{http_code}' "$BASE$path" "$@"); code=${out##*$'\n'}
  if [ "$code" -ge 400 ]; then echo "HTTP $code from $path: ${out%$'\n'*}" >&2; return 1; fi; printf '%s' "${out%$'\n'*}"; }
tc_get /search --data-urlencode "q=${1:?query}" --data-urlencode limit=5 \
 | jq --arg site "$SITE" '{query, window_days: .days,
     clusters: [.clusters[] | {title: .ai_title, threat_score, latest_activity: (.date_range_latest[:10]), url: "\($site)/cluster/\(.slug)"}],
     entities: [.entities[] | {type: .entity_type, value: .entity_value, clusters: .cluster_count}],
     darkweb:  [.darkweb[]  | {type, name, victim_count}],
     citations: [.clusters[] | "\($site)/cluster/\(.slug)"]}'

Output

From a run on 29 September 2026. Yours will differ as the corpus moves.

{
  "query": "citrix netscaler",
  "window_days": 3650,
  "answer": "5 incident cluster(s) in the last 3650 days match 'citrix netscaler'; highest-scoring: 'Critical Zero-Day Exploits in Citrix NetScaler Confirmed by CISA' (score 80.0, 2026-09-29).",
  "clusters": [
    {
      "title": "Critical Zero-Day Exploits in Citrix NetScaler Confirmed by CISA",
      "threat_score": 80.0,
      "urgency": "medium",
      "latest_activity": "2026-09-29",
      "summary": "On September 26, 2026, CISA confirmed the active exploitation of two critical zero-day vulnerabilities in Citrix NetScaler, identified as CVE-2026-88771 and CVE-2026-88772, both with a CVSS score of 9.5. These vulnerabilities allow remote code execution and affect all default configurations of NetScaler ADC and Gateway. Reports indicated that administrators were advised to disconnect their systems without prior explanation, signaling ongoing exploitation. Citrix released security updates on September 27, 2026, but the vulnerabilities had already been exploited before patches were available. The vulnerabilities stem from improper input validation and a buffer overflow, impacting numerous organizations worldwide. The National Cybersecurity Centre of the Netherlands alerted local organizations based on information from a European CERT. The situation underscores the urgency for affected organizations to apply the patches immediately.\n\nKey Points:\n\u2022 CISA confirmed active exploitation of two critical zero-day vulnerabilities in Citrix NetScaler.\n\u2022 Vulnerabilities CVE-2026-88771 and CVE-2026-88772 allow remote code execution with a CVSS score of 9.5.\n\u2022 Administrators were urged to disconnect systems before official patch release, indicating ongoing attacks.",
      "url": "https://threatcluster.io/cluster/critical-zero-day-exploits-in-citrix-netscaler-confirmed-by--5b100c3f"
    },
    {
      "title": "Critical Zero-Day Vulnerabilities in Citrix NetScaler Under Active Exploitation",
      "threat_score": 82.25,
      "urgency": "medium",
      "latest_activity": "2026-09-29",
      "summary": "Citrix has disclosed multiple critical vulnerabilities in its NetScaler ADC and Gateway products, with two zero-day vulnerabilities (CVE-2026-88771 and CVE-2026-88772) actively exploited in the wild. These vulnerabilities allow unauthenticated remote code execution and denial of service, affecting all default configurations. Citrix released patches on September 27, 2026, and cybersecurity agencies, including CISA, have urged immediate updates. The vulnerabilities have been added to CISA's Known Exploited Vulnerabilities Catalog. Other vulnerabilities disclosed include CVE-2026-88773 through CVE-2026-88778, with CVSS scores ranging from 7 to 9.5. Organizations are advised to check for signs of compromise before applying updates, as exploitation is confirmed globally. The situation is critical, with significant implications for organizations using Citrix products.\n\nKey Points:\n\u2022 Two critical zero-day vulnerabilities (CVE-2026-88771, CVE-2026-88772) are actively exploited.\n\u2022 Citrix has released urgent patches for eight vulnerabilities affecting NetScaler ADC and Gateway.\n\u2022 CISA has added the vulnerabilities to its Known Exploited Vulnerabilities Catalog.",
      "url": "https://threatcluster.io/cluster/critical-zero-day-vulnerabilities-in-citrix-netscaler-under--16a7b04e"
    },
    {
      "title": "Citrix NetScaler Zero-Days Exploited: Urgent Patches Released",
      "threat_score": 74.75,
      "urgency": "medium",
      "latest_activity": "2026-09-29",
      "summary": "Citrix confirmed active exploitation of two critical zero-day vulnerabilities in its NetScaler products, CVE-2026-88771 and CVE-2026-88772, on September 27, 2026. These vulnerabilities, rated 9.5 on the CVSS scale, allow remote code execution and affect all NetScaler appliances in default configurations. The earliest known exploitation attempt occurred on September 24, 2026, when GreyNoise detected a malicious cyber actor attempting to exploit a Citrix NetScaler Gateway. Citrix's delayed communication left customers relying on unofficial channels for threat information. As of September 29, 2026, over 50,000 instances of Citrix NetScaler devices were identified as potentially vulnerable. Citrix has released patches and provided indicators of compromise to assist customers. Attribution for the attacks remains under investigation, with concerns about financially motivated cybercriminals and state-sponsored groups targeting these vulnerabilities.\n\nKey Points:\n\u2022 Citrix confirmed active exploitation of two critical zero-days in NetScaler products.\n\u2022 Patches were released on September 27, 2026, after a significant communication delay.\n\u2022 Over 50,000 Citrix NetScaler instances are potentially vulnerable to these exploits.",
      "url": "https://threatcluster.io/cluster/citrix-netscaler-zero-days-exploited-urgent-patches-released-10cc0704"
    },
    {
      "title": null,
      "threat_score": 62.13,
      "urgency": "medium",
      "latest_activity": "2026-09-28",
      "summary": null,
      "url": "https://threatcluster.io/cluster/086b5e0b"
    },
    {
      "title": "Critical Zero-Day Exploits Target Citrix and Other Major Platforms",
…

The file on GitHub: agent-tool.pyand agent-tool.sh.

Goes well with

All recipes: /build. Endpoint reference: /api/public/v1/docs.