How to Build an AI Pentesting Assistant with LangChain

Table of Contents

• How to Build an AI Pentesting Assistant with LangChain
  • Understanding the Pentesting Assistant Architecture
    • Why This Matters in Production
  • Prerequisites and Environment Setup
• Create and activate virtual environment
• Install core dependencies
• Install security tools (requires sudo on Linux)
  • Environment Variables
  • Building the Core Pentesting Assistant
    • Step 1: Tool Wrappers with Error Handling

📺 Watch: Neural Networks Explained

Video by 3Blue1Brown

---

As of June 2026, security teams face an unprecedented challenge: the averag [2]e enterprise deploys over 500 new code changes daily, yet traditional penetration testing cycles take weeks. According to the 2025 Verizon Data Breach Investigations Report, 74% of breaches involve the exploitation of vulnerabilities that were known but untested. This gap between deployment velocity and security validation has created an urgent need for automated, intelligent pentesting tools.

In this tutorial, you'll build a production-ready AI pentesting assistant that combines LangChain [8]'s orchestration capabilities with real security tools like Nmap and SQLMap. Unlike generic AI wrappers, this assistant will execute actual security scans, interpret results, and generate prioritized remediation plans. We'll cover the architecture decisions that matter in production: rate limiting, result caching, and handling false positives.

Understanding the Pentesting Assistant Architecture

Before writing code, let's examine why traditional approaches fail and how our architecture addresses these limitations. Most automated pentesting tools operate as rigid scripts—they run the same tests regardless of context. Our AI assistant uses a different paradigm: it treats each target as a unique system requiring adaptive testing strategies.

The core architecture consists of three layers:

1. Orchestration Layer: LangChain agents that plan and coordinate testing sequences
2. Tool Execution Layer: Wrappers around real security tools (Nmap, SQLMap, Nikto)
3. Analysis Layer: LLM-powered interpretation of scan results with context awareness

The key insight is that LLMs excel at pattern recognition and decision-making but fail at precise tool execution. By separating these concerns, we get the best of both worlds: AI-driven strategy with deterministic tool outputs.

Why This Matters in Production

In production environments, you cannot afford false positives that waste engineering time or false negatives that leave vulnerabilities exposed. According to the 2025 SANS Penetration Testing Report, organizations waste an average of 40% of security team time validating false positives from automated scanners. Our assistant addresses this by:

• Cross-referencing findings across multiple tools
• Applying context-aware severity scoring
• Generating actionable remediation steps with code examples

Prerequisites and Environment Setup

You'll need Python 3.10+ and the following tools installed on your system. We'll use a virtual environment to isolate dependencies.

# Create and activate virtual environment
python3 -m venv pentest-ai
source pentest-ai/bin/activate

# Install core dependencies
pip install langchain==0.3.0 langchain-community==0.3.0 langchain-openai [7]==0.2.0
pip install python-nmap==0.7.1 requests==2.31.0 beautifulsoup4==4.12.2
pip install pydantic==2.5.0 typing-extensions==4.8.0

# Install security tools (requires sudo on Linux)
sudo apt-get install nmap sqlmap nikto -y

Important: Running security tools against systems you don't own is illegal. All examples in this tutorial should be tested against your own lab environment or authorized targets like HackTheBox or TryHackMe.

Environment Variables

Create a .env file for your API keys:

OPENAI_API_KEY=sk-your-key-here
TARGET_DOMAIN=testphp.vulnweb.com  # Example vulnerable target

Building the Core Pentesting Assistant

Now we'll implement the assistant step by step. Each component is designed for production use with proper error handling, rate limiting, and result caching.

Step 1: Tool Wrappers with Error Handling

First, we create robust wrappers around security tools. These handle edge cases like timeouts, permission errors, and malformed outputs.

# tools/security_tools.py
import subprocess
import json
import time
from typing import Dict, List, Optional
from dataclasses import dataclass, field
from datetime import datetime

@dataclass
class ScanResult:
    """Standardized scan result structure"""
    tool_name: str
    target: str
    findings: List[Dict]
    raw_output: str
    scan_duration: float
    timestamp: datetime = field(default_factory=datetime.now)
    error: Optional[str] = None

class NmapScanner:
    """Production-grade Nmap wrapper with rate limiting"""

    def __init__(self, rate_limit: float = 1.0):
        self.rate_limit = rate_limit  # Seconds between scans
        self.last_scan_time = 0.0

    def scan_port(self, target: str, port: int = 80) -> ScanResult:
        """Scan a single port with rate limiting"""
        # Enforce rate limiting
        current_time = time.time()
        time_since_last = current_time - self.last_scan_time
        if time_since_last < self.rate_limit:
            time.sleep(self.rate_limit - time_since_last)

        start_time = time.time()
        try:
            # Use -T4 for faster scanning, -sV for version detection
            cmd = [
                'nmap', '-T4', '-sV', '-p', str(port),
                '--open', '--script', 'http-title',
                target
            ]

            result = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                timeout=120  # 2-minute timeout
            )

            self.last_scan_time = time.time()

            if result.returncode != 0:
                return ScanResult(
                    tool_name='nmap',
                    target=target,
                    findings=[],
                    raw_output=result.stderr,
                    scan_duration=time.time() - start_time,
                    error=f"Nmap exited with code {result.returncode}"
                )

            # Parse findings from output
            findings = self._parse_nmap_output(result.stdout)

            return ScanResult(
                tool_name='nmap',
                target=target,
                findings=findings,
                raw_output=result.stdout,
                scan_duration=time.time() - start_time
            )

        except subprocess.TimeoutExpired:
            return ScanResult(
                tool_name='nmap',
                target=target,
                findings=[],
                raw_output='',
                scan_duration=120.0,
                error='Scan timed out after 120 seconds'
            )
        except FileNotFoundError:
            return ScanResult(
                tool_name='nmap',
                target=target,
                findings=[],
                raw_output='',
                scan_duration=0.0,
                error='Nmap not installed. Run: sudo apt-get install nmap'
            )

    def _parse_nmap_output(self, output: str) -> List[Dict]:
        """Parse Nmap output into structured findings"""
        findings = []
        for line in output.split('\n'):
            if '/tcp' in line and 'open' in line:
                parts = line.split()
                if len(parts) >= 4:
                    findings.append({
                        'port': parts[0].split('/')[0],
                        'protocol': parts[0].split('/')[1],
                        'state': parts[1],
                        'service': parts[2],
                        'version': ' '.join(parts[3:]) if len(parts) > 3 else 'unknown'
                    })
        return findings

Step 2: SQL Injection Detection with SQLMap

SQLMap is powerful but can be destructive. Our wrapper adds safety checks and result parsing.

# tools/sqlmap_wrapper.py
import subprocess
import json
import os
from typing import Dict, List, Optional

