🛰️
SOC Alert Triage Autogen
blueprint · 7 agents · 4 MCP servers
1385★ stars
186forks
4.8rating
What it does
Three analysts enrich the alert in parallel: an intel analyst checks the indicators against threat intelligence, an activity analyst pulls the surrounding metric series, and a context analyst reconstructs what the affected account or host was doing from the logs. A verdict agent weighs the combined evidence and a skeptic challenges the call, looping briefly until the verdict survives challenge. The flow then branches: benign alerts are closed with an auditable written rationale; real or uncertain ones are escalated to the on-call analyst with the evidence already assembled. Nothing is dropped without a written reason.
The cast · 7 agents
Closercloserfast
Close the alert with a written rationale: the verdict, the evidence behind it, and what would have changed the call. Post it to the triage log channel so the closure is auditable.
chat.post_message
Skepticskepticreasoning
Attack the verdict. If it says benign, argue how this could be an attack the evidence supports; if suspicious, argue the innocent explanation. Return an object with a 'settled' boolean, the surviving 'benign' call, and confidence. Settle only when the verdict genuinely survives your best challenge; when the evidence is truly ambiguous, settle with benign=false so a human looks.
no tools · reasoning only
Escalatorescalatorfast
Escalate to the on-call analyst with the case already built: the verdict and confidence, the three reports condensed, the timeline, and the first two things to check. Page, don't bury.
chat.post_message
Intel AnalystintelAnalystfast
Check each indicator in the alert — URLs, hashes, IPs — against threat intelligence. Report reputation, first-seen, and related campaigns. Facts only.
intel.url_reportintel.file_report
VerdictverdictAgentreasoning
Weigh the three reports and call it: benign with a specific innocent explanation, or suspicious with the attack pattern it fits. If the skeptic challenged you, answer each point or change your call.
no tools · reasoning only
Context AnalystcontextAnalystfast
Reconstruct from the logs what the affected account or host was doing before and after the alert: logins, locations, unusual commands or queries. Read-only. Report the timeline.
db.query_dsl
Activity AnalystactivityAnalystfast
Pull the metric series around the alert window for the affected service or host: traffic, auth failures, egress volume. Report what deviated from baseline and when.
metrics.query_range
Flow
⇉ parallel
→ Intel Analyst → intelFindings
→ Activity Analyst → activityFindings
→ Context Analyst → contextFindings
↻ loop until verdict.settled == true · max 2
→ Verdict → verdict
→ Skeptic → verdict
⑂ branch
verdict.benign == true
→ Closer → outcome
Interface
Inputs
alertobject
The detection: rule name, indicators, and the affected account or host.
Outputs
outcomestring
Closed-with-rationale, or the escalation package, whichever path ran.
Depends on · 4 MCP servers
🛡️Virustotalio.github.BurtTheCoder/virustotalintelBVerified🔌io.github.grafana/grafana-mcpnot in the registry yetmetrics
🔌io.github.supabase/postgres-mcpnot in the registry yetdb
🔌io.github.slack/slack-mcpnot in the registry yetchat
⤓ Export runnable code
# Generated by @socketcat/compiler for target: langgraph
# blueprint: com.socketcat/soc-alert-triage v1.0.0 schema: socketcat.dev/blueprint/v0
# This code is yours. Edit it freely. The socketcat_runtime helper is optional and can be vendored.
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
import socketcat_runtime as _rt
class State(TypedDict, total=False):
alert: object
outcome: object
intelFindings: object
activityFindings: object
contextFindings: object
__loop_5: object
verdict: object
def node_Fork_0(state):
return {}
def node_Join_1(state):
return {}
def node_AgentInvoke_2(state):
return _rt.run_agent("intelAnalyst", state, {"alert":"alert"}, "intelFindings")
def node_AgentInvoke_3(state):
return _rt.run_agent("activityAnalyst", state, {"alert":"alert"}, "activityFindings")
def node_AgentInvoke_4(state):
return _rt.run_agent("contextAnalyst", state, {"alert":"alert"}, "contextFindings")
def node_LoopHeader_5(state):
return {}
def node_LoopTick_6(state):
return {"__loop_5": state.get("__loop_5", 0) + 1}
def node_Nop_7(state):
return {}
def node_AgentInvoke_8(state):
return _rt.run_agent("verdictAgent", state, {"verdict":"verdict","intelFindings":"intelFindings","contextFindings":"contextFindings","activityFindings":"activityFindings"}, "verdict")
def node_AgentInvoke_9(state):
return _rt.run_agent("skeptic", state, {"verdict":"verdict","intelFindings":"intelFindings","contextFindings":"contextFindings","activityFindings":"activityFindings"}, "verdict")
def node_Branch_10(state):
return {}
def node_Join_11(state):
return {}
def node_AgentInvoke_12(state):
return _rt.run_agent("closer", state, {"verdict":"verdict"}, "outcome")
def node_AgentInvoke_13(state):
return _rt.run_agent("escalator", state, {"verdict":"verdict"}, "outcome")
def route_LoopHeader_5(state):
count = state.get("__loop_5", 0)
return "exit" if (_rt.cond("verdict.settled == true", state) or count >= 2) else "loop"
def route_Branch_10(state):
if _rt.cond("verdict.benign == true", state): return "AgentInvoke_12"
return "AgentInvoke_13"
def build():
b = StateGraph(State)
b.add_node("Fork_0", node_Fork_0)
b.add_node("Join_1", node_Join_1)
b.add_node("AgentInvoke_2", node_AgentInvoke_2)
b.add_node("AgentInvoke_3", node_AgentInvoke_3)
b.add_node("AgentInvoke_4", node_AgentInvoke_4)
b.add_node("LoopHeader_5", node_LoopHeader_5)
b.add_node("LoopTick_6", node_LoopTick_6)
b.add_node("Nop_7", node_Nop_7)
b.add_node("AgentInvoke_8", node_AgentInvoke_8)
b.add_node("AgentInvoke_9", node_AgentInvoke_9)
b.add_node("Branch_10", node_Branch_10)
b.add_node("Join_11", node_Join_11)
b.add_node("AgentInvoke_12", node_AgentInvoke_12)
b.add_node("AgentInvoke_13", node_AgentInvoke_13)
b.add_edge(START, "Fork_0")
b.add_edge("Fork_0", "AgentInvoke_2")
b.add_edge("Fork_0", "AgentInvoke_3")
b.add_edge("Fork_0", "AgentInvoke_4")
b.add_edge("Join_1", "AgentInvoke_8")
b.add_edge("AgentInvoke_2", "Join_1")
b.add_edge("AgentInvoke_3", "Join_1")
b.add_edge("AgentInvoke_4", "Join_1")
b.add_conditional_edges("LoopHeader_5", route_LoopHeader_5, {"loop": "AgentInvoke_8", "exit": "Nop_7"})
b.add_edge("LoopTick_6", "LoopHeader_5")
b.add_edge("Nop_7", "Branch_10")
b.add_edge("AgentInvoke_8", "AgentInvoke_9")
b.add_edge("AgentInvoke_9", "LoopTick_6")
b.add_conditional_edges("Branch_10", route_Branch_10, {"AgentInvoke_12": "AgentInvoke_12", "AgentInvoke_13": "AgentInvoke_13"})
b.add_edge("Join_11", END)
b.add_edge("AgentInvoke_12", "Join_11")
b.add_edge("AgentInvoke_13", "Join_11")
return b.compile()
INPUTS = ["alert"]
OUTPUTS = ["outcome"]
if __name__ == "__main__":
_rt.main(build, INPUTS, OUTPUTS)
▸blueprint.json (the portable format)
{
"id": "com.socketcat/soc-alert-triage",
"flow": {
"type": "sequence",
"blocks": [
{
"type": "parallel",
"blocks": [
{
"in": {
"alert": "alert"
},
"out": "intelFindings",
"use": "intelAnalyst",
"type": "agent"
},
{
"in": {
"alert": "alert"
},
"out": "activityFindings",
"use": "activityAnalyst",
"type": "agent"
},
{
"in": {
"alert": "alert"
},
"out": "contextFindings",
"use": "contextAnalyst",
"type": "agent"
}
]
},
{
"max": 2,
"type": "loop",
"until": "verdict.settled == true",
"blocks": [
{
"in": {
"verdict": "verdict",
"intelFindings": "intelFindings",
"contextFindings": "contextFindings",
"activityFindings": "activityFindings"
},
"out": "verdict",
"use": "verdictAgent",
"type": "agent"
},
{
"in": {
"verdict": "verdict",
"intelFindings": "intelFindings",
"contextFindings": "contextFindings",
"activityFindings": "activityFindings"
},
"out": "verdict",
"use": "skeptic",
"type": "agent"
}
]
},
{
"else": [
{
"in": {
"verdict": "verdict"
},
"out": "outcome",
"use": "escalator",
"type": "agent"
}
],
"type": "branch",
"cases": [
{
"when": "verdict.benign == true",
"blocks": [
{
"in": {
"verdict": "verdict"
},
"out": "outcome",
"use": "closer",
"type": "agent"
}
]
}
]
}
]
},
"tags": [
"security",
"soc",
"triage",
"debate"
],
"title": "SOC Alert Triage",
"agents": {
"closer": {
"model": {
"hint": "fast"
},
"title": "Closer",
"tools": [
"chat.post_message"
],
"instructions": "Close the alert with a written rationale: the verdict, the evidence behind it, and what would have changed the call. Post it to the triage log channel so the closure is auditable."
},
"skeptic": {
"model": {
"hint": "reasoning"
},
"title": "Skeptic",
"output": {
"type": "object",
"required": [
"settled",
"benign"
],
"properties": {
"benign": {
"type": "boolean"
},
"settled": {
"type": "boolean"
},
"confidence": {
"type": "string"
}
}
},
"instructions": "Attack the verdict. If it says benign, argue how this could be an attack the evidence supports; if suspicious, argue the innocent explanation. Return an object with a 'settled' boolean, the surviving 'benign' call, and confidence. Settle only when the verdict genuinely survives your best challenge; when the evidence is truly ambiguous, settle with benign=false so a human looks."
},
"escalator": {
"model": {
"hint": "fast"
},
"title": "Escalator",
"tools": [
"chat.post_message"
],
"instructions": "Escalate to the on-call analyst with the case already built: the verdict and confidence, the three reports condensed, the timeline, and the first two things to check. Page, don't bury."
},
"intelAnalyst": {
"model": {
"hint": "fast"
},
"title": "Intel Analyst",
"tools": [
"intel.url_report",
"intel.file_report"
],
"instructions": "Check each indicator in the alert — URLs, hashes, IPs — against threat intelligence. Report reputation, first-seen, and related campaigns. Facts only."
},
"verdictAgent": {
"model": {
"hint": "reasoning"
},
"title": "Verdict",
"instructions": "Weigh the three reports and call it: benign with a specific innocent explanation, or suspicious with the attack pattern it fits. If the skeptic challenged you, answer each point or change your call."
},
"contextAnalyst": {
"model": {
"hint": "fast"
},
"title": "Context Analyst",
"tools": [
"db.query_dsl"
],
"instructions": "Reconstruct from the logs what the affected account or host was doing before and after the alert: logins, locations, unusual commands or queries. Read-only. Report the timeline."
},
"activityAnalyst": {
"model": {
"hint": "fast"
},
"title": "Activity Analyst",
"tools": [
"metrics.query_range"
],
"instructions": "Pull the metric series around the alert window for the affected service or host: traffic, auth failures, egress volume. Report what deviated from baseline and when."
}
},
"estate": {
"verdict": {
"type": "object",
"description": "The judged verdict, including a 'benign' flag and confidence."
},
"intelFindings": {
"type": "object",
"description": "Indicator reputations from threat intelligence."
},
"contextFindings": {
"type": "object",
"description": "What the account or host was doing before and after."
},
"activityFindings": {
"type": "object",
"description": "The metric series around the alert window."
}
},
"$schema": "socketcat.dev/blueprint/v0",
"authors": [
{
"url": "https://socketcat.com",
"name": "SocketCat"
}
],
"license": "MIT",
"servers": [
{
"ref": "io.github.BurtTheCoder/virustotal",
"alias": "intel"
},
{
"ref": "io.github.grafana/grafana-mcp",
"alias": "metrics"
},
{
"ref": "io.github.supabase/postgres-mcp",
"alias": "db"
},
{
"ref": "io.github.slack/slack-mcp",
"alias": "chat"
}
],
"summary": "Enriches a security alert from three angles in parallel, judges it with a verdict-plus-skeptic pair, and closes with rationale or escalates with a built case.",
"targets": [
"*"
],
"version": "1.0.0",
"interface": {
"inputs": {
"alert": {
"type": "object",
"description": "The detection: rule name, indicators, and the affected account or host."
}
},
"outputs": {
"outcome": {
"type": "string",
"description": "Closed-with-rationale, or the escalation package, whichever path ran."
}
}
},
"extensions": {
"com.autogen": {
"chat": "group"
}
},
"description": "Three analysts enrich the alert in parallel: an intel analyst checks the indicators against threat intelligence, an activity analyst pulls the surrounding metric series, and a context analyst reconstructs what the affected account or host was doing from the logs. A verdict agent weighs the combined evidence and a skeptic challenges the call, looping briefly until the verdict survives challenge. The flow then branches: benign alerts are closed with an auditable written rationale; real or uncertain ones are escalated to the on-call analyst with the evidence already assembled. Nothing is dropped without a written reason."
}