Extending the MCP Server

This guide explains how to extend the Prowler MCP Server with new tools and features.

New to Prowler MCP Server? Start with the user documentation:

• Overview - Key capabilities, use cases, and deployment options
• Installation - Install locally or use the managed server
• Configuration - Configure Claude Desktop, Cursor, and other MCP hosts
• Tools Reference - Complete list of all available tools

Introduction

The Prowler MCP Server brings the entire Prowler ecosystem to AI assistants through the Model Context Protocol (MCP). It enables seamless integration with AI tools like Claude Desktop, Cursor, and other MCP clients.

The server follows a modular architecture with three independent sub-servers:

Sub-Server	Tool Prefix	Auth Required	Description
Prowler	prowler_	Yes	Full access to Prowler Cloud, Prowler Private Cloud, and Prowler Local Server features
Prowler Hub	prowler_hub_	No	Security checks catalog with over 2,000 checks, fixers, and 70+ compliance frameworks
Prowler Documentation	prowler_docs_	No	Full-text search and retrieval of official documentation

The core Prowler sub-server is served under the prowler_ tool prefix, while its source lives in the prowler_app/ module for historical reasons. Tool names use the prefix; import paths use the module.

For a complete list of tools and their descriptions, see the Tools Reference.

Architecture Overview

The MCP Server architecture is illustrated in the Overview documentation. AI assistants connect through the MCP protocol to access Prowler’s three main components.

Server Structure

The main server orchestrates three sub-servers with prefixed namespacing:

mcp_server/prowler_mcp_server/
├── server.py                 # Main orchestrator
├── main.py                   # CLI entry point
├── prowler_hub/
├── prowler_app/
│   ├── tools/                # Tool implementations
│   ├── models/               # Pydantic models
│   └── utils/                # API client, auth, loader
└── prowler_documentation/

Tool Registration Patterns

The MCP Server uses two patterns for tool registration:

1. Direct Decorators (Prowler Hub/Docs): Tools are registered using @mcp.tool() decorators
2. Auto-Discovery (prowler_app): All public methods of BaseTool subclasses are auto-registered

Adding Tools to the prowler_app Sub-Server

Step 1: Create the Tool Class

Create a new file or add to an existing file in prowler_app/tools/:

# prowler_app/tools/new_feature.py
from typing import Any

from pydantic import Field

from prowler_mcp_server.prowler_app.models.new_feature import (
    FeatureListResponse,
    DetailedFeature,
)
from prowler_mcp_server.prowler_app.tools.base import BaseTool


class NewFeatureTools(BaseTool):
    """Tools for managing new features."""

    async def list_features(
        self,
        status: str | None = Field(
            default=None,
            description="Filter by status (active, inactive, pending)"
        ),
        page_size: int = Field(
            default=50,
            description="Number of results per page (1-100)"
        ),
    ) -> dict[str, Any]:
        """List all features with optional filtering.

        Returns a lightweight list of features optimized for LLM consumption.
        Use get_feature for complete information about a specific feature.
        """
        # Validate parameters
        self.api_client.validate_page_size(page_size)

        # Build query parameters
        params: dict[str, Any] = {"page[size]": page_size}
        if status:
            params["filter[status]"] = status

        # Make API request
        clean_params = self.api_client.build_filter_params(params)
        response = await self.api_client.get("/api/v1/features", params=clean_params)

        # Transform to LLM-friendly format
        return FeatureListResponse.from_api_response(response).model_dump()

    async def get_feature(
        self,
        feature_id: str = Field(description="The UUID of the feature"),
    ) -> dict[str, Any]:
        """Get detailed information about a specific feature.

        Returns complete feature details including configuration and metadata.
        """
        response = await self.api_client.get(f"/api/v1/features/{feature_id}")
        return DetailedFeature.from_api_response(response["data"]).model_dump()

There is no try/except here on purpose. A failed request raises, and Error Handling explains what turns that raise into a message the agent can act on.

Step 2: Create the Models

Create corresponding models in prowler_app/models/:

# prowler_app/models/new_feature.py
from typing import Any

from pydantic import Field

from prowler_mcp_server.prowler_app.models.base import MinimalSerializerMixin


class SimplifiedFeature(MinimalSerializerMixin):
    """Lightweight feature for list operations."""

    id: str = Field(description="Unique feature identifier")
    name: str = Field(description="Feature name")
    status: str = Field(description="Current status")

    @classmethod
    def from_api_response(cls, data: dict[str, Any]) -> "SimplifiedFeature":
        """Transform API response to simplified format."""
        attributes = data.get("attributes", {})
        return cls(
            id=data["id"],
            name=attributes["name"],
            status=attributes["status"],
        )


class DetailedFeature(SimplifiedFeature):
    """Extended feature with complete details."""

    description: str | None = Field(default=None, description="Feature description")
    configuration: dict[str, Any] | None = Field(default=None, description="Configuration")
    created_at: str = Field(description="Creation timestamp")
    updated_at: str = Field(description="Last update timestamp")

    @classmethod
    def from_api_response(cls, data: dict[str, Any]) -> "DetailedFeature":
        """Transform API response to detailed format."""
        attributes = data.get("attributes", {})
        return cls(
            id=data["id"],
            name=attributes["name"],
            status=attributes["status"],
            description=attributes.get("description"),
            configuration=attributes.get("configuration"),
            created_at=attributes["created_at"],
            updated_at=attributes["updated_at"],
        )


class FeatureListResponse(MinimalSerializerMixin):
    """Response wrapper for feature list operations."""

    count: int = Field(description="Total number of features")
    features: list[SimplifiedFeature] = Field(description="List of features")

    @classmethod
    def from_api_response(cls, response: dict[str, Any]) -> "FeatureListResponse":
        """Transform API response to list format."""
        data = response.get("data", [])
        features = [SimplifiedFeature.from_api_response(item) for item in data]
        return cls(count=len(features), features=features)

Step 3: Verify Auto-Discovery

No manual registration is needed. The tool_loader.py automatically discovers and registers all BaseTool subclasses. Verify your tool is loaded by checking the server logs:

INFO - Auto-registered 2 tools from NewFeatureTools
INFO - Loaded and registered: NewFeatureTools

Adding Tools to Prowler Hub/Docs

For Prowler Hub or Documentation tools, use the @mcp.tool() decorator directly:

# prowler_hub/server.py
from fastmcp import FastMCP

hub_mcp_server = FastMCP("prowler-hub")

@hub_mcp_server.tool()
async def get_new_artifact(
    artifact_id: str,
) -> dict:
    """Fetch a specific artifact from Prowler Hub.

    Args:
        artifact_id: The unique identifier of the artifact

    Returns:
        Dictionary containing artifact details
    """
    response = prowler_hub_client.get(f"/artifact/{artifact_id}")
    response.raise_for_status()
    return response.json()

Model Design Patterns

MinimalSerializerMixin

All models should use MinimalSerializerMixin to optimize responses for LLM consumption:

from prowler_mcp_server.prowler_app.models.base import MinimalSerializerMixin

class MyModel(MinimalSerializerMixin):
    """Model that excludes empty values from serialization."""
    required_field: str
    optional_field: str | None = None  # Excluded if None
    empty_list: list = []              # Excluded if empty

This mixin automatically excludes:

• None values
• Empty strings
• Empty lists
• Empty dictionaries

Two-Tier Model Pattern

Use two-tier models for efficient responses:

• Simplified: Lightweight models for list operations
• Detailed: Extended models for single-item retrieval

class SimplifiedItem(MinimalSerializerMixin):
    """Use for list operations - minimal fields."""
    id: str
    name: str
    status: str

class DetailedItem(SimplifiedItem):
    """Use for get operations - extends simplified with details."""
    description: str | None = None
    configuration: dict | None = None
    created_at: str
    updated_at: str

Factory Method Pattern

Always implement from_api_response() for API transformation:

@classmethod
def from_api_response(cls, data: dict[str, Any]) -> "MyModel":
    """Transform API response to model.

    This method handles the JSON:API format used by Prowler API,
    extracting attributes and relationships as needed.
    """
    attributes = data.get("attributes", {})
    return cls(
        id=data["id"],
        name=attributes["name"],
        # ... map other fields
    )

API Client Usage

The ProwlerAPIClient is a singleton that handles authentication and HTTP requests:

class MyTools(BaseTool):
    async def my_tool(self) -> dict:
        # GET request
        response = await self.api_client.get("/api/v1/endpoint", params={"key": "value"})

        # POST request
        response = await self.api_client.post(
            "/api/v1/endpoint",
            json_data={"data": {"type": "items", "attributes": {...}}}
        )

        # PATCH request
        response = await self.api_client.patch(
            f"/api/v1/endpoint/{id}",
            json_data={"data": {"attributes": {...}}}
        )

        # DELETE request
        response = await self.api_client.delete(f"/api/v1/endpoint/{id}")

Helper Methods

The API client provides useful helper methods:

# Validate page size (1-1000)
self.api_client.validate_page_size(page_size)

# Normalize date range with max days limit
date_range = self.api_client.normalize_date_range(date_from, date_to, max_days=2)

# Build filter parameters (handles type conversion)
clean_params = self.api_client.build_filter_params({
    "filter[status]": "active",
    "filter[severity__in]": ["high", "critical"],  # Converts to comma-separated
    "filter[muted]": True,  # Converts to "true"
})

# Poll async task until completion
result = await self.api_client.poll_task_until_complete(
    task_id=task_id,
    timeout=60,
    poll_interval=1.0
)

Best Practices

Tool Docstrings

Tool docstrings become the description that is going to be read by the LLM. Provide clear usage instructions and common workflows:

async def search_items(self, status: str = Field(...)) -> dict:
    """Search items with advanced filtering.

    Returns a lightweight list optimized for LLM consumption.
    Use get_item for complete details about a specific item.

    Common workflows:
    - Find critical items: status="critical"
    - Find recent items: Use date_from parameter
    """

Error Handling

Raise, never return. A returned {"error": ...} dict is reported to the client as isError: false — a successful tool call whose payload happens to mention a failure. Clients and models read that as success. A raised exception becomes a spec-correct tool execution error instead.

The common case therefore needs no handler at all:

async def get_item(self, item_id: str) -> dict:
    response = await self.api_client.get(f"/api/v1/items/{item_id}")
    return DetailedItem.from_api_response(response["data"]).model_dump()

prowler_mcp_server/lib/errors.py classifies the failures every tool shares — a rejected credential, a missing permission, a rate limit, an outage, an unreachable API, a bad argument — and gives each one a message that says what went wrong and what to do about it. Anything it does not recognise is masked, because mask_error_details=True is set on every sub-server and upstream response bodies must never be replayed into a model’s context.

Three ways to raise, in the order to reach for them:

from fastmcp.exceptions import ToolError

from prowler_mcp_server.lib.errors import InvalidArgument

# 1. An argument this server rejected before any request went out. The message
#    is repeated to the agent verbatim, so write it for one to read.
if not 1 <= page_size <= 1000:
    raise InvalidArgument("page_size must be between 1 and 1000.")

# 2. A request the API answered or never answered: let it propagate untouched.
#    `ProwlerAPIError` and `ProwlerAPIUnreachable` are what the classifier keys
#    on, and the second one is what stops a retry from duplicating a write.
response = await self.api_client.get(f"/api/v1/items/{item_id}")
data = response["data"]

# 3. A sentence the classifier cannot know -- a resource name, a precondition,
#    the next tool to call. NOTE the absent `from` clause: it is what marks the
#    message as already final. With `from e` the classifier would replace it.
if not data:
    raise ToolError(
        f"No item with the ID {item_id!r} exists. Use prowler_list_items to "
        "find a valid one."
    )

The one thing that still returns rather than raises is a write whose outcome is genuinely unknown. prowler_send_findings_to_jira is the worked example: work items are created one at a time and Prowler cannot delete them, so a dispatch that stopped halfway answers with a result object carrying safe_to_retry: false. “This may have been applied” is a fact about the world, not an error, and squashing it into one loses the only thing that stops a retry from duplicating the write.

Parameter Descriptions

Use Pydantic Field() with clear descriptions. This also helps LLMs understand the purpose of each parameter, so be as descriptive as possible:

async def list_items(
    self,
    severity: list[str] = Field(
        default=[],
        description="Filter by severity levels (critical, high, medium, low)"
    ),
    status: str | None = Field(
        default=None,
        description="Filter by status (PASS, FAIL, MANUAL)"
    ),
    page_size: int = Field(
        default=50,
        description="Results per page"
    ),
) -> dict:

Development Commands

# Navigate to MCP server directory
cd mcp_server

# Run in STDIO mode (default)
uv run prowler-mcp

# Run in HTTP mode
uv run prowler-mcp --transport http --host 0.0.0.0 --port 8000

# Run with environment variables
PROWLER_API_KEY="pk_xxx" uv run prowler-mcp

For complete installation and deployment options, see:

• Installation Guide - Development setup instructions
• Configuration Guide - MCP client configuration

For development I recommend to use the Model Context Protocol Inspector as MCP client to test and debug your tools.

Testing

Tests live in mcp_server/tests/, mirroring the source tree, and use the test_*.py prefix (the same convention as the API, not the SDK’s *_test.py suffix).

From mcp_server/:

cd mcp_server

uv run pytest                              # Whole suite
uv run pytest tests/prowler_app/models     # One area
uv run pytest --cov=./prowler_mcp_server   # With coverage

From the repository root:

make test-mcp   # Runs the MCP suite exactly as CI does

Async tests need no marker — asyncio_mode is set to auto.

Reading the Coverage Numbers

Coverage here has a high floor that means nothing. coverage.py measures statements, and in a Pydantic model module nearly every statement is a class-body field declaration that runs at import time. prowler_app/server.py imports every tool module — and therefore every model module — when it is first imported, so all of those declarations execute and count as covered before a single test runs.

Importing the package and executing no tests at all already reports 36% overall, with individual model modules between 54% and 84%. A model module sitting at ~68% with no tests written for it has none of its behaviour covered: the covered lines are its imports, class statements and Field(...) declarations, and the missing ranges are its from_api_response() bodies.

Judge a module against that import-only floor, not against zero, and do not set a Codecov target from the raw total.

Shared Fixtures

All fixtures live in mcp_server/tests/conftest.py. Three are autouse and apply to every test: the environment is pinned to deterministic values, real socket connections are blocked, and the API client singleton registry is snapshotted and restored.

Fixture	What it gives you
mock_api_client	The API client singleton with its transport mocked. The workhorse.
mock_router	Route registry and request recorder
mcp_root_server	The mounted root server, for in-memory client tests
health_client	Starlette TestClient for the /health route
http_request_headers	Injects request headers for HTTP-transport auth tests
hub_router / docs_router	Mock the Hub and Docs sub-servers’ sync HTTP clients
api_client / isolated_api_client	The live singleton / a freshly-constructed one

Helpers live in mcp_server/tests/helpers/: JSON:API document builders (jsonapi.py), the MockRouter (http.py), tool-contract assertions (assertions.py) and fake credentials (tokens.py).

Writing a Tool Test

Drive tools through an in-memory MCP client, and open the client inside the test — FastMCP warns that holding a client in a fixture causes event-loop problems.

from fastmcp import Client

from tests.helpers.jsonapi import jsonapi_collection, jsonapi_resource

FINDING_ATTRIBUTES = {
    "uid": "prowler-aws-s3_bucket_public_access-123456789012-us-east-1-my-bucket",
    "status": "FAIL",
    "severity": "high",
    "status_extended": "S3 bucket my-bucket is publicly accessible.",
    "delta": "new",
    "muted": False,
    "muted_reason": None,
    "check_metadata": {"checkid": "s3_bucket_public_access"},
}


async def test_search_without_dates_queries_the_latest_scan_endpoint(
    mcp_root_server, mock_api_client, mock_router
):
    """With no date range the tool targets the cheaper `/findings/latest`."""
    mock_router.add(
        "GET",
        "/api/v1/findings/latest",
        json=jsonapi_collection(
            [jsonapi_resource("findings", "f1", FINDING_ATTRIBUTES)]
        ),
    )

    async with Client(mcp_root_server) as client:
        result = await client.call_tool("prowler_search_security_findings", {})

    assert result.data["findings"][0]["check_id"] == "s3_bucket_public_access"
    assert mock_router.paths() == ["GET /api/v1/findings/latest"]

The exemplar suite covers findings end to end — tests/prowler_app/models/test_findings.py and tests/prowler_app/tools/test_findings.py. It is deliberately one feature across both layers rather than a scattering of unrelated samples, and findings is the feature that exercises the whole foundation: two-tier models, nested sub-models, both relationship shapes, endpoint switching on a date range, list-to-CSV filter encoding, and a tool that returns prose instead of a model.

Note the two files share a name. That is why __init__.py is required in every tests/ subdirectory here — without it they would collide on import.

Tool parameters are declared with pydantic Field(default=...), and only FastMCP’s tool wrapper resolves those defaults. Calling a tool method directly with an argument omitted leaves it as a raw FieldInfo object, which is truthy — so a filter such as if email: silently builds a query out of the FieldInfo repr. Call tools through the client, or pass every argument explicitly.

Why the API Key Is Pinned, Not Stripped

prowler_app/server.py builds every tool at import time. Constructing a tool reaches ProwlerAppAuth, which raises when PROWLER_API_KEY is missing, and load_all_tools swallows that error per tool class. The result is that the whole prowler_* namespace registers zero tools while the server still logs “Successfully mounted Prowler tools server”.

The suite therefore pins a fake key in [tool.pytest_env], which is applied before any test module is imported, and tests/test_server.py asserts each namespace is non-empty so this failure can never return silently.

ProwlerAppAuth resolves PROWLER_MCP_TRANSPORT_MODE and API_BASE_URL in its default arguments, which Python evaluates once at module import. monkeypatch.setenv cannot change them — pass mode= and base_url= explicitly in auth tests.

For the full set of rules and templates, see the prowler-test-mcp skill and the official FastMCP testing guide.

Related Documentation

MCP Server Overview

Key capabilities, use cases, and deployment options

Tools Reference

Complete reference of all available tools

Prowler Hub

Security checks and compliance frameworks catalog

Lighthouse AI

AI-powered security analyst

Additional Resources

• MCP Protocol Specification - Model Context Protocol details
• Prowler API Documentation - API reference
• Prowler Hub API - Hub API reference
• GitHub Repository - Source code