MCP-Powered Product Security Agent

Build a local AI agent that automates security assessment intake using MCP tools.

Introduction

30 Second Summary

Every time a product team ships a new feature or launches a new service, someone on the security side has to figure out what testing it needs, whether all the right evidence has been submitted, and what is blocking progress. Today that process lives in spreadsheets, Slack threads, and tribal knowledge.

In this project, you will build a local AI-powered Product Security Operations agent that automates assessment intake and readiness evaluation using Model Context Protocol (MCP), Ollama, and a configurable rules engine.

What You'll Build

You will type a natural language question into a CLI, watch the agent reason about which tools to call, and receive back a structured security assessment readiness report complete with required services, missing evidence, and identified blockers.

By the end of this project, you'll have:

  • An MCP server exposing 8 product security tools that an AI agent can discover and invoke on its own.
  • An agent loop connecting Llama 3.1 to the MCP server via tool calling, letting the LLM decide which tools to use based on your question.
  • A YAML-based rules engine that deterministically scopes required security services and evidence, keeping audit-critical decisions out of the LLM's hands.
  • Secret Mission: Add a second MCP server for SBOM/SCA analysis and orchestrate the agent across multiple servers simultaneously.

Are there any prerequisites?

You should be comfortable with Python fundamentals and have a basic understanding of REST APIs and JSON. You will also need a machine with at least 32GB of RAM to run the Llama 3.1 model locally through Ollama.

Before We Start

Building an AI agent that calls tools requires three layers: a local LLM that supports tool calling, the MCP SDK that exposes tools in a standard way, and a Python project that wires them together.

In this step, you will install uv (a fast Python package manager), scaffold the project, pull the Llama 3.1 model into Ollama, and verify that every dependency resolves correctly.

In this step, get ready to:
  • Install uv and scaffold the project with all required dependencies.
  • Pull the Llama 3.1 8B model for local tool calling.
  • Verify that the MCP SDK, Ollama SDK, and model are all reachable.
Install uv and create the project

uv is a fast Python package manager that replaces pip and venv with a single tool.

  • Check whether uv is already installed by running this command in your terminal:
uv --version

✔️ I see a version number

uv is ready to go. Skip ahead to creating the project below.

ⓧ Command not found

Install uv by running the command for your operating system.

macOS

  • Install uv by running:
curl -LsSf https://astral.sh/uv/install.sh | sh
  • Close and reopen your terminal so the new PATH takes effect.

Windows

  • Install uv by running:
powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"
  • Close and reopen your terminal so the new PATH takes effect.
  • Confirm the install succeeded by running:
uv --version

You should see a version number printed (e.g. uv 0.12.1).

  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create the project and move into it by running these commands:
uv init prodsec-agent
cd prodsec-agent

What did uv init do?

uv init prodsec-agent created a new prodsec-agent directory on your Desktop with a pyproject.toml, a main.py, a .python-version file, and a Git repository already initialised.

  • Confirm you are inside the project by running:
ls

You should see pyproject.toml and main.py listed in the output.

  • Add the three project dependencies by running:
uv add mcp ollama pyyaml

uv downloads and locks the packages. You will see output listing the resolved versions for mcp, ollama, and pyyaml.

What are these dependencies?

mcp is the Model Context Protocol Python SDK. It lets you define tools that any MCP-compatible client can discover and invoke.

ollama is the official Python client for the Ollama local LLM runtime. It sends prompts and tool schemas to Llama 3.1.

pyyaml parses YAML files. You will use it later for the rules engine configuration.

  • Create the config directory for later use by running:
mkdir config
  • Open the project in VS Code by running:
code .

uv add failed or timed out?

Make sure you are inside the prodsec-agent folder (the one containing pyproject.toml). Run ls to confirm.

If you see a network error, check your internet connection and retry uv add mcp ollama pyyaml.

ask me for help debugging my uv add failure

Pull the Llama 3.1 model

Ollama needs the model weights downloaded before it can respond to prompts. The llama3.1:8b model supports tool calling, which means it can decide which functions to invoke.

  • Pull the model by running:
ollama pull llama3.1:8b

This downloads approximately 4.7 GB. It may take a few minutes depending on your connection speed.

  • Verify the model responds by running:
ollama run llama3.1:8b "Say hello"

You should see Llama 3.1 print a greeting. Press Ctrl+D (macOS/Linux) or Ctrl+Z then Enter (Windows) to exit if the model enters interactive mode.

Ollama not responding?

Make sure Ollama is running. Open a separate terminal and run ollama serve if it is not already active. On Windows, check that the Ollama icon appears in your system tray.

help me troubleshoot Ollama not responding

Verify the full stack

Before you write any project code, confirm that both SDKs import correctly and that Ollama is reachable from Python.

What do you expect to see when you import both SDKs and print a success message?

  • Run the verification command:
uv run python -c "from mcp.server import MCPServer; from ollama import chat; print('Dependencies OK')"

You should see Dependencies OK printed with no errors.

Seeing an import error?

If you see ModuleNotFoundError, make sure you run the command with uv run python (not bare python). The uv run prefix activates the virtual environment that contains your installed packages.

help me debug my MCP or Ollama import error

✔️ Awesome, I've got everything!

Great. Open pyproject.toml in VS Code and confirm your dependencies are listed under [project].

ⓧ I'd like to double check the full code

Your pyproject.toml should look like this:

[project]
name = "prodsec-agent"
version = "0.1.0"
description = "MCP-Powered Product Security Operations Agent"
requires-python = ">=3.12"
dependencies = [
    "mcp",
    "ollama",
    "pyyaml",
]

Your environment is ready. Next up, you will build the data layer that gives the agent something real to query.

Build the Data Layer

Your environment is ready, with uv, the MCP SDK, and Ollama all confirmed working.

Now you need data for the agent to operate on. Without a data layer, the MCP tools you build later would have nothing to query. You would type a question and get empty results. In this step, you'll create a SQLite database populated with realistic Product Security records: products, assessments, evidence artifacts, and service results.

In this step, get ready to:
  • Create the SQLite schema with four tables for product security data.
  • Seed the database with realistic products, assessments, and service results.
  • Run and verify the data layer prints a summary of seeded records.
Create the SQLite schema

The database module needs to handle connection management and define the tables the agent will query. You'll start with the file setup, then add each table group.

  • Create a new file called database.py in the root of your prodsec-agent folder.
  • Add the imports and connection helper by pasting this code:
import sqlite3
import json
from pathlib import Path

DB_PATH = Path(__file__).parent / "data" / "prodsec.db"


def get_connection():
    DB_PATH.parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(str(DB_PATH))
    conn.row_factory = sqlite3.Row
    return conn

What does this code do?

DB_PATH points to a data/ subdirectory that will be created automatically when the database is first accessed.

get_connection() creates the directory if it does not exist, opens a SQLite connection, and sets row_factory = sqlite3.Row so query results behave like dictionaries instead of plain tuples.

  • Below get_connection, add the first half of the init_db function with the products and assessments tables:
def init_db():
    conn = get_connection()
    conn.executescript("""
        CREATE TABLE IF NOT EXISTS products (
            id TEXT PRIMARY KEY,
            name TEXT NOT NULL,
            version TEXT,
            owner TEXT,
            product_type TEXT,
            has_web_interface INTEGER DEFAULT 0,
            has_rest_api INTEGER DEFAULT 0,
            has_mobile_app INTEGER DEFAULT 0,
            has_embedded_firmware INTEGER DEFAULT 0,
            has_third_party_components INTEGER DEFAULT 0,
            has_ai_ml INTEGER DEFAULT 0,
            has_internet_connectivity INTEGER DEFAULT 0,
            deployment_environment TEXT,
            repository_url TEXT,
            test_environment_url TEXT
        );

        CREATE TABLE IF NOT EXISTS assessments (
            id TEXT PRIMARY KEY,
            product_id TEXT NOT NULL,
            status TEXT DEFAULT 'intake',
            readiness TEXT DEFAULT 'not_evaluated',
            created_at TEXT,
            updated_at TEXT,
            required_services TEXT,
            blockers TEXT,
            progress_pct INTEGER DEFAULT 0,
            FOREIGN KEY (product_id) REFERENCES products(id)
        );

What do these tables represent?

products stores each application or device your security team assesses. The boolean columns (has_web_interface, has_embedded_firmware, etc.) let the rules engine determine which security services are required.

assessments tracks each security assessment engagement, linking back to a product and storing workflow state like progress percentage and required services.

  • Save database.py and verify no syntax errors by running:
uv run python -c "import database; print('Import OK')"

You should see Import OK printed. This confirms the file parses correctly so far.

Seeing a SyntaxError?

Make sure the triple-quoted string """ on line 3 of init_db is not accidentally closed early. The SQL string stays open until you add the closing """ in the next chunk.

help me fix a SyntaxError in my database.py file

  • Directly below the assessments table (continuing the same SQL string), add the evidence and service_results tables, then close the function:

        CREATE TABLE IF NOT EXISTS evidence (
            id TEXT PRIMARY KEY,
            assessment_id TEXT NOT NULL,
            evidence_type TEXT NOT NULL,
            provided INTEGER DEFAULT 0,
            file_path TEXT,
            notes TEXT,
            FOREIGN KEY (assessment_id) REFERENCES assessments(id)
        );

        CREATE TABLE IF NOT EXISTS service_results (
            id TEXT PRIMARY KEY,
            assessment_id TEXT NOT NULL,
            service_name TEXT NOT NULL,
            status TEXT DEFAULT 'not_started',
            started_at TEXT,
            completed_at TEXT,
            blocked_reason TEXT,
            blocked_since TEXT,
            FOREIGN KEY (assessment_id) REFERENCES assessments(id)
        );
    """)
    conn.commit()
    conn.close()

What do these tables track?

evidence records artifacts a product team must provide before assessment can begin (architecture diagrams, threat models, SBOMs). The provided flag tracks whether each item has been submitted.

service_results tracks each security testing service (SAST, DAST, SCA, etc.) and its execution state. A service can be not_started, running, complete, or blocked.

  • Save database.py and confirm the schema works by running:
uv run python -c "import database; database.init_db(); print('Schema created')"

You should see Schema created printed with no errors.

  • Confirm the database file now exists by running:
ls data/

You should see prodsec.db listed in the output.

Don't see the data folder?

Check that get_connection() includes the DB_PATH.parent.mkdir(parents=True, exist_ok=True) line. Without it, SQLite cannot create the file because the data/ directory does not exist yet.

help me debug why my data directory isn't being created

Seed with realistic Product Security data

Empty tables are not useful for testing. You need realistic records so the agent can answer questions about real products and assessments.

  • Below init_db, add the first part of the seed_data function that inserts two products and one assessment:
def seed_data():
    conn = get_connection()
    existing = conn.execute("SELECT COUNT(*) FROM products").fetchone()[0]
    if existing > 0:
        conn.close()
        return

    conn.execute("""
        INSERT INTO products VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
    """, ("PROD-001", "CloudSync Pro", "4.2", "Alice Chen", "SaaS Application",
          1, 1, 1, 0, 1, 1, 1, "AWS", "https://github.com/acme/cloudsync",
          "https://staging.cloudsync.acme.io"))

    conn.execute("""
        INSERT INTO products VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
    """, ("PROD-002", "IoT Gateway Hub", "2.0", "Bob Martinez", "Embedded Device",
          1, 1, 0, 1, 1, 0, 1, "On-premise", "https://github.com/acme/iot-gateway",
          None))

    conn.execute("""
        INSERT INTO assessments VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
    """, ("ASM-001", "PROD-001", "in_progress", "ready", "2026-07-15", "2026-07-30",
          json.dumps(["SAST", "DAST", "SCA", "API Security Testing", "AI Security Assessment"]),
          json.dumps([]), 65))

What does this seed data represent?

PROD-001 (CloudSync Pro) is a SaaS application with web, API, mobile, third-party components, and AI/ML capabilities. PROD-002 (IoT Gateway Hub) is an embedded device with firmware. These contrasting products let the rules engine demonstrate different scoping decisions later.

The idempotency check (if existing > 0: return) ensures running the script multiple times does not duplicate data.

  • Continue seed_data by adding evidence items and service results, then closing the connection:
    evidence_items = [
        ("EVD-001", "ASM-001", "architecture_diagram", 1, None, "Provided via Confluence"),
        ("EVD-002", "ASM-001", "threat_model", 1, None, "STRIDE-based threat model"),
        ("EVD-003", "ASM-001", "sbom", 1, None, "CycloneDX format"),
        ("EVD-004", "ASM-001", "test_credentials", 1, None, "Stored in vault"),
    ]
    conn.executemany("""
        INSERT INTO evidence VALUES (?, ?, ?, ?, ?, ?)
    """, evidence_items)

    services = [
        ("SVC-001", "ASM-001", "SAST", "complete", "2026-07-16", "2026-07-18", None, None),
        ("SVC-002", "ASM-001", "SCA", "complete", "2026-07-16", "2026-07-17", None, None),
        ("SVC-003", "ASM-001", "DAST", "running", "2026-07-25", None, None, None),
        ("SVC-004", "ASM-001", "API Security Testing", "complete", "2026-07-18", "2026-07-22", None, None),
        ("SVC-005", "ASM-001", "AI Security Assessment", "in_progress", "2026-07-28", None, None, None),
    ]
    conn.executemany("""
        INSERT INTO service_results VALUES (?, ?, ?, ?, ?, ?, ?, ?)
    """, services)

    conn.commit()
    conn.close()

Why these specific service statuses?

The mix of statuses (complete, running, in_progress) gives the agent realistic scenarios to report on. When you later ask "What is the status of ASM-001?", the agent will describe a partially-complete assessment with SAST and SCA done, DAST actively scanning, and AI Security Assessment still underway.

  • Save database.py and verify the seed logic runs without errors:
uv run python -c "import database; database.init_db(); database.seed_data(); print('Seed OK')"

You should see Seed OK printed with no errors.

Seeing an IntegrityError?

If you see UNIQUE constraint failed, the data was already inserted from a previous run. Delete the database file and try again:

rm data/prodsec.db

help me fix an IntegrityError in my database seed

Run and verify the full data layer

The module needs an entry point that initializes the database, seeds it, and prints a summary so you can confirm everything is working.

  • At the bottom of database.py, add the __main__ block:
if __name__ == "__main__":
    init_db()
    seed_data()
    conn = get_connection()
    products = conn.execute("SELECT id, name FROM products").fetchall()
    assessments = conn.execute("SELECT id, product_id, status, progress_pct FROM assessments").fetchall()
    services = conn.execute("SELECT service_name, status FROM service_results").fetchall()
    conn.close()
    print(f"Database initialized at: {DB_PATH}")
    print(f"\nProducts ({len(products)}):")
    for p in products:
        print(f"  {p['id']}: {p['name']}")
    print(f"\nAssessments ({len(assessments)}):")
    for a in assessments:
        print(f"  {a['id']}: product={a['product_id']}, status={a['status']}, progress={a['progress_pct']}%")
    print(f"\nService Results ({len(services)}):")
    for s in services:
        print(f"  {s['service_name']}: {s['status']}")

What does this block do?

The if __name__ == "__main__" guard means this code only runs when you execute the file directly. When other modules import database.py later, this block is skipped.

It calls init_db() and seed_data(), then queries each table and prints a human-readable summary.

Before you run this, what do you expect to see printed? Think about how many products and assessments the seed data creates.

  • First delete the existing database so the script starts fresh:
rm data/prodsec.db
  • Save database.py and run it:
uv run python database.py

You should see output listing 2 products (CloudSync Pro and IoT Gateway Hub), 1 assessment at 65% progress, and 5 service results with mixed statuses.

Seeing an empty output or errors?

Make sure you deleted the old data/prodsec.db file before running. The idempotency check skips seeding if products already exist from your earlier test commands.

If you see ModuleNotFoundError, confirm you are running the command from inside the prodsec-agent directory.

help me debug my database.py output

✔️ Awesome, I've got everything!

Great. Double check you have saved database.py and that your output shows 2 products, 1 assessment, and 5 service results.

ⓧ I'd like to double check the full code

Scroll up and confirm your database.py contains each of these sections in order:

  • Imports (sqlite3, json, Path) and DB_PATH definition.
  • get_connection() function that creates the directory and returns a connection.
  • init_db() function with four CREATE TABLE statements: products, assessments, evidence, service_results.
  • seed_data() function inserting 2 products, 1 assessment, 4 evidence items, and 5 service results.
  • if __name__ == "__main__" block that calls both functions and prints a summary.

Your data layer is live. Next up, you'll build the MCP server that exposes this data as tools an AI agent can discover and invoke.

Build the MCP Server

You have a database full of products and assessments, but right now only your Python code can access it directly. For an AI agent to use this data, it needs a standardized way to discover and invoke your operations.

The Model Context Protocol solves this by turning your Python functions into tools that any MCP-compatible client can find and call. In this step, you will build an MCP server that exposes nine Product Security tools, then verify it works with an in-memory test client.

In this step, get ready to:
  • Create an MCP server with eight assessment management tools.
  • Add a readiness check tool that evaluates assessment preparedness.
  • Test the server using an in-memory MCP client.
Create the MCP server with assessment tools

The MCP Python SDK lets you register tools with a simple decorator pattern. You write a normal Python function with type hints, add @mcp.tool(), and the SDK auto-generates the JSON schema that clients use to discover your tools.

  • Create a new file called mcp_server.py in your prodsec-agent folder.
  • Add the imports and server initialization by pasting this code:
import json
from datetime import datetime
from mcp.server import MCPServer
from database import get_connection, init_db, seed_data

mcp = MCPServer("ProductSecurity")

init_db()
seed_data()

What does this code do?

MCPServer("ProductSecurity") creates a named MCP server instance. The init_db() and seed_data() calls at module level ensure the database exists whenever this module is imported.

Any function you decorate with @mcp.tool() becomes discoverable to MCP clients. The SDK reads the function's type hints and docstring to generate the tool's input schema and description automatically.

  • Add the first tool, create_assessment, below the initialization code:
@mcp.tool()
def create_assessment(product_id: str) -> str:
    """Create a new security assessment for a product.

    Args:
        product_id: The ID of the product to assess (e.g., PROD-001)

    Returns:
        JSON with the created assessment details including its new ID
    """
    conn = get_connection()
    product = conn.execute("SELECT * FROM products WHERE id = ?", (product_id,)).fetchone()
    if not product:
        conn.close()
        return json.dumps({"error": f"Product {product_id} not found"})

    assessment_id = f"ASM-{datetime.now().strftime('%Y%m%d%H%M%S')}"
    now = datetime.now().isoformat()

    conn.execute("""
        INSERT INTO assessments (id, product_id, status, readiness, created_at, updated_at,
            required_services, blockers, progress_pct)
        VALUES (?, ?, 'intake', 'not_evaluated', ?, ?, '[]', '[]', 0)
    """, (assessment_id, product_id, now, now))
    conn.commit()
    conn.close()

    return json.dumps({"assessment_id": assessment_id, "product_id": product_id, "status": "intake"})
  • Save mcp_server.py and confirm the module loads by running:
uv run python -c "from mcp_server import mcp; print('Server loaded with 1 tool')"

You should see Server loaded with 1 tool printed. This confirms the imports resolve and the decorator registered your function.

Seeing an ImportError?

  • Make sure you are running from inside the prodsec-agent directory where pyproject.toml lives.
  • Confirm database.py exists in the same folder as mcp_server.py.
  • Check that mcp is in your pyproject.toml dependencies and run uv sync if needed.

Why can't my MCP server module load?

Now add the remaining query and management tools. These follow the same pattern: decorated function, docstring describing the tool, and JSON-serialized return value.

  • Add the get_assessment tool below create_assessment:
@mcp.tool()
def get_assessment(assessment_id: str) -> str:
    """Get full details of a security assessment.

    Args:
        assessment_id: The assessment ID (e.g., ASM-001)

    Returns:
        JSON with assessment details including status, progress, required services, and blockers
    """
    conn = get_connection()
    assessment = conn.execute("SELECT * FROM assessments WHERE id = ?", (assessment_id,)).fetchone()
    if not assessment:
        conn.close()
        return json.dumps({"error": f"Assessment {assessment_id} not found"})

    result = dict(assessment)
    result["required_services"] = json.loads(result["required_services"] or "[]")
    result["blockers"] = json.loads(result["blockers"] or "[]")
    conn.close()
    return json.dumps(result)
  • Add the get_product_info and list_assessments tools:
@mcp.tool()
def get_product_info(product_id: str) -> str:
    """Get detailed information about a product including its attributes and capabilities.

    Args:
        product_id: The product ID (e.g., PROD-001 or PROD-002)

    Returns:
        JSON with product details: name, version, owner, type, capabilities, and environment info
    """
    conn = get_connection()
    product = conn.execute("SELECT * FROM products WHERE id = ?", (product_id,)).fetchone()
    conn.close()
    if not product:
        return json.dumps({"error": f"Product {product_id} not found"})
    return json.dumps(dict(product))


@mcp.tool()
def list_assessments() -> str:
    """List all security assessments with their current status.

    Returns:
        JSON array of all assessments with id, product_id, status, readiness, and progress
    """
    conn = get_connection()
    rows = conn.execute(
        "SELECT id, product_id, status, readiness, progress_pct, created_at FROM assessments"
    ).fetchall()
    conn.close()
    return json.dumps([dict(r) for r in rows])
  • Save mcp_server.py and verify four tools load:
uv run python -c "from mcp_server import mcp; print('4 tools registered')"

You should see 4 tools registered printed with no errors.

  • Add the get_assessment_status tool:
@mcp.tool()
def get_assessment_status(assessment_id: str) -> str:
    """Get the current status and progress of an assessment including all individual service results.

    Args:
        assessment_id: The assessment ID (e.g., ASM-001)

    Returns:
        JSON with overall status, progress percentage, and each service's individual status
    """
    conn = get_connection()
    assessment = conn.execute("SELECT * FROM assessments WHERE id = ?", (assessment_id,)).fetchone()
    if not assessment:
        conn.close()
        return json.dumps({"error": f"Assessment {assessment_id} not found"})

    services = conn.execute(
        "SELECT * FROM service_results WHERE assessment_id = ?", (assessment_id,)
    ).fetchall()
    conn.close()

    return json.dumps({
        "assessment_id": assessment_id,
        "status": assessment["status"],
        "readiness": assessment["readiness"],
        "progress_pct": assessment["progress_pct"],
        "services": [dict(s) for s in services],
    })
  • Add the get_evidence_status tool:
@mcp.tool()
def get_evidence_status(assessment_id: str) -> str:
    """Get the evidence and artifacts status for an assessment. Shows what has been provided and what is missing.

    Args:
        assessment_id: The assessment ID (e.g., ASM-001)

    Returns:
        JSON listing provided evidence, missing evidence, and counts
    """
    conn = get_connection()
    evidence = conn.execute(
        "SELECT * FROM evidence WHERE assessment_id = ?", (assessment_id,)
    ).fetchall()
    conn.close()

    provided = [dict(e) for e in evidence if e["provided"]]
    missing = [dict(e) for e in evidence if not e["provided"]]

    return json.dumps({
        "assessment_id": assessment_id,
        "provided": provided,
        "missing": missing,
        "total": len(evidence),
        "provided_count": len(provided),
        "missing_count": len(missing),
    })
  • Save and verify six tools are loaded:
uv run python -c "from mcp_server import mcp; print('6 tools registered')"

You should see 6 tools registered printed.

  • Add the get_blockers tool:
@mcp.tool()
def get_blockers(assessment_id: str) -> str:
    """Get current blockers preventing assessment progress.

    Args:
        assessment_id: The assessment ID (e.g., ASM-001)

    Returns:
        JSON with blocked services and missing evidence items
    """
    conn = get_connection()
    blocked_services = conn.execute(
        "SELECT * FROM service_results WHERE assessment_id = ? AND status = 'blocked'",
        (assessment_id,),
    ).fetchall()

    missing_evidence = conn.execute(
        "SELECT * FROM evidence WHERE assessment_id = ? AND provided = 0",
        (assessment_id,),
    ).fetchall()
    conn.close()

    return json.dumps({
        "assessment_id": assessment_id,
        "blocked_services": [dict(s) for s in blocked_services],
        "missing_evidence": [dict(e) for e in missing_evidence],
    })
  • Add the update_assessment_status tool:
@mcp.tool()
def update_assessment_status(assessment_id: str, status: str) -> str:
    """Update the workflow status of an assessment.

    Args:
        assessment_id: The assessment ID (e.g., ASM-001)
        status: New status. One of: intake, scoping, in_progress, blocked, complete, cancelled

    Returns:
        JSON confirming the status update
    """
    valid_statuses = ["intake", "scoping", "in_progress", "blocked", "complete", "cancelled"]
    if status not in valid_statuses:
        return json.dumps({"error": f"Invalid status. Must be one of: {valid_statuses}"})

    conn = get_connection()
    conn.execute(
        "UPDATE assessments SET status = ?, updated_at = ? WHERE id = ?",
        (status, datetime.now().isoformat(), assessment_id),
    )
    conn.commit()
    conn.close()
    return json.dumps({"assessment_id": assessment_id, "new_status": status})

What do these tools cover?

Together, these eight tools give the agent full CRUD access to the assessment workflow. It can create assessments, query their status, check evidence completeness, identify blockers, and move assessments through workflow stages.

Each tool returns JSON strings because MCP tool responses are text-based. The LLM will parse these responses to formulate its next action or final answer.

  • Save mcp_server.py and confirm all eight tools load:
uv run python -c "from mcp_server import mcp; print('8 tools registered')"

You should see 8 tools registered printed with no errors.

Add the readiness check tool

The eight tools above handle data retrieval and updates. The ninth tool is different: it applies business logic to evaluate whether a product is actually ready for a security assessment.

For now, this tool will import a rules engine module that does not exist yet. Python's lazy import pattern (importing inside the function body) means the server loads fine. The import only triggers when the tool is actually called.

  • Add the check_assessment_readiness tool at the bottom of mcp_server.py:
@mcp.tool()
def check_assessment_readiness(product_id: str) -> str:
    """Check if a product has all required evidence and information to begin a security assessment.
    Uses the configurable rules engine to determine required services and evidence based on product attributes.

    Args:
        product_id: The product ID to evaluate (e.g., PROD-001 or PROD-002)

    Returns:
        JSON with readiness verdict (READY/NOT READY), required services with justification,
        required evidence, provided evidence, missing evidence, and blockers
    """
    from rules_engine import evaluate_readiness

    conn = get_connection()
    product = conn.execute("SELECT * FROM products WHERE id = ?", (product_id,)).fetchone()
    if not product:
        conn.close()
        return json.dumps({"error": f"Product {product_id} not found"})

    assessment = conn.execute(
        "SELECT id FROM assessments WHERE product_id = ? ORDER BY created_at DESC LIMIT 1",
        (product_id,),
    ).fetchone()

    evidence = []
    if assessment:
        evidence = conn.execute(
            "SELECT * FROM evidence WHERE assessment_id = ?", (assessment["id"],)
        ).fetchall()
    conn.close()

    result = evaluate_readiness(dict(product), [dict(e) for e in evidence])
    return json.dumps(result)

Why is the import inside the function?

The from rules_engine import evaluate_readiness line is inside the function body, not at the top of the file. This means Python only tries to find rules_engine.py when someone actually calls this tool.

You will create the rules engine in a later step. Until then, the server loads and all other tools work fine.

  • Save mcp_server.py and confirm all nine tools load:
uv run python -c "from mcp_server import mcp; print('9 tools registered')"

You should see 9 tools registered printed. The lazy import means the server loads successfully even though rules_engine.py does not exist yet.

Test the server with an in-memory client

The MCP SDK includes a Client class that can connect directly to a server object in memory. This means you can test your tools without spinning up a subprocess or HTTP server.

  • Create a new file called test_server.py in your prodsec-agent folder and paste this code:
import asyncio
from mcp import Client
from mcp_server import mcp


async def main():
    async with Client(mcp) as client:
        tools = await client.list_tools()
        print(f"Connected to MCP server with {len(tools.tools)} tools:")
        for tool in tools.tools:
            print(f"  - {tool.name}: {tool.description[:60]}...")

        print("\n--- Testing get_product_info ---")
        result = await client.call_tool("get_product_info", {"product_id": "PROD-001"})
        print(f"Result: {result.structured_content}")

        print("\n--- Testing list_assessments ---")
        result = await client.call_tool("list_assessments", {})
        print(f"Result: {result.structured_content}")


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

What does this test script do?

  • Client(mcp) connects directly to the server object in memory. No network transport needed.
  • client.list_tools() asks the server to enumerate all registered tools. This is how an agent discovers what it can do.
  • client.call_tool("get_product_info", {...}) invokes a specific tool with arguments and returns the result.

Before you run this, what do you expect to see? Think about how many tools are registered and what product data is in your database.

  • Run the test script:
uv run python test_server.py

You should see output listing all 9 tools by name and description, followed by the product data for PROD-001 (CloudSync Pro) and the list of assessments.

Seeing an error about rules_engine?

If you see an error mentioning rules_engine, the test script is trying to call check_assessment_readiness directly. The test only calls get_product_info and list_assessments, so this error should not appear. Double-check your test_server.py matches the code above.

Why is my test_server.py failing?

✔️ Awesome, I've got everything!

Great. Double-check you have saved both mcp_server.py and test_server.py.

ⓧ I'd like to double check the full code

Your mcp_server.py should contain these sections in order. Scroll up and confirm each is present:

  • Imports (json, datetime, MCPServer, get_connection, init_db, seed_data).
  • Server initialization: mcp = MCPServer("ProductSecurity") plus init_db() and seed_data().
  • Tool 1: create_assessment.
  • Tool 2: get_assessment.
  • Tool 3: get_product_info.
  • Tool 4: list_assessments.
  • Tool 5: get_assessment_status.
  • Tool 6: get_evidence_status.
  • Tool 7: get_blockers.
  • Tool 8: update_assessment_status.
  • Tool 9: check_assessment_readiness (with from rules_engine import evaluate_readiness inside the function body).

Your test_server.py should import asyncio and Client, connect with async with Client(mcp) as client, list tools, and call get_product_info and list_assessments.

Your MCP server is live and tested. Any MCP-compatible client can now discover and call your nine Product Security tools. Next up, you will build the agent loop that connects Ollama's Llama 3.1 to these tools so the LLM can decide which ones to invoke.

Build the Agent Loop

Your MCP server exposes 9 tools, but nothing is orchestrating them yet. A user's natural language question needs a layer that decides which tools to call, executes them, and assembles a coherent answer.

That orchestration layer is the agent loop. It sends the user's question to Ollama (Llama 3.1), lets the LLM decide which MCP tools to invoke, executes those calls via the MCP client, feeds results back, and repeats until it produces a final text answer.

In this step, get ready to:
  • Convert MCP tool schemas to Ollama's expected format.
  • Build the agent loop with iterative tool calling.
  • Add an interactive CLI and test end-to-end.
Build the tool conversion layer

Ollama expects tools described in a specific JSON format. The MCP server already exposes tool schemas, but you need a function that translates them into the shape Ollama understands.

  • Create a new file called agent.py in your prodsec-agent folder.
  • Add the imports and system prompt by pasting this code:
import asyncio
import json
from mcp import Client
from ollama import chat, ChatResponse
from mcp_server import mcp


# System prompt instructs the LLM how to behave and when to use tools
SYSTEM_PROMPT = """You are a Product Security Operations Agent. You help security engineers and product teams manage security assessments efficiently.

You have access to tools that let you:
- Create and manage security assessments
- Look up product information and attributes
- Evaluate assessment readiness using deterministic business rules
- Check assessment status and identify blockers
- Update workflow status

IMPORTANT RULES:
1. When asked about assessment readiness, ALWAYS use the check_assessment_readiness tool. Never guess.
2. When reporting readiness, clearly list: the verdict, required services with reasons, and any missing items.
3. Use get_assessment_status to check progress on existing assessments.
4. Be concise but thorough. Always use tools rather than making assumptions about data."""

What does this code do?

The imports bring in the MCP Client for calling tools, the Ollama chat function for LLM interaction, and the mcp server object you built in the previous step.

The SYSTEM_PROMPT tells Llama 3.1 what role it plays and sets strict rules. Rule 1 is key: it tells the LLM to always use the readiness tool rather than guessing. You will see later why this instruction alone is not enough.

  • Below the SYSTEM_PROMPT, add the function that converts MCP tool schemas to Ollama's format:
async def get_tools_from_mcp(client: Client) -> list[dict]:
    """Get tools from MCP server and convert to Ollama tool-calling format."""
    tool_list = await client.list_tools()
    ollama_tools = []

    # Reshape each MCP tool into Ollama's expected structure
    for tool in tool_list.tools:
        ollama_tools.append({
            "type": "function",
            "function": {
                "name": tool.name,
                "description": tool.description,
                "parameters": tool.input_schema,
            },
        })

    return ollama_tools

What does this function do?

The MCP client's list_tools() returns tool metadata (name, description, JSON schema for parameters). Ollama expects tools in a specific format with type, function.name, function.description, and function.parameters.

This function bridges the two formats so your MCP tools are automatically discoverable by the LLM without any manual registration.

  • Save agent.py.
  • Verify the imports and function definition are valid by running:
uv run python -c "from agent import get_tools_from_mcp; print('Tool conversion layer OK')"

You should see Tool conversion layer OK printed with no errors.

Seeing an ImportError?

  • If you see an error about mcp or ollama, confirm your dependencies were installed by running uv pip list and checking both packages appear.
  • If you see an error about mcp_server, make sure you are running the command from inside the prodsec-agent folder where mcp_server.py lives.
  • help me debug this ImportError in agent.py
Implement the agent loop

The agent loop is the core orchestration logic. It sends the user's message plus available tools to Ollama, checks whether the LLM wants to call a tool, executes it via MCP, feeds the result back, and repeats.

  • In agent.py, below get_tools_from_mcp, add the first half of the agent loop function:
async def run_agent_loop(user_message: str) -> str:
    """Run the agent loop: connect to MCP, get tools, query Ollama, execute tool calls."""
    async with Client(mcp) as client:
        ollama_tools = await get_tools_from_mcp(client)

        # Build the conversation with system instructions and user query
        messages = [
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": user_message},
        ]

        max_iterations = 10
        iteration = 0

        while iteration < max_iterations:
            iteration += 1

            # Send messages and tools to Ollama for reasoning
            response: ChatResponse = chat(
                model="llama3.1:8b",
                messages=messages,
                tools=ollama_tools,
            )

            messages.append(response.message)

What is happening here?

The function opens an in-memory MCP client connection, fetches the tool list, and enters a loop. Each iteration calls Ollama's chat() with the full conversation history and the available tools.

The max_iterations = 10 guard prevents infinite loops if the LLM keeps requesting tools without reaching a conclusion.

  • Directly below the messages.append(response.message) line (still inside the while loop), add the tool execution logic:
            # If the LLM wants to call tools, execute each one
            if response.message.tool_calls:
                for tc in response.message.tool_calls:
                    tool_name = tc.function.name
                    tool_args = tc.function.arguments

                    print(f"  [Tool call: {tool_name}({json.dumps(tool_args)})]")

                    result = await client.call_tool(tool_name, tool_args)

                    # Extract text from the MCP result object
                    result_text = ""
                    if hasattr(result, "structured_content") and result.structured_content is not None:
                        result_text = json.dumps(result.structured_content)
                    elif hasattr(result, "content"):
                        result_text = "\n".join(
                            block.text for block in result.content if hasattr(block, "text")
                        )
                    else:
                        result_text = str(result)

                    # Feed the tool result back so the LLM can reason about it
                    messages.append({
                        "role": "tool",
                        "content": result_text,
                    })
            else:
                return response.message.content

        return "Agent reached maximum iterations without a final answer."

What does the tool execution do?

When Ollama's response contains tool_calls, the LLM is saying "I need data from these tools before I can answer." The loop executes each tool call via client.call_tool() and appends the result as a tool message.

The result extraction handles multiple MCP response formats: structured_content (parsed JSON) and content blocks (text segments). When there are no tool calls, the LLM has its final answer and the loop exits.

  • Save agent.py.
  • Verify the module still loads cleanly by running:
uv run python -c "import agent; print('Agent module loaded OK')"

You should see Agent module loaded OK with no syntax errors.

Seeing a SyntaxError?

  • Check that the if response.message.tool_calls: block is indented at the same level as response: ChatResponse = chat(...) inside the while loop (three levels of indentation: function, async with, while).
  • Make sure the final return "Agent reached maximum iterations..." line is inside the function but outside the while loop (same level as while iteration < max_iterations:).
  • help me fix this SyntaxError in my agent loop
Add the CLI interface and test

The agent loop function is async, so you need a synchronous wrapper and a CLI loop to interact with it from the terminal.

  • At the bottom of agent.py, below run_agent_loop, add the synchronous wrapper and interactive CLI:
def run(user_message: str) -> str:
    """Synchronous wrapper for the agent loop."""
    return asyncio.run(run_agent_loop(user_message))


if __name__ == "__main__":
    print("=" * 60)
    print("  Product Security Operations Agent")
    print("  Powered by Llama 3.1 + MCP")
    print("  Type 'quit' to exit")
    print("=" * 60)
    print()

    while True:
        try:
            user_input = input("You: ").strip()
            if not user_input:
                continue
            if user_input.lower() in ("quit", "exit", "q"):
                print("Goodbye!")
                break

            print()
            response = run(user_input)
            print(f"\nAgent: {response}\n")
        except KeyboardInterrupt:
            print("\nGoodbye!")
            break

What does the CLI do?

The run() wrapper lets you call the async agent loop from synchronous code. The __main__ block creates a read-eval-print loop: it takes your input, passes it to the agent, prints the response, and waits for the next question.

Tool call logs (printed by the agent loop) appear between your question and the final answer so you can see exactly which tools the LLM chose to invoke.

  • Save agent.py.

✔️ Awesome, I've got everything!

Great. Confirm you have saved agent.py with all four sections: imports + SYSTEM_PROMPT, get_tools_from_mcp, run_agent_loop, and the CLI block.

ⓧ I'd like to double check the full code

Scroll up and confirm your agent.py contains each of these sections in order:

  • Imports (asyncio, json, Client, chat, ChatResponse, mcp) at the top.
  • SYSTEM_PROMPT multi-line string with 4 rules.
  • get_tools_from_mcp function that loops through tool_list.tools and builds the Ollama format.
  • run_agent_loop function with async with Client(mcp), the while loop, chat() call, tool execution block, and max iteration guard.
  • run() wrapper calling asyncio.run().
  • if __name__ == "__main__" block with the interactive input loop.

Before you run the agent, what do you think will happen when you ask "What products are available?" Will the LLM answer from memory, or will it call a tool first?

  • Start the agent by running:
uv run python agent.py

You should see the banner appear:

============================================================
  Product Security Operations Agent
  Powered by Llama 3.1 + MCP
  Type 'quit' to exit
============================================================
  • Type the following question and press Enter:
What products are available?

You should see tool call logs appear (such as [Tool call: list_assessments(...)] and [Tool call: get_product_info(...)]), followed by a formatted answer listing CloudSync Pro and IoT Gateway Hub with their details.

Why does this prove the agent loop works?

The LLM did not have the product data in its training. It had to discover the MCP tools, decide to call list_assessments or get_product_info, receive the results, and compose a human-readable answer. That entire flow happened automatically through your agent loop.

Now try a readiness question. Before you type it, predict: will the agent give a precise, structured readiness verdict?

  • Type this question into the agent:
Is product PROD-001 ready for a security assessment?

The agent gives an answer, but notice what happens. It might retrieve some data and attempt to reason about readiness on its own. The answer is vague or inconsistent because the LLM is guessing what "ready" means rather than applying deterministic business rules.

Why is the readiness answer unreliable?

The system prompt tells the LLM to use the check_assessment_readiness tool, but that tool currently imports from a rules_engine module that does not exist yet. Without the rules engine, the LLM either skips the tool (and guesses) or hits an import error.

Security scoping decisions must be deterministic and auditable. An LLM guessing "probably ready" is not acceptable for a real Product Security workflow. The next step fixes this.

  • Type quit to exit the agent.

