Best Practices
Development best practices and patterns for MCP.
Follow these core principles when designing your MCP servers: 1. Single Responsibility: Each server should have a focused, well-defined purpose 2. Interface Segregation: Create specific interfaces for different use cases 3. Dependency Injection: Inject dependencies rather than creating them internally 4. Error Handling: Implement consistent error handling across all methods 5. Configuration Management: Externalize configuration for flexibility
# Good: Single responsibility
class WeatherServer(Server):
def __init__(self):
super().__init__()
self.register_tool_handler("get_weather", self.get_weather)
self.register_tool_handler("get_forecast", self.get_forecast)
# Bad: Mixed responsibilities
class WeatherAndDatabaseServer(Server):
def __init__(self):
super().__init__()
self.register_tool_handler("get_weather", self.get_weather)
self.register_tool_handler("query_users", self.query_users) # Wrong!Implement robust error handling to ensure your servers are reliable and provide meaningful feedback to clients. Key strategies: - Use structured error responses - Classify errors by type and severity - Implement graceful degradation - Avoid exposing sensitive information - Log errors for debugging
async def safe_tool_call(self, request: CallToolRequest) -> CallToolResult:
try:
# Validate input
if not request.arguments.get("required_field"):
return CallToolResult(
content=[{
"type": "text",
"text": "Error: required_field is missing"
}],
isError=True
)
# Perform operation
result = await self.perform_operation(request.arguments)
return CallToolResult(
content=[{
"type": "text",
"text": f"Success: {result}"
}]
)
except Exception as e:
# Log error for debugging
logger.error(f"Tool call failed: {e}")
return CallToolResult(
content=[{
"type": "text",
"text": "An error occurred while processing your request"
}],
isError=True
)Optimize your MCP servers for better performance and scalability. Key optimization techniques: - Implement caching strategies - Use async/await patterns - Optimize database queries - Implement connection pooling - Monitor and profile performance
import asyncio
from functools import lru_cache
class OptimizedServer(Server):
def __init__(self):
super().__init__()
self.cache = {}
self.register_tool_handler("get_data", self.get_data)
@lru_cache(maxsize=100)
def expensive_calculation(self, input_data):
# Cache expensive calculations
return complex_calculation(input_data)
async def get_data(self, request: CallToolRequest) -> CallToolResult:
# Use connection pooling
async with self.db_pool.acquire() as conn:
result = await conn.fetch(
"SELECT * FROM data WHERE id = $1",
request.arguments.get("id")
)
return CallToolResult(
content=[{
"type": "text",
"text": f"Data: {result}"
}]
)Implement security measures to protect your MCP servers and the data they handle. Security considerations: - Validate all inputs - Implement authentication and authorization - Use secure communication channels - Handle sensitive data carefully - Regular security audits
import hashlib
import secrets
class SecureServer(Server):
def __init__(self):
super().__init__()
self.api_keys = set() # Store valid API keys
self.register_tool_handler("secure_operation", self.secure_operation)
def validate_api_key(self, api_key: str) -> bool:
return api_key in self.api_keys
def sanitize_input(self, input_data: str) -> str:
# Remove potentially dangerous characters
return input_data.replace("<", "<").replace(">", ">")
async def secure_operation(self, request: CallToolRequest) -> CallToolResult:
# Validate API key
api_key = request.arguments.get("api_key")
if not self.validate_api_key(api_key):
return CallToolResult(
content=[{"type": "text", "text": "Invalid API key"}],
isError=True
)
# Sanitize input
user_input = self.sanitize_input(request.arguments.get("input", ""))
# Perform secure operation
result = await self.perform_secure_operation(user_input)
return CallToolResult(
content=[{"type": "text", "text": f"Secure result: {result}"}]
)Implement comprehensive testing to ensure your MCP servers work correctly and reliably. Testing approaches: - Unit tests for individual methods - Integration tests for server interactions - Performance tests for load handling - Security tests for vulnerabilities - Automated testing in CI/CD
import pytest
import asyncio
from unittest.mock import Mock
class TestMCPServer:
@pytest.fixture
async def server(self):
return MyMCPServer()
@pytest.fixture
async def session(self, server):
transport = Mock()
session = ClientSession(transport)
return session
async def test_list_tools(self, server):
request = ListToolsRequest()
result = await server.list_tools(request)
assert len(result.tools) > 0
assert any(tool.name == "hello" for tool in result.tools)
async def test_tool_call(self, server):
request = CallToolRequest(
name="hello",
arguments={"name": "Test"}
)
result = await server.hello(request)
assert result.content[0]["text"] == "Hello, Test! Welcome to MCP."
async def test_error_handling(self, server):
request = CallToolRequest(
name="hello",
arguments={"invalid": "data"}
)
result = await server.hello(request)
# Should handle gracefully
assert "Hello, World" in result.content[0]["text"]Server Development Guide
Building MCP servers
API Reference
Complete API documentation