Building a Model Context Protocol Server in Python: A Complete Guide

The Model Context Protocol (MCP) represents a significant advancement in how we interact with AI models. By creating custom MCP servers, developers can extend AI capabilities with specialized tools and resources, enabling more powerful and context-aware applications. This comprehensive guide will walk you through building your own MCP server in Python, from basic concepts to advanced implementations.

What is the Model Context Protocol?

The Model Context Protocol is a standardized way for AI systems to communicate with external tools and resources. An MCP server acts as a bridge between AI models and various capabilities, such as:

• Accessing specialized databases or knowledge bases
• Performing complex calculations or simulations
• Interacting with external APIs and services
• Processing and analyzing data in specialized ways
• Generating content like images, code, or other media

Why Build an MCP Server in Python?

Python offers several advantages for developing MCP servers:

• Rich Ecosystem: Access to thousands of libraries for data processing, API integration, and more
• Simplicity: Clean syntax and readability make development faster
• AI/ML Integration: Seamless compatibility with popular AI and machine learning frameworks
• Cross-Platform: Works across different operating systems
• Community Support: Vast community and resources for troubleshooting

Prerequisites for Building an MCP Server

Before diving into development, ensure you have:

1. Python Environment: Python 3.8+ installed
2. Development Tools: A code editor (VS Code, PyCharm, etc.)
3. Basic Knowledge: Familiarity with Python, APIs, and web servers
4. Understanding of MCP: Basic knowledge of how the Model Context Protocol works
5. Required Libraries: fastapi, uvicorn, pydantic, and other dependencies

MCP Server Architecture Overview

An MCP server typically consists of these core components:

1. Server Framework: Handles HTTP/WebSocket communication
2. Tool Definitions: Specifies the capabilities your server provides
3. Resource Providers: Manages access to data sources
4. Execution Engine: Processes requests and executes tool functions
5. Authentication & Security: Ensures secure access to your server

Step-by-Step Guide to Building an MCP Server

Step 1: Setting Up Your Project (15 minutes)

Start by creating a new project directory and setting up a virtual environment:

# Create project directory
mkdir mcp-server
cd mcp-server

# Create and activate virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Install required packages
pip install fastapi uvicorn pydantic python-dotenv

Create a basic project structure:

mcp-server/
├── venv/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── tools/
│   │   ├── __init__.py
│   │   └── base.py
│   ├── resources/
│   │   ├── __init__.py
│   │   └── base.py
│   └── config.py
├── .env
└── requirements.txt

Step 2: Defining the MCP Server Configuration (30 minutes)

Create a configuration file to manage your server settings:

# app/config.py
import os
from pydantic import BaseSettings
from dotenv import load_dotenv

load_dotenv()

class Settings(BaseSettings):
    APP_NAME: str = "Python MCP Server"
    APP_VERSION: str = "0.1.0"
    APP_DESCRIPTION: str = "A Model Context Protocol server implemented in Python"

    # Server configuration
    HOST: str = os.getenv("HOST", "0.0.0.0")
    PORT: int = int(os.getenv("PORT", "8000"))

    # Security settings
    API_KEY: str = os.getenv("API_KEY", "")
    ENABLE_AUTH: bool = os.getenv("ENABLE_AUTH", "True").lower() == "true"

    # Logging
    LOG_LEVEL: str = os.getenv("LOG_LEVEL", "INFO")

settings = Settings()

Step 3: Creating the Base Tool and Resource Classes (45 minutes)

Define the base classes for tools and resources:

# app/tools/base.py
from typing import Dict, Any, List, Optional
from pydantic import BaseModel, Field

class ToolParameter(BaseModel):
    name: str
    type: str
    description: str
    required: bool = True
    default: Optional[Any] = None

class ToolSchema(BaseModel):
    name: str
    description: str
    parameters: List[ToolParameter]

class BaseTool:
    """Base class for all MCP tools"""

    @property
    def schema(self) -> ToolSchema:
        """Return the tool's schema definition"""
        raise NotImplementedError("Tool must implement schema property")

    async def execute(self, params: Dict[str, Any]) -> Any:
        """Execute the tool with the provided parameters"""
        raise NotImplementedError("Tool must implement execute method")

# app/resources/base.py
from typing import Dict, Any, Optional
from pydantic import BaseModel

class ResourceSchema(BaseModel):
    name: str
    description: str
    uri_format: str

class BaseResource:
    """Base class for all MCP resources"""

    @property
    def schema(self) -> ResourceSchema:
        """Return the resource's schema definition"""
        raise NotImplementedError("Resource must implement schema property")

    async def get(self, uri: str, params: Optional[Dict[str, Any]] = None) -> Any:
        """Retrieve the resource identified by the URI"""
        raise NotImplementedError("Resource must implement get method")

Step 4: Implementing a Sample Tool (1 hour)

Let's create a simple weather forecast tool:

# app/tools/weather.py
from typing import Dict, Any, List
import aiohttp
from app.tools.base import BaseTool, ToolSchema, ToolParameter

class WeatherForecastTool(BaseTool):
    """Tool for fetching weather forecasts"""

    @property
    def schema(self) -> ToolSchema:
        return ToolSchema(
            name="get_forecast",
            description="Get weather forecast for a specific location",
            parameters=[
                ToolParameter(
                    name="city",
                    type="string",
                    description="City name (e.g., 'San Francisco')",
                    required=True
                ),
                ToolParameter(
                    name="days",
                    type="integer",
                    description="Number of days to forecast",
                    required=False,
                    default=5
                )
            ]
        )

    async def execute(self, params: Dict[str, Any]) -> Any:
        """Execute the weather forecast tool"""
        city = params.get("city")
        days = params.get("days", 5)

        # In a real implementation, you would call a weather API
        # This is a simplified example
        async with aiohttp.ClientSession() as session:
            # Replace with actual weather API endpoint
            api_url = f"https://api.weatherapi.com/v1/forecast.json?key=YOUR_API_KEY&q={city}&days={days}"

            async with session.get(api_url) as response:
                if response.status == 200:
                    data = await response.json()
                    return {
                        "location": data["location"]["name"],
                        "country": data["location"]["country"],
                        "current": {
                            "temp_c": data["current"]["temp_c"],
                            "condition": data["current"]["condition"]["text"]
                        },
                        "forecast": [
                            {
                                "date": day["date"],
                                "max_temp_c": day["day"]["maxtemp_c"],
                                "min_temp_c": day["day"]["mintemp_c"],
                                "condition": day["day"]["condition"]["text"]
                            }
                            for day in data["forecast"]["forecastday"]
                        ]
                    }
                else:
                    return {"error": f"Failed to fetch weather data: {response.status}"}

Step 5: Implementing a Sample Resource (1 hour)

Now, let's create a resource for accessing weather data:

# app/resources/weather.py
from typing import Dict, Any, Optional
import aiohttp
from app.resources.base import BaseResource, ResourceSchema

class WeatherResource(BaseResource):
    """Resource for accessing weather data"""

    @property
    def schema(self) -> ResourceSchema:
        return ResourceSchema(
            name="weather",
            description="Access current and historical weather data",
            uri_format="weather://{city}/{data_type}"
        )

    async def get(self, uri: str, params: Optional[Dict[str, Any]] = None) -> Any:
        """Retrieve weather data based on the URI"""
        # Parse the URI (format: weather://city/data_type)
        parts = uri.replace("weather://", "").split("/")

        if len(parts) != 2:
            return {"error": "Invalid URI format. Expected: weather://city/data_type"}

        city, data_type = parts

        # In a real implementation, you would call a weather API
        # This is a simplified example
        async with aiohttp.ClientSession() as session:
            # Replace with actual weather API endpoint
            api_url = f"https://api.weatherapi.com/v1/current.json?key=YOUR_API_KEY&q={city}"

            async with session.get(api_url) as response:
                if response.status == 200:
                    data = await response.json()

                    if data_type == "current":
                        return {
                            "location": data["location"]["name"],
                            "country": data["location"]["country"],
                            "temperature": data["current"]["temp_c"],
                            "condition": data["current"]["condition"]["text"],
                            "humidity": data["current"]["humidity"],
                            "wind_kph": data["current"]["wind_kph"]
                        }
                    else:
                        return {"error": f"Unknown data type: {data_type}"}
                else:
                    return {"error": f"Failed to fetch weather data: {response.status}"}

Step 6: Creating the Main Server Application (1 hour)

Now, let's build the main FastAPI application:

# app/main.py
from fastapi import FastAPI, Depends, HTTPException, Header
from fastapi.middleware.cors import CORSMiddleware
from typing import Dict, Any, List, Optional
import logging

from app.config import settings
from app.tools.weather import WeatherForecastTool
from app.resources.weather import WeatherResource

# Configure logging
logging.basicConfig(
    level=getattr(logging, settings.LOG_LEVEL),
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger("mcp_server")

# Initialize FastAPI app
app = FastAPI(
    title=settings.APP_NAME,
    description=settings.APP_DESCRIPTION,
    version=settings.APP_VERSION,
)

# Add CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Initialize tools and resources
tools = {
    "get_forecast": WeatherForecastTool(),
}

resources = {
    "weather": WeatherResource(),
}

# Authentication dependency
async def verify_api_key(x_api_key: str = Header(None)):
    if not settings.ENABLE_AUTH:
        return True

    if x_api_key != settings.API_KEY:
        raise HTTPException(status_code=401, detail="Invalid API key")
    return True

# Server info endpoint
@app.get("/")
async def get_server_info():
    return {
        "name": settings.APP_NAME,
        "version": settings.APP_VERSION,
        "description": settings.APP_DESCRIPTION,
    }

# List available tools
@app.get("/tools")
async def list_tools(_: bool = Depends(verify_api_key)):
    return {
        "tools": [tool.schema for tool in tools.values()]
    }

# Execute a tool
@app.post("/tools/{tool_name}")
async def execute_tool(
    tool_name: str,
    params: Dict[str, Any],
    _: bool = Depends(verify_api_key)
):
    if tool_name not in tools:
        raise HTTPException(status_code=404, detail=f"Tool '{tool_name}' not found")

    try:
        logger.info(f"Executing tool: {tool_name} with params: {params}")
        result = await tools[tool_name].execute(params)
        return {"result": result}
    except Exception as e:
        logger.error(f"Error executing tool {tool_name}: {str(e)}")
        raise HTTPException(status_code=500, detail=f"Error executing tool: {str(e)}")

# List available resources
@app.get("/resources")
async def list_resources(_: bool = Depends(verify_api_key)):
    return {
        "resources": [resource.schema for resource in resources.values()]
    }

# Access a resource
@app.get("/resources")
async def access_resource(
    uri: str,
    params: Optional[Dict[str, Any]] = None,
    _: bool = Depends(verify_api_key)
):
    # Parse the URI to determine which resource handler to use
    resource_type = uri.split("://")[0] if "://" in uri else None

    if not resource_type or resource_type not in resources:
        raise HTTPException(status_code=404, detail=f"Resource type '{resource_type}' not found")

    try:
        logger.info(f"Accessing resource: {uri}")
        result = await resources[resource_type].get(uri, params)
        return {"result": result}
    except Exception as e:
        logger.error(f"Error accessing resource {uri}: {str(e)}")
        raise HTTPException(status_code=500, detail=f"Error accessing resource: {str(e)}")

# Run the server
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(
        "app.main:app",
        host=settings.HOST,
        port=settings.PORT,
        reload=True,
    )

Step 7: Running and Testing Your MCP Server (30 minutes)

Create a script to run your server:

# Create run.py in the project root
echo 'from app.main import app
import uvicorn
from app.config import settings

if __name__ == "__main__":
    uvicorn.run(
        app,
        host=settings.HOST,
        port=settings.PORT,
        reload=True,
    )' > run.py

Run your server:

python run.py

Your MCP server should now be running at http://localhost:8000. You can test it using curl or any API client:

# Get server info
curl http://localhost:8000/

# List available tools
curl http://localhost:8000/tools

# Execute the weather forecast tool
curl -X POST http://localhost:8000/tools/get_forecast \
  -H "Content-Type: application/json" \
  -d '{"city": "San Francisco", "days": 3}'

Advanced MCP Server Features

Adding Authentication and Security (45 minutes)

Enhance your server's security with proper authentication:

# app/auth.py
from fastapi import Depends, HTTPException, status
from fastapi.security import APIKeyHeader
from app.config import settings

api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)

async def get_api_key(api_key: str = Depends(api_key_header)):
    if not settings.ENABLE_AUTH:
        return True

    if api_key != settings.API_KEY:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid API key",
        )
    return True

Update your main.py to use this authentication module.

Implementing Streaming Responses (1 hour)

For tools that generate content incrementally, implement streaming responses:

# app/tools/text_generator.py
from typing import Dict, Any, AsyncGenerator
from fastapi import WebSocket
import json
from app.tools.base import BaseTool, ToolSchema, ToolParameter

class TextGeneratorTool(BaseTool):
    """Tool for generating text incrementally"""

    @property
    def schema(self) -> ToolSchema:
        return ToolSchema(
            name="generate_text",
            description="Generate text based on a prompt",
            parameters=[
                ToolParameter(
                    name="prompt",
                    type="string",
                    description="The prompt to generate text from",
                    required=True
                ),
                ToolParameter(
                    name="max_tokens",
                    type="integer",
                    description="Maximum number of tokens to generate",
                    required=False,
                    default=100
                )
            ]
        )

    async def execute_stream(self, params: Dict[str, Any]) -> AsyncGenerator[str, None]:
        """Execute the text generator with streaming output"""
        prompt = params.get("prompt")
        max_tokens = params.get("max_tokens", 100)

        # In a real implementation, you would call a text generation API
        # This is a simplified example that just returns the prompt word by word
        words = prompt.split()
        for i, word in enumerate(words):
            if i >= max_tokens:
                break
            yield word
            # Simulate processing time
            await asyncio.sleep(0.1)

Add a WebSocket endpoint to your main.py:

@app.websocket("/ws/tools/{tool_name}")
async def websocket_tool(websocket: WebSocket, tool_name: str):
    await websocket.accept()

    try:
        # Receive parameters
        data = await websocket.receive_text()
        params = json.loads(data)

        if tool_name not in tools:
            await websocket.send_json({"error": f"Tool '{tool_name}' not found"})
            await websocket.close()
            return

        tool = tools[tool_name]

        # Check if the tool supports streaming
        if hasattr(tool, "execute_stream"):
            async for chunk in tool.execute_stream(params):
                await websocket.send_json({"chunk": chunk})
            await websocket.send_json({"done": True})
        else:
            # Fall back to non-streaming execution
            result = await tool.execute(params)
            await websocket.send_json({"result": result, "done": True})

    except Exception as e:
        logger.error(f"WebSocket error: {str(e)}")
        await websocket.send_json({"error": str(e)})

    finally:
        await websocket.close()

Implementing Tool Composition (1 hour)

Create a system for composing multiple tools together:

# app/tools/composer.py
from typing import Dict, Any, List
from app.tools.base import BaseTool, ToolSchema, ToolParameter

class ToolComposer(BaseTool):
    """Tool for composing multiple tools together"""

    def __init__(self, available_tools: Dict[str, BaseTool]):
        self.available_tools = available_tools

    @property
    def schema(self) -> ToolSchema:
        return ToolSchema(
            name="compose_tools",
            description="Execute multiple tools in sequence, passing outputs as inputs",
            parameters=[
                ToolParameter(
                    name="steps",
                    type="array",
                    description="Array of tool execution steps",
                    required=True
                )
            ]
        )

    async def execute(self, params: Dict[str, Any]) -> Any:
        """Execute a sequence of tools"""
        steps = params.get("steps", [])
        context = {}  # Stores intermediate results

        for i, step in enumerate(steps):
            if "tool" not in step or step["tool"] not in self.available_tools:
                return {"error": f"Invalid tool in step {i}: {step.get('tool')}"}

            tool = self.available_tools[step["tool"]]
            tool_params = step.get("params", {})

            # Process parameter references (e.g., $results.step1.temperature)
            for param_name, param_value in tool_params.items():
                if isinstance(param_value, str) and param_value.startswith("$results."):
                    # Extract the reference path
                    ref_path = param_value[9:].split(".")
                    ref_value = context
                    for path_part in ref_path:
                        if path_part in ref_value:
                            ref_value = ref_value[path_part]
                        else:
                            return {"error": f"Invalid reference in step {i}: {param_value}"}

                    tool_params[param_name] = ref_value

            # Execute the tool
            result = await tool.execute(tool_params)

            # Store the result in context
            step_id = step.get("id", f"step{i}")
            context[step_id] = result

        return {"results": context}

Best Practices for MCP Server Development

1. Modular Design

Structure your MCP server with modularity in mind:

• Separate tools and resources into individual modules
• Use dependency injection for configuration and shared services
• Create abstract base classes for common functionality
• Implement plugin systems for extensibility

2. Error Handling and Logging

Implement robust error handling:

try:
    # Operation that might fail
    result = await perform_operation()
    return result
except ValidationError as e:
    logger.warning(f"Validation error: {str(e)}")
    return {"error": "Invalid input", "details": str(e)}
except ExternalAPIError as e:
    logger.error(f"External API error: {str(e)}")
    return {"error": "External service unavailable", "details": str(e)}
except Exception as e:
    logger.exception(f"Unexpected error: {str(e)}")
    return {"error": "Internal server error", "details": str(e)}

3. Performance Optimization

Optimize your MCP server for performance:

• Use asynchronous programming for I/O-bound operations
• Implement caching for frequently accessed data
• Use connection pooling for database and API connections
• Consider horizontal scaling for high-traffic scenarios

4. Documentation

Document your MCP server thoroughly:

• Generate OpenAPI documentation for your API endpoints
• Document tool and resource schemas
• Provide usage examples for each tool and resource
• Create a comprehensive README with setup and deployment instructions

Common Challenges and Solutions

Challenge: Handling Large Responses

Solution: Implement pagination and streaming responses:

async def get_large_dataset(page: int = 0, page_size: int = 100):
    # Calculate offset
    offset = page * page_size

    # Fetch data with pagination
    data = await database.fetch_many(
        query="SELECT * FROM large_table LIMIT :limit OFFSET :offset",
        values={"limit": page_size, "offset": offset}
    )

    # Get total count
    total = await database.fetch_val("SELECT COUNT(*) FROM large_table")

    return {
        "data": data,
        "pagination": {
            "page": page,
            "page_size": page_size,
            "total": total,
            "pages": (total + page_size - 1) // page_size
        }
    }

Challenge: Managing External API Rate Limits

Solution: Implement rate limiting and backoff strategies:

class RateLimitedAPI:
    def __init__(self, requests_per_minute: int = 60):
        self.requests_per_minute = requests_per_minute
        self.request_times = []

    async def make_request(self, url: str, method: str = "GET", **kwargs):
        # Check if we're exceeding rate limits
        now = time.time()
        self.request_times = [t for t in self.request_times if now - t < 60]

        if len(self.request_times) >= self.requests_per_minute:
            # Calculate wait time
            oldest_request = min(self.request_times)
            wait_time = 60 - (now - oldest_request)
            logger.info(f"Rate limit reached, waiting {wait_time:.2f} seconds")
            await asyncio.sleep(wait_time)

        # Make the request
        async with aiohttp.ClientSession() as session:
            async with session.request(method, url, **kwargs) as response:
                self.request_times.append(time.time())
                return await response.json()

Challenge: Securing Sensitive Data

Solution: Implement proper secrets management:

# Use environment variables for sensitive data
API_KEY = os.getenv("API_KEY")

# Use a secrets manager for more complex scenarios
from azure.keyvault.secrets import SecretClient
from azure.identity import DefaultAzureCredential

credential = DefaultAzureCredential()
secret_client = SecretClient(vault_url="https://your-vault.vault.azure.net/", credential=credential)
api_key = secret_client.get_secret("api-key").value

Real-World MCP Server Use Cases

Use Case 1: AI-Powered Data Analysis

Create an MCP server that provides data analysis tools:

• Time series forecasting
• Anomaly detection
• Sentiment analysis
• Data visualization generation
• Statistical analysis

Use Case 2: Content Generation

Build an MCP server for content generation:

• Image generation and manipulation
• Text-to-speech conversion
• Language translation
• Code generation and formatting
• Document summarization

Use Case 3: Knowledge Base Access

Develop an MCP server that provides access to specialized knowledge:

• Medical research database
• Legal document analysis
• Financial data and analysis
• Scientific literature search
• Technical documentation

Conclusion

Building a Model Context Protocol server in Python opens up a world of possibilities for extending AI capabilities. By following this guide, you've learned how to create a robust MCP server that can provide specialized tools and resources to AI models.

The key to success lies in thoughtful design, proper error handling, and a focus on security and performance. As you continue to develop your MCP server, consider adding more tools and resources that leverage your unique domain expertise or data sources.

The future of AI lies in its ability to seamlessly integrate with specialized tools and knowledge sources. By building MCP servers, you're contributing to this ecosystem and enabling more powerful, context-aware AI applications.

Start small, focus on solving specific problems well, and gradually expand your MCP server's capabilities as you gain experience. The modular architecture outlined in this guide will make it easy to add new features and scale your server as needed.