Agent hangs or gives no response?

  • Make sure Ollama is running in the background. Open a separate terminal and run ollama list to confirm llama3.1:8b is available.
  • If the first response takes a long time, that is normal. Llama 3.1 8B needs to load into memory on the first call, which can take 10-30 seconds depending on your hardware.
  • If you see a connection error mentioning port 11434, Ollama's server is not running. Start it with ollama serve in a separate terminal.
  • help me get the agent loop working with Ollama

Your agent loop is working. It can discover tools, call them, and compose answers from real data. Next up, you will build the rules engine that gives the readiness tool deterministic, auditable logic instead of letting the LLM guess.

Add the Rules Engine

Your agent can answer questions about products and assessments. But try asking it whether a product is ready for a security assessment, and you will get inconsistent results.

Security scoping decisions must be deterministic and auditable. A different engineer asking the same question on a different day must get the same answer. In this step, you will build a YAML-based rules engine that provides structured business logic the LLM invokes but does not control.

In this step, get ready to:
  • Create a YAML rules configuration that maps product attributes to required security services.
  • Build a rules engine module that evaluates readiness deterministically.
  • Test the full agent flow with rules-based assessment readiness.
See the problem firsthand

The check_assessment_readiness tool in mcp_server.py already tries to import rules_engine. That module does not exist yet.

  • Start the agent by running:
uv run python agent.py
  • When you see the You: prompt, type this question:
Is product PROD-002 ready for a security assessment?

Before you press Enter, what do you think will happen? The tool tries to import a module that does not exist.

You will see the agent either report an error from the tool call or attempt to guess an answer from other data it retrieves. Either way, the answer is unreliable. Type quit to exit the agent.

Why is guessing a problem?

If the LLM guesses which security services are required, different runs produce different answers. An auditor cannot trace the decision back to a documented rule. The fix is a deterministic rules engine that the LLM calls as a tool but cannot override.

Create the YAML rules configuration

The rules configuration maps product attributes (like has_web_interface or has_embedded_firmware) to the security services that must be performed. Each rule fires independently based on a single attribute.

  • Open the config folder in the VS Code Explorer sidebar.
  • Create a new file called scoping_rules.yaml inside the config folder.
  • Add the first three scoping rules by pasting this content:
scoping_rules:
  - id: RULE-WEB-001
    description: "Web applications require SAST, DAST, and Penetration Testing"
    condition:
      field: has_web_interface
      equals: 1
    services:
      - SAST
      - DAST
      - Penetration Testing

  - id: RULE-API-001
    description: "REST APIs require API Security Testing"
    condition:
      field: has_rest_api
      equals: 1
    services:
      - API Security Testing

  - id: RULE-SCA-001
    description: "Third-party components require SCA and SBOM Review"
    condition:
      field: has_third_party_components
      equals: 1
    services:
      - SCA
      - SBOM Review

What does this structure do?

Each rule has a condition that checks one product attribute against a value. When the condition matches, the listed services are required for that product's assessment.

The rule ID makes every decision traceable back to a documented policy.

  • Below the SCA rule you just pasted, add the firmware and AI rules:
  - id: RULE-FW-001
    description: "Embedded firmware requires Firmware Security Testing"
    condition:
      field: has_embedded_firmware
      equals: 1
    services:
      - Firmware Security Testing

  - id: RULE-AI-001
    description: "AI/ML functionality requires AI Security Assessment"
    condition:
      field: has_ai_ml
      equals: 1
    services:
      - AI Security Assessment
  • Below the scoping rules array, add the base evidence and evidence rules sections:
base_evidence:
  - architecture_diagram
  - test_credentials

evidence_rules:
  - condition:
      field: has_embedded_firmware
      equals: 1
    evidence:
      - firmware_image
      - threat_model

  - condition:
      field: has_third_party_components
      equals: 1
    evidence:
      - sbom

  - condition:
      field: has_web_interface
      equals: 1
    evidence:
      - threat_model

  - condition:
      field: has_ai_ml
      equals: 1
    evidence:
      - threat_model

What are evidence rules?

The base_evidence list defines items every product must provide. The evidence_rules add extra requirements based on attributes.

A firmware product needs a firmware_image and threat_model in addition to the base set.

  • Save config/scoping_rules.yaml.
  • Confirm the file exists by running:
ls config/

You should see scoping_rules.yaml listed in the output.

Don't see the file?

Make sure you created the file inside the config directory, not in the project root. The path should be config/scoping_rules.yaml.

Help me find my missing config file.

✔️ Awesome, I've got everything!

Great. Double check that you have saved the file before moving on.

ⓧ I'd like to double check the full code

Your config/scoping_rules.yaml should contain three sections in order: scoping_rules (5 rules from RULE-WEB-001 through RULE-AI-001), base_evidence (2 items), and evidence_rules (4 conditional rules). Scroll up and confirm each chunk is present and in order.

Build the rules engine module

The rules engine loads the YAML configuration, evaluates a product's attributes against each rule, and produces a structured readiness verdict. It is pure Python with no LLM involvement.

  • Create a new file called rules_engine.py in your project root (the prodsec-agent folder).
  • Add the imports, path constant, and the service-determination function:
import yaml
from pathlib import Path

RULES_PATH = Path(__file__).parent / "config" / "scoping_rules.yaml"


def load_rules() -> dict:
    """Load scoping rules from YAML configuration."""
    with open(RULES_PATH, "r") as f:
        return yaml.safe_load(f)


def determine_required_services(product: dict) -> list[dict]:
    """Determine which security services are required based on product attributes and rules."""
    rules = load_rules()
    required_services = []
    for rule in rules["scoping_rules"]:
        condition_field = rule["condition"]["field"]
        condition_value = rule["condition"]["equals"]
        product_value = product.get(condition_field)
        if product_value == condition_value:
            for service in rule["services"]:
                required_services.append({"service": service, "triggered_by": f"{condition_field} = {condition_value}", "rule_id": rule["id"]})
    # Deduplicate while preserving order
    seen = set()
    unique_services = []
    for svc in required_services:
        if svc["service"] not in seen:
            seen.add(svc["service"])
            unique_services.append(svc)
    return unique_services

What does determine_required_services do?

It loops through every scoping rule, checks whether the product's attribute matches the rule's condition, and collects the triggered services. Each result includes which attribute triggered it and the rule ID for audit traceability.

  • Below determine_required_services, add the evidence determination and the top-level readiness evaluator:
def determine_required_evidence(product: dict) -> list[str]:
    """Determine which evidence items are required based on product attributes."""
    rules = load_rules()
    required_evidence = set(rules.get("base_evidence", []))
    for rule in rules.get("evidence_rules", []):
        condition_field = rule["condition"]["field"]
        condition_value = rule["condition"]["equals"]
        if product.get(condition_field) == condition_value:
            for item in rule["evidence"]:
                required_evidence.add(item)
    return sorted(required_evidence)


