DEVELOPMENT.mdโข10.9 kB
# Development Guide - MCP Wikipedia Server
This guide provides detailed information for developers who want to contribute to or extend the MCP Wikipedia Server project.
## ๐๏ธ Architecture Overview
### Core Components
```
โโโโโโโโโโโโโโโโโโโ โโโโโโโโโโโโโโโโโโโ โโโโโโโโโโโโโโโโโโโ
โ MCP Client โโโโโโ MCP Server โโโโโโ Wikipedia API โ
โ (AI Assistant) โ โ (FastMCP) โ โ (wikipedia) โ
โโโโโโโโโโโโโโโโโโโ โโโโโโโโโโโโโโโโโโโ โโโโโโโโโโโโโโโโโโโ
```
### Technology Stack
- **Protocol**: Model Context Protocol (MCP) v1.0
- **Framework**: FastMCP for server implementation
- **Language**: Python 3.11+ (with type hints)
- **API**: Wikipedia Python library
- **Async**: Full asynchronous support with asyncio
## ๐ง Development Setup
### 1. Development Environment
```bash
# Clone the repository
git clone <repository-url>
cd MCPClientServer
# Set up development environment
python -m venv .venv311-dev
source .venv311-dev/bin/activate
# Install development dependencies
pip install --upgrade pip
pip install wikipedia mcp fastmcp
pip install pytest pytest-asyncio black flake8 mypy
```
### 2. IDE Configuration
#### VS Code Settings
```json
{
"python.pythonPath": "./.venv311-dev/bin/python",
"python.linting.enabled": true,
"python.linting.flake8Enabled": true,
"python.formatting.provider": "black",
"python.testing.pytestEnabled": true
}
```
## ๐ Code Structure
### Server Implementation (`mcp_server.py`)
```python
from fastmcp import FastMCP
from typing import Dict, Any, List
import wikipedia
class WikipediaServer:
"""Main MCP server class for Wikipedia tools."""
def __init__(self):
self.app = FastMCP("Wikipedia Tools")
self._register_tools()
def _register_tools(self):
"""Register all available tools with the MCP server."""
# Tool registration logic
```
### Key Design Patterns
1. **Singleton Pattern**: Server instance management
2. **Factory Pattern**: Tool creation and registration
3. **Strategy Pattern**: Different search strategies
4. **Observer Pattern**: Event handling and logging
## ๐ ๏ธ Adding New Tools
### Step 1: Define Tool Schema
```python
from fastmcp import FastMCP
@app.tool()
async def new_wikipedia_tool(
query: str,
limit: int = 5
) -> Dict[str, Any]:
"""
Description of what this tool does.
Args:
query: Search query parameter
limit: Maximum number of results
Returns:
Dictionary containing tool results
"""
# Implementation here
pass
```
### Step 2: Implement Tool Logic
```python
async def new_wikipedia_tool(query: str, limit: int = 5) -> Dict[str, Any]:
try:
# Your tool implementation
results = await some_async_operation(query, limit)
return {
"success": True,
"data": results,
"metadata": {
"query": query,
"limit": limit,
"timestamp": datetime.now().isoformat()
}
}
except Exception as e:
return {
"success": False,
"error": str(e),
"metadata": {"query": query}
}
```
### Step 3: Add Tests
```python
import pytest
from mcp_server import WikipediaServer
@pytest.mark.asyncio
async def test_new_wikipedia_tool():
server = WikipediaServer()
result = await server.new_wikipedia_tool("test query")
assert result["success"] is True
assert "data" in result
assert result["metadata"]["query"] == "test query"
```
## ๐งช Testing Framework
### Running Tests
```bash
# Run all tests
python -m pytest tests/ -v
# Run specific test file
python -m pytest tests/test_server.py -v
# Run with coverage
pip install pytest-cov
python -m pytest tests/ --cov=src --cov-report=html
```
### Test Categories
1. **Unit Tests**: Individual tool functionality
2. **Integration Tests**: MCP protocol compliance
3. **Performance Tests**: Response time and throughput
4. **Error Handling Tests**: Edge cases and failures
### Example Test Structure
```python
import pytest
import asyncio
from unittest.mock import Mock, patch
from src.mcp_server.mcp_server import WikipediaServer
class TestWikipediaServer:
@pytest.fixture
def server(self):
return WikipediaServer()
@pytest.mark.asyncio
async def test_fetch_wikipedia_info_success(self, server):
"""Test successful Wikipedia search."""
result = await server.fetch_wikipedia_info("Python programming")
assert result["success"] is True
assert "summary" in result["data"]
assert len(result["data"]["summary"]) > 0
@pytest.mark.asyncio
async def test_fetch_wikipedia_info_not_found(self, server):
"""Test handling of non-existent articles."""
result = await server.fetch_wikipedia_info("NonExistentArticle12345")
# Should handle gracefully
assert "error" in result or result["data"]["summary"] == ""
```
## ๐ Debugging and Logging
### Development Logging
```python
import logging
import sys
# Configure development logging
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('mcp_server.log'),
logging.StreamHandler(sys.stdout)
]
)
logger = logging.getLogger(__name__)
```
### Debug Mode
```python
# Enable debug mode in development
DEBUG = True
if DEBUG:
# Additional debug information
logger.debug(f"Processing request: {request}")
logger.debug(f"Wikipedia API response: {response}")
```
## ๐ Performance Optimization
### Async Best Practices
```python
import asyncio
import aiohttp
from typing import List, Dict
async def batch_wikipedia_requests(queries: List[str]) -> List[Dict]:
"""Process multiple Wikipedia queries concurrently."""
async def fetch_single(query: str) -> Dict:
# Single request implementation
pass
# Process requests concurrently
tasks = [fetch_single(query) for query in queries]
results = await asyncio.gather(*tasks, return_exceptions=True)
return results
```
### Caching Strategy
```python
from functools import lru_cache
import time
class WikipediaCache:
def __init__(self, ttl: int = 300): # 5 minutes TTL
self.cache = {}
self.ttl = ttl
def get(self, key: str):
if key in self.cache:
data, timestamp = self.cache[key]
if time.time() - timestamp < self.ttl:
return data
else:
del self.cache[key]
return None
def set(self, key: str, value):
self.cache[key] = (value, time.time())
```
## ๐ Deployment Considerations
### Production Configuration
```python
# production_config.py
import os
PRODUCTION_CONFIG = {
"host": os.getenv("MCP_HOST", "localhost"),
"port": int(os.getenv("MCP_PORT", 8000)),
"workers": int(os.getenv("MCP_WORKERS", 4)),
"timeout": int(os.getenv("MCP_TIMEOUT", 30)),
"max_requests": int(os.getenv("MCP_MAX_REQUESTS", 1000)),
"log_level": os.getenv("MCP_LOG_LEVEL", "INFO")
}
```
### Docker Support
```dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY src/ ./src/
COPY pyproject.toml .
EXPOSE 8000
CMD ["python", "src/mcp_server/mcp_server.py"]
```
## ๐ง Code Quality
### Pre-commit Hooks
```bash
# Install pre-commit
pip install pre-commit
# Set up pre-commit hooks
pre-commit install
```
Create `.pre-commit-config.yaml`:
```yaml
repos:
- repo: https://github.com/psf/black
rev: 23.3.0
hooks:
- id: black
language_version: python3.11
- 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
```
### Code Style Guidelines
1. **PEP 8**: Follow Python style guide
2. **Type Hints**: Use comprehensive type annotations
3. **Docstrings**: Google-style docstrings for all functions
4. **Error Handling**: Explicit exception handling
5. **Async/Await**: Proper async patterns throughout
## ๐ Monitoring and Metrics
### Basic Metrics Collection
```python
import time
from collections import defaultdict
class MetricsCollector:
def __init__(self):
self.request_count = defaultdict(int)
self.response_times = defaultdict(list)
self.error_count = defaultdict(int)
def record_request(self, tool_name: str, response_time: float, success: bool):
self.request_count[tool_name] += 1
self.response_times[tool_name].append(response_time)
if not success:
self.error_count[tool_name] += 1
def get_stats(self) -> Dict:
return {
"total_requests": sum(self.request_count.values()),
"average_response_time": {
tool: sum(times) / len(times)
for tool, times in self.response_times.items()
},
"error_rates": {
tool: self.error_count[tool] / self.request_count[tool]
for tool in self.request_count.keys()
}
}
```
## ๐ค Contributing Guidelines
### Pull Request Process
1. **Fork** the repository
2. **Create** a feature branch from `main`
3. **Implement** your changes with tests
4. **Run** the full test suite
5. **Format** code with black
6. **Submit** pull request with clear description
### Commit Message Format
```
type(scope): brief description
Detailed explanation of changes if needed.
- Bullet points for specific changes
- Reference issues with #123
```
Types: `feat`, `fix`, `docs`, `style`, `refactor`, `test`, `chore`
## ๐ Resources for Contributors
- [MCP Specification](https://spec.modelcontextprotocol.io/)
- [FastMCP Documentation](https://github.com/jlowin/fastmcp)
- [Wikipedia API Docs](https://wikipedia.readthedocs.io/)
- [Python Async Programming](https://docs.python.org/3/library/asyncio.html)
- [pytest Documentation](https://docs.pytest.org/)
## ๐ Common Development Issues
| Issue | Solution |
|-------|---------|
| Import errors in tests | Ensure PYTHONPATH includes src directory |
| Async test failures | Use `@pytest.mark.asyncio` decorator |
| Wikipedia API rate limits | Implement proper delays and retry logic |
| Type checking errors | Add proper type hints and mypy configuration |
---
Happy coding! ๐