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?