def evaluate_readiness(product: dict, evidence: list[dict]) -> dict:
    """Evaluate whether a product is ready to begin a security assessment."""
    required_services = determine_required_services(product)
    required_evidence = determine_required_evidence(product)
    provided_evidence_types = {e["evidence_type"] for e in evidence if e.get("provided")}
    missing_evidence = [e for e in required_evidence if e not in provided_evidence_types]
    is_ready = len(missing_evidence) == 0
    return {
        "product_id": product["id"],
        "product_name": product["name"],
        "readiness": "READY" if is_ready else "NOT READY",
        "required_services": [{"service": s["service"], "reason": s["triggered_by"], "rule": s["rule_id"]} for s in required_services],
        "required_evidence": required_evidence,
        "provided_evidence": sorted(provided_evidence_types),
        "missing_evidence": missing_evidence,
        "blockers": [f"Missing: {item}" for item in missing_evidence],
    }

What does evaluate_readiness return?

It combines service determination and evidence checking into a single structured verdict. The output includes the readiness status (READY or NOT READY), every required service with the rule that triggered it, and a list of missing evidence items as blockers.

This is the function that check_assessment_readiness in your MCP server already imports.

  • Save rules_engine.py.
  • Verify the module loads correctly by running:
uv run python -c "from rules_engine import evaluate_readiness; print('Rules engine OK')"

You should see Rules engine OK printed with no errors.

Seeing a FileNotFoundError?

This means rules_engine.py cannot find config/scoping_rules.yaml. Make sure you are running the command from inside the prodsec-agent directory, and that the YAML file is saved at config/scoping_rules.yaml (not config/scoping_rules.yml).

Help me fix this FileNotFoundError.

✔️ Awesome, I've got everything!

Great. Confirm the file is saved before continuing.

ⓧ I'd like to double check the full code

Your rules_engine.py should contain four functions in this order: load_rules, determine_required_services, determine_required_evidence, and evaluate_readiness. Scroll up and confirm both chunks are present in order.

Update the MCP tool description and test end-to-end

The MCP Python SDK exposes your function's docstring as the tool description that the LLM sees. A richer description helps the agent decide when to call the tool.

  • In mcp_server.py, find the check_assessment_readiness function. Locate its docstring:
def check_assessment_readiness(product_id: str) -> str:
    """Check if a product has all required evidence and information to begin a security assessment."""
  • Replace that single-line docstring with this expanded version:
def check_assessment_readiness(product_id: str) -> str:
    """Check if a product has all required evidence and information to begin a security assessment.
    Uses the configurable rules engine to determine required services and evidence based on product attributes.

    Args:
        product_id: The product ID to evaluate (e.g., PROD-001 or PROD-002)

    Returns:
        JSON with readiness verdict (READY/NOT READY), required services with justification,
        required evidence, provided evidence, missing evidence, and blockers
    """

Why expand the docstring?

When the agent calls list_tools(), the MCP SDK returns each tool's name and description. The description comes directly from the Python docstring.

By adding Args and Returns detail, the LLM gets clearer guidance on when and how to use this tool.

Expand your other tool docstrings too

For best results, add Args and Returns sections to every tool function in mcp_server.py. Richer descriptions help the LLM choose the right tool on the first try.

  • Save mcp_server.py.
  • Start the agent by running:
uv run python agent.py

Before you type your question, predict: will the agent return a READY or NOT READY verdict for PROD-002 (the IoT Gateway Hub that has embedded firmware but no existing assessment)?

  • At the You: prompt, type:
Is product PROD-002 ready for a security assessment?

You will see tool call logs as the agent invokes check_assessment_readiness. The rules engine evaluates PROD-002's attributes, determines it requires Firmware Security Testing (among other services), and reports missing evidence including firmware_image, threat_model, and test_credentials. The final verdict is NOT READY.

What just happened?

The LLM decided to call check_assessment_readiness based on your question. The MCP server invoked evaluate_readiness from the rules engine, which loaded the YAML rules, matched PROD-002's attributes, and returned a deterministic verdict.

This is the core pattern: the LLM orchestrates (decides when to check readiness), but the rules engine decides (what services are required and whether evidence is complete). Every answer is traceable back to a rule ID.

  • Type quit to exit the agent.

Agent not calling check_assessment_readiness?

If the agent uses other tools (like get_product_info) instead of check_assessment_readiness, check that your system prompt in agent.py includes the rule about always using the readiness tool. Also confirm that the expanded docstring is saved in mcp_server.py.

Help me debug why the agent isn't using the right tool.

Your rules engine is live. The agent now gives deterministic, auditable readiness verdicts backed by documented rules instead of LLM guesses. Next up, you will extend this architecture with a second MCP server in the Secret Mission.

Secret mission

Multi-Server Agent with SBOM/SCA

Real enterprise security platforms don't run on a single server. Extend your agent to orchestrate across TWO MCP servers simultaneously, proving the modular architecture pattern that scales to production.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

  • Project directory prodsec-agent on your Desktop.
  • SQLite database file at data/prodsec.db inside the project directory.
  • Ollama model llama3.1:8b stored locally (approximately 4.7GB disk space).

Keep everything running

No action needed. Choose this if you are still actively building or want to experiment further with the agent.

  • Your prodsec-agent directory stays on your Desktop ready to use.
  • The llama3.1:8b model remains available for any Ollama project.
  • Ollama itself stays installed on your machine.

Pause - I'll come back to this later

Stop Ollama from running in the background to free up memory. Your files and model stay on disk.

  • Close any running agent scripts by pressing Ctrl+C in the terminal.
  • Quit Ollama from your system tray (right-click the Ollama icon and select Quit).
  • When you return, start Ollama again and run uv run python agent.py from your prodsec-agent folder.

Delete - I don't want to use this again

Remove all project resources and free up disk space completely.

  • Remove the Ollama model by running this command in your terminal:
ollama rm llama3.1:8b

This frees approximately 4.7GB of disk space.

  • Delete the project directory. On macOS/Linux, run:
cd ~/Desktop
rm -rf prodsec-agent
  • On Windows, run:
cd ~/Desktop
Remove-Item -Recurse -Force prodsec-agent

This removes all Python files, the SQLite database, the YAML rules configuration, and the virtual environment.

Well Done!

Well Done!

Nice work. You just built a fully local AI-powered Product Security Operations agent that takes natural language queries, reasons about which tools to call, and returns auditable security assessment reports.

Here's what you accomplished:

  • Built an MCP server exposing 9 Product Security operations tools using the MCP Python SDK v2, with automatic tool discovery via the @mcp.tool() decorator.
  • Built an agent loop connecting Ollama (Llama 3.1) to MCP tools via tool calling, where the LLM autonomously decides which tools to invoke and feeds results back until it reaches a final answer.
  • Implemented a YAML-based rules engine that provides deterministic, auditable security assessment scoping decisions separate from the LLM. The AI orchestrates, the rules decide.
  • Secret Mission: Orchestrated across multiple MCP servers simultaneously, proving the modular enterprise architecture pattern where independent domain servers expose tools to a single agent.

Ready to quiz yourself?