prowler/docs/developer-guide/mcp-server.mdx

---
title: 'Extending the MCP Server'
---

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

<Info>
**New to Prowler MCP Server?** Start with the user documentation:
- [Overview](/getting-started/products/prowler-mcp) - Key capabilities, use cases, and deployment options
- [Installation](/getting-started/installation/prowler-mcp) - Install locally or use the managed server
- [Configuration](/getting-started/basic-usage/prowler-mcp) - Configure Claude Desktop, Cursor, and other MCP hosts
- [Tools Reference](/getting-started/basic-usage/prowler-mcp-tools) - Complete list of all available tools
</Info>

## Introduction

The Prowler MCP Server brings the entire Prowler ecosystem to AI assistants through the [Model Context Protocol (MCP)](https://modelcontextprotocol.io). 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 |

<Note>
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.
</Note>

<Note>
For a complete list of tools and their descriptions, see the [Tools Reference](/getting-started/basic-usage/prowler-mcp-tools).
</Note>

## Architecture Overview

The MCP Server architecture is illustrated in the [Overview documentation](/getting-started/products/prowler-mcp#mcp-server-architecture). 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
├── lib/
│   ├── server.py             # ProwlerMCP, the base class of every sub-server
│   ├── errors.py             # Exception types and the one error renderer
│   ├── logger.py
│   └── analytics.py
├── 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

Both funnel through `ProwlerMCP.tool` (`lib/server.py`), which is what applies the error contract to every tool no matter how it was registered. Build sub-servers with `ProwlerMCP`, never `FastMCP` directly.

## 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/`:

```python
# 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.
        """
        # No try/except: a failure here raises, and the tool wrapper turns it into a
        # ToolError the client sees as `isError: true`. See "Error Handling" below.
        response = await self.api_client.get(f"/api/v1/features/{feature_id}")
        return DetailedFeature.from_api_response(response["data"]).model_dump()
```

### Step 2: Create the Models

Create corresponding models in `prowler_app/models/`:

```python
# 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:

```python
# 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:

```python
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

```python
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:

```python
@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:

```python
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:

```python
# 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:

```python
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

Let failures raise. Every sub-server is a `ProwlerMCP` (`prowler_mcp_server/lib/server.py`),
whose `tool()` wraps whatever it registers in `tool_errors`, turning any exception into a
`ToolError`. The client sees `isError: true` and a message it can act on.

That wrapping is not something you apply — the two registration styles (the `@mcp.tool()`
decorators, and the direct `mcp.tool(fn)` call `BaseTool` uses) both funnel through
`ProwlerMCP.tool`. Build sub-servers with `ProwlerMCP`, never `FastMCP` directly: masking
is on everywhere, so a tool that escaped the funnel would answer `Error calling tool 'x'`
with no detail at all.

Never `return {"error": ...}`: a returned payload is `isError: false` at the MCP protocol
level, so the client is told the call succeeded and only finds out otherwise if it happens
to inspect the right key.

```python
async def get_item(self, item_id: str) -> dict:
    """A rejected request, a timeout and a malformed payload all raise from here.

    Each is rendered with the API's own words plus what it implies about retrying.
    """
    response = await self.api_client.get(f"/api/v1/items/{item_id}")
    return DetailedItem.from_api_response(response["data"]).model_dump()
```

Raise `ToolError` whenever the message is one you wrote for the caller. Its text reaches
the client verbatim, so anything they need in order to recover has to be *in* the message
— an error carries nothing else:

```python
from fastmcp.exceptions import ToolError

if not data:
    raise ToolError(
        f"Item '{item_id}' was not found. Use prowler_list_items to find valid IDs."
    )
```

**Do not raise `ValueError` from a tool.** The two are not interchangeable: anything that
is not a `ToolError` is described as a bug in this server. That is right for a model
factory rejecting an API payload or a pydantic `ValidationError`, and wrong for a
refusal — so the exception type is what carries the distinction:

```text
Date range cannot exceed 2 days. Requested range: 2025-01-01 to 2025-01-10 (10 days)

The Prowler MCP Server hit an unexpected ValueError: Missing pagination metadata in API
response. This is a bug in the server, not something you can fix by changing the
arguments.
```

If you surface an exception yourself — into a `ToolError` you build, or into a field of a
structured result — pass it through `render_tool_error(e)` rather than `str(e)`, so the
same failure is never described two ways. Pass `warn=False` when the result already
reports the outcome.

#### Deciding between an error and a result

Ask two questions, in order:

1. **Did the tool finish its own job?** `test_integration_connection`'s job is to run the
   check and report what happened, so `connected: false` is the job finished.
   `get_finding_details`' job is to return the finding, so no finding means it did not.
2. **Is the reported state a fact about the remote world or about our call?** The world
   (Jira refused the credentials, 3 of 40 items failed, a discovery found nothing) is a
   **result**. Our call (403, connection reset, invalid UUID, a bug in a model factory) is
   an **error**.

One rule overrides both: **if a write may have partially landed, that fact travels in a
successful structured result, never in an error.** An agent reads `isError: true` as
"nothing happened, safe to retry"; reporting "I may have created 17 Jira issues" that way
invites a duplicate dispatch.

#### What the client reads

`render_tool_error` describes the failure in one plain sentence: the call, the status and
whatever the API said, with the field named when it named one.

```text
GET /findings/b1ca536c failed with HTTP 404. No Finding matches the given query.
POST /integrations failed with HTTP 400. This field may not be blank. (/data/attributes/configuration/bucket_name); Enter a valid URL.
Date range cannot exceed 2 days. Requested range: 2025-01-01 to 2025-01-10 (10 days)
```

Nothing is added that the status code already implies. The one exception is a request that
could have changed something and never came back with a verdict — a 5xx or a timeout on a
write — which gets a warning, because an agent otherwise reads any failure as "nothing
happened" and sends the write again:

```text
DELETE /integrations/i1 failed with HTTP 500. A server error occurred. It may have been carried out anyway, so check the current state before retrying.
```

Every server sets `mask_error_details=True`. That costs nothing, because `ToolError`
bypasses masking; it only stops raw internals escaping from code paths outside a tool.

### Parameter Descriptions

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

```python
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

```bash
# 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](/getting-started/installation/prowler-mcp#from-source-development) - Development setup instructions
- [Configuration Guide](/getting-started/basic-usage/prowler-mcp) - MCP client configuration

For development I recommend to use the [Model Context Protocol Inspector](https://github.com/modelcontextprotocol/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/`:

```bash
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:

```bash
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

<Warning>
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.
</Warning>

### 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.

```python
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.

<Warning>
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.
</Warning>

### 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.

<Note>
`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.
</Note>

For the full set of rules and templates, see the
[`prowler-test-mcp` skill](https://github.com/prowler-cloud/prowler/blob/master/skills/prowler-test-mcp/SKILL.md)
and the [official FastMCP testing guide](https://gofastmcp.com/development/tests).

## Related Documentation

<CardGroup cols={2}>
  <Card title="MCP Server Overview" icon="circle-info" href="/getting-started/products/prowler-mcp">
    Key capabilities, use cases, and deployment options
  </Card>
  <Card title="Tools Reference" icon="wrench" href="/getting-started/basic-usage/prowler-mcp-tools">
    Complete reference of all available tools
  </Card>
  <Card title="Prowler Hub" icon="database" href="/getting-started/products/prowler-hub">
    Security checks and compliance frameworks catalog
  </Card>
  <Card title="Lighthouse AI" icon="robot" href="/getting-started/products/prowler-lighthouse-ai">
    AI-powered security analyst
  </Card>
</CardGroup>

## Additional Resources

- [MCP Protocol Specification](https://modelcontextprotocol.io) - Model Context Protocol details
- [Prowler API Documentation](https://api.prowler.com/api/v1/docs) - API reference
- [Prowler Hub API](https://hub.prowler.com/api/docs) - Hub API reference
- [GitHub Repository](https://github.com/prowler-cloud/prowler) - Source code