How to Build an MCP Server: Step-by-Step Tutorial

Prerequisites

Before we begin, make sure you have the following:

• Python 3.8+ installed on your system
• Basic understanding of Python programming
• Familiarity with JSON and REST APIs
• A code editor (VS Code, PyCharm, etc.)
• Git for version control

Understanding MCP Server Architecture

An MCP server consists of several key components that work together to provide AI models with access to external data and tools:

1. Server Interface

The server interface defines how your MCP server communicates with AI models. It handles:

• Request/response processing
• Authentication and authorization
• Error handling and validation
• Capability discovery

2. Resource Handlers

Resource handlers implement the actual functionality your server provides. These can include:

• Database connections and queries
• File system operations
• API integrations
• Tool execution
• Data transformation

3. Configuration Management

Configuration management handles server settings, environment variables, and runtime configuration.

Step 1: Set Up Your Development Environment

Let's start by setting up a clean development environment for your MCP server:

mkdir my-mcp-server
cd my-mcp-server
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate
pip install mcp-server-sdk
pip install fastapi uvicorn

Step 2: Create Your First MCP Server

Let's create a simple MCP server that provides weather information. This will demonstrate the core concepts:

from mcp.server import Server
from mcp.server.models import InitializationOptions
from mcp.server.stdio import stdio_server
import asyncio
import json

class WeatherMCPServer(Server):
    def __init__(self):
        super().__init__()
        
    async def initialize(self, options: InitializationOptions) -> None:
        """Initialize the server with capabilities."""
        await super().initialize(options)
        
        # Register our weather tool
        self.register_tool(
            "get_weather",
            "Get current weather information for a location",
            {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "City name or coordinates"
                    }
                },
                "required": ["location"]
            }
        )
    
    async def call_tool(self, name: str, arguments: dict) -> str:
        """Handle tool calls from the AI model."""
        if name == "get_weather":
            location = arguments.get("location", "Unknown")
            # In a real implementation, you would call a weather API
            return json.dumps({
                "location": location,
                "temperature": "22°C",
                "condition": "Sunny",
                "humidity": "65%"
            })
        
        raise ValueError(f"Unknown tool: {name}")

async def main():
    server = WeatherMCPServer()
    async with stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            InitializationOptions(
                server_name="weather-mcp-server",
                server_version="1.0.0",
                capabilities=server.get_capabilities()
            )
        )

if __name__ == "__main__":
    asyncio.run(main())

Step 3: Add Advanced Features

Now let's enhance our server with more advanced features like error handling, logging, and configuration management:

import logging
import os
from typing import Optional
from mcp.server import Server
from mcp.server.models import InitializationOptions
from mcp.server.stdio import stdio_server
import asyncio
import json

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class EnhancedWeatherMCPServer(Server):
    def __init__(self):
        super().__init__()
        self.api_key = os.getenv("WEATHER_API_KEY")
        self.base_url = "https://api.weatherapi.com/v1"
        
    async def initialize(self, options: InitializationOptions) -> None:
        """Initialize the server with enhanced capabilities."""
        await super().initialize(options)
        
        logger.info("Initializing Enhanced Weather MCP Server")
        
        # Register multiple tools
        self.register_tool(
            "get_current_weather",
            "Get current weather conditions for a location",
            {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "City name or coordinates"
                    }
                },
                "required": ["location"]
            }
        )
        
        self.register_tool(
            "get_weather_forecast",
            "Get weather forecast for a location",
            {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "City name or coordinates"
                    },
                    "days": {
                        "type": "integer",
                        "description": "Number of days to forecast (1-7)",
                        "default": 3
                    }
                },
                "required": ["location"]
            }
        )
    
    async def call_tool(self, name: str, arguments: dict) -> str:
        """Handle tool calls with enhanced error handling."""
        try:
            logger.info(f"Tool call: {name} with arguments: {arguments}")
            
            if name == "get_current_weather":
                return await self._get_current_weather(arguments)
            elif name == "get_weather_forecast":
                return await self._get_weather_forecast(arguments)
            else:
                raise ValueError(f"Unknown tool: {name}")
                
        except Exception as e:
            logger.error(f"Error in tool call {name}: {str(e)}")
            return json.dumps({
                "error": str(e),
                "status": "error"
            })
    
    async def _get_current_weather(self, arguments: dict) -> str:
        """Get current weather for a location."""
        location = arguments.get("location")
        if not location:
            raise ValueError("Location is required")
            
        # In a real implementation, you would make an API call here
        # For this example, we'll return mock data
        weather_data = {
            "location": location,
            "current": {
                "temperature": "22°C",
                "condition": "Sunny",
                "humidity": "65%",
                "wind_speed": "12 km/h"
            },
            "timestamp": "2024-01-20T10:30:00Z"
        }
        
        return json.dumps(weather_data)
    
    async def _get_weather_forecast(self, arguments: dict) -> str:
        """Get weather forecast for a location."""
        location = arguments.get("location")
        days = arguments.get("days", 3)
        
        if not location:
            raise ValueError("Location is required")
            
        # Mock forecast data
        forecast_data = {
            "location": location,
            "forecast": [
                {
                    "date": "2024-01-21",
                    "high": "24°C",
                    "low": "18°C",
                    "condition": "Partly Cloudy"
                },
                {
                    "date": "2024-01-22",
                    "high": "26°C",
                    "low": "20°C",
                    "condition": "Sunny"
                },
                {
                    "date": "2024-01-23",
                    "high": "23°C",
                    "low": "17°C",
                    "condition": "Rainy"
                }
            ]
        }
        
        return json.dumps(forecast_data)

async def main():
    server = EnhancedWeatherMCPServer()
    async with stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            InitializationOptions(
                server_name="enhanced-weather-mcp-server",
                server_version="1.0.0",
                capabilities=server.get_capabilities()
            )
        )

if __name__ == "__main__":
    asyncio.run(main())

Step 4: Add Configuration and Environment Management

Let's add proper configuration management to make our server more flexible and secure:

# config.py
import os
from dataclasses import dataclass
from typing import Optional

@dataclass
class ServerConfig:
    server_name: str
    server_version: str
    api_key: Optional[str] = None
    base_url: str = "https://api.weatherapi.com/v1"
    log_level: str = "INFO"
    max_requests_per_minute: int = 60
    
    @classmethod
    def from_env(cls) -> "ServerConfig":
        return cls(
            server_name=os.getenv("MCP_SERVER_NAME", "weather-mcp-server"),
            server_version=os.getenv("MCP_SERVER_VERSION", "1.0.0"),
            api_key=os.getenv("WEATHER_API_KEY"),
            base_url=os.getenv("WEATHER_API_BASE_URL", "https://api.weatherapi.com/v1"),
            log_level=os.getenv("LOG_LEVEL", "INFO"),
            max_requests_per_minute=int(os.getenv("MAX_REQUESTS_PER_MINUTE", "60"))
        )

# .env file example:
# MCP_SERVER_NAME=my-weather-server
# MCP_SERVER_VERSION=1.0.0
# WEATHER_API_KEY=your_api_key_here
# LOG_LEVEL=DEBUG
# MAX_REQUESTS_PER_MINUTE=100

Step 5: Testing Your MCP Server

Testing is crucial for ensuring your MCP server works correctly. Let's create a comprehensive test suite:

# test_weather_server.py
import pytest
import json
import asyncio
from unittest.mock import patch
from your_server import EnhancedWeatherMCPServer

class TestWeatherMCPServer:
    @pytest.fixture
    async def server(self):
        return EnhancedWeatherMCPServer()
    
    @pytest.mark.asyncio
    async def test_get_current_weather(self, server):
        """Test current weather functionality."""
        arguments = {"location": "New York"}
        result = await server._get_current_weather(arguments)
        
        data = json.loads(result)
        assert data["location"] == "New York"
        assert "current" in data
        assert "temperature" in data["current"]
    
    @pytest.mark.asyncio
    async def test_get_weather_forecast(self, server):
        """Test weather forecast functionality."""
        arguments = {"location": "London", "days": 3}
        result = await server._get_weather_forecast(arguments)
        
        data = json.loads(result)
        assert data["location"] == "London"
        assert len(data["forecast"]) == 3
    
    @pytest.mark.asyncio
    async def test_missing_location(self, server):
        """Test error handling for missing location."""
        arguments = {}
        
        with pytest.raises(ValueError, match="Location is required"):
            await server._get_current_weather(arguments)
    
    @pytest.mark.asyncio
    async def test_unknown_tool(self, server):
        """Test handling of unknown tools."""
        result = await server.call_tool("unknown_tool", {})
        data = json.loads(result)
        
        assert data["status"] == "error"
        assert "Unknown tool" in data["error"]

Step 6: Deployment and Distribution

Once your MCP server is working correctly, you'll want to deploy it and make it available to others:

Package Your Server

Create a proper Python package for distribution:

from setuptools import setup, find_packages

setup(
    name="weather-mcp-server",
    version="1.0.0",
    description="A weather MCP server for AI models",
    author="Your Name",
    author_email="your.email@example.com",
    packages=find_packages(),
    install_requires=[
        "mcp-server-sdk>=1.0.0",
        "requests>=2.25.0",
        "python-dotenv>=0.19.0",
    ],
    entry_points={
        "console_scripts": [
            "weather-mcp-server=weather_mcp_server.main:main",
        ],
    },
    python_requires=">=3.8",
    classifiers=[
        "Development Status :: 4 - Beta",
        "Intended Audience :: Developers",
        "License :: OSI Approved :: MIT License",
        "Programming Language :: Python :: 3",
        "Programming Language :: Python :: 3.8",
        "Programming Language :: Python :: 3.9",
        "Programming Language :: Python :: 3.10",
        "Programming Language :: Python :: 3.11",
    ],
)

Docker Deployment

Create a Docker container for easy deployment:

FROM python:3.11-slim

WORKDIR /app

# Copy requirements and install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy server code
COPY . .

# Set environment variables
ENV PYTHONPATH=/app
ENV LOG_LEVEL=INFO

# Run the server
CMD ["python", "-m", "weather_mcp_server.main"]

Best Practices for MCP Server Development

🔒 Security Considerations

• Always validate and sanitize input data
• Implement proper authentication and authorization
• Use environment variables for sensitive configuration
• Implement rate limiting to prevent abuse
• Log security-relevant events

⚡ Performance Optimization

• Implement caching for frequently accessed data
• Use connection pooling for database connections
• Implement async/await patterns for I/O operations
• Monitor and optimize memory usage
• Use appropriate data structures for your use case

🧪 Testing and Quality Assurance

• Write comprehensive unit tests
• Implement integration tests
• Use mocking for external dependencies
• Test error conditions and edge cases
• Implement continuous integration

📚 Documentation

• Write clear API documentation
• Include usage examples
• Document configuration options
• Provide troubleshooting guides
• Keep documentation up to date

Next Steps

Congratulations! You've successfully built your first MCP server. Here are some suggestions for what to do next: