# ContextForge Development Guide

> This guide provides comprehensive information for developers working on ContextForge project.

- Skill: `tools-only/contextforge-development-guide` (Agent Skill, multi-file: 3 files)
- Install (CLI): `npx skillmds@latest add tools-only/contextforge-development-guide`
- Raw SKILL.md: https://api.skillmd.com/api/skills/tools-only/contextforge-development-guide/raw
- Safety review: pending (external: skill-scanner WARNING, skillspector FAIL)
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Coding & Dev Tools
- Author: tools-only (https://skillmd.com/u/tools-only)
- Updated: 2026-09-29
- Page: https://skillmd.com/skills/tools-only/contextforge-development-guide

---

# ContextForge Development Guide

This guide provides comprehensive information for developers working on ContextForge project.

## Table of Contents
- [Quick Start](#quick-start)
- [Development Setup](#development-setup)
- [Project Architecture](#project-architecture)
- [Development Workflow](#development-workflow)
- [Code Quality](#code-quality)
- [Database Management](#database-management)
- [API Development](#api-development)
- [Plugin Development](#plugin-development)
- [Testing MCP Servers](#testing-mcp-servers)
- [Debugging](#debugging)
- [Performance Optimization](#performance-optimization)
- [Contributing](#contributing)

## Quick Start

```bash
# Clone and setup
git clone https://github.com/IBM/mcp-context-forge.git
cd mcp-context-forge

# Complete setup with uv (recommended)
cp .env.example .env && make venv install-dev check-env

# Start development server with hot-reload
make dev

# Run quality checks before committing
make autoflake isort black pre-commit
make doctest test htmlcov flake8 pylint verify
```

## Development Setup

### Prerequisites

- **Python 3.11+** (3.10 minimum)
- **uv** (recommended) or pip/virtualenv
- **Make** for automation
- **Docker/Podman** (optional, for container development)
- **Node.js 18+** (for UI development and MCP Inspector)
- **PostgreSQL/MySQL** (optional, for production database testing)

### Environment Setup

#### Using uv (Recommended)

```bash
# Install uv
curl -LsSf https://astral.sh/uv/install.sh | sh

# Create virtual environment and install dependencies
make venv install-dev

# Verify environment
make check-env
```

#### Traditional Setup

```bash
# Create virtual environment
python3 -m venv .venv
source .venv/bin/activate

# Install in editable mode with all extras
pip install -e ".[dev,test,docs,otel,redis]"
```

### Configuration

```bash
# Copy example configuration
cp .env.example .env

# Edit configuration
vim .env

# Key development settings
ENVIRONMENT=development          # Enables debug features
DEV_MODE=true                   # Additional development helpers
DEBUG=true                      # Verbose error messages
RELOAD=true                     # Auto-reload on code changes
LOG_LEVEL=DEBUG                 # Maximum logging verbosity
MCPGATEWAY_UI_ENABLED=true      # Enable Admin UI
MCPGATEWAY_ADMIN_API_ENABLED=true  # Enable Admin API
```

## Project Architecture

### Directory Structure

```
mcp-context-forge/
├── mcpgateway/                 # Main application package
│   ├── main.py                # FastAPI application entry
│   ├── cli.py                 # CLI commands
│   ├── config.py              # Settings management
│   ├── models.py              # SQLAlchemy models
│   ├── schemas.py             # Pydantic schemas
│   ├── admin.py               # Admin UI routes
│   ├── auth.py                # Authentication logic
│   ├── services/              # Business logic layer
│   │   ├── gateway_service.py    # Federation management
│   │   ├── server_service.py     # Virtual server composition
│   │   ├── tool_service.py       # Tool registry
│   │   ├── a2a_service.py        # Agent-to-Agent
│   │   └── export_service.py     # Bulk operations
│   ├── transports/            # Protocol implementations
│   │   ├── sse_transport.py      # Server-Sent Events
│   │   ├── websocket_transport.py # WebSocket
│   │   └── stdio_transport.py    # Standard I/O wrapper
│   ├── plugins/               # Plugin framework
│   │   ├── framework/            # Core plugin system
│   │   └── [plugin_dirs]/       # Individual plugins
│   ├── validation/            # Input validation
│   ├── utils/                 # Utility modules
│   ├── templates/             # Jinja2 templates (Admin UI)
│   └── static/                # Static assets
├── tests/                     # Test suites
│   ├── unit/                  # Unit tests
│   ├── integration/           # Integration tests
│   ├── e2e/                   # End-to-end tests
│   ├── playwright/            # UI tests
│   └── conftest.py            # Pytest fixtures
├── alembic/                   # Database migrations
├── docs/                      # Documentation
├── plugins/                   # Plugin configurations
└── mcp-servers/               # Example MCP servers
```

### Technology Stack

- **Web Framework**: FastAPI 0.115+
- **Database ORM**: SQLAlchemy 2.0+
- **Validation**: Pydantic 2.0+
- **Admin UI**: HTMX + Alpine.js
- **Testing**: Pytest + Playwright
- **Package Management**: uv (or pip)
- **Database**: SQLite (dev), PostgreSQL/MySQL (production)
- **Caching**: Redis (optional)
- **Observability**: OpenTelemetry

### Key Components

#### 1. Core Services
- **GatewayService**: Manages federation and peer discovery
- **ServerService**: Handles virtual server composition
- **ToolService**: Tool registry and invocation
- **A2AService**: Agent-to-Agent integration
- **AuthService**: JWT authentication and authorization

#### 2. Transport Layers
- **SSE Transport**: Server-Sent Events for streaming
- **WebSocket Transport**: Bidirectional real-time communication
- **HTTP Transport**: Standard JSON-RPC over HTTP
- **Stdio Wrapper**: Bridge for stdio-based MCP clients

#### 3. Plugin System
- **Hook-based**: Pre/post request/response hooks
- **Filters**: PII, deny-list, regex, resource filtering
- **Custom plugins**: Extensible framework for custom logic

## Development Workflow

### Running the Development Server

```bash
# Development server with hot-reload (port 8000)
make dev

# Production-like server (port 4444)
make serve

# With SSL/TLS
make certs serve-ssl

# Custom host/port
python3 -m mcpgateway --host 0.0.0.0 --port 8080
```

### Code Formatting and Linting

```bash
# Auto-format code (run before committing)
make autoflake isort black pre-commit

# Comprehensive linting
make flake8 bandit interrogate pylint verify

# Quick lint for changed files only
make lint-changed

# Watch mode for auto-linting
make lint-watch

# Fix common issues automatically
make lint-fix
```

### Pre-commit Workflow

```bash
# Install git hooks
make pre-commit-install

# Run pre-commit checks manually
make pre-commit

# Complete quality pipeline (recommended before commits)
make autoflake isort black pre-commit
make doctest test htmlcov smoketest
make flake8 bandit interrogate pylint verify
```

## Code Quality

### Style Guidelines

- **Python**: PEP 8 with Black formatting (line length 200)
- **Type hints**: Required for all public APIs
- **Docstrings**: Google style, required for all public functions
- **Imports**: Organized with isort (black profile)
- **Naming**:
  - Functions/variables: `snake_case`
  - Classes: `PascalCase`
  - Constants: `UPPER_SNAKE_CASE`

### Quality Tools

```bash
# Format code
make black              # Python formatter
make isort              # Import sorter
make autoflake          # Remove unused imports

# Lint code
make flake8             # Style checker
make pylint             # Advanced linting
make mypy               # Type checking
make bandit             # Security analysis

# Documentation
make interrogate        # Docstring coverage
make doctest            # Test code examples

# All checks
make verify             # Run all quality checks
```

## Database Management

### Migrations with Alembic

```bash
# Create a new migration
alembic revision --autogenerate -m "Add new feature"

# Apply migrations
alembic upgrade head

# Rollback one revision
alembic downgrade -1

# Show migration history
alembic history

# Reset database (CAUTION: destroys data)
alembic downgrade base && alembic upgrade head
```

### Database Operations

```bash
# Different database backends
DATABASE_URL=sqlite:///./dev.db make dev           # SQLite
DATABASE_URL=postgresql://localhost/mcp make dev   # PostgreSQL
DATABASE_URL=mysql+pymysql://localhost/mcp make dev # MySQL

# Database utilities
python3 -m mcpgateway.cli db upgrade    # Apply migrations
python3 -m mcpgateway.cli db reset      # Reset database
python3 -m mcpgateway.cli db seed       # Seed test data
```

## API Development

### Adding New Endpoints

```python
# mcpgateway/main.py or separate router file
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from mcpgateway.database import get_db
from mcpgateway.schemas import MySchema

router = APIRouter(prefix="/api/v1")

@router.post("/my-endpoint", response_model=MySchema)
async def my_endpoint(
    data: MySchema,
    db: Session = Depends(get_db),
    current_user = Depends(get_current_user)
):
    """
    Endpoint description.

    Args:
        data: Input data
        db: Database session
        current_user: Authenticated user

    Returns:
        MySchema: Response data
    """
    # Implementation
    return result

# Register router in main.py
app.include_router(router, tags=["my-feature"])
```

### Schema Validation

```python
# mcpgateway/schemas.py
from pydantic import BaseModel, Field, validator

class MySchema(BaseModel):
    """Schema for my feature."""

    name: str = Field(..., min_length=1, max_length=255)
    value: int = Field(..., gt=0, le=100)

    @validator('name')
    def validate_name(cls, v):
        """Custom validation logic."""
        if not v.isalnum():
            raise ValueError('Name must be alphanumeric')
        return v

    class Config:
        """Pydantic config."""
        str_strip_whitespace = True
        use_enum_values = True
```

### Testing APIs

```python
# tests/integration/test_my_endpoint.py
import pytest
from fastapi.testclient import TestClient

def test_my_endpoint(test_client: TestClient, auth_headers):
    """Test my endpoint."""
    response = test_client.post(
        "/api/v1/my-endpoint",
        json={"name": "test", "value": 50},
        headers=auth_headers
    )
    assert response.status_code == 200
    assert response.json()["name"] == "test"
```

## Plugin Development

### Creating a Plugin

```yaml
# plugins/my_plugin/plugin-manifest.yaml
name: my_plugin
version: 1.0.0
description: Custom plugin for X functionality
enabled: true
hooks:
  - type: pre_request
    handler: my_plugin.hooks:pre_request_hook
  - type: post_response
    handler: my_plugin.hooks:post_response_hook
config:
  setting1: value1
  setting2: value2
```

```python
# plugins/my_plugin/hooks.py
from typing import Dict, Any
import logging

logger = logging.getLogger(__name__)

async def pre_request_hook(request: Dict[str, Any], config: Dict[str, Any]) -> Dict[str, Any]:
    """Process request before handling."""
    logger.info(f"Pre-request hook: {request.get('method')}")
    # Modify request if needed
    return request

async def post_response_hook(response: Dict[str, Any], config: Dict[str, Any]) -> Dict[str, Any]:
    """Process response before sending."""
    logger.info(f"Post-response hook: {response.get('result')}")
    # Modify response if needed
    return response
```

### Registering Plugins

```yaml
# plugins/config.yaml
plugins:
  - path: plugins/my_plugin
    enabled: true
    config:
      custom_setting: value
```

```bash
# Enable plugin system
export PLUGINS_ENABLED=true
export PLUGIN_CONFIG_FILE=plugins/config.yaml

# Test plugin
make dev
```

## Testing MCP Servers

### Using MCP Inspector

```bash
# Setup environment
export MCP_GATEWAY_BASE_URL=http://localhost:4444
export MCP_SERVER_URL=http://localhost:4444/servers/UUID/mcp
export MCP_AUTH="Bearer $(python3 -m mcpgateway.utils.create_jwt_token --username admin --exp 0 --secret my-test-key)"

# Launch Inspector with SSE (direct)
npx @modelcontextprotocol/inspector

# Launch with stdio wrapper
npx @modelcontextprotocol/inspector python3 -m mcpgateway.wrapper

# Open browser to http://localhost:5173
# Add server: http://localhost:4444/servers/UUID/sse
# Add header: Authorization: Bearer <token>
```

### Using mcpgateway.translate

```bash
# Expose stdio server over HTTP/SSE
python3 -m mcpgateway.translate \
    --stdio "uvx mcp-server-git" \
    --expose-sse \
    --port 9000

# Test with curl
curl http://localhost:9000/sse

# Register with gateway
curl -X POST http://localhost:4444/gateways \
    -H "Authorization: Bearer $TOKEN" \
    -H "Content-Type: application/json" \
    -d '{"name":"git_server","url":"http://localhost:9000/sse"}'
```

### Using SuperGateway Bridge

```bash
# Install and run SuperGateway
npm install -g supergateway
npx supergateway --stdio "uvx mcp-server-git"

# Register with ContextForge
curl -X POST http://localhost:4444/gateways \
    -H "Authorization: Bearer $TOKEN" \
    -H "Content-Type: application/json" \
    -d '{"name":"supergateway","url":"http://localhost:8000/sse"}'
```

## Debugging

### Debug Mode

```bash
# Enable debug mode
export DEBUG=true
export LOG_LEVEL=DEBUG
export DEV_MODE=true

# Run with debugger
python3 -m debugpy --listen 5678 --wait-for-client -m mcpgateway

# Or use IDE debugger with launch.json (VS Code)
```

### VS Code Configuration

```json
// .vscode/launch.json
{
    "version": "0.2.0",
    "configurations": [
        {
            "name": "Debug ContextForge",
            "type": "python",
            "request": "launch",
            "module": "mcpgateway",
            "args": ["--host", "0.0.0.0", "--port", "8000"],
            "env": {
                "DEBUG": "true",
                "LOG_LEVEL": "DEBUG",
                "ENVIRONMENT": "development"
            },
            "console": "integratedTerminal"
        }
    ]
}
```

### Logging

```python
# Add debug logging in code
import logging
logger = logging.getLogger(__name__)

def my_function():
    logger.debug(f"Debug info: {variable}")
    logger.info("Operation started")
    logger.warning("Potential issue")
    logger.error("Error occurred", exc_info=True)
```

```bash
# View logs
tail -f mcpgateway.log          # If LOG_TO_FILE=true
journalctl -u mcpgateway -f     # Systemd service
docker logs -f mcpgateway        # Docker container
```

### Request Tracing

```bash
# Enable OpenTelemetry tracing
export OTEL_ENABLE_OBSERVABILITY=true
export OTEL_TRACES_EXPORTER=console  # Or otlp, jaeger

# Run with tracing
make dev

# View traces in console or tracing backend
```

### Database Debugging

```bash
# Enable SQL echo
export DATABASE_ECHO=true

# Query database directly
sqlite3 mcp.db "SELECT * FROM tools LIMIT 10;"
psql mcp -c "SELECT * FROM servers;"

# Database profiling
python3 -m mcpgateway.utils.db_profiler
```

## Performance Optimization

### Profiling

```python
# Profile code execution
import cProfile
import pstats

def profile_function():
    profiler = cProfile.Profile()
    profiler.enable()

    # Code to profile
    expensive_operation()

    profiler.disable()
    stats = pstats.Stats(profiler)
    stats.sort_stats('cumulative')
    stats.print_stats(10)
```

### Caching Strategies

```python
# Use Redis caching
from mcpgateway.cache import cache_get, cache_set

async def get_expensive_data(key: str):
    # Try cache first
    cached = await cache_get(f"data:{key}")
    if cached:
        return cached

    # Compute if not cached
    result = expensive_computation()
    await cache_set(f"data:{key}", result, ttl=3600)
    return result
```

### Database Optimization

```python
# Use eager loading to avoid N+1 queries
from sqlalchemy.orm import joinedload

def get_servers_with_tools(db: Session):
    return db.query(Server)\
        .options(joinedload(Server.tools))\
        .all()

# Use bulk operations
def bulk_insert_tools(db: Session, tools: List[Dict]):
    db.bulk_insert_mappings(Tool, tools)
    db.commit()
```

### Async Best Practices

```python
# Use async/await properly
import asyncio
from typing import List

async def process_items(items: List[str]):
    # Process concurrently
    tasks = [process_item(item) for item in items]
    results = await asyncio.gather(*tasks)
    return results

# Use connection pooling
from aiohttp import ClientSession

async def make_requests():
    async with ClientSession() as session:
        # Reuse session for multiple requests
        async with session.get(url1) as resp1:
            data1 = await resp1.json()
        async with session.get(url2) as resp2:
            data2 = await resp2.json()
```

## Contributing

### Development Process

1. **Fork and clone** the repository
2. **Create a feature branch**: `git checkout -b feature/my-feature`
3. **Set up environment**: `make venv install-dev`
4. **Make changes** and write tests
5. **Run quality checks**: `make verify`
6. **Commit with sign-off**: `git commit -s -m "feat: add new feature"`
7. **Push and create PR**: `git push origin feature/my-feature`

### Commit Guidelines

Follow [Conventional Commits](https://www.conventionalcommits.org/):

- `feat:` New feature
- `fix:` Bug fix
- `docs:` Documentation changes
- `style:` Code style changes (formatting, etc.)
- `refactor:` Code refactoring
- `test:` Test additions or changes
- `chore:` Build process or auxiliary tool changes

### Code Review Process

1. **Self-review** your changes
2. **Run all tests**: `make test`
3. **Update documentation** if needed
4. **Ensure CI passes**
5. **Address review feedback**
6. **Squash commits** if requested

### Getting Help

- **Documentation**: [docs/](docs/)
- **Issues**: [GitHub Issues](https://github.com/IBM/mcp-context-forge/issues)
- **Discussions**: [GitHub Discussions](https://github.com/IBM/mcp-context-forge/discussions)
- **Contributing Guide**: [CONTRIBUTING.md](CONTRIBUTING.md)

## Advanced Topics

### Multi-tenancy Development

```python
# Implement tenant isolation
from mcpgateway.auth import get_current_tenant

@router.get("/tenant-data")
async def get_tenant_data(
    tenant = Depends(get_current_tenant),
    db: Session = Depends(get_db)
):
    # Filter by tenant
    return db.query(Model).filter(Model.tenant_id == tenant.id).all()
```

### Custom Transport Implementation

```python
# mcpgateway/transports/custom_transport.py
from mcpgateway.transports.base import BaseTransport

class CustomTransport(BaseTransport):
    """Custom transport implementation."""

    async def connect(self, url: str):
        """Establish connection."""
        # Implementation

    async def send(self, message: dict):
        """Send message."""
        # Implementation

    async def receive(self) -> dict:
        """Receive message."""
        # Implementation
```

### Federation Development

```python
# Test federation locally
# Start multiple instances
PORT=4444 make dev  # Instance 1
PORT=4445 make dev  # Instance 2

# Register peers
curl -X POST http://localhost:4444/gateways \
    -H "Authorization: Bearer $TOKEN" \
    -d '{"name":"peer2","url":"http://localhost:4445/sse"}'
```

## Security Considerations

### Authentication Testing

```bash
# Generate test tokens
python3 -m mcpgateway.utils.create_jwt_token \
    --username test@example.com \
    --exp 60 \
    --secret test-key

# Test with different auth methods
curl -H "Authorization: Bearer $TOKEN" http://localhost:4444/api/test
curl -u admin:changeme http://localhost:4444/api/test
```

### Security Scanning

```bash
# Static analysis
make bandit

# Dependency scanning
make security-scan

# OWASP checks
pip install safety
safety check
```

## Troubleshooting

### Common Issues

1. **Import errors**: Ensure package installed with `pip install -e .`
2. **Database locked**: Use PostgreSQL for concurrent access
3. **Port in use**: Change with `PORT=8001 make dev`
4. **Missing dependencies**: Run `make install-dev`
5. **Permission errors**: Check file permissions and user context

### Debug Commands

```bash
# Check environment
make check-env

# Verify installation
python3 -c "import mcpgateway; print(mcpgateway.__version__)"

# Test configuration
python3 -m mcpgateway.config

# Database status
alembic current

# Clear caches
redis-cli FLUSHDB
```

## Resources

- [MCP Specification](https://modelcontextprotocol.io/)
- [FastAPI Documentation](https://fastapi.tiangolo.com/)
- [SQLAlchemy Documentation](https://docs.sqlalchemy.org/)
- [Pydantic Documentation](https://docs.pydantic.dev/)
- [HTMX Documentation](https://htmx.org/)
- [Alpine.js Documentation](https://alpinejs.dev/)