StackWiz MCP Development Guide¶
This guide is for developers who want to extend, customize, or contribute to StackWiz MCP.
Architecture Overview¶
Project Structure¶
stackwiz-mcp/
├── stackwiz_mcp/ # Main package
│ ├── __init__.py # Package initialization
│ ├── __main__.py # Entry point
│ ├── server.py # FastMCP server implementation
│ ├── mcp_server.py # Standard MCP implementation
│ ├── config.py # Configuration management
│ ├── tools/ # Tool implementations
│ │ ├── create_stack.py # Stack creation logic
│ │ ├── list_stacks.py # Stack listing
│ │ ├── manage_stack.py # Stack operations
│ │ ├── dns_operations.py # DNS management
│ │ └── validate_config.py # Configuration validation
│ ├── resources/ # Resource providers
│ │ ├── stack_configs.py # Stack configuration access
│ │ ├── templates.py # Template management
│ │ └── infrastructure.py # Infrastructure info
│ ├── prompts/ # Interactive prompts
│ │ └── deployment_prompts.py
│ ├── models/ # Data models
│ │ └── stack_models.py # Pydantic models
│ └── utils/ # Utilities
│ ├── stack_utils.py # Stack helpers
│ ├── validation.py # Input validation
│ ├── logging.py # Logging setup
│ └── health.py # Health checks
├── tests/ # Test suite
├── docs/ # Documentation
└── examples/ # Usage examples
Core Components¶
- MCP Server Layer
- Handles MCP protocol communication
- Routes requests to appropriate tools
-
Manages resources and prompts
-
Tool Layer
- Implements individual MCP tools
- Validates inputs
-
Executes Docker operations
-
Utility Layer
- Common functionality
- Docker API interactions
- File system operations
Development Setup¶
1. Clone and Setup Environment¶
# Clone repository
cd /srv/dockerdata/stackwiz-mcp
# Create virtual environment
python -m venv venv
source venv/bin/activate
# Install in development mode
pip install -e ".[dev]"
# Install pre-commit hooks
pre-commit install
2. Development Dependencies¶
# Install all development tools
pip install -r requirements-dev.txt
# Tools included:
# - pytest: Testing framework
# - black: Code formatter
# - flake8: Linter
# - mypy: Type checker
# - pytest-cov: Coverage reporting
3. Environment Configuration¶
Create .env.development:
STACKWIZ_ENV=development
STACKWIZ_LOG_LEVEL=DEBUG
STACKWIZ_BASE_DIR=/tmp/stackwiz-dev
DOCKER_HOST=unix:///var/run/docker.sock
Adding New Tools¶
1. Create Tool Module¶
Create stackwiz_mcp/tools/my_new_tool.py:
"""My new tool for StackWiz MCP."""
from typing import Dict, Any, Optional
from pydantic import BaseModel, Field
from fastmcp import Tool
from ..utils.validation import validate_stack_name
from ..utils.stack_utils import get_stack_path
from ..utils.logging import get_logger
logger = get_logger(__name__)
class MyToolInput(BaseModel):
"""Input model for my new tool."""
stack_name: str = Field(
...,
description="Name of the stack",
pattern="^[a-z][a-z0-9-]*$"
)
option1: Optional[str] = Field(
None,
description="Optional parameter"
)
option2: bool = Field(
False,
description="Boolean flag"
)
async def my_new_tool(input_data: MyToolInput) -> Dict[str, Any]:
"""
Execute my new tool operation.
Args:
input_data: Validated input parameters
Returns:
Dictionary with operation results
"""
try:
# Validate stack name
validate_stack_name(input_data.stack_name)
# Get stack path
stack_path = get_stack_path(input_data.stack_name)
# Implement your logic here
logger.info(f"Executing my_new_tool for {input_data.stack_name}")
# Perform operations...
result = {
"success": True,
"message": f"Operation completed for {input_data.stack_name}",
"data": {
"stack_name": input_data.stack_name,
"option1": input_data.option1,
"option2": input_data.option2
}
}
return result
except Exception as e:
logger.error(f"Error in my_new_tool: {str(e)}")
return {
"success": False,
"error": {
"code": "TOOL_ERROR",
"message": str(e)
}
}
# Create the tool instance
tool = Tool(
name="my_new_tool",
description="Description of what this tool does",
input_model=MyToolInput,
fn=my_new_tool
)
2. Register the Tool¶
Update stackwiz_mcp/tools/__init__.py:
from .create_stack import tool as create_stack_tool
from .list_stacks import tool as list_stacks_tool
from .manage_stack import tool as manage_stack_tool
from .my_new_tool import tool as my_new_tool_tool # Add this
__all__ = [
"create_stack_tool",
"list_stacks_tool",
"manage_stack_tool",
"my_new_tool_tool", # Add this
]
3. Add to Server¶
Update stackwiz_mcp/server.py:
# Import the new tool
from .tools import (
create_stack_tool,
list_stacks_tool,
manage_stack_tool,
my_new_tool_tool, # Add this
)
# In the server setup
mcp.add_tool(create_stack_tool)
mcp.add_tool(list_stacks_tool)
mcp.add_tool(manage_stack_tool)
mcp.add_tool(my_new_tool_tool) # Add this
Testing¶
1. Unit Tests¶
Create tests/test_my_new_tool.py:
"""Tests for my_new_tool."""
import pytest
from unittest.mock import Mock, patch
from stackwiz_mcp.tools.my_new_tool import my_new_tool, MyToolInput
@pytest.mark.asyncio
async def test_my_new_tool_success():
"""Test successful execution."""
# Arrange
input_data = MyToolInput(
stack_name="test-stack",
option1="value1",
option2=True
)
# Act
with patch('stackwiz_mcp.tools.my_new_tool.get_stack_path') as mock_path:
mock_path.return_value = "/tmp/test-stack"
result = await my_new_tool(input_data)
# Assert
assert result["success"] is True
assert result["data"]["stack_name"] == "test-stack"
assert result["data"]["option1"] == "value1"
assert result["data"]["option2"] is True
@pytest.mark.asyncio
async def test_my_new_tool_invalid_name():
"""Test with invalid stack name."""
# Arrange
input_data = MyToolInput(
stack_name="Invalid-Name", # Capital letters not allowed
option1="value1"
)
# Act
result = await my_new_tool(input_data)
# Assert
assert result["success"] is False
assert "error" in result
2. Integration Tests¶
Create tests/integration/test_my_new_tool_integration.py:
"""Integration tests for my_new_tool."""
import pytest
import tempfile
import shutil
from pathlib import Path
from stackwiz_mcp.tools.my_new_tool import my_new_tool, MyToolInput
@pytest.fixture
def temp_base_dir():
"""Create temporary base directory."""
temp_dir = tempfile.mkdtemp()
yield temp_dir
shutil.rmtree(temp_dir)
@pytest.mark.asyncio
async def test_my_new_tool_integration(temp_base_dir, monkeypatch):
"""Test tool with real file system."""
# Set base directory
monkeypatch.setenv("STACKWIZ_BASE_DIR", temp_base_dir)
# Create test stack
stack_path = Path(temp_base_dir) / "test-stack"
stack_path.mkdir(parents=True)
# Execute tool
input_data = MyToolInput(stack_name="test-stack")
result = await my_new_tool(input_data)
# Verify
assert result["success"] is True
assert stack_path.exists()
3. Running Tests¶
# Run all tests
pytest
# Run with coverage
pytest --cov=stackwiz_mcp --cov-report=html
# Run specific test
pytest tests/test_my_new_tool.py -v
# Run only unit tests
pytest tests/unit/ -v
# Run only integration tests
pytest tests/integration/ -v
Code Style and Quality¶
1. Formatting¶
# Format code with black
black stackwiz_mcp tests
# Check without modifying
black --check stackwiz_mcp tests
2. Linting¶
# Run flake8
flake8 stackwiz_mcp tests
# Run with specific rules
flake8 --select=E9,F63,F7,F82 stackwiz_mcp
3. Type Checking¶
4. Pre-commit Hooks¶
.pre-commit-config.yaml:
repos:
- repo: https://github.com/psf/black
rev: 23.3.0
hooks:
- id: black
- repo: https://github.com/pycqa/flake8
rev: 6.0.0
hooks:
- id: flake8
- repo: https://github.com/pre-commit/mirrors-mypy
rev: v1.3.0
hooks:
- id: mypy
Debugging¶
1. Enable Debug Logging¶
# In your code
from ..utils.logging import get_logger
logger = get_logger(__name__)
logger.debug(f"Debug info: {variable}")
2. Interactive Debugging¶
3. Remote Debugging¶
Best Practices¶
1. Input Validation¶
Always validate inputs thoroughly:
from pydantic import BaseModel, Field, validator
class StackConfig(BaseModel):
name: str = Field(..., pattern="^[a-z][a-z0-9-]*$")
port: int = Field(..., ge=1, le=65535)
@validator('name')
def validate_name_length(cls, v):
if len(v) > 50:
raise ValueError("Name too long")
return v
2. Error Handling¶
Use consistent error format:
try:
# Operation
pass
except SpecificError as e:
return {
"success": False,
"error": {
"code": "SPECIFIC_ERROR",
"message": str(e),
"details": {...},
"suggestion": "Try this instead..."
}
}
3. Async Operations¶
Use async/await properly:
async def long_operation():
# For I/O operations
async with aiofiles.open(path) as f:
content = await f.read()
# For subprocess
proc = await asyncio.create_subprocess_exec(
'docker', 'compose', 'up',
stdout=asyncio.subprocess.PIPE
)
stdout, _ = await proc.communicate()
4. Documentation¶
Always include docstrings:
def process_stack(name: str, config: Dict[str, Any]) -> StackResult:
"""
Process a stack configuration and create Docker resources.
Args:
name: The stack name, must be lowercase alphanumeric with hyphens
config: Stack configuration dictionary containing:
- image: Docker image to use
- port: Port to expose
- environment: Environment variables
Returns:
StackResult object containing:
- success: Whether operation succeeded
- path: Path to created stack
- url: Access URL if applicable
Raises:
ValidationError: If configuration is invalid
DockerError: If Docker operations fail
Example:
>>> result = process_stack("my-app", {"image": "nginx", "port": 80})
>>> print(result.url)
https://my-app.example.com
"""
Contributing¶
1. Fork and Branch¶
# Fork on GitHub, then:
git clone https://github.com/YOUR_USERNAME/stackwiz-mcp.git
cd stackwiz-mcp
git checkout -b feature/my-new-feature
2. Make Changes¶
- Follow code style guidelines
- Add tests for new functionality
- Update documentation
- Add changelog entry
3. Submit Pull Request¶
- Push to your fork
- Create pull request
- Describe changes clearly
- Link related issues
Pull Request Template¶
## Description
Brief description of changes
## Type of Change
- [ ] Bug fix
- [ ] New feature
- [ ] Breaking change
- [ ] Documentation update
## Testing
- [ ] Unit tests pass
- [ ] Integration tests pass
- [ ] Manual testing completed
## Checklist
- [ ] Code follows style guidelines
- [ ] Self-review completed
- [ ] Documentation updated
- [ ] Tests added/updated
Advanced Topics¶
Custom Resource Providers¶
from fastmcp import Resource
class CustomResource(Resource):
async def get(self, uri: str) -> str:
# Implement resource retrieval
pass
# Register resource
mcp.add_resource("custom://", CustomResource())
Custom Prompts¶
from fastmcp import Prompt
deployment_prompt = Prompt(
name="deploy-custom",
description="Deploy custom application",
template="""
Let's deploy your {app_type} application.
Requirements:
- {requirements}
Steps:
1. {step1}
2. {step2}
"""
)
Extending Docker Operations¶
import docker
from docker.types import ServiceMode, Resources
class AdvancedStackManager:
def __init__(self):
self.client = docker.from_env()
def create_swarm_service(self, name: str, image: str):
service = self.client.services.create(
image=image,
name=name,
mode=ServiceMode("replicated", replicas=3),
resources=Resources(
cpu_limit=1000000000, # 1 CPU
mem_limit=512 * 1024 * 1024 # 512MB
)
)
return service
Performance Optimization¶
1. Caching¶
from functools import lru_cache
@lru_cache(maxsize=128)
def get_docker_images():
"""Cache Docker image list."""
client = docker.from_env()
return [img.tags[0] for img in client.images.list()]
2. Async Operations¶
import asyncio
async def parallel_operations(stacks: List[str]):
"""Execute operations in parallel."""
tasks = [process_stack(stack) for stack in stacks]
results = await asyncio.gather(*tasks)
return results
Monitoring and Metrics¶
Add Prometheus Metrics¶
from prometheus_client import Counter, Histogram, Gauge
# Define metrics
stack_created = Counter('stackwiz_stacks_created_total',
'Total stacks created')
operation_duration = Histogram('stackwiz_operation_duration_seconds',
'Operation duration')
active_stacks = Gauge('stackwiz_active_stacks',
'Number of active stacks')
# Use in code
@operation_duration.time()
def create_stack_with_metrics(...):
result = create_stack(...)
if result["success"]:
stack_created.inc()
active_stacks.inc()
return result
Security Considerations¶
1. Input Sanitization¶
import re
from pathlib import Path
def sanitize_path(user_input: str) -> Path:
"""Sanitize user input for file paths."""
# Remove any path traversal attempts
clean = re.sub(r'\.\./', '', user_input)
clean = re.sub(r'^/', '', clean)
# Ensure within base directory
base = Path(os.environ.get('STACKWIZ_BASE_DIR'))
full_path = base / clean
# Verify it's within base
if not full_path.resolve().is_relative_to(base):
raise ValueError("Invalid path")
return full_path
2. Secret Management¶
import secrets
from cryptography.fernet import Fernet
class SecretManager:
def __init__(self):
self.key = Fernet.generate_key()
self.cipher = Fernet(self.key)
def encrypt_env_var(self, value: str) -> str:
"""Encrypt sensitive environment variable."""
return self.cipher.encrypt(value.encode()).decode()
def decrypt_env_var(self, encrypted: str) -> str:
"""Decrypt environment variable."""
return self.cipher.decrypt(encrypted.encode()).decode()
Next Steps¶
- Examples - See implementation examples
- Tool Reference - API documentation
- Troubleshooting - Debug common issues