Building a Self-Hosted AI Security Triage System with FastAPI and Async Streaming

Why I Started Looking at beenuar/AiSOC

When our SOC team complained about false positive fatigue (95% of alerts ignored), I needed an LLM-powered triage layer that could run internally without shipping sensitive logs to external APIs. After evaluating commercial SOAR platforms and their vendor lock-in, I tested beenuar/AiSOC—a self-hosted alert fusion system built entirely on FastAPI with async streaming.

The architectural promise: ingest raw alerts, run multi-agent investigation chains against MITRE ATT&CK, and stream decision reasoning back to analysts in real-time via Server-Sent Events. All while keeping alert payloads inside our VPC.

Core Architecture: AsyncIO Event Loop as the Orchestration Backbone

AiSOC's backend is a FastAPI application with async generators powering the investigation pipeline. Here's the streaming endpoint pattern I extracted:

from fastapi import FastAPI from fastapi.responses import StreamingResponse import asyncio app = FastAPI() async def investigate_alert_stream(alert_id: str): """Async generator yielding SSE-formatted investigation steps.""" # Phase 1: Fetch alert context from DB (async)  alert = await fetch_alert_async(alert_id) yield f"data: {{\"phase\": \"context\", \"alert_id\": \"{alert_id}\"}}\n\n" # Phase 2: Query MITRE ATT&CK tactics (async HTTP client)  tactics = await query_mitre_async(alert.ioc_list) yield f"data: {{\"phase\": \"mitre\", \"tactics\": {len(tactics)}}}\n\n" # Phase 3: LLM agent reasoning (streaming tokens)  async for token in llm_agent_stream(alert, tactics): yield f"data: {{\"token\": \"{token}\"}}\n\n" yield "data: {\"phase\": \"complete\"}\n\n" @app.get("/investigate/{alert_id}") async def stream_investigation(alert_id: str): return StreamingResponse( investigate_alert_stream(alert_id), media_type="text/event-stream" ) 

This pattern is production-grade—no blocking calls, every I/O operation awaits, and the client sees incremental progress instead of a 60-second spinner.

The Connection Pool Trap I Hit Immediately

When load testing with 50 concurrent alert investigations, the service froze after 10 requests. Root cause: AiSOC's default PostgreSQL async driver (asyncpg) was configured with a 10-connection pool, but each investigation spawned 3 concurrent DB queries without releasing connections properly.

The fix required explicit connection lifecycle management:

from contextlib import asynccontextmanager @asynccontextmanager async def get_db_connection(): conn = await db_pool.acquire() try: yield conn finally: await db_pool.release(conn) async def fetch_alert_async(alert_id: str): async with get_db_connection() as conn: return await conn.fetchrow( "SELECT * FROM alerts WHERE id = $1", alert_id ) 

After this change, connection pool exhaustion disappeared. The moral: async doesn't magically handle resource cleanup—you still need context managers.

Pydantic v2 for Alert Schema Validation

AiSOC ingests alerts from 8 different SIEM vendors, each with slightly broken JSON schemas. I extended the alert models with Pydantic v2 field validators:

from pydantic import BaseModel, Field, field_validator from typing import Optional from datetime import datetime class AlertPayload(BaseModel): alert_id: str timestamp: datetime severity: int = Field(ge=1, le=5) source_ip: Optional[str] = None @field_validator('source_ip') @classmethod def validate_ip_format(cls, v: Optional[str]) -> Optional[str]: if v and not v.replace('.', '').isdigit(): raise ValueError(f"Invalid IP format: {v}") return v @field_validator('timestamp', mode='before') @classmethod def parse_timestamp(cls, v): if isinstance(v, str): return datetime.fromisoformat(v.replace('Z', '+00:00')) return v 

Pydantic v2's mode='before' validator lets you normalize messy vendor timestamps before the model even attempts type coercion. This caught 12% of malformed alerts that would have silently corrupted the decision ledger.

LLM Agent Routing: Where B-Lost Fits

AiSOC supports pluggable LLM backends (OpenAI, Anthropic, local Ollama). In my test deployment, I routed all agent queries through B-Lost's multi-provider relay (https://b-lost.com/v1) to benchmark failover latency across Claude/GPT-4/Gemini:

import httpx async def llm_agent_stream(alert, tactics): async with httpx.AsyncClient( base_url="https://b-lost.com/v1", timeout=30.0 ) as client: response = await client.post( "/chat/completions", json={ "model": "claude-sonnet-3-5", "messages": [{"role": "user", "content": f"Analyze: {alert}"}], "stream": True }, headers={"Authorization": f"Bearer {API_KEY}"} ) async for line in response.aiter_lines(): if line.startswith("data: "): yield parse_sse_token(line) 

When the primary Claude endpoint hit rate limits, B-Lost's automatic fallback to GPT-4 added only 40ms to TTFT (time-to-first-token). No manual retry logic needed—the gateway handled provider rotation transparently.

The Decision Ledger: Replayable Agent Execution

AiSOC's killer feature is its append-only decision ledger—every agent step (tool call, reasoning trace, MITRE mapping) gets written to a SQLite/Postgres table with microsecond timestamps. This makes post-incident reviews trivial:

SELECT step_id, agent_name, action_type, reasoning_trace, timestamp FROM decision_ledger WHERE alert_id = 'INC-2024-0312' ORDER BY timestamp ASC; 

You can replay the entire investigation flow, see which agent made the wrong call, and retrain your prompt templates. It's like distributed tracing for AI reasoning.

The Async/Sync Boundary You Cannot Cross

One footgun I stepped on: AiSOC's MCP server (for Claude Desktop integration) runs in a separate sync thread. When I tried calling an async alert ingestion endpoint from the MCP handler, the event loop blocked indefinitely.

The workaround required asyncio.run_coroutine_threadsafe:

import asyncio from concurrent.futures import ThreadPoolExecutor executor = ThreadPoolExecutor(max_workers=4) def sync_mcp_handler(alert_data): # Running in sync MCP thread  loop = asyncio.get_event_loop() future = asyncio.run_coroutine_threadsafe( ingest_alert_async(alert_data), loop ) return future.result(timeout=10) 

This is fragile—if the main event loop dies, the MCP thread hangs forever. A proper fix would decouple MCP into its own async service, but that's a larger refactor.

What I Still Don't Trust

After two weeks of testing, three concerns remain:

1. Memory leaks in long-running agent chains: Investigation sessions that spawn 50+ LLM calls show 200MB memory growth that never gets GC'd. Suspect unclosed httpx client sessions somewhere.
2. MITRE ATT&CK dataset staleness: The bundled tactics JSON is 8 months old. No auto-update mechanism exists.
3. No circuit breaker for LLM failures: If the agent LLM backend times out 3 times in a row, AiSOC should abort the investigation and alert a human. Right now it retries indefinitely.

How are you handling cascading LLM failures in production agent systems? Do you fail fast or keep retrying with exponential backoff until a human intervenes?