--- title: '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](/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 ## 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 | 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](/getting-started/basic-usage/prowler-mcp-tools). ## 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 ├── 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/`: ```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. """ 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](#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/`: ```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 **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: ```python 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: ```python 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: ```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 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. ```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. 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](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 Key capabilities, use cases, and deployment options Complete reference of all available tools Security checks and compliance frameworks catalog AI-powered security analyst ## 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