agentic-soc-platform

name	agentic-soc-platform
description	Build AI-driven security operations automation with ASP's agent-centric SIRP, modules, and playbooks
triggers	["set up an agentic SOC platform","create security automation workflows with ASP","integrate AI agents for security operations","build SIRP modules and playbooks","configure ASP for security alert processing","develop custom security automation modules","implement AI-driven threat detection","automate security incident response"]

Agentic SOC Platform Skill

Skill by ara.so — AI Agent Skills collection

Overview

Agentic SOC Platform (ASP) is an open-source, AI-driven security operations automation platform that combines SIEM integration, AI agents (Langgraph/Dify), and a built-in Security Incident Response Platform (SIRP). It processes security alerts through Redis streams, enriches them with AI analysis, and enables automated response workflows.

Key Components:

• Modules: Streaming processors that consume alerts from Redis streams and perform AI-driven analysis
• Playbooks: Event-driven automation tasks triggered manually from the SIRP UI
• SIRP Platform: Built on Nocoly for case management, alerts, and artifacts
• AI Agents: Support for Langgraph, Dify, and local LLMs

Installation

Docker Deployment (Recommended)

# Clone repository
git clone https://github.com/FunnyWolf/agentic-soc-platform.git
cd agentic-soc-platform

# Start with Docker Compose
cd Docker
docker-compose up -d

# Services will be available at:
# - SIRP Platform: http://localhost:8000
# - Redis: localhost:6379
# - Webhook Receiver: http://localhost:5000

Manual Installation

# Python 3.9+ required
git clone https://github.com/FunnyWolf/agentic-soc-platform.git
cd agentic-soc-platform

# Install dependencies
pip install -r requirements.txt

# Configure environment
cp .env.example .env
# Edit .env with your settings

# Initialize database
python manage.py migrate

# Start services
python manage.py runserver  # SIRP platform
python module_engine.py     # Module processor
python playbook_loader.py   # Playbook executor
python webhook_receiver.py  # Alert ingestion

Configuration

Environment Variables

# .env file
REDIS_HOST=localhost
REDIS_PORT=6379
REDIS_DB=0

# Database
DATABASE_URL=postgresql://user:pass@localhost:5432/asp_db

# AI Agent Configuration
OPENAI_API_KEY=${OPENAI_API_KEY}
OPENAI_API_BASE=https://api.openai.com/v1

# Dify Configuration
DIFY_API_URL=http://localhost:5001
DIFY_API_KEY=${DIFY_API_KEY}

# Local LLM (Ollama)
OLLAMA_BASE_URL=http://localhost:11434

# SIRP Configuration
SIRP_API_URL=http://localhost:8000
SIRP_API_KEY=${SIRP_API_KEY}

# Webhook Settings
WEBHOOK_PORT=5000
WEBHOOK_SECRET=${WEBHOOK_SECRET}

Redis Stream Configuration

# config/redis_streams.py
ALERT_STREAMS = {
    'edr_alerts': 'stream:edr:alerts',
    'ndr_alerts': 'stream:ndr:alerts',
    'siem_alerts': 'stream:siem:alerts',
    'email_threats': 'stream:email:threats',
}

CONSUMER_GROUPS = {
    'edr_analyzer': ['stream:edr:alerts'],
    'ndr_analyzer': ['stream:ndr:alerts'],
    'threat_enricher': ['stream:siem:alerts', 'stream:email:threats'],
}

Core Architecture

Alert Processing Flow

# webhook_receiver.py - Receiving alerts from SIEM
from flask import Flask, request, jsonify
import redis
import json

app = Flask(__name__)
r = redis.Redis(host='localhost', port=6379, db=0)

@app.route('/webhook/alert', methods=['POST'])
def receive_alert():
    """Receive alert from SIEM and push to Redis stream"""
    alert_data = request.json
    
    # Determine stream based on alert source
    source = alert_data.get('source', 'unknown')
    stream_key = f"stream:{source}:alerts"
    
    # Push to Redis stream
    message_id = r.xadd(
        stream_key,
        {
            'alert_id': alert_data.get('id'),
            'payload': json.dumps(alert_data),
            'timestamp': alert_data.get('timestamp'),
            'severity': alert_data.get('severity', 'medium')
        }
    )
    
    return jsonify({
        'status': 'success',
        'stream': stream_key,
        'message_id': message_id.decode()
    }), 200

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