class SQLMapScanner:
    """Safe SQLMap wrapper with result caching"""

    def __init__(self, cache_dir: str = './scan_cache'):
        self.cache_dir = cache_dir
        os.makedirs(cache_dir, exist_ok=True)

    def test_endpoint(self, url: str, method: str = 'GET') -> Dict:
        """Test a single endpoint for SQL injection"""
        # Generate cache key from URL and method
        cache_key = f"{method}_{url.replace('/', '_').replace(':', '')}"
        cache_path = os.path.join(self.cache_dir, f"{cache_key}.json")

        # Check cache first
        if os.path.exists(cache_path):
            with open(cache_path, 'r') as f:
                return json.load(f)

        # Safety checks before running SQLMap
        if not url.startswith(('http://', 'https://')):
            return {'error': 'Invalid URL scheme', 'vulnerable': False}

        try:
            # Use --batch for non-interactive, --level 1 for safety
            cmd = [
                'sqlmap', '-u', url,
                '--batch',  # Non-interactive mode
                '--level', '1',  # Minimal testing depth
                '--risk', '1',   # Low risk (no heavy time-based)
                '--time-sec', '5',  # Max 5 seconds per test
                '--output-dir', '/tmp/sqlmap_output',
                '--flush-session'  # Clean session for each test
            ]

            result = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                timeout=300  # 5-minute timeout
            )

            # Parse results
            is_vulnerable = 'is vulnerable' in result.stdout.lower()
            findings = self._parse_sqlmap_output(result.stdout)

            scan_result = {
                'url': url,
                'method': method,
                'vulnerable': is_vulnerable,
                'findings': findings,
                'raw_output': result.stdout[:5000],  # Truncate for storage
                'error': None
            }

            # Cache results
            with open(cache_path, 'w') as f:
                json.dump(scan_result, f, indent=2)

            return scan_result

        except subprocess.TimeoutExpired:
            return {
                'url': url,
                'method': method,
                'vulnerable': False,
                'findings': [],
                'error': 'Scan timed out after 5 minutes'
            }
        except FileNotFoundError:
            return {
                'url': url,
                'method': method,
                'vulnerable': False,
                'findings': [],
                'error': 'SQLMap not installed. Run: sudo apt-get install sqlmap'
            }

    def _parse_sqlmap_output(self, output: str) -> List[Dict]:
        """Extract structured findings from SQLMap output"""
        findings = []
        lines = output.split('\n')

        for i, line in enumerate(lines):
            if 'Parameter:' in line and 'is vulnerable' in lines[min(i+1, len(lines)-1)]:
                findings.append({
                    'parameter': line.split('Parameter:')[1].strip() if 'Parameter:' in line else 'unknown',
                    'type': 'SQL Injection',
                    'evidence': lines[i+1] if i+1 < len(lines) else ''
                })

        return findings

Step 3: LangChain Agent Integration

Now we connect our tools to LangChain's agent framework. This enables the AI to plan testing sequences and interpret results.

# agent/pentest_agent.py
from langchain.agents import Tool, AgentExecutor, create_react_agent
from langchain.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from langchain.schema import SystemMessage, HumanMessage
from typing import Dict, List, Any
import json

class PentestAgent:
    """AI-powered pentesting assistant using LangChain"""

    def __init__(self, openai_api_key: str, model: str = "gpt [4]-4"):
        self.llm = ChatOpenAI(
            temperature=0.2,  # Low temperature for deterministic outputs
            model=model,
            api_key=openai_api_key
        )

        # Initialize tool wrappers
        self.nmap = NmapScanner()
        self.sqlmap = SQLMapScanner()

        # Define tools for the agent
        self.tools = self._create_tools()

        # Create the agent
        self.agent = self._build_agent()

    def _create_tools(self) -> List[Tool]:
        """Create LangChain tools from our security wrappers"""

        def nmap_scan(target: str) -> str:
            """Scan target for open ports and services"""
            result = self.nmap.scan_port(target, 80)
            if result.error:
                return f"Error: {result.error}"
            return json.dumps(result.findings, indent=2)

        def sqlmap_test(url: str) -> str:
            """Test URL for SQL injection vulnerabilities"""
            result = self.sqlmap.test_endpoint(url)
            if result.get('error'):
                return f"Error: {result['error']}"
            return json.dumps(result, indent=2)

        def analyze_findings(findings_json: str) -> str:
            """Analyze scan findings and generate remediation steps"""
            try:
                findings = json.loads(findings_json)
            except json.JSONDecodeError:
                return "Invalid findings format. Please provide valid JSON."

            prompt = f"""Analyze these security findings and provide:
            1. Severity assessment (Critical/High/Medium/Low)
            2. Potential impact if exploited
            3. Specific remediation steps with code examples
            4. Priority order for fixing

            Findings: {json.dumps(findings, indent=2)}

            Format your response as a structured report."""

            response = self.llm.invoke([HumanMessage(content=prompt)])
            return response.content

        return [
            Tool(
                name="nmap_scan",
                func=nmap_scan,
                description="Scan a target IP or domain for open ports and running services. Input should be a domain or IP address."
            ),
            Tool(
                name="sqlmap_test",
                func=sqlmap_test,
                description="Test a URL for SQL injection vulnerabilities. Input should be a full URL including http:// or https://"
            ),
            Tool(
                name="analyze_findings",
                func=analyze_findings,
                description="Analyze security findings and generate remediation steps. Input should be a JSON string of findings."
            )
        ]

    def _build_agent(self) -> AgentExecutor:
        """Build the ReAct agent with custom prompt"""

        system_prompt = """You are an expert penetration testing assistant. Your role is to:
        1. Plan and execute security scans methodically
        2. Interpret results in context of the target's environment
        3. Provide actionable remediation advice

        Always start with reconnaissance (nmap scan) before testing for vulnerabilities.
        Never run destructive tests without explicit authorization.
        If you encounter errors, explain them clearly and suggest fixes.

        Available tools:
        {tools}

        Use the following format:
        Thought: Consider what to do next
        Action: tool_name
        Action Input: input for the tool
        Observation: result from the tool
        .. (repeat as needed)
        Final Answer: comprehensive analysis and recommendations"""

        prompt = PromptTemplate.from_template(system_prompt)

        agent = create_react_agent(
            llm=self.llm,
            tools=self.tools,
            prompt=prompt
        )

        return AgentExecutor(
            agent=agent,
            tools=self.tools,
            verbose=True,
            max_iterations=10,  # Prevent infinite loops
            handle_parsing_errors=True
        )

    def run_pentest(self, target: str) -> Dict[str, Any]:
        """Execute a complete penetration test against a target"""

        user_request = f"""Perform a security assessment of {target}. Follow these steps:
        1. First, scan for open ports and services using nmap
        2. Based on findings, test for SQL injection vulnerabilities
        3. Analyze all findings and provide remediation recommendations

        Target: {target}"""

        try:
            result = self.agent.invoke({"input": user_request})
            return {
                'target': target,
                'status': 'completed',
                'output': result['output'],
                'intermediate_steps': result.get('intermediate_steps', [])
            }
        except Exception as e:
            return {
                'target': target,
                'status': 'failed',
                'error': str(e)
            }

Step 4: FastAPI Web Interface

For production deployment, we'll wrap our agent in a FastAPI application with proper authentication and rate limiting.

# api/main.py
from fastapi import FastAPI, HTTPException, Depends, Security
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel, Field
from typing import Optional
import time
from collections import defaultdict

app = FastAPI(title="AI Pentesting Assistant API")
security = HTTPBearer()

# Rate limiting configuration
class RateLimiter:
    def __init__(self, max_requests: int = 10, window_seconds: int = 60):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.requests = defaultdict(list)

    def check_rate_limit(self, client_id: str) -> bool:
        now = time.time()
        window_start = now - self.window_seconds

        # Clean old requests
        self.requests[client_id] = [
            req_time for req_time in self.requests[client_id]
            if req_time > window_start
        ]

        if len(self.requests[client_id]) >= self.max_requests:
            return False

        self.requests[client_id].append(now)
        return True

rate_limiter = RateLimiter()

class PentestRequest(BaseModel):
    target: str = Field(.., description="Target domain or IP address")
    scan_type: str = Field(default="full", description="full, quick, or custom")
    timeout_minutes: int = Field(default=10, ge=1, le=60)

class PentestResponse(BaseModel):
    target: str
    status: str
    output: Optional[str] = None
    error: Optional[str] = None

@app.post("/pentest", response_model=PentestResponse)
async def run_pentest(
    request: PentestRequest,
    credentials: HTTPAuthorizationCredentials = Security(security)
):
    """Execute a penetration test against the specified target"""

    # Rate limiting check
    client_id = credentials.credentials
    if not rate_limiter.check_rate_limit(client_id):
        raise HTTPException(
            status_code=429,
            detail="Rate limit exceeded. Maximum 10 requests per minute."
        )

    # Validate target
    if not request.target or len(request.target) > 255:
        raise HTTPException(
            status_code=400,
            detail="Invalid target. Must be between 1 and 255 characters."
        )

    # Initialize agent
    agent = PentestAgent(openai_api_key=os.getenv("OPENAI_API_KEY"))

    # Execute pentest
    result = agent.run_pentest(request.target)

    if result['status'] == 'failed':
        raise HTTPException(
            status_code=500,
            detail=result.get('error', 'Unknown error during pentest')
        )

    return PentestResponse(
        target=result['target'],
        status=result['status'],
        output=result['output']
    )

@app.get("/health")
async def health_check():
    """Health check endpoint"""
    return {"status": "healthy", "timestamp": time.time()}

Handling Edge Cases and Production Concerns

Memory Management

Security scans can consume significant memory. Implement these safeguards:

# utils/memory_manager.py
import psutil
import os
import signal

class MemoryGuard:
    """Prevent memory exhaustion from large scans"""

    def __init__(self, max_memory_mb: int = 1024):
        self.max_memory = max_memory_mb * 1024 * 1024  # Convert to bytes

    def check_memory(self) -> bool:
        """Check if current process exceeds memory limit"""
        process = psutil.Process(os.getpid())
        memory_info = process.memory_info()

        if memory_info.rss > self.max_memory:
            return False
        return True

    def enforce_limit(self):
        """Kill process if memory exceeded"""
        if not self.check_memory():
            os.kill(os.getpid(), signal.SIGTERM)

API Rate Limiting and Cost Control

LLM API calls can be expensive. Implement cost tracking:

# utils/cost_tracker.py
from dataclasses import dataclass
from datetime import datetime, timedelta

@dataclass
class CostTracker:
    """Track and limit API costs"""
    daily_budget: float = 10.0  # USD
    costs: list = None

    def __post_init__(self):
        self.costs = []

    def add_cost(self, amount: float, model: str):
        self.costs.append({
            'amount': amount,
            'model': model,
            'timestamp': datetime.now()
        })

    def get_daily_cost(self) -> float:
        """Calculate total cost for today"""
        today = datetime.now().date()
        daily_costs = [
            c['amount'] for c in self.costs
            if c['timestamp'].date() == today
        ]
        return sum(daily_costs)

    def can_proceed(self) -> bool:
        """Check if within daily budget"""
        return self.get_daily_cost() < self.daily_budget

Testing Your Pentesting Assistant

Create a test script to verify everything works:

# test_assistant.py
import asyncio
from agent.pentest_agent import PentestAgent
import os
from dotenv import load_dotenv

load_dotenv()

async def test_pentest():
    """Test the pentesting assistant against a known vulnerable target"""

    agent = PentestAgent(
        openai_api_key=os.getenv("OPENAI_API_KEY")
    )

    # Test against a safe target (testphp.vulnweb.com is intentionally vulnerable)
    result = agent.run_pentest("testphp.vulnweb.com")

    print(f"Target: {result['target']}")
    print(f"Status: {result['status']}")
    print(f"Output:\n{result['output']}")

if __name__ == "__main__":
    asyncio.run(test_pentest())

What's Next

Your AI pentesting assistant is now functional, but production deployment requires additional considerations:

1. Add authentication: Implement JWT-based authentication for the API
2. Database integration: Store scan results in PostgreSQL for historical analysis
3. Webhook notifications: Alert teams when critical vulnerabilities are found
4. CI/CD integration: Add a GitHub Action that runs scans on pull requests

For further learning, explore our guides on building AI agents for security automation and deploying LangChain in production.

The future of pentesting is AI-augmented, not AI-replaced. Your assistant handles the repetitive scanning and analysis, freeing human experts to focus on complex logic flaws and business logic vulnerabilities that AI cannot yet understand. As of June 2026, this hybrid approach represents the state of the art in security testing—combining the speed of automation with the insight of human expertise.

---