Tool-Calling Patterns for Security AI

import anthropic import json client = anthropic.Anthropic() # Uses ANTHROPIC_API_KEY env var # ── Define your tools ────────────────────────────────────────── TOOLS = [ { "name": "query_siem", "description": "Execute a Splunk SPL query and return matching events as JSON", "input_schema": { "type": "object", "properties": { "spl": {"type": "string", "description": "SPL query to execute"}, "earliest": {"type": "string", "description": "Time range e.g. -24h, -7d", "default": "-24h"} }, "required": ["spl"] } }, { "name": "lookup_ip", "description": "Check an IP address against threat intelligence feeds", "input_schema": { "type": "object", "properties": { "ip": {"type": "string", "description": "IPv4 or IPv6 address to check"} }, "required": ["ip"] } } ] # ── Implement your tools ─────────────────────────────────────── def query_siem(spl: str, earliest: str = "-24h") -> dict: """Replace with your actual Splunk API call""" # from splunklib import client as splunk_client # service = splunk_client.connect(host=HOST, port=8089, username=USER, password=PASS) # job = service.jobs.create(spl, earliest_time=earliest) # return {"events": list(job.results()), "count": job["resultCount"]} return {"events": [], "count": 0, "note": "Replace with real Splunk connection"} def lookup_ip(ip: str) -> dict: """Replace with VirusTotal, AbuseIPDB, or CERT-In TI feed""" # import requests # resp = requests.get(f"https://www.virustotal.com/api/v3/ip_addresses/{ip}", # headers={"x-apikey": VT_KEY}) # data = resp.json()["data"]["attributes"] # return {"malicious": data["last_analysis_stats"]["malicious"] > 0, "score": ...} return {"malicious": False, "score": 0, "note": "Replace with real TI lookup"} # ── Tool dispatcher ──────────────────────────────────────────── def dispatch_tool(name: str, inputs: dict) -> str: """Call the right tool and return result as JSON string""" try: if name == "query_siem": result = query_siem(**inputs) elif name == "lookup_ip": result = lookup_ip(**inputs) else: result = {"error": f"Unknown tool: {name}"} return json.dumps(result) except Exception as e: return json.dumps({"error": str(e)}) # ── The agentic loop ─────────────────────────────────────────── def run_security_agent(alert: str, max_tool_calls: int = 10) -> str: """ Run the agent to completion. Returns the agent's final text response. max_tool_calls prevents runaway agents from consuming budget. """ messages = [{"role": "user", "content": alert}] tool_call_count = 0 system_prompt = """You are a SOC Level 2 analyst. Use your tools to investigate alerts. Ground every factual claim in tool results. Return your findings as JSON with fields: verdict, confidence, evidence[], mitre_techniques[], next_steps[]""" while True: response = client.messages.create( model="claude-sonnet-4-6", max_tokens=2048, system=system_prompt, tools=TOOLS, messages=messages ) # Agent finished reasoning if response.stop_reason == "end_turn": return "".join(b.text for b in response.content if hasattr(b, "text")) # Safety: prevent infinite loops if tool_call_count >= max_tool_calls: messages.append({"role": "assistant", "content": response.content}) messages.append({ "role": "user", "content": [{"type": "text", "text": "Max tool calls reached. Please provide your best verdict with the information gathered so far."}] }) continue # Process tool calls messages.append({"role": "assistant", "content": response.content}) tool_results = [] for block in response.content: if block.type == "tool_use": tool_call_count += 1 result_json = dispatch_tool(block.name, block.input) tool_results.append({ "type": "tool_result", "tool_use_id": block.id, "content": result_json }) messages.append({"role": "user", "content": tool_results}) # ── Usage ────────────────────────────────────────────────────── if __name__ == "__main__": alert = """ EventCode=4625 TargetUserName=admin IpAddress=185.220.101.45 FailureReason=Wrong password LogonType=3 Count=147 TimeGenerated=2024-01-15 03:42:18 IST """ result = run_security_agent(alert) print(result) Copy

import requests import time import json def query_splunk(spl: str, earliest: str = "-24h", latest: str = "now", host: str = "https://splunk:8089", username: str = "admin", password: str = "...", max_results: int = 100) -> dict: """ Execute a Splunk SPL query and return results. For an agent: keep max_results low (100-500) to control token usage. """ session = requests.Session() session.auth = (username, password) session.verify = False # For self-signed certs — use proper CA in production # Create search job create_resp = session.post(f"{host}/services/search/jobs", data={ "search": f"search {spl}", "earliest_time": earliest, "latest_time": latest, "output_mode": "json", "count": max_results }) sid = create_resp.json()["sid"] # Poll until complete while True: status = session.get(f"{host}/services/search/jobs/{sid}", params={"output_mode": "json"}).json() state = status["entry"][0]["content"]["dispatchState"] if state in ("DONE", "FAILED"): break time.sleep(1) # Fetch results results = session.get(f"{host}/services/search/jobs/{sid}/results", params={"output_mode": "json", "count": max_results}).json() return { "count": len(results.get("results", [])), "events": results.get("results", [])[:max_results], "query": spl, "time_range": f"{earliest} to {latest}" } # Tool definition for the agent SPLUNK_TOOL = { "name": "query_splunk", "description": """Execute a Splunk SPL query to search security events. Use for: finding related alerts, checking login history, searching for IOCs. IMPORTANT: Keep queries specific — avoid broad searches that return millions of events. Always use time constraints (earliest=-24h) to limit scope.""", "input_schema": { "type": "object", "properties": { "spl": {"type": "string", "description": "SPL query (without the leading 'search' keyword)"}, "earliest": {"type": "string", "description": "Time range start: -1h, -24h, -7d", "default": "-24h"}, "max_results": {"type": "integer", "description": "Maximum events to return (default 50, max 500)", "default": 50} }, "required": ["spl"] } } Copy

from azure.identity import DefaultAzureCredential from azure.monitor.query import LogsQueryClient from datetime import timedelta import json credential = DefaultAzureCredential() client = LogsQueryClient(credential) def query_sentinel(kql: str, workspace_id: str, timespan_hours: int = 24, max_rows: int = 100) -> dict: """ Execute KQL in Microsoft Sentinel Log Analytics workspace. Requires: azure-monitor-query, azure-identity packages. Auth: DefaultAzureCredential uses env vars, managed identity, or CLI login. """ try: response = client.query_workspace( workspace_id=workspace_id, query=kql + f" | limit {max_rows}", timespan=timedelta(hours=timespan_hours) ) rows = [] for table in response.tables: for row in table.rows: rows.append(dict(zip(table.columns, row))) return { "count": len(rows), "events": rows, "query": kql, "timespan_hours": timespan_hours } except Exception as e: return {"error": str(e), "count": 0, "events": []} SENTINEL_TOOL = { "name": "query_sentinel", "description": """Execute KQL to query Microsoft Sentinel security events. Use for: searching SecurityEvent, SigninLogs, AzureActivity, OfficeActivity tables. Always limit results with | limit N or | summarize before passing to me.""", "input_schema": { "type": "object", "properties": { "kql": {"type": "string", "description": "KQL query (without time filter — timespan set separately)"}, "timespan_hours": {"type": "integer", "description": "Hours to look back (default 24)", "default": 24}, "max_rows": {"type": "integer", "description": "Maximum rows to return (default 50)", "default": 50} }, "required": ["kql"] } } # Environment: set AZURE_WORKSPACE_ID WORKSPACE_ID = "your-log-analytics-workspace-id" Copy

import requests import os VT_KEY = os.environ["VT_API_KEY"] # VirusTotal API key ABUSEIPDB_KEY = os.environ["ABUSEIPDB_KEY"] # AbuseIPDB API key def check_ip_reputation(ip: str) -> dict: """Check IP against VirusTotal and AbuseIPDB""" result = {"ip": ip, "sources": {}} # VirusTotal try: vt = requests.get( f"https://www.virustotal.com/api/v3/ip_addresses/{ip}", headers={"x-apikey": VT_KEY} ).json() attrs = vt.get("data", {}).get("attributes", {}) stats = attrs.get("last_analysis_stats", {}) result["virustotal"] = { "malicious": stats.get("malicious", 0), "suspicious": stats.get("suspicious", 0), "harmless": stats.get("harmless", 0), "country": attrs.get("country"), "asn": attrs.get("asn"), "as_owner": attrs.get("as_owner"), "tags": attrs.get("tags", []) } except Exception as e: result["virustotal"] = {"error": str(e)} # AbuseIPDB try: abuse = requests.get( "https://api.abuseipdb.com/api/v2/check", headers={"Key": ABUSEIPDB_KEY, "Accept": "application/json"}, params={"ipAddress": ip, "maxAgeInDays": 90} ).json() data = abuse.get("data", {}) result["abuseipdb"] = { "abuse_confidence": data.get("abuseConfidenceScore", 0), "total_reports": data.get("totalReports", 0), "country": data.get("countryCode"), "isp": data.get("isp"), "is_tor": data.get("isTor", False) } except Exception as e: result["abuseipdb"] = {"error": str(e)} # Derive verdict vt_mal = result.get("virustotal", {}).get("malicious", 0) abuse_conf = result.get("abuseipdb", {}).get("abuse_confidence", 0) is_tor = result.get("abuseipdb", {}).get("is_tor", False) if vt_mal >= 5 or abuse_conf >= 75 or is_tor: result["verdict"] = "MALICIOUS" result["confidence"] = 90 elif vt_mal >= 2 or abuse_conf >= 25: result["verdict"] = "SUSPICIOUS" result["confidence"] = 65 else: result["verdict"] = "CLEAN" result["confidence"] = 70 return result def check_file_hash(sha256: str) -> dict: """Check a file hash against VirusTotal""" try: vt = requests.get( f"https://www.virustotal.com/api/v3/files/{sha256}", headers={"x-apikey": VT_KEY} ).json() attrs = vt.get("data", {}).get("attributes", {}) stats = attrs.get("last_analysis_stats", {}) return { "hash": sha256, "malicious_detections": stats.get("malicious", 0), "total_engines": sum(stats.values()), "malware_family": attrs.get("popular_threat_classification", {}).get("suggested_threat_label"), "first_seen": attrs.get("first_submission_date"), "verdict": "MALICIOUS" if stats.get("malicious", 0) > 3 else "CLEAN" } except Exception as e: return {"hash": sha256, "error": str(e)} TI_TOOLS = [ { "name": "check_ip_reputation", "description": "Check an IP address against VirusTotal and AbuseIPDB threat intelligence. Returns malicious verdict, country, ASN, Tor status.", "input_schema": {"type": "object", "properties": {"ip": {"type": "string"}}, "required": ["ip"]} }, { "name": "check_file_hash", "description": "Check a SHA256 file hash against VirusTotal. Returns detection count, malware family name, and verdict.", "input_schema": {"type": "object", "properties": {"sha256": {"type": "string", "description": "SHA256 hash of the file"}}, "required": ["sha256"]} } ] Copy

import requests import os SLACK_WEBHOOK = os.environ["SLACK_WEBHOOK_URL"] SLACK_BOT_TOKEN = os.environ["SLACK_BOT_TOKEN"] SOC_CHANNEL = "#soc-alerts" def notify_analyst( alert_id: str, verdict: str, confidence: int, summary: str, evidence: list, proposed_action: str = None ) -> dict: """ Send agent triage result to Slack. If proposed_action is set, include Approve/Reject interactive buttons. This is the human-in-the-loop gate before any write actions. """ COLOURS = {"TRUE_POSITIVE": "#ef4444", "FALSE_POSITIVE": "#4db8a0", "NEEDS_INVESTIGATION": "#f59e0b"} colour = COLOURS.get(verdict, "#818cf8") blocks = [ {"type": "section", "text": {"type": "mrkdwn", "text": f"*🔔 AI Triage Result — Alert {alert_id}*"}}, {"type": "section", "fields": [ {"type": "mrkdwn", "text": f"*Verdict:*\n{verdict}"}, {"type": "mrkdwn", "text": f"*Confidence:*\n{confidence}%"} ]}, {"type": "section", "text": {"type": "mrkdwn", "text": f"*Summary:*\n{summary}"}}, {"type": "section", "text": {"type": "mrkdwn", "text": "*Evidence:*\n" + "\n".join(f"• {e}" for e in evidence[:5])}} ] # Add HITL approval buttons if a write action is proposed if proposed_action: blocks.append({ "type": "section", "text": {"type": "mrkdwn", "text": f"*⚠️ Proposed action:* `{proposed_action}`"} }) blocks.append({ "type": "actions", "elements": [ {"type": "button", "text": {"type": "plain_text", "text": "✅ Approve"}, "style": "primary", "action_id": "approve_action", "value": f"{alert_id}:{proposed_action}"}, {"type": "button", "text": {"type": "plain_text", "text": "❌ Reject"}, "style": "danger", "action_id": "reject_action", "value": f"{alert_id}:rejected"} ] }) response = requests.post(SLACK_WEBHOOK, json={ "channel": SOC_CHANNEL, "attachments": [{"color": colour, "blocks": blocks}] }) return {"sent": response.status_code == 200, "alert_id": alert_id} NOTIFY_TOOL = { "name": "notify_analyst", "description": "Send triage results to the SOC Slack channel. Use this as the FINAL step after completing investigation. Include proposed_action only if you recommend a specific containment action requiring analyst approval.", "input_schema": { "type": "object", "properties": { "alert_id": {"type": "string"}, "verdict": {"type": "string", "enum": ["TRUE_POSITIVE","FALSE_POSITIVE","NEEDS_INVESTIGATION"]}, "confidence": {"type": "integer", "minimum": 0, "maximum": 100}, "summary": {"type": "string"}, "evidence": {"type": "array", "items": {"type": "string"}}, "proposed_action": {"type": "string", "description": "Optional: action requiring analyst approval e.g. 'Block IP 185.220.101.45 on perimeter firewall'"} }, "required": ["alert_id", "verdict", "confidence", "summary", "evidence"] } } Copy

import requests import os from datetime import datetime JIRA_URL = os.environ["JIRA_URL"] # e.g. https://yourorg.atlassian.net JIRA_EMAIL = os.environ["JIRA_EMAIL"] JIRA_TOKEN = os.environ["JIRA_TOKEN"] JIRA_PROJECT = os.environ["JIRA_PROJECT"] # e.g. SOC JIRA_AUTH = (JIRA_EMAIL, JIRA_TOKEN) def create_security_ticket( summary: str, description: str, severity: str, alert_id: str, iocs: list = None, mitre_techniques: list = None ) -> dict: """ Create a Jira ticket for a security incident. Includes full investigation context from the agent. """ priority_map = {"CRITICAL": "Highest", "HIGH": "High", "MEDIUM": "Medium", "LOW": "Low"} priority = priority_map.get(severity.upper(), "Medium") # Build description with investigation context full_desc = f"""{description} --- *AI Triage Details* Alert ID: {alert_id} Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S IST')} *Indicators of Compromise* {chr(10).join(f'* {ioc}' for ioc in (iocs or []))} *MITRE ATT&CK Techniques* {chr(10).join(f'* {t}' for t in (mitre_techniques or []))} _This ticket was created by the AI Triage Agent. An analyst must review and validate before remediation._ """ payload = { "fields": { "project": {"key": JIRA_PROJECT}, "summary": f"[AI Triage] {summary}", "description": {"type": "doc", "version": 1, "content": [{"type": "paragraph", "content": [{"type": "text", "text": full_desc}]}]}, "issuetype": {"name": "Incident"}, "priority": {"name": priority}, "labels": ["ai-triage", "security", severity.lower()] } } resp = requests.post( f"{JIRA_URL}/rest/api/3/issue", json=payload, auth=JIRA_AUTH, headers={"Content-Type": "application/json"} ) if resp.status_code == 201: issue = resp.json() return {"success": True, "ticket_id": issue["key"], "url": f"{JIRA_URL}/browse/{issue['key']}"} else: return {"success": False, "error": resp.text} JIRA_TOOL = { "name": "create_security_ticket", "description": "Create a Jira security incident ticket with investigation findings. Use this for TRUE_POSITIVE verdicts to ensure proper tracking and follow-up. Do NOT create tickets for FALSE_POSITIVE alerts.", "input_schema": { "type": "object", "properties": { "summary": {"type": "string", "description": "One-line incident description"}, "description": {"type": "string", "description": "Full investigation summary"}, "severity": {"type": "string", "enum": ["CRITICAL","HIGH","MEDIUM","LOW"]}, "alert_id": {"type": "string"}, "iocs": {"type": "array", "items": {"type": "string"}}, "mitre_techniques": {"type": "array", "items": {"type": "string"}} }, "required": ["summary", "description", "severity", "alert_id"] } } Copy

import requests import os # CrowdStrike Falcon API CS_CLIENT_ID = os.environ["CS_CLIENT_ID"] CS_CLIENT_SECRET = os.environ["CS_CLIENT_SECRET"] CS_BASE_URL = "https://api.crowdstrike.com" def get_falcon_token() -> str: resp = requests.post(f"{CS_BASE_URL}/oauth2/token", data={"client_id": CS_CLIENT_ID, "client_secret": CS_CLIENT_SECRET}) return resp.json()["access_token"] def query_falcon_host(hostname: str) -> dict: """ Get host details and recent process activity from CrowdStrike Falcon. Read-only — no containment action without explicit human approval. """ token = get_falcon_token() headers = {"Authorization": f"Bearer {token}"} # Find device by hostname device_resp = requests.get(f"{CS_BASE_URL}/devices/queries/devices/v1", params={"filter": f"hostname:'{hostname}'"}, headers=headers).json() if not device_resp.get("resources"): return {"error": f"Host {hostname} not found in Falcon"} device_id = device_resp["resources"][0] device_detail = requests.get(f"{CS_BASE_URL}/devices/entities/devices/v2", params={"ids": device_id}, headers=headers).json() device = device_detail["resources"][0] return { "hostname": hostname, "device_id": device_id, "os": device.get("os_version"), "last_seen": device.get("last_seen"), "status": device.get("status"), "external_ip": device.get("external_ip"), "local_ip": device.get("local_ip"), "agent_version": device.get("agent_version") } # Microsoft Defender for Endpoint (MDE) via Graph API MDE_TENANT_ID = os.environ.get("MDE_TENANT_ID") MDE_CLIENT_ID = os.environ.get("MDE_CLIENT_ID") MDE_CLIENT_SECRET = os.environ.get("MDE_CLIENT_SECRET") def get_mde_token() -> str: resp = requests.post( f"https://login.microsoftonline.com/{MDE_TENANT_ID}/oauth2/v2.0/token", data={"client_id": MDE_CLIENT_ID, "client_secret": MDE_CLIENT_SECRET, "scope": "https://api.securitycenter.microsoft.com/.default", "grant_type": "client_credentials"} ) return resp.json()["access_token"] def query_mde_device(hostname: str) -> dict: """Query Microsoft Defender for Endpoint for host information""" token = get_mde_token() headers = {"Authorization": f"Bearer {token}"} resp = requests.get( "https://api.securitycenter.microsoft.com/api/machines", params={"$filter": f"computerDnsName eq '{hostname}'"}, headers=headers ).json() if not resp.get("value"): return {"error": f"Host {hostname} not found in MDE"} device = resp["value"][0] return { "hostname": hostname, "device_id": device.get("id"), "os": device.get("osPlatform"), "last_seen": device.get("lastSeen"), "risk_score": device.get("riskScore"), "exposure_level": device.get("exposureLevel"), "health_status": device.get("healthStatus"), "aad_device_id": device.get("aadDeviceId") } EDR_TOOLS = [ { "name": "query_falcon_host", "description": "Get CrowdStrike Falcon information for a specific host — OS, last seen, risk status. READ-ONLY.", "input_schema": {"type": "object", "properties": {"hostname": {"type": "string"}}, "required": ["hostname"]} }, { "name": "query_mde_device", "description": "Get Microsoft Defender for Endpoint information for a host — risk score, exposure level, health status. READ-ONLY.", "input_schema": {"type": "object", "properties": {"hostname": {"type": "string"}}, "required": ["hostname"]} } ] Copy