Creating Modules

Modules are streaming processors that consume alerts from Redis streams.

Basic Module Structure

# modules/edr_analyzer.py
from core.module_base import ModuleBase
from agents.langgraph_agent import LanggraphAgent
import json

class EDRAnalyzer(ModuleBase):
    """Analyze EDR alerts using AI agent"""
    
    def __init__(self):
        super().__init__(
            name='edr_analyzer',
            streams=['stream:edr:alerts'],
            consumer_group='edr_analysis_group'
        )
        self.agent = LanggraphAgent(
            model='gpt-4',
            system_prompt="""You are a security analyst specializing in EDR alerts.
            Analyze the alert and determine:
            1. Is this a true positive or false positive?
            2. What is the MITRE ATT&CK technique?
            3. What is the recommended response?
            """
        )
    
    async def process_message(self, message_id, data):
        """Process individual alert from stream"""
        alert_payload = json.loads(data['payload'])
        
        # AI analysis
        analysis = await self.agent.analyze({
            'alert_type': alert_payload.get('type'),
            'process_name': alert_payload.get('process_name'),
            'command_line': alert_payload.get('command_line'),
            'user': alert_payload.get('user'),
            'host': alert_payload.get('host')
        })
        
        
        sirp_case =  .create_sirp_case({
            : ,
            : ._map_severity(analysis[]),
            : analysis[],
            : analysis.get(),
            : analysis.get(),
            : [
                {
                    : ,
                    : alert_payload.get()
                },
                {
                    : ,
                    : alert_payload.get()
                }
            ]
        })
        
        
         .ack_message(message_id)
        
         sirp_case
    
     ():
        
         confidence > :
             
         confidence > :
             
         confidence > :
             
         

Module Base Class

# core/module_base.py
import redis
import asyncio
from abc import ABC, abstractmethod
from typing import List, Dict
import logging

class ModuleBase(ABC):
    """Base class for all ASP modules"""
    
    def __init__(self, name: str, streams: List[str], consumer_group: str):
        self.name = name
        self.streams = streams
        self.consumer_group = consumer_group
        self.redis_client = redis.Redis(
            host='localhost',
            port=6379,
            db=0,
            decode_responses=True
        )
        self.logger = logging.getLogger(f"module.{name}")
        self._setup_consumer_groups()
    
    def _setup_consumer_groups(self):
        """Create consumer groups for streams"""
        for stream in self.streams:
            try:
                self.redis_client.xgroup_create(
                    stream,
                    self.consumer_group,
                    id='0',
                    mkstream=
                )
             redis.exceptions.ResponseError  e:
                    (e):
                    
    
      ():
        
        .logger.info()
        
         :
             stream  .streams:
                messages = .redis_client.xreadgroup(
                    .consumer_group,
                    .name,  
                    {stream: },
                    count=,
                    block=
                )
                
                 stream_name, stream_messages  messages:
                     message_id, data  stream_messages:
                        :
                             .process_message(message_id, data)
                         Exception  e:
                            .logger.error()
            
             asyncio.sleep()
    

      ():
        
        
    
      ():
        
         stream  .streams:
            .redis_client.xack(stream, .consumer_group, message_id)
    
      ():
        
         clients.sirp_client  SIRPClient
        
        client = SIRPClient()
          client.create_case(case_data)

Creating Playbooks

Playbooks are event-driven automation tasks triggered from the SIRP UI.

Basic Playbook Structure

# playbooks/threat_intel_enrichment.py
from core.playbook_base import PlaybookBase
from typing import Dict, List

class ThreatIntelEnrichment(PlaybookBase):
    """Enrich indicators with threat intelligence"""
    
    metadata = {
        'name': 'Threat Intel Enrichment',
        'description': 'Query VirusTotal, AbuseIPDB, and other TI sources',
        'input_types': ['ip', 'domain', 'hash', 'url'],
        'output_type': 'enrichment_report'
    }
    
    async def execute(self, artifact: Dict) -> Dict:
        """Execute playbook on artifact"""
        artifact_type = artifact['type']
        artifact_value = artifact['value']
        
        results = {}
        
        if artifact_type == 'ip':
            results['virustotal'] = await self._query_virustotal_ip(artifact_value)
            results['abuseipdb'] = await self._query_abuseipdb(artifact_value)
            results['shodan'] = await self._query_shodan(artifact_value)
        
        elif artifact_type == 'domain':
            results[] =  ._query_virustotal_domain(artifact_value)
            results[] =  ._query_urlscan(artifact_value)
        
         artifact_type == :
            results[] =  ._query_virustotal_hash(artifact_value)
            results[] =  ._query_hybrid_analysis(artifact_value)
        
        
         .update_artifact(artifact[], {
            : results,
            : ._calculate_reputation(results),
            : ._extract_tags(results)
        })
        
         {
            : ,
            : artifact[],
            : results
        }
    
      () -> :
        
         aiohttp
         os
        
        api_key = os.getenv()
        url = 
        
          aiohttp.ClientSession()  session:
              session.get(
                url,
                headers={: api_key}
            )  response:
                 response.status == :
                    data =  response.json()
                     {
                        : data[][][][],
                        : data[][].get(, ),
                        : data[][].get(),
                        : data[][].get()
                    }
                 {: }
    
      () -> :
        
         aiohttp
         os
        
        api_key = os.getenv()
        url = 
        
          aiohttp.ClientSession()  session:
              session.get(
                url,
                headers={: api_key},
                params={: ip, : }
            )  response:
                 response.status == :
                    data =  response.json()
                     {
                        : data[][],
                        : data[][],
                        : data[][]
                    }
                 {: }
    
     () -> :
        
        score = 
        
           results    results[]:
            score -= results[][] * 
        
           results    results[]:
            score -= results[][]
        
         (, score)
    
     () -> []:
        
        tags = []
        
           results:
             results[].get(, ) > :
                tags.append()
        
           results:
             results[].get(, ) > :
                tags.append()
        
         tags

Playbook Base Class

# core/playbook_base.py
from abc import ABC, abstractmethod
from typing import Dict, Any
import logging

class PlaybookBase(ABC):
    """Base class for all playbooks"""
    
    metadata = {
        'name': 'Base Playbook',
        'description': '',
        'input_types': [],
        'output_type': 'generic'
    }
    
    def __init__(self):
        self.logger = logging.getLogger(f"playbook.{self.metadata['name']}")
    
    @abstractmethod
    async def execute(self, artifact: Dict) -> Dict:
        """Execute playbook - must be implemented by subclass"""
        pass
    
    async def update_artifact(self, artifact_id: str, updates: Dict):
        """Update artifact in SIRP"""
        from clients.sirp_client import SIRPClient
        
        client = SIRPClient()
        return await client.update_artifact(artifact_id, updates)
    
    async def ():
        
         clients.sirp_client  SIRPClient
        
        client = SIRPClient()
          client.add_case_note(case_id, note)

AI Agent Integration

Langgraph Agent

# agents/langgraph_agent.py
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from typing import Dict, TypedDict
import os

class AnalysisState(TypedDict):
    alert_data: Dict
    analysis: Dict
    confidence: float
    mitre_technique: str
    recommendation: str

class LanggraphAgent:
    """AI agent using Langgraph for alert analysis"""
    
    def __init__(self, model: str = 'gpt-4', system_prompt: str = ''):
        self.llm = ChatOpenAI(
            model=model,
            api_key=os.getenv('OPENAI_API_KEY'),
            temperature=0.2
        )
        self.system_prompt = system_prompt
        self.graph = self._build_graph()
    
    def _build_graph(self) -> StateGraph:
        """Build Langgraph workflow"""
        workflow = StateGraph(AnalysisState)
        
        # Add nodes
        workflow.add_node("extract_iocs", self._extract_iocs)
        workflow.add_node("analyze_behavior", self._analyze_behavior)
        workflow.add_node("map_mitre", ._map_mitre)
        workflow.add_node(, ._generate_recommendation)
        
        
        workflow.set_entry_point()
        workflow.add_edge(, )
        workflow.add_edge(, )
        workflow.add_edge(, )
        workflow.add_edge(, END)
        
         workflow.()
    
      () -> :
        
        initial_state = {
            : alert_data,
            : {},
            : ,
            : ,
            : 
        }
        
        result =  .graph.ainvoke(initial_state)
         result
    
      () -> AnalysisState:
        
        prompt = 
        
        response =  .llm.ainvoke(prompt)
        state[][] = response.content
         state
    
      () -> AnalysisState:
        
        prompt = 
        
        response =  .llm.ainvoke(prompt)
        
        state[] =   
        state[][] = response.content
         state
    
      () -> AnalysisState:
        
        prompt = 
        
        response =  .llm.ainvoke(prompt)
        state[] = response.content.strip()
         state
    
      () -> AnalysisState:
        
        prompt = 
        
        response =  .llm.ainvoke(prompt)
        state[] = response.content
         state

Dify Agent Integration

# agents/dify_agent.py
import aiohttp
import os
from typing import Dict

class DifyAgent:
    """Integration with Dify workflow platform"""
    
    def __init__(self):
        self.api_url = os.getenv('DIFY_API_URL')
        self.api_key = os.getenv('DIFY_API_KEY')
    
    async def run_workflow(self, workflow_id: str, inputs: Dict) -> Dict:
        """Execute Dify workflow"""
        url = f"{self.api_url}/v1/workflows/run"
        
        async with aiohttp.ClientSession() as session:
            async with session.post(
                url,
                headers={
                    'Authorization': f'Bearer {self.api_key}',
                    'Content-Type': 'application/json'
                },
                json={
                    'workflow_id': workflow_id,
                    'inputs': inputs
                }
            ) as response:
                return await response.json()
    
    async def analyze_alert(self, alert_data: Dict) -> :
        
        result =  .run_workflow(
            workflow_id=,
            inputs={
                : alert_data.get(),
                : alert_data,
                : 
            }
        )
        
         {
            : result[][][],
            : result[][][],
            : result[][][],
            : result[][][]
        }

SIRP Client

# clients/sirp_client.py
import aiohttp
import os
from typing import Dict, List

class SIRPClient:
    """Client for interacting with SIRP platform"""
    
    def __init__(self):
        self.base_url = os.getenv('SIRP_API_URL', 'http://localhost:8000')
        self.api_key = os.getenv('SIRP_API_KEY')
    
    async def create_case(self, case_data: Dict) -> Dict:
        """Create new case in SIRP"""
        async with aiohttp.ClientSession() as session:
            async with session.post(
                f"{self.base_url}/api/v1/cases",
                headers={'Authorization': f'Bearer {self.api_key}'},
                json=case_data
            ) as response:
                return await response.json()
    
    async def update_case(self, case_id: str, updates: Dict) -> Dict:
        """Update existing case"""
        async with aiohttp.ClientSession()  session:
              session.patch(
                ,
                headers={: },
                json=updates
            )  response:
                  response.json()
    
      () -> :
        
          aiohttp.ClientSession()  session:
              session.post(
                ,
                headers={: },
                json=artifact
            )  response:
                  response.json()
    
      () -> :
        
          aiohttp.ClientSession()  session:
              session.patch(
                ,
                headers={: },
                json=updates
            )  response:
                  response.json()
    
      () -> :
        
          aiohttp.ClientSession()  session:
              session.post(
                ,
                headers={: },
                json={: note}
            )  response:
                  response.json()

Running the Platform

Module Engine

# module_engine.py
import asyncio
import importlib
import os
from pathlib import Path

async def load_modules():
    """Dynamically load all modules"""
    modules = []
    module_dir = Path('modules')
    
    for module_file in module_dir.glob('*.py'):
        if module_file.stem.startswith('_'):
            continue
        
        module = importlib.import_module(f'modules.{module_file.stem}')
        
        # Find module class
        for attr_name in dir(module):
            attr = getattr(module, attr_name)
            if (isinstance(attr, type) and 
                hasattr(attr, '__bases__') and 
                'ModuleBase' in [b.__name__ for b in attr.__bases__]):
                modules.append(attr())
    
    return modules

async def main():
    """Run all modules concurrently"""
    modules = await load_modules()
    
    print(f"Loaded {len(modules)} modules")
    for module in modules:
        ()
    
    
     asyncio.gather(*[module.run()  module  modules])

 __name__ == :
    asyncio.run(main())

Playbook Loader

# playbook_loader.py
import asyncio
import importlib
from pathlib import Path
from flask import Flask, request, jsonify

app = Flask(__name__)
playbooks = {}

def load_playbooks():
    """Load all playbooks"""
    playbook_dir = Path('playbooks')
    
    for playbook_file in playbook_dir.glob('*.py'):
        if playbook_file.stem.startswith('_'):
            continue
        
        module = importlib.import_module(f'playbooks.{playbook_file.stem}')
        
        for attr_name in dir(module):
            attr = getattr(module, attr_name)
            if (isinstance(attr, type) and 
                hasattr(attr, '__bases__') and 
                'PlaybookBase' in [b.__name__ for b in attr.__bases__]):
                instance = attr()
                playbooks[instance.metadata['name']] = instance

@app.route('/api/v1/playbooks', methods=['GET'])
def list_playbooks():
    """List available playbooks"""
    return jsonify([
        {
            'name': pb.metadata['name'],
            : pb.metadata[],
            : pb.metadata[]
        }
         pb  playbooks.values()
    ])


  ():
    
     playbook_name   playbooks:
         jsonify({: }), 
    
    artifact = request.json
    playbook = playbooks[playbook_name]
    
    result =  playbook.execute(artifact)
     jsonify(result)

 __name__ == :
    load_playbooks()
    ()
    app.run(host=, port=)

SIEM Integration

Splunk Integration

# integrations/splunk_forwarder.py
import requests
import json
import os

class SplunkForwarder:
    """Forward Splunk alerts to ASP"""
    
    def __init__(self):
        self.asp_webhook_url = os.getenv('ASP_WEBHOOK_URL', 'http://localhost:5000/webhook/alert')
        self.webhook_secret = os.getenv('WEBHOOK_SECRET')
    
    def format_alert(self, splunk_result: dict) -> dict:
        """Format Splunk alert for ASP"""
        return {
            'source': 'splunk',
            'id': splunk_result.get('sid'),
            'timestamp': splunk_result.get('_time'),
            'severity': self._map_severity(splunk_result.get('urgency')),
            'type': splunk_result.get('search_name'),
            'raw_data': splunk_result,
            'host': splunk_result.get('host'),
            'user': splunk_result.get('user'),
            'description': splunk_result.get('description')
        }
    
    def forward_alert(self, splunk_result: dict):
        """Send alert to ASP webhook"""
        alert = .format_alert(splunk_result)
        
        response = requests.post(
            .asp_webhook_url,
            json=alert,
            headers={
                : ,
                : .webhook_secret
            }
        )
        
         response.json()
    
     () -> :
        
        mapping = {
            : ,
            : ,
            : ,
            : ,
            : 
        }
         mapping.get(urgency.lower(), )

Kibana/ELK Integration

# integrations/elk_forwarder.py
from elasticsearch import Elasticsearch
import os
import requests

class ELKForwarder:
    """Forward Elasticsearch alerts to ASP"""
    
    def __init__(self):
        self.es = Elasticsearch(
            [os.getenv('ELASTICSEARCH_URL', 'http://localhost:9200')],
            api_key=os.getenv('ELASTICSEARCH_API_KEY')
        )
        self.asp_webhook_url = os.getenv('ASP_WEBHOOK_URL')