Human-in-the-Loop Design Patterns

import anthropic, time, requests client = anthropic.Anthropic() def request_approval(action: dict, channel: str = "slack") -> bool: """Send approval request and wait for response. Returns True if approved.""" approval_id = f"approval_{int(time.time())}" # Store pending approval in your state store (Redis, DB, etc.) pending_approvals[approval_id] = { "action": action, "status": "pending", "requested_at": time.time(), "timeout_seconds": 300 # 5 minutes } # Send to analyst (Slack, email, dashboard - see other tabs) send_approval_request(approval_id, action, channel) # Poll for response (or use webhook callback pattern) start = time.time() while time.time() - start < 300: # 5 min timeout status = pending_approvals[approval_id]["status"] if status == "approved": return True if status == "rejected": return False time.sleep(5) # Timeout — escalate or auto-reject escalate_timeout(approval_id, action) return False def run_soc_agent_with_hitl(alert: str): messages = [{"role": "user", "content": f"Triage this alert: {alert}"}] while True: response = client.messages.create( model="claude-sonnet-4-6", max_tokens=1000, tools=SOC_TOOLS, messages=messages ) if response.stop_reason == "end_turn": break for block in response.content: if block.type == "tool_use": tool = block.name args = block.input # Check if this action requires approval if requires_approval(tool, args): approved = request_approval({ "tool": tool, "args": args, "alert_context": alert }) if not approved: # Inject rejection into conversation messages.append({ "role": "user", "content": f"Action {tool} was REJECTED by analyst. Do not retry. Document the alert for manual investigation." }) continue # Execute approved or autonomous action result = execute_tool(tool, args) messages.append({"role": "assistant", "content": response.content}) messages.append({"role": "user", "content": [{"type": "tool_result", "tool_use_id": block.id, "content": str(result)}]}) Copy

import json, redis r = redis.Redis() def suspend_agent_for_approval(agent_state: dict, action: dict, approval_id: str): """Suspend agent — save state, send approval request, exit.""" # Serialise agent state including full conversation history r.setex( f"agent_state:{approval_id}", 86400, # 24h TTL json.dumps({ "messages": agent_state["messages"], "pending_action": action, "alert_id": agent_state["alert_id"], "agent_run_id": agent_state["run_id"] }) ) # Send approval request (analyst gets Slack message with Approve/Reject buttons) send_slack_approval(approval_id, action) # Agent exits — will be resumed by webhook return {"status": "suspended", "approval_id": approval_id} # Webhook endpoint — called when analyst clicks Approve or Reject def on_approval_callback(approval_id: str, decision: str, analyst: str): state_json = r.get(f"agent_state:{approval_id}") if not state_json: return {"error": "Approval expired or not found"} state = json.loads(state_json) r.delete(f"agent_state:{approval_id}") if decision == "approved": # Execute the pending action result = execute_tool(state["pending_action"]["tool"], state["pending_action"]["args"]) # Resume agent with action result injected into conversation resume_agent(state, result, analyst) else: # Resume agent with rejection notice resume_agent(state, f"Action rejected by {analyst}", analyst) Copy

APPROVAL_TIERS = { "l1": { "actions": ["block_ip", "quarantine_host", "kill_process"], "approvers": ["l1_analyst_group"], "timeout_seconds": 300, # 5 min — L1 always on shift "on_timeout": "escalate_l2" }, "l2": { "actions": ["disable_account", "revoke_mfa", "network_isolation"], "approvers": ["l2_analyst", "soc_lead"], "timeout_seconds": 900, # 15 min "on_timeout": "escalate_ciso" }, "ciso": { "actions": ["data_deletion", "cert_in_notification", "customer_notification"], "approvers": ["ciso", "ciso_delegate"], "timeout_seconds": 1800, # 30 min "on_timeout": "page_ciso" # PagerDuty / phone call } } def route_approval(action_type: str, action_args: dict): for tier, config in APPROVAL_TIERS.items(): if action_type in config["actions"]: return send_tiered_approval(tier, config, action_type, action_args) # Default — require L2 if action not in any tier return send_tiered_approval("l2", APPROVAL_TIERS["l2"], action_type, action_args) Copy

import os, json, hmac, hashlib, time from slack_sdk import WebClient from flask import Flask, request, jsonify app = Flask(__name__) slack = WebClient(token=os.environ["SLACK_BOT_TOKEN"]) SLACK_SIGNING_SECRET = os.environ["SLACK_SIGNING_SECRET"] APPROVAL_CHANNEL = "#soc-approvals" def send_slack_approval(approval_id: str, action: dict, context: str = ""): """Send interactive Slack approval message with Approve/Reject buttons.""" severity_emoji = {"block_ip": "🔴", "quarantine_host": "🔴", "disable_account": "🟠", "kill_process": "🟡"}.get(action["tool"], "⚪") blocks = [ {"type": "header", "text": {"type": "plain_text", "text": f"{severity_emoji} Agent Action Requires Approval"}}, {"type": "section", "fields": [ {"type": "mrkdwn", "text": f"*Action:*\n`{action['tool']}`"}, {"type": "mrkdwn", "text": f"*Parameters:*\n```{json.dumps(action['args'], indent=2)}```"}, ]}, {"type": "section", "text": {"type": "mrkdwn", "text": f"*Context:*\n{context or action.get('alert_context', 'N/A')}"}}, {"type": "section", "text": {"type": "mrkdwn", "text": f"*Approval ID:* `{approval_id}`\n*Expires:* 5 minutes from now"}}, {"type": "actions", "elements": [ {"type": "button", "text": {"type": "plain_text", "text": "✅ Approve"}, "style": "primary", "action_id": "approve_action", "value": json.dumps({"approval_id": approval_id, "decision": "approved"})}, {"type": "button", "text": {"type": "plain_text", "text": "❌ Reject"}, "style": "danger", "action_id": "reject_action", "value": json.dumps({"approval_id": approval_id, "decision": "rejected"})}, {"type": "button", "text": {"type": "plain_text", "text": "🔍 Investigate First"}, "action_id": "defer_action", "value": json.dumps({"approval_id": approval_id, "decision": "deferred"})} ]} ] slack.chat_postMessage(channel=APPROVAL_CHANNEL, blocks=blocks, text=f"Agent action requires approval: {action['tool']}") @app.route("/slack/interactions", methods=["POST"]) def slack_interaction(): """Webhook endpoint for Slack button interactions.""" # Verify Slack signature timestamp = request.headers.get("X-Slack-Request-Timestamp", "") signature = request.headers.get("X-Slack-Signature", "") body = request.get_data(as_text=True) sig_basestring = f"v0:{timestamp}:{body}" expected = "v0=" + hmac.new( SLACK_SIGNING_SECRET.encode(), sig_basestring.encode(), hashlib.sha256 ).hexdigest() if not hmac.compare_digest(expected, signature): return jsonify({"error": "Invalid signature"}), 403 payload = json.loads(request.form["payload"]) action = payload["actions"][0] analyst = payload["user"]["name"] data = json.loads(action["value"]) # Process the decision on_approval_callback(data["approval_id"], data["decision"], analyst) # Update the Slack message to show decision decision_text = {"approved": "✅ Approved", "rejected": "❌ Rejected", "deferred": "🔍 Deferred for manual review"}[data["decision"]] slack.chat_update( channel=payload["channel"]["id"], ts=payload["message"]["ts"], text=f"{decision_text} by {analyst}", blocks=[{"type": "section", "text": {"type": "mrkdwn", "text": f"*{decision_text}* by @{analyst} at "}}] ) return jsonify({"response_action": "clear"}) Copy

from flask import Flask, jsonify, request from datetime import datetime import uuid, redis, json app = Flask(__name__) r = redis.Redis() def create_approval_request(action: dict, agent_id: str, alert_id: str) -> str: """Create an approval request and store in Redis queue.""" approval_id = str(uuid.uuid4())[:8].upper() r.setex(f"approval:{approval_id}", 600, json.dumps({ # 10min TTL "id": approval_id, "action": action, "agent_id": agent_id, "alert_id": alert_id, "status": "pending", "created_at": datetime.utcnow().isoformat(), "expires_at": (datetime.utcnow().timestamp() + 600) })) r.lpush("approval_queue", approval_id) return approval_id @app.route("/api/approvals", methods=["GET"]) def get_pending_approvals(): """Return all pending approvals for dashboard display.""" ids = r.lrange("approval_queue", 0, -1) approvals = [] for aid in ids: data = r.get(f"approval:{aid.decode()}") if data: approval = json.loads(data) if approval["status"] == "pending": approvals.append(approval) return jsonify({"approvals": approvals, "count": len(approvals)}) @app.route("/api/approvals//decide", methods=["POST"]) def decide_approval(approval_id): """Process approve or reject decision from dashboard.""" data = r.get(f"approval:{approval_id}") if not data: return jsonify({"error": "Approval not found or expired"}), 404 approval = json.loads(data) body = request.get_json() decision = body.get("decision") # "approved" or "rejected" analyst = body.get("analyst") approval.update({"status": decision, "decided_by": analyst, "decided_at": datetime.utcnow().isoformat()}) r.setex(f"approval:{approval_id}", 86400, json.dumps(approval)) # Keep record 24h r.lrem("approval_queue", 0, approval_id) # Trigger agent resumption on_approval_callback(approval_id, decision, analyst) return jsonify({"status": "ok", "decision": decision}) Copy

// Frontend — approval queue polling and display async function loadApprovalQueue() { const res = await fetch('/api/approvals'); const data = await res.json(); document.getElementById('queue-count').textContent = data.count; document.getElementById('approval-list').innerHTML = data.approvals.map(a => { const expiresIn = Math.max(0, Math.floor((a.expires_at - Date.now()/1000))); const urgency = expiresIn < 60 ? 'border-color:#ef4444' : expiresIn < 180 ? 'border-color:#f59e0b' : ''; return ``; }).join('') || '

No pending approvals

'; } async function decide(approvalId, decision) { const analyst = getCurrentUser(); // Your auth system await fetch(`/api/approvals/${approvalId}/decide`, { method: 'POST', headers: {'Content-Type': 'application/json'}, body: JSON.stringify({decision, analyst}) }); loadApprovalQueue(); // Refresh } // Poll every 10 seconds setInterval(loadApprovalQueue, 10000); loadApprovalQueue(); Copy

def extract_confidence(model_response: str) -> float: """ Ask the model to output structured JSON including confidence. Always request this format in your system prompt. """ import json, re # System prompt should include: # "Always respond in JSON: {verdict, confidence (0.0-1.0), reasoning, recommended_action}" try: # Extract JSON from response (model may wrap in markdown) json_match = re.search(r'\{.*\}', model_response, re.DOTALL) if json_match: data = json.loads(json_match.group()) confidence = float(data.get('confidence', 0.0)) return min(max(confidence, 0.0), 1.0) # Clamp to [0,1] except (json.JSONDecodeError, ValueError): pass return 0.0 # Default to zero confidence if unparseable — always escalate def should_require_approval(action_type: str, confidence: float) -> bool: THRESHOLDS = { "block_ip": 0.92, "quarantine_host": 0.95, "true_positive_verdict": 0.90, "alert_triage": 0.80, "disable_account": 2.0, # Never autonomous (> 1.0 = impossible) "delete_file": 2.0, } threshold = THRESHOLDS.get(action_type, 0.90) # Default 0.90 return confidence < threshold Copy