# Falcon MCP Server Module Development Guide
This guide provides instructions for implementing new modules for the Falcon MCP server.
## Module Structure
Each module should:
1. Inherit from the `BaseModule` class
2. Implement the `register_tools` method
3. Define tool methods that interact with the Falcon API
4. Use common utilities for configuration, logging, error handling, and API interactions
## Step-by-Step Implementation Guide
### 1. Create a New Module File
Create a new file in the `falcon_mcp/modules` directory:
```python
"""
[Module Name] module for Falcon MCP Server
This module provides tools for [brief description].
"""
from typing import Any
from mcp.server import FastMCP
from pydantic import Field
from falcon_mcp.common.logging import get_logger
from falcon_mcp.modules.base import BaseModule
logger = get_logger(__name__)
class YourModule(BaseModule):
"""Module for [description]."""
def register_tools(self, server: FastMCP) -> None:
"""Register tools with the MCP server.
Args:
server: MCP server instance
"""
# Register tools
self._add_tool(
server=server,
method=self.your_tool_method,
name="your_tool_name",
)
# Add more tools as needed
def your_tool_method(
self,
param1: str = Field(
description="Description of param1. Explain what values are valid.",
examples=["example_value_1", "example_value_2"],
),
param2: int | None = Field(
default=None,
ge=1,
le=100,
description="Description of param2. Include constraints if applicable.",
),
) -> dict[str, Any]:
"""Description of what your tool does.
Provide context on when to use this tool and any important notes.
Tool descriptions are derived from this docstring.
"""
# Use base class methods for common patterns
results = self._base_search_api_call(
operation="YourFalconAPIOperation",
search_params={
"param1": param1,
"param2": param2,
},
error_message="Failed to perform operation",
)
if self._is_error(results):
return results
return results
```
### 2. Update API Scope Requirements
Add your API operations to the `API_SCOPE_REQUIREMENTS` dictionary in `falcon_mcp/common/api_scopes.py`:
```python
API_SCOPE_REQUIREMENTS = {
# Existing operations...
"YourFalconAPIOperation": ["required:scope"],
# Add more operations as needed
}
```
### 3. Module Auto-Discovery
Modules are automatically discovered by the registry system. You don't need to call any registration functions or add imports:
1. Create your module class in the `falcon_mcp/modules` directory (e.g., `your_module.py`)
2. Make sure it inherits from `BaseModule`
3. **Modules are automatically discovered** - no manual imports or registration needed
The server will automatically discover and register your module during initialization. The module name will be derived
from the class name (e.g., `YourModule` becomes `your`).
During server initialization, the registry system will:
1. Scan the modules directory using `pkgutil.iter_modules()`
2. Dynamically import each module file using `importlib.import_module()`
3. Find classes that end with "Module" (excluding BaseModule) via introspection
4. Register them in the `AVAILABLE_MODULES` dictionary
5. Make them available to the server
This approach eliminates manual registration while maintaining a clean architecture that avoids cyclic imports.
### 4. Add Tests
Create a test file in the `tests/modules` directory that inherits from the `TestModules` base class:
```python
"""
Tests for the YourModule module.
"""
from falcon_mcp.modules.your_module import YourModule
from tests.modules.utils.test_modules import TestModules
class TestYourModule(TestModules):
"""Test cases for the YourModule module."""
def setUp(self):
"""Set up test fixtures."""
self.setup_module(YourModule)
def test_register_tools(self):
"""Test registering tools with the server."""
expected_tools = [
"falcon_your_tool_name",
# Add other tools here
]
self.assert_tools_registered(expected_tools)
def test_your_tool_method(self):
"""Test your tool method."""
# Setup mock response
mock_response = {
"status_code": 200,
"body": {
"resources": [{"id": "test", "name": "Test Resource"}]
}
}
self.mock_client.command.return_value = mock_response
# Call your tool method
result = self.module.your_tool_method("test_param", 123)
# Verify client command was called correctly
self.mock_client.command.assert_called_once_with(
"YourFalconAPIOperation",
parameters={"param1": "test_param", "param2": 123}
)
# Verify result
expected_result = [{"id": "test", "name": "Test Resource"}]
self.assertEqual(result, expected_result)
def test_your_tool_method_error(self):
"""Test your tool method with API error."""
# Setup mock response with error
mock_response = {
"status_code": 403,
"body": {
"errors": [{"message": "Access denied"}]
}
}
self.mock_client.command.return_value = mock_response
# Call your tool method
result = self.module.your_tool_method("test_param")
# Verify result contains error
self.assertIn("error", result)
self.assertIn("details", result)
```
The `TestModules` base class provides:
1. A `setup_module()` method that handles the common setup of mocking the client and server
2. An `assert_tools_registered()` helper method to verify tool registration
This approach simplifies test code and ensures consistency across all module tests.
## Contributing Module Changes
When contributing new modules or changes to existing modules, please follow these guidelines:
### Conventional Commits for Modules
This project uses [Conventional Commits](https://www.conventionalcommits.org/) for automated releases and clear commit history. When contributing module-related changes, use these commit message patterns:
**Adding New Modules:**
```bash
git commit -m "feat(modules): add [module-name] module for [functionality]"
# Examples:
git commit -m "feat(modules): add spotlight module for vulnerability management"
git commit -m "feat(modules): add intel module for threat intelligence analysis"
```
**Adding Tools to Existing Modules (Preferred - More Specific Scoping):**
```bash
git commit -m "feat(modules/[module-name]): add [specific-functionality]"
# Examples:
git commit -m "feat(modules/cloud): add list kubernetes clusters tool"
git commit -m "feat(modules/hosts): add list devices tool"
git commit -m "feat(modules/detections): add advanced filtering capability"
```
**Modifying Existing Modules:**
```bash
git commit -m "feat(modules/[module-name]): enhance [specific-functionality]"
git commit -m "fix(modules/[module-name]): resolve [specific-issue]"
# Examples:
git commit -m "feat(modules/detections): enhance FQL filtering with new operators"
git commit -m "fix(modules/hosts): resolve authentication error in search function"
```
**General Module Changes (Less Preferred but Acceptable):**
```bash
git commit -m "feat(modules): enhance [module-name] with [new-functionality]"
git commit -m "fix(modules): resolve [issue] in [module-name] module"
# Examples:
git commit -m "feat(modules): enhance detections module with FQL filtering"
git commit -m "fix(modules): resolve authentication error in hosts module"
```
**Module Tests and Documentation:**
```bash
git commit -m "test(modules): add comprehensive tests for [module-name] module"
git commit -m "docs(modules): update [module-name] module documentation"
```
See the main [CONTRIBUTING.md](CONTRIBUTING.md) guide for complete conventional commits guidelines.
## Best Practices
### Error Handling
1. **Use Common Error Utilities**: Always use `handle_api_response` for API responses instead of manual status code checks
2. **Provide Operation Names**: Include the operation name for better error messages and permission handling
3. **Custom Error Messages**: Use descriptive error messages for each operation
4. **Consistent Error Format**: Ensure error responses follow the standard format with `error` and optional `details` fields
Example of proper error handling:
```python
# Make the API request
response = self.client.command(operation, parameters=params)
# Use handle_api_response to process the response
result = handle_api_response(
response,
operation=operation,
error_message="Failed to perform operation",
default_result=[]
)
# Check if the result is an error response
if isinstance(result, dict) and "error" in result:
# Handle error case
return result # or wrap it in a list if returning to a tool expecting a list
```
### Parameter Handling
1. **Use prepare_api_parameters**: Filter out None values and format parameters
2. **Type Annotations**: Always include type annotations for parameters and return values
3. **Default Values**: Provide sensible defaults for optional parameters
### Response Processing
1. **Use extract_resources**: Extract resources from API responses
2. **Handle Empty Results**: Provide appropriate defaults for empty results
3. **Return Structured Data**: Return well-structured data that follows consistent patterns
### Documentation
1. **Docstrings**: Include detailed docstrings for all classes and methods. Tool descriptions are derived from method docstrings, so make sure they are comprehensive and well-written.
2. **Parameter Descriptions**: Document all parameters and return values
3. **Examples**: Include examples in docstrings where helpful
### Type Hints
1. **Use Modern Syntax**: Use Python 3.10+ type hints (PEP 585/604)
2. **Built-in Generics**: Use `list[...]`, `dict[...]` instead of `List[...]`, `Dict[...]`
3. **Union Syntax**: Use `X | None` instead of `Optional[X]`
4. **Minimal Imports**: Only import `Any` from typing module when needed
Example:
```python
# ✅ Correct - Modern type hints
from typing import Any
def search_entities(self, filter: str | None = None) -> list[dict[str, Any]]:
...
# ❌ Avoid - Legacy type hints
from typing import Dict, List, Optional, Any
def search_entities(self, filter: Optional[str] = None) -> List[Dict[str, Any]]:
...
```
### Testing
1. **Test All Tools**: Write tests for all tools in your module
2. **Test Error Cases**: Include tests for error scenarios
3. **Mock API Responses**: Use mock responses for testing
## Common Utilities Reference
### Configuration (`falcon_mcp/common/config.py`)
- `FalconConfig`: Configuration class for the Falcon MCP server
- `load_config`: Load configuration from environment variables and arguments
### Logging (`falcon_mcp/common/logging.py`)
- `configure_logging`: Configure logging for the Falcon MCP server
- `get_logger`: Get a logger with the specified name
### Error Handling (`falcon_mcp/common/errors.py`)
- `is_success_response`: Check if an API response indicates success
- `_format_error_response`: Format an error as a standardized response
- `handle_api_response`: Handle an API response, returning either the result or an error
### API Scopes (`falcon_mcp/common/api_scopes.py`)
- `get_required_scopes`: Get the required API scopes for a specific operation
### Utilities (`falcon_mcp/common/utils.py`)
- `filter_none_values`: Remove None values from a dictionary
- `prepare_api_parameters`: Prepare parameters for Falcon API requests
- `extract_resources`: Extract resources from an API response
- `extract_first_resource`: Extract the first resource from an API response
## Example: Implementing a Hosts Module
Here's an example of implementing a Hosts module that provides tools for accessing and managing hosts in the Falcon platform:
```python
"""
Hosts module for Falcon MCP Server
This module provides tools for accessing and managing CrowdStrike Falcon hosts.
"""
from typing import Any
from mcp.server import FastMCP
from pydantic import Field
from falcon_mcp.common.logging import get_logger
from falcon_mcp.modules.base import BaseModule
logger = get_logger(__name__)
class HostsModule(BaseModule):
"""Module for accessing and managing CrowdStrike Falcon hosts."""
def register_tools(self, server: FastMCP) -> None:
"""Register tools with the MCP server.
Args:
server: MCP server instance
"""
self._add_tool(
server=server,
method=self.search_hosts,
name="search_hosts",
)
self._add_tool(
server=server,
method=self.get_host_details,
name="get_host_details",
)
def search_hosts(
self,
filter: str | None = Field(
default=None,
description="FQL Syntax formatted string used to limit the results.",
examples={"platform_name:'Windows'", "hostname:'PC*'"},
),
limit: int = Field(
default=10,
ge=1,
le=5000,
description="The maximum records to return. [1-5000]",
),
offset: int | None = Field(
default=None,
description="The offset to start retrieving records from.",
),
sort: str | None = Field(
default=None,
description=dedent("""
Sort hosts using these options:
hostname: Host name/computer name
last_seen: Timestamp when the host was last seen
first_seen: Timestamp when the host was first seen
modified_timestamp: When the host record was last modified
platform_name: Operating system platform
agent_version: CrowdStrike agent version
os_version: Operating system version
external_ip: External IP address
Sort either asc (ascending) or desc (descending).
Both formats are supported: 'hostname.desc' or 'hostname|desc'
Examples: 'hostname.asc', 'last_seen.desc', 'platform_name.asc'
""").strip(),
examples={"hostname.asc", "last_seen.desc"},
),
) -> list[dict[str, Any]]:
"""Search for hosts in your CrowdStrike environment.
"""
device_ids = self._base_search_api_call(
operation="QueryDevicesByFilter",
search_params={
"filter": filter,
"limit": limit,
"offset": offset,
"sort": sort,
},
error_message="Failed to search hosts",
)
# If handle_api_response returns an error dict instead of a list,
# it means there was an error, so we return it wrapped in a list
if self._is_error(device_ids):
return [device_ids]
# If we have device IDs, get the details for each one
if device_ids:
details = self._base_get_by_ids(
operation="PostDeviceDetailsV2",
ids=device_ids,
id_key="ids",
)
# If handle_api_response returns an error dict instead of a list,
# it means there was an error, so we return it wrapped in a list
if self._is_error(details):
return [details]
return details
return []
def get_host_details(
self,
ids: list[str] = Field(
description="Host device IDs to retrieve details for. You can get device IDs from the search_hosts operation, the Falcon console, or the Streaming API. Maximum: 5000 IDs per request."
),
) -> list[dict[str, Any]] | dict[str, Any]:
"""Retrieve detailed information for specified host device IDs.
This tool returns comprehensive host details for one or more device IDs.
Use this when you already have specific device IDs and need their full details.
For searching/discovering hosts, use the `falcon_search_hosts` tool instead.
"""
logger.debug("Getting host details for IDs: %s", ids)
# Handle empty list case - return empty list without making API call
if not ids:
return []
return self._base_get_by_ids(
operation="PostDeviceDetailsV2",
ids=ids,
id_key="ids",
)
```
The module will be automatically discovered by the registry system - no manual imports or registration needed.
