Skip to content

module_development.md

Latest commit

 

History

History
552 lines (422 loc) · 18.6 KB

module_development.md

File metadata and controls

552 lines (422 loc) · 18.6 KB

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:

"""
[Module Name] module for Falcon MCP Server

This module provides tools for [brief description].
"""
from typing import Any

from mcp.server import FastMCP
from mcp.types import ToolAnnotations
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
        """
        # Read-only tool — default annotations apply automatically
        self._add_tool(
            server=server,
            method=self.your_tool_method,
            name="your_tool_name",
        )

        # Mutating tool — must override annotations
        self._add_tool(
            server=server,
            method=self.your_create_method,
            name="your_create_name",
            annotations=ToolAnnotations(
                readOnlyHint=False,
                destructiveHint=False,
                idempotentHint=False,
                openWorldHint=True,
            ),
        )

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

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:

"""
Tests for the YourModule module.
"""
from mcp.types import ToolAnnotations

from falcon_mcp.modules.base import READ_ONLY_ANNOTATIONS
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_tool_annotations(self):
        """Test tool annotations are correctly set."""
        self.module.register_tools(self.mock_server)
        # Read-only tools should use defaults
        self.assert_tool_annotations("falcon_your_tool_name", READ_ONLY_ANNOTATIONS)
        # Mutating tools must have explicit annotations
        self.assert_tool_annotations(
            "falcon_your_create_name",
            ToolAnnotations(
                readOnlyHint=False,
                destructiveHint=False,
                idempotentHint=False,
                openWorldHint=True,
            ),
        )

    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 for automated releases and clear commit history. When contributing module-related changes, use these commit message patterns:

Adding New Modules:

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):

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:

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):

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:

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

# 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

Tool Annotations

Tools are annotated with MCP ToolAnnotations to inform clients about tool behavior:

  • readOnlyHint: Does the tool only read data? (default: True)
  • destructiveHint: Does the tool delete or destroy data? (default: False)
  • idempotentHint: Is repeating the call safe? (default: True)
  • openWorldHint: Does the tool interact with external systems? (default: True)

The _add_tool() helper applies READ_ONLY_ANNOTATIONS by default. Override for any tool that creates, modifies, deletes, or triggers actions.

Tool Type readOnlyHint destructiveHint idempotentHint Example
Search/Get/List True False True falcon_search_hosts
Create/Write False False False falcon_add_ioc
Delete/Remove False True True falcon_remove_iocs
Launch/Trigger False False False falcon_launch_scheduled_report

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:

# ✅ 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:

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