Lab Guide: Unit 5 — Multi-Agent Orchestration for Security

pip install anthropic
mkdir -p ~/noctua-labs/unit5/week1 && cd ~/noctua-labs/unit5/week1
# SDK uses: from anthropic import Anthropic

# Claude Code prompt:
# "Build a Supervisor agent using the Anthropic SDK that:
# 1. Receives a security alert as input
# 2. Routes to one of three specialists: threat_analyst, containment_advisor, compliance
# 3. Each specialist is a separate Claude call with a specialized system prompt
# 4. Supervisor synthesizes specialist output into final recommendation
# 5. Returns structured JSON: {routing, specialist_output, synthesis, confidence}"
claude

import anthropic, os

client = anthropic.Anthropic()

# Create the agent (run once — save agent.id to .env or config file)
agent = client.beta.agents.create(
    name="SOC Investigator",
    model="claude-sonnet-4-6",
    system="""You are a senior SOC analyst conducting incident investigations.
Given a security alert:
1. Extract and enrich all IoCs (IPs, domains, file hashes) using web search
2. Apply CCT analysis: map to MITRE ATT&CK, generate 3 hypotheses with probabilities
3. Assess severity (Critical/High/Medium/Low) with justification
4. Produce a structured JSON report followed by a concise narrative summary""",
    tools=[{"type": "agent_toolset_20260401"}],  # includes bash, read, write, grep, web_search
)

# Create the environment (run once — save environment.id)
environment = client.beta.environments.create(
    name="soc-investigation-env",
    config={"type": "cloud", "networking": {"type": "unrestricted"}},
)

print(f"AGENT_ID={agent.id}")
print(f"AGENT_VERSION={agent.version}")
print(f"ENVIRONMENT_ID={environment.id}")
print("Save these to your .env file — do not re-create on each run")

import anthropic, os, json

client = anthropic.Anthropic()

AGENT_ID = os.environ["AGENT_ID"]
ENVIRONMENT_ID = os.environ["ENVIRONMENT_ID"]

def investigate(alert_text: str, title: str = "SOC Investigation") -> dict:
    """Run a Managed Agent investigation and return the structured report."""
    session = client.beta.sessions.create(
        agent=AGENT_ID,
        environment_id=ENVIRONMENT_ID,
        title=title,
    )
    print(f"Session: {session.id}")

    report_text = []
    with client.beta.sessions.events.stream(session.id) as stream:
        # Send the alert — stream must be open before sending
        client.beta.sessions.events.send(
            session.id,
            events=[{"type": "user.message", "content": [
                {"type": "text", "text": alert_text}
            ]}],
        )
        for event in stream:
            if event.type == "agent.message":
                for block in event.content:
                    if hasattr(block, "text"):
                        print(block.text, end="", flush=True)
                        report_text.append(block.text)
            elif event.type == "agent.tool_use":
                print(f"\n[Tool: {event.name}]", flush=True)
            elif event.type == "session.status_idle":
                print("\n── Investigation complete ──")
                break

    return {"session_id": session.id, "report": "".join(report_text)}

# Test with the Meridian Financial incident
with open("test_incidents.json") as f:
    incidents = json.load(f)

result = investigate(incidents[0]["alert_text"], title=incidents[0]["id"])
print(f"\nSession ID for audit: {result['session_id']}")

import time, json

results = []
for incident in incidents:
    start = time.time()
    result = investigate(incident["alert_text"], title=incident["id"])
    elapsed = time.time() - start

    # Evaluate output
    report = result["report"]
    results.append({
        "id": incident["id"],
        "framework": "managed_agents",
        "latency_s": round(elapsed, 1),
        "severity_correct": incident["expected_severity"].lower() in report.lower(),
        "ioc_count_found": sum(1 for ioc in incident["expected_iocs"] if ioc in report),
        "session_id": result["session_id"],  # audit trail
    })

with open("results_managed_agents.json", "w") as f:
    json.dump(results, f, indent=2)

print(f"Accuracy: {sum(r['severity_correct'] for r in results)}/{len(results)}")
print(f"Avg latency: {sum(r['latency_s'] for r in results)/len(results):.1f}s")

# AGENTS.md — [Your Capstone System Name]
# Human-written. Last updated: [date]

## Patterns & Conventions
- [e.g., All agent outputs return structured JSON with keys: status, data, confidence, error]
- [e.g., Use claude-haiku-4-5 for triage agents, claude-sonnet-4-6 for analysis agents]

## Anti-Patterns — DO NOT
- DO NOT add tool calls without input validation — every parameter must be sanitized
- DO NOT write agent state to disk except via the approved SQLite helper in utils/state.py
- [Add what would go wrong in YOUR system without this guidance]

## Known Traps
- [e.g., The log parser returns UTC timestamps but the alert format uses local time — normalize before comparison]
- [e.g., Staging environment uses a reduced threat intel feed — test results differ from production]