威胁情报富化代理：使用Claude自动化安全分析
本教程指导开发者构建一个由Claude驱动的威胁情报富化代理，自动化安全分析师的手动关联工作。代理接收原始IOC（IP、哈希、域名），利用Claude的工具使用能力自主决定查询哪些情报源，跨源关联发现，并生成结构化报告。教程涵盖工具模式设计、代理循环构建、多源关联提示和结构化输出生成，并提供模拟后端以便替换为真实API。适合希望学习AI工程实践的安全开发者。
技术背景
安全分析师每天需要手动查询多个威胁情报源（如VirusTotal、AbuseIPDB、MITRE ATT&CK）来关联分析IOC，过程耗时且易出错。自动化此流程可显著提升效率，但传统脚本难以处理动态决策和多源关联。Claude的工具使用能力允许模型自主决定调用哪些工具及顺序，实现智能代理。本教程展示如何构建这样的代理，解决手动关联的痛点，适用于安全产品集成或SOC自动化。主要限制包括API成本、延迟和模型可能产生错误工具调用，需通过循环上限和错误处理缓解。
核心原理
核心原理是利用Claude的工具使用（function calling）能力，在代理循环中让模型根据用户输入和工具结果动态决定下一步操作。系统提示将模型定位为高级威胁分析师，引导其系统性调查。工具定义包含清晰的描述和输入模式，帮助模型选择正确工具。代理循环持续运行，直到模型停止调用工具（end_turn）或达到最大轮次。每次工具调用结果作为新消息返回给模型，形成多轮推理。最终，模型生成自由文本分析，再通过第二次Claude调用转换为结构化JSON报告。
架构分析
系统分为四层：1) 工具定义层：声明四个工具（IP信誉、文件哈希、域名查询、MITRE映射），每个工具包含名称、描述和输入模式，描述对模型决策至关重要。2) 模拟后端层：用Python函数模拟真实API，返回逼真数据结构，便于替换。3) 代理编排层：核心循环，调用Claude API，处理工具调用，维护消息历史。4) 报告生成层：使用第二个Claude调用将分析文本转换为符合预定义模式的JSON。设计决策包括：模块化工具便于扩展；模拟后端隔离外部依赖；循环上限防止成本失控；结构化输出供下游系统消费。
完整实战教程
安全分析师花费大量时间手动跨威胁情报源进行关联分析——查询VirusTotal获取文件哈希，检查AbuseIPDB获取IP信息，交叉引用MITRE ATT&CK，然后将所有信息综合成报告。本教程展示如何构建一个由Claude驱动的代理，自动化整个工作流程。
该代理接收原始威胁指标（IOC）——IP地址、文件哈希或域名——并利用Claude的工具使用能力决定查询哪些情报源，跨源关联发现，生成结构化、分析师就绪的威胁报告。本示例中的工具是模拟的，但架构设计使得您可以替换为真实API（如VirusTotal、AbuseIPDB、Shodan等）而无需更改编排逻辑。
无论您是构建安全产品中的威胁情报功能（ISV），还是为企业SOC自动化富化工作流，此模式都直接适用。
您将学到什么
- 如何设计工具模式，为Claude提供足够的上下文以选择正确的情报源
- 如何构建代理循环，让Claude根据发现链式调用工具
- 如何提示进行多源关联和MITRE ATT&CK映射
- 如何将自由文本分析转换为结构化JSON报告，供下游系统使用
先决条件
- Python 3.10+
- Anthropic API密钥——在
.env文件中设置为ANTHROPIC_API_KEY=sk-ant-... - 熟悉Claude的工具使用会有所帮助，但不是必需的
步骤 1：设置环境
安装 Anthropic SDK 和 python-dotenv。在与本笔记本相同的目录下创建一个 .env 文件，并填入你的 API 密钥：
ANTHROPIC_API_KEY=sk-ant-...
%pip install anthropic python-dotenv --quiet
import json
import anthropic
from dotenv import load_dotenv
load_dotenv()
client = anthropic.Anthropic()
MODEL_NAME = "claude-sonnet-4-6"
步骤 2：定义威胁情报工具
我们定义了四个工具，代表常见的威胁情报数据源。每个工具都有清晰的描述，帮助 Claude 理解何时以及为何使用它——这对于有效的智能体行为至关重要。在生产环境中，这些工具将封装真实的 API 调用；但模式保持不变。
# Define tools for threat intelligence gathering
tools = [
{
"name": "lookup_ip_reputation",
"description": "Query IP reputation database to get geolocation, ISP information, abuse history, open ports, and known malicious associations for an IP address. Returns threat types, malware associations, and abuse confidence scoring.",
"input_schema": {
"type": "object",
"properties": {
"ip_address": {
"type": "string",
"description": "The IPv4 or IPv6 address to investigate.",
}
},
"required": ["ip_address"],
},
},
{
"name": "lookup_file_hash",
"description": "Query file reputation service with a cryptographic hash. Returns detection ratio across antivirus engines, malware family classification, behavioral summary, contacted infrastructure, and temporal indicators (first/last seen).",
"input_schema": {
"type": "object",
"properties": {
"file_hash": {
"type": "string",
"description": "The MD5, SHA1, or SHA256 hash of the suspicious file.",
},
"hash_type": {
"type": "string",
"enum": ["md5", "sha1", "sha256"],
"description": "The type of hash provided.",
},
},
"required": ["file_hash", "hash_type"],
},
},
{
"name": "lookup_domain",
"description": "Investigate a domain's reputation including registration details, DNS records, SSL certificate information, hosting provider, and threat categorization. Useful for analyzing phishing domains, malware distribution sites, and C2 infrastructure.",
"input_schema": {
"type": "object",
"properties": {
"domain": {
"type": "string",
"description": "The domain name to investigate (e.g., example.com).",
}
},
"required": ["domain"],
},
},
{
"name": "get_mitre_techniques",
"description": "Map observed behaviors, malware families, or attack patterns to the MITRE ATT&CK framework. Returns matching technique IDs, tactic categories, associated threat groups, and detection recommendations.",
"input_schema": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Description of the behavior, malware family, or attack pattern to map (e.g., 'command and control beaconing', 'credential theft', 'lateral movement').",
}
},
"required": ["query"],
},
},
]
print(f"Defined {len(tools)} threat intelligence tools:")
for tool in tools:
print(f" - {tool['name']}: {tool['description'][:80]}...")
步骤 3：构建模拟威胁情报后端
这些函数模拟真实的威胁情报 API。每个函数返回逼真的数据结构，与从 VirusTotal、AbuseIPDB 或 MITRE ATT&CK 查询等服务获得的数据结构相似。要投入生产，只需将每个函数的主体替换为 API 调用——接口保持不变。
# Simulated threat intelligence data sources
# In production, replace each function body with real API calls:
# lookup_ip_reputation -> AbuseIPDB, GreyNoise, Shodan
# lookup_file_hash -> VirusTotal, MalwareBazaar, Hybrid Analysis
# lookup_domain -> URLhaus, DomainTools, WHOIS
# get_mitre_techniques -> MITRE ATT&CK STIX/TAXII feed
def lookup_ip_reputation(ip_address: str) -> dict:
"""Query IP reputation. In production: AbuseIPDB, GreyNoise, or Shodan API."""
ip_database = {
"203.0.113.42": {
"ip": "203.0.113.42",
"country": "Russia",
"city": "Saint Petersburg",
"asn": "AS48666",
"isp": "MnogoByte LLC",
"abuse_confidence_score": 87,
"total_reports": 1243,
"last_reported": "2026-03-10T14:22:00Z",
"threat_types": ["botnet_c2", "malware_distribution", "brute_force"],
"known_malware_associations": ["Emotet", "Trickbot"],
"open_ports": [443, 8080, 4444],
"is_tor_exit_node": False,
"is_known_proxy": True,
"first_seen": "2025-08-15T00:00:00Z",
"tags": ["banking-trojan-c2", "spam-source"],
},
"198.51.100.17": {
"ip": "198.51.100.17",
"country": "China",
"city": "Shanghai",
"asn": "AS4134",
"isp": "ChinaNet",
"abuse_confidence_score": 94,
"total_reports": 3891,
"last_reported": "2026-03-12T09:15:00Z",
"threat_types": ["apt_c2", "data_exfiltration", "scanning"],
"known_malware_associations": ["PlugX", "ShadowPad"],
"open_ports": [443, 8443, 53],
"is_tor_exit_node": False,
"is_known_proxy": False,
"first_seen": "2024-11-02T00:00:00Z",
"tags": ["apt-infrastructure", "state-sponsored"],
},
}
return ip_database.get(
ip_address,
{
"ip": ip_address,
"country": "Unknown",
"abuse_confidence_score": 0,
"total_reports": 0,
"threat_types": [],
"note": "No records found for this IP",
},
)
def lookup_file_hash(file_hash: str, hash_type: str) -> dict:
"""Query file reputation. In production: VirusTotal or MalwareBazaar API."""
hash_database = {
"d131dd02c5e6eec4693d9a0698aff95c": {
"hash": "d131dd02c5e6eec4693d9a0698aff95c",
"hash_type": "md5",
"sha256": "a1b2c3d4e5f6789012345678abcdef0123456789abcdef0123456789abcdef01",
"detections": 58,
"total_engines": 72,
"detection_rate": "80.6%",
"malware_family": "Emotet",
"malware_type": "banking_trojan",
"severity": "critical",
"file_type": "PE32 executable (DLL)",
"file_size_bytes": 237568,
"file_name": "update_service.dll",
"first_seen": "2025-12-01T08:30:00Z",
"last_seen": "2026-03-09T22:14:00Z",
"ssdeep": "6144:Kl2a8JG1oPRqMDFlOGnA8g0ZFBFSqBKiDEF:Kl2a8Q1oPRDF3",
"tags": ["emotet", "epoch5", "banking-trojan", "dropper"],
"behavior_summary": "Drops secondary payload via regsvr32, establishes persistence via scheduled task, communicates with C2 over HTTPS on non-standard ports",
"contacted_ips": ["203.0.113.42", "203.0.113.88", "192.0.2.101"],
"contacted_domains": ["update-service-cdn.ru", "cdn-api-gateway.cc"],
},
"7b3a0c8f2e1d4a5b6c9d8e7f0a1b2c3d": {
"hash": "7b3a0c8f2e1d4a5b6c9d8e7f0a1b2c3d",
"hash_type": "md5",
"sha256": "f0e1d2c3b4a5968778695a4b3c2d1e0f9a8b7c6d5e4f3a2b1c0d9e8f7a6b5c4d",
"detections": 47,
"total_engines": 70,
"detection_rate": "67.1%",
"malware_family": "Trickbot",
"malware_type": "modular_trojan",
"severity": "critical",
"file_type": "PE32 executable (EXE)",
"file_size_bytes": 524288,
"file_name": "windefupdate.exe",
"first_seen": "2025-10-18T12:00:00Z",
"last_seen": "2026-03-11T06:45:00Z",
"tags": ["trickbot", "gtag-mor84", "credential-stealer", "lateral-movement"],
"behavior_summary": "Injects into svchost.exe, harvests browser credentials, performs network reconnaissance, attempts lateral movement via EternalBlue",
"contacted_ips": ["198.51.100.17", "198.51.100.22"],
"contacted_domains": ["api-check-update.cc", "telemetry-cdn.ru"],
},
}
return hash_database.get(
file_hash,
{
"hash": file_hash,
"hash_type": hash_type,
"detections": 0,
"total_engines": 72,
"malware_family": "Unknown",
"severity": "unknown",
"note": "No records found for this hash",
},
)
def lookup_domain(domain: str) -> dict:
"""Query domain reputation. In production: URLhaus, DomainTools, or WHOIS API."""
domain_database = {
"secure-bankofamerica-login.com": {
"domain": "secure-bankofamerica-login.com",
"reputation_score": 98,
"category": "phishing",
"subcategory": "credential_harvesting",
"active": True,
"registrar": "NameSilo LLC",
"registration_date": "2026-02-28T00:00:00Z",
"registrant_country": "Panama",
"hosting_provider": "BulletProof Hosting Ltd",
"hosting_country": "Moldova",
"ip_addresses": ["192.0.2.55", "192.0.2.56"],
"mx_records": [],
"ssl_issuer": "Let's Encrypt",
"ssl_valid_from": "2026-02-28T00:00:00Z",
"targeted_brand": "Bank of America",
"similar_domains_found": 12,
"urlhaus_reference": "https://urlhaus.abuse.ch/url/2345678/",
"tags": ["phishing-kit", "credential-harvest", "typosquat"],
"associated_ips_with_other_malicious_domains": True,
"dns_records": {
"A": ["192.0.2.55"],
"NS": ["ns1.bulletproof-dns.cc", "ns2.bulletproof-dns.cc"],
"TXT": [],
},
},
"update-service-cdn.ru": {
"domain": "update-service-cdn.ru",
"reputation_score": 91,
"category": "malware",
"subcategory": "c2_server",
"active": True,
"registrar": "REG.RU LLC",
"registration_date": "2025-07-14T00:00:00Z",
"registrant_country": "Russia",
"hosting_provider": "MnogoByte LLC",
"hosting_country": "Russia",
"ip_addresses": ["203.0.113.42"],
"ssl_issuer": "Self-signed",
"tags": ["emotet-c2", "malware-distribution", "fast-flux"],
"associated_malware": ["Emotet", "Trickbot"],
"dns_records": {
"A": ["203.0.113.42", "203.0.113.88"],
"NS": ["ns1.reg.ru", "ns2.reg.ru"],
"TXT": [],
},
},
}
return domain_database.get(
domain,
{
"domain": domain,
"reputation_score": 0,
"category": "unknown",
"active": None,
"note": "No records found for this domain",
},
)
def get_mitre_techniques(query: str) -> dict:
"""Map behaviors to MITRE ATT&CK. In production: query ATT&CK STIX/TAXII feed or local DB."""
mitre_mappings = {
"command and control": {
"techniques": [
{
"id": "T1071.001",
"name": "Web Protocols",
"tactic": "Command and Control",
"description": "Adversaries communicate using application layer protocols associated with web traffic to avoid detection",
},
{
"id": "T1573.002",
"name": "Asymmetric Cryptography",
"tactic": "Command and Control",
"description": "Use asymmetric encryption for C2 communications",
},
{
"id": "T1008",
"name": "Fallback Channels",
"tactic": "Command and Control",
"description": "Use alternate communication channels if primary C2 is disrupted",
},
],
"associated_groups": ["APT28", "APT29", "Lazarus Group", "Wizard Spider"],
"detection_suggestions": [
"Monitor for unusual outbound HTTPS to non-standard ports",
"Inspect TLS certificates for self-signed or recently issued certs",
"Track beaconing patterns in network flow data",
],
},
"credential theft": {
"techniques": [
{
"id": "T1056.001",
"name": "Keylogging",
"tactic": "Collection",
"description": "Log keystrokes to intercept credentials as they are typed",
},
{
"id": "T1555.003",
"name": "Credentials from Web Browsers",
"tactic": "Credential Access",
"description": "Acquire credentials from web browser credential stores",
},
{
"id": "T1003.001",
"name": "LSASS Memory",
"tactic": "Credential Access",
"description": "Access credential material stored in LSASS process memory",
},
],
"associated_groups": ["Trickbot operators", "Emotet operators", "FIN7"],
"detection_suggestions": [
"Monitor for LSASS access by unusual processes",
"Alert on credential store file access",
"Deploy credential guard on endpoints",
],
},
"phishing": {
"techniques": [
{
"id": "T1566.001",
"name": "Spearphishing Attachment",
"tactic": "Initial Access",
"description": "Send emails with a malicious attachment to gain access",
},
{
"id": "T1566.002",
"name": "Spearphishing Link",
"tactic": "Initial Access",
"description": "Send emails with malicious links to credential-harvesting sites",
},
{
"id": "T1598.003",
"name": "Spearphishing Link (for Information)",
"tactic": "Reconnaissance",
"description": "Send spearphishing links to gather information for targeting",
},
],
"associated_groups": ["APT28", "Lazarus Group", "Kimsuky", "TA505"],
"detection_suggestions": [
"Implement DMARC/DKIM/SPF for email authentication",
"Deploy URL rewriting and sandboxing for links",
"Monitor for newly registered look-alike domains",
],
},
"lateral movement": {
"techniques": [
{
"id": "T1210",
"name": "Exploitation of Remote Services",
"tactic": "Lateral Movement",
"description": "Exploit remote services to move laterally within the environment",
},
{
"id": "T1021.002",
"name": "SMB/Windows Admin Shares",
"tactic": "Lateral Movement",
"description": "Use valid accounts to interact with remote network shares using SMB",
},
],
"associated_groups": ["Wizard Spider", "FIN6", "Sandworm Team"],
"detection_suggestions": [
"Monitor for anomalous SMB traffic patterns",
"Alert on use of PsExec or similar tools",
"Track authentication events across endpoints",
],
},
}
query_lower = query.lower()
for key, mapping in mitre_mappings.items():
if key in query_lower:
return mapping
# Fuzzy fallback: check if any keyword appears
for key, mapping in mitre_mappings.items():
for word in key.split():
if word in query_lower:
return mapping
return {
"techniques": [],
"associated_groups": [],
"note": "No direct MITRE ATT&CK mapping found for this query. Try broader terms like 'command and control', 'credential theft', 'phishing', or 'lateral movement'.",
}
print(
"Simulated threat intel backends ready. In production, replace function bodies with real API calls."
)
第四步：创建代理循环
这是核心编排部分。我们给 Claude 一个系统提示，将其定位为高级威胁情报分析师，然后让它决定调用哪些工具以及以何种顺序调用。while 循环会一直持续，只要 Claude 想要调用工具——它可能会调用一个工具，检查结果，然后根据发现决定调用另一个工具。这种多轮推理正是使其成为代理而非简单 API 封装的原因。
SYSTEM_PROMPT = """You are a senior threat intelligence analyst. When given an Indicator of Compromise (IOC), you systematically investigate it by:
1. Identifying the IOC type and querying the most relevant intelligence source first
2. Analyzing initial results to determine what follow-up queries would be valuable
3. Cross-referencing findings across multiple sources when related indicators are found
4. Mapping observed behaviors and malware families to MITRE ATT&CK techniques
5. Synthesizing all findings into a clear, evidence-based assessment
Always query multiple sources when possible. If an IP lookup reveals associated malware, look up the MITRE techniques for that malware. If a hash lookup reveals contacted domains or IPs, investigate those too. Build the fullest picture you can before forming your assessment.
State your confidence level (low/medium/high) and explain what evidence supports it."""
MAX_TURNS = 10 # cap the agent loop to prevent runaway costs
def process_tool_call(tool_name: str, tool_input: dict) -> str:
"""Route tool calls to the appropriate backend function."""
handlers = {
"lookup_ip_reputation": lambda inp: lookup_ip_reputation(inp["ip_address"]),
"lookup_file_hash": lambda inp: lookup_file_hash(inp["file_hash"], inp["hash_type"]),
"lookup_domain": lambda inp: lookup_domain(inp["domain"]),
"get_mitre_techniques": lambda inp: get_mitre_techniques(inp["query"]),
}
handler = handlers.get(tool_name)
if handler is None:
return json.dumps({"error": f"Unknown tool: {tool_name}"})
return json.dumps(handler(tool_input), indent=2)
def run_threat_intel_agent(ioc: str, ioc_type: str) -> tuple[str, list[dict]]:
"""
Run the threat intel enrichment agent on a single IOC.
Returns (analysis_text, list_of_tool_calls_made).
"""
user_message = (
f"Investigate this indicator of compromise and provide a threat assessment:\n"
f" IOC: {ioc}\n"
f" Type: {ioc_type}\n\n"
f"Query all relevant intelligence sources, cross-reference findings, "
f"map to MITRE ATT&CK where applicable, and provide your assessment with "
f"severity rating, confidence score, and recommended response actions."
)
messages = [{"role": "user", "content": user_message}]
tool_calls_made = []
for _turn in range(MAX_TURNS):
response = client.messages.create(
model=MODEL_NAME,
max_tokens=4096,
system=SYSTEM_PROMPT,
tools=tools,
messages=messages,
)
# If Claude is done reasoning, extract final text
if response.stop_reason == "end_turn":
final_text = next(
(block.text for block in response.content if hasattr(block, "text")),
"No analysis generated.",
)
return final_text, tool_calls_made
# If Claude wants to use tools, process all tool calls in this turn
if response.stop_reason == "tool_use":
# Add assistant's response (contains both text and tool_use blocks)
messages.append({"role": "assistant", "content": response.content})
# Process each tool call and collect results
tool_results = []
for block in response.content:
if block.type == "tool_use":
print(f" -> Calling {block.name}({json.dumps(block.input)})")
tool_calls_made.append({"tool": block.name, "input": block.input})
result = process_tool_call(block.name, block.input)
tool_results.append(
{
"type": "tool_result",
"tool_use_id": block.id,
"content": result,
}
)
# Send tool results back to Claude
messages.append({"role": "user", "content": tool_results})
else:
# Unexpected stop reason
return f"Agent stopped unexpectedly: {response.stop_reason}", tool_calls_made
# Hit turn limit without end_turn
return (
f"Agent reached max_turns limit ({MAX_TURNS}) without completing analysis. "
f"Consider raising MAX_TURNS or simplifying the investigation scope.",
tool_calls_made,
)
print("Agent loop ready.")
步骤 5：在示例 IOC 上运行代理
让我们用三种不同的 IOC 类型来测试代理。观察工具调用轨迹——Claude 将决定查询哪些来源，并可能根据发现的结果进行后续查询（例如，在哈希分析中发现的域名进行查找）。
# Three test cases covering different IOC types
test_cases = [
{
"ioc": "203.0.113.42",
"type": "ip_address",
"description": "Suspicious IP flagged by firewall for outbound C2 beaconing",
},
{
"ioc": "d131dd02c5e6eec4693d9a0698aff95c",
"type": "file_hash",
"description": "File hash from endpoint detection alert (suspected Emotet)",
},
{
"ioc": "secure-bankofamerica-login.com",
"type": "domain",
"description": "Domain found in phishing email reported by employee",
},
]
# Store results for structured report generation
all_results = []
for i, test in enumerate(test_cases, 1):
print(f"\n{'=' * 70}")
print(f" Test Case {i}: {test['description']}")
print(f" IOC: {test['ioc']} (Type: {test['type']})")
print(f"{'=' * 70}\n")
print("Tool call trace:")
analysis, tools_used = run_threat_intel_agent(test["ioc"], test["type"])
all_results.append(
{
"ioc": test["ioc"],
"type": test["type"],
"description": test["description"],
"analysis": analysis,
"tools_used": tools_used,
}
)
print(f"\n--- Agent queried {len(tools_used)} tool(s) ---")
print(f"\nAnalysis:\n{analysis}")
print()
步骤 6：生成结构化威胁报告
原始代理分析对于人类分析师来说非常有用，但下游系统（SIEM、工单系统、SOC 仪表板）需要结构化数据。此步骤将代理的自由文本分析转换为标准化的 JSON 报告。在生产环境中，您可以输出 STIX 2.1 对象或直接输入到您的事件响应平台。
REPORT_SCHEMA = {
"type": "object",
"properties": {
"ioc": {
"type": "string",
"description": "The indicator of compromise that was investigated",
},
"ioc_type": {
"type": "string",
"enum": ["ip_address", "file_hash", "domain", "url", "email"],
},
"severity": {
"type": "string",
"enum": ["critical", "high", "medium", "low", "informational"],
},
"confidence": {"type": "integer", "description": "Confidence score from 0-100"},
"threat_classification": {
"type": "string",
"description": "Primary threat category (e.g., 'banking_trojan_c2', 'phishing', 'apt_infrastructure')",
},
"summary": {"type": "string", "description": "One-paragraph executive summary of findings"},
"related_malware": {"type": "array", "items": {"type": "string"}},
"related_threat_groups": {"type": "array", "items": {"type": "string"}},
"mitre_techniques": {
"type": "array",
"items": {
"type": "object",
"properties": {
"technique_id": {"type": "string"},
"technique_name": {"type": "string"},
"tactic": {"type": "string"},
},
},
},
"recommended_actions": {"type": "array", "items": {"type": "string"}},
"related_iocs": {
"type": "array",
"items": {"type": "string"},
"description": "Other IOCs discovered during investigation",
},
},
"required": [
"ioc",
"ioc_type",
"severity",
"confidence",
"threat_classification",
"summary",
"recommended_actions",
],
}
def generate_structured_report(analysis: str, ioc: str, ioc_type: str) -> dict:
"""Transform free-text agent analysis into a structured JSON report."""
response = client.messages.create(
model=MODEL_NAME,
max_tokens=4096,
system=(
"You are a threat intelligence report formatter. Convert the analyst's "
"findings into a structured JSON report matching the provided schema EXACTLY. "
"Include ONLY the fields listed in the schema — do not add extra fields, "
"nested objects, or elaboration. Keep array values as simple strings, not "
"objects. Keep the summary to 2-3 sentences. Return ONLY valid JSON, no "
"markdown fences."
),
messages=[
{
"role": "user",
"content": (
f"Convert this threat intelligence analysis into a structured JSON report.\n\n"
f"IOC investigated: {ioc} (type: {ioc_type})\n\n"
f"Analysis:\n{analysis}\n\n"
f"Schema (include ONLY these fields):\n{json.dumps(REPORT_SCHEMA, indent=2)}\n\n"
f"Return the JSON object directly with no markdown formatting."
),
}
],
)
response_text = response.content[0].text
# Extract JSON from response (handle markdown code blocks if present)
if "```json" in response_text:
json_str = response_text.split("```json")[1].split("```")[0].strip()
elif "```" in response_text:
json_str = response_text.split("```")[1].split("```")[0].strip()
else:
json_str = response_text.strip()
try:
return json.loads(json_str)
except json.JSONDecodeError as e:
return {"error": f"Failed to parse structured report: {e}", "raw": response_text}
# Generate structured reports for all test cases
print("=" * 70)
print(" STRUCTURED THREAT INTELLIGENCE REPORTS")
print("=" * 70)
structured_reports = []
for result in all_results:
report = generate_structured_report(result["analysis"], result["ioc"], result["type"])
structured_reports.append(report)
print(f"\n--- {result['ioc']} ---")
print(json.dumps(report, indent=2))
print()
总结与后续步骤
本教程演示了如何构建一个威胁情报增强代理，该代理通过查询多个数据源、交叉引用发现并生成结构化报告来自主调查 IOC。您可以借鉴的关键模式：
- 工具使用代理循环：Claude 决定调用哪些工具，并根据结果进行后续操作——无需硬编码查询序列
- 模块化工具设计：每个情报源都是一个具有明确定义模式的干净函数，使得替换为真实 API 变得简单直接
- 多源关联：系统提示鼓励 Claude 跨源切换（例如，哈希 → 联系的 IP → IP 信誉 → MITRE 映射）
- 结构化输出：原始分析被转换为机器可读的报告，供下游系统使用
使其可用于生产
替换为真实 API：将模拟函数替换为对 VirusTotal、AbuseIPDB、Shodan、GreyNoise、URLhaus、DomainTools 或您组织的内部威胁情报源的调用。工具模式和代理循环保持不变。
添加更多工具：WHOIS 查询、被动 DNS、SSL 证书透明度日志、沙箱引爆（例如，Joe Sandbox、Hybrid Analysis）以及 EDR/SIEM 增强查询。
使用异步扩展：对于批量 IOC 处理，使用 asyncio 并行运行工具调用并处理 IOC 队列。考虑使用 TTL 缓存结果以避免冗余 API 调用。
强化结构化输出：使用 Claude 的工具使用能力强制 JSON 模式合规，而不是解析自由文本，或根据第 6 步中定义的 REPORT_SCHEMA 验证报告。
与您的技术栈集成：将结构化报告导出为 STIX 2.1 包，推送到 Splunk/Elastic/SOAR 平台，或馈送到工单系统以自动创建事件。
添加置信度校准：根据来源可靠性、数据新鲜度和跨源佐证对置信度分数进行加权。独立一致同意的来源应提升置信度。
进一步阅读
- 构建高效代理 — Anthropic 关于代理模式的研究
- 工具使用文档 — Claude 工具使用能力的完整指南
- MITRE ATT&CK — 本教程中引用的框架
常见问题
常见错误：1) API密钥未设置或无效，导致认证失败——检查.env文件并确保已加载。2) 工具调用格式错误，如缺少必填参数——检查工具输入模式与调用是否匹配。3) 模型陷入循环或达到MAX_TURNS——提高上限或优化提示词。4) 结构化报告解析失败——模型可能返回非JSON或额外字段，需增强提示词或使用工具强制模式。5) 模拟数据未命中——确保IOC在模拟数据库中。排查方法：打印工具调用轨迹，检查消息历史，验证API响应。生产优化：使用异步并发、缓存结果、设置超时和重试、监控成本。
企业应用扩展
生产化建议：1) 替换模拟函数为真实API调用，保持接口不变。2) 添加更多工具如WHOIS、被动DNS、沙箱。3) 使用asyncio并行处理批量IOC。4) 使用Claude的工具强制JSON模式输出，而非解析文本。5) 集成STIX 2.1导出、SIEM推送或工单系统。6) 实现置信度加权和缓存。7) 参考Anthropic的代理构建指南和工具使用文档。
总结
本教程指导开发者构建一个由Claude驱动的威胁情报富化代理，自动化安全分析师的手动关联工作。代理接收原始IOC（IP、哈希、域名），利用Claude的工具使用能力自主决定查询哪些情报源，跨源关联发现，并生成结构化报告。教程涵盖工具模式设计、代理循环构建、多源关联提示和结构化输出生成，并提供模拟后端以便替换为真实API。适合希望学习AI工程实践的安全开发者。
- --
内容来源与验证
- 内容等级：PREMIUM_A
- 质量等级：A
- 难度：中级
- 验证等级：V1
- 原文作者：jannet-park
- 原文日期：2026-04-07
- 来源许可：MIT
- 来源版本：
882ebded174d7c51689128dfb728f0de204a8b37 - 查看原文
站内还有更多完整实战教程
加入会员，解锁全部教程、完整代码与持续更新内容。
加入会员