Building MCP Servers with Python: Complete Guide

Prerequisites

Before we begin, make sure you have the following installed:

• Python 3.8 or higher
• pip (Python package installer)
• A code editor (VS Code, PyCharm, etc.)
• Basic understanding of Python and async programming

Setting Up Your Development Environment

1. Install the MCP Python SDK

The MCP Python SDK provides all the tools you need to build MCP servers:

# Install the MCP Python SDK
pip install mcp

# Or install from source
pip install git+https://github.com/modelcontextprotocol/python-sdk.git

# For development, you might also want
pip install mcp[dev]

2. Create Your Project Structure

Set up a clean project structure for your MCP server:

my-mcp-server/
├── src/
│   ├── __init__.py
│   ├── server.py
│   ├── tools/
│   │   ├── __init__.py
│   │   ├── database_tools.py
│   │   └── api_tools.py
│   └── resources/
│       ├── __init__.py
│       └── file_resources.py
├── tests/
│   ├── __init__.py
│   ├── test_server.py
│   └── test_tools.py
├── requirements.txt
├── README.md
└── mcp_config.json

Building Your First MCP Server

1. Basic Server Setup

Let's start with a simple MCP server that provides basic tools:

import asyncio
from mcp.server import Server
from mcp.server.models import InitializationOptions
from mcp.server.stdio import stdio_server
from mcp.types import (
    CallToolRequest,
    CallToolResult,
    ListToolsRequest,
    ListToolsResult,
    Tool,
)

# Create a simple MCP server
server = Server("my-python-server")

# Define a simple tool
@server.list_tools()
async def handle_list_tools() -> ListToolsResult:
    return ListToolsResult(
        tools=[
            Tool(
                name="hello_world",
                description="A simple hello world tool",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "name": {
                            "type": "string",
                            "description": "Name to greet"
                        }
                    },
                    "required": ["name"]
                }
            )
        ]
    )

# Handle tool calls
@server.call_tool()
async def handle_call_tool(name: str, arguments: dict) -> CallToolResult:
    if name == "hello_world":
        name = arguments.get("name", "World")
        return CallToolResult(
            content=[
                {
                    "type": "text",
                    "text": f"Hello, {name}! Welcome to MCP!"
                }
            ]
        )
    
    raise ValueError(f"Unknown tool: {name}")

# Run the server
async def main():
    async with stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            InitializationOptions(
                server_name="my-python-server",
                server_version="1.0.0",
                capabilities=server.get_capabilities(
                    notification_options=None,
                    experimental_capabilities={}
                )
            )
        )

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

2. Adding Database Tools

Let's add more practical tools for database operations:

import sqlite3
from typing import List, Dict, Any

class DatabaseTools:
    def __init__(self, db_path: str = ":memory:"):
        self.db_path = db_path
    
    def get_connection(self):
        return sqlite3.connect(self.db_path)
    
    async def execute_query(self, query: str) -> List[Dict[str, Any]]:
        """Execute a SQL query and return results"""
        conn = self.get_connection()
        try:
            cursor = conn.cursor()
            cursor.execute(query)
            
            if query.strip().upper().startswith("SELECT"):
                columns = [description[0] for description in cursor.description]
                rows = cursor.fetchall()
                return [dict(zip(columns, row)) for row in rows]
            else:
                conn.commit()
                return [{"message": f"Query executed successfully. Rows affected: {cursor.rowcount}"}]
        finally:
            conn.close()

# Add to your server
db_tools = DatabaseTools()

@server.list_tools()
async def handle_list_tools() -> ListToolsResult:
    return ListToolsResult(
        tools=[
            Tool(
                name="execute_sql",
                description="Execute a SQL query on the database",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "query": {
                            "type": "string",
                            "description": "SQL query to execute"
                        }
                    },
                    "required": ["query"]
                }
            )
        ]
    )

@server.call_tool()
async def handle_call_tool(name: str, arguments: dict) -> CallToolResult:
    if name == "execute_sql":
        query = arguments.get("query")
        results = await db_tools.execute_query(query)
        return CallToolResult(
            content=[
                {
                    "type": "text",
                    "text": f"Query results: {results}"
                }
            ]
        )
    
    # ... other tool handlers

3. Adding API Integration Tools

Create tools for integrating with external APIs:

import aiohttp
import json
from typing import Dict, Any

class APITools:
    def __init__(self):
        self.session = None
    
    async def get_session(self):
        if self.session is None:
            self.session = aiohttp.ClientSession()
        return self.session
    
    async def make_request(self, method: str, url: str, headers: Dict = None, data: Dict = None) -> Dict[str, Any]:
        """Make HTTP requests to external APIs"""
        session = await self.get_session()
        
        try:
            if method.upper() == "GET":
                async with session.get(url, headers=headers) as response:
                    return {
                        "status": response.status,
                        "data": await response.json(),
                        "headers": dict(response.headers)
                    }
            elif method.upper() == "POST":
                async with session.post(url, headers=headers, json=data) as response:
                    return {
                        "status": response.status,
                        "data": await response.json(),
                        "headers": dict(response.headers)
                    }
        except Exception as e:
            return {"error": str(e)}
    
    async def close(self):
        if self.session:
            await self.session.close()

# Add to your server
api_tools = APITools()

@server.list_tools()
async def handle_list_tools() -> ListToolsResult:
    return ListToolsResult(
        tools=[
            Tool(
                name="api_request",
                description="Make HTTP requests to external APIs",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "method": {
                            "type": "string",
                            "enum": ["GET", "POST", "PUT", "DELETE"],
                            "description": "HTTP method"
                        },
                        "url": {
                            "type": "string",
                            "description": "API endpoint URL"
                        },
                        "headers": {
                            "type": "object",
                            "description": "Request headers"
                        },
                        "data": {
                            "type": "object",
                            "description": "Request body data"
                        }
                    },
                    "required": ["method", "url"]
                }
            )
        ]
    )

Advanced Patterns

1. Resource Management

MCP servers can also provide resources (files, data, etc.):

from mcp.types import (
    ListResourcesRequest,
    ListResourcesResult,
    ReadResourceRequest,
    ReadResourceResult,
    Resource,
)

@server.list_resources()
async def handle_list_resources() -> ListResourcesResult:
    return ListResourcesResult(
        resources=[
            Resource(
                uri="file:///data/config.json",
                name="Configuration File",
                description="Application configuration",
                mimeType="application/json"
            ),
            Resource(
                uri="file:///data/logs.txt",
                name="Log File",
                description="Application logs",
                mimeType="text/plain"
            )
        ]
    )

@server.read_resource()
async def handle_read_resource(uri: str) -> ReadResourceResult:
    if uri == "file:///data/config.json":
        return ReadResourceResult(
            contents=[
                {
                    "uri": uri,
                    "mimeType": "application/json",
                    "text": '{"setting": "value"}'
                }
            ]
        )
    
    raise ValueError(f"Unknown resource: {uri}")

2. Error Handling and Validation

Implement robust error handling for your tools:

from mcp.types import CallToolResult
import logging

logger = logging.getLogger(__name__)

@server.call_tool()
async def handle_call_tool(name: str, arguments: dict) -> CallToolResult:
    try:
        if name == "execute_sql":
            query = arguments.get("query")
            if not query:
                return CallToolResult(
                    content=[
                        {
                            "type": "text",
                            "text": "Error: SQL query is required"
                        }
                    ],
                    isError=True
                )
            
            results = await db_tools.execute_query(query)
            return CallToolResult(
                content=[
                    {
                        "type": "text",
                        "text": f"Query executed successfully: {results}"
                    }
                ]
            )
        
        elif name == "api_request":
            method = arguments.get("method", "GET")
            url = arguments.get("url")
            
            if not url:
                return CallToolResult(
                    content=[
                        {
                            "type": "text",
                            "text": "Error: URL is required"
                        }
                    ],
                    isError=True
                )
            
            headers = arguments.get("headers", {})
            data = arguments.get("data")
            
            result = await api_tools.make_request(method, url, headers, data)
            return CallToolResult(
                content=[
                    {
                        "type": "text",
                        "text": f"API request completed: {result}"
                    }
                ]
            )
        
        else:
            return CallToolResult(
                content=[
                    {
                        "type": "text",
                        "text": f"Error: Unknown tool '{name}'"
                    }
                ],
                isError=True
            )
    
    except Exception as e:
        logger.error(f"Error in tool {name}: {str(e)}")
        return CallToolResult(
            content=[
                {
                    "type": "text",
                    "text": f"Error executing tool: {str(e)}"
                }
            ],
            isError=True
        )

3. Configuration Management

Handle configuration and environment variables:

import os
from typing import Optional
from dataclasses import dataclass

@dataclass
class ServerConfig:
    db_path: str
    api_timeout: int
    max_connections: int
    log_level: str
    
    @classmethod
    def from_env(cls) -> "ServerConfig":
        return cls(
            db_path=os.getenv("DB_PATH", ":memory:"),
            api_timeout=int(os.getenv("API_TIMEOUT", "30")),
            max_connections=int(os.getenv("MAX_CONNECTIONS", "10")),
            log_level=os.getenv("LOG_LEVEL", "INFO")
        )

# Initialize configuration
config = ServerConfig.from_env()

# Use configuration in your tools
db_tools = DatabaseTools(config.db_path)
api_tools = APITools()  # You could pass config here too

Testing Your MCP Server

1. Unit Tests

Write tests for your tools and server:

import pytest
import asyncio
from unittest.mock import AsyncMock, patch

class TestDatabaseTools:
    @pytest.fixture
    def db_tools(self):
        return DatabaseTools(":memory:")
    
    @pytest.mark.asyncio
    async def test_execute_query_select(self, db_tools):
        # Create a test table
        await db_tools.execute_query("CREATE TABLE test (id INTEGER, name TEXT)")
        await db_tools.execute_query("INSERT INTO test VALUES (1, 'test')")
        
        # Test SELECT query
        results = await db_tools.execute_query("SELECT * FROM test")
        assert len(results) == 1
        assert results[0]["id"] == 1
        assert results[0]["name"] == "test"

class TestAPITools:
    @pytest.fixture
    def api_tools(self):
        return APITools()
    
    @pytest.mark.asyncio
    async def test_make_request(self, api_tools):
        with patch("aiohttp.ClientSession.get") as mock_get:
            mock_response = AsyncMock()
            mock_response.status = 200
            mock_response.json.return_value = {"test": "data"}
            mock_response.headers = {"content-type": "application/json"}
            mock_get.return_value.__aenter__.return_value = mock_response
            
            result = await api_tools.make_request("GET", "http://test.com")
            assert result["status"] == 200
            assert result["data"] == {"test": "data"}

2. Integration Tests

Test your server end-to-end:

import pytest
from mcp.server.stdio import stdio_server
from mcp.types import CallToolRequest, ListToolsRequest

@pytest.mark.asyncio
async def test_server_integration():
    async with stdio_server() as (read_stream, write_stream):
        # Start server in background
        server_task = asyncio.create_task(
            server.run(read_stream, write_stream, init_options)
        )
        
        # Test listing tools
        list_request = ListToolsRequest()
        await write_stream.write(list_request.model_dump_json() + "\n")
        
        # Test calling a tool
        call_request = CallToolRequest(
            name="hello_world",
            arguments={"name": "Test"}
        )
        await write_stream.write(call_request.model_dump_json() + "\n")
        
        # Clean up
        server_task.cancel()
        try:
            await server_task
        except asyncio.CancelledError:
            pass

Deployment and Distribution

1. Packaging Your Server

Package your MCP server for distribution:

from setuptools import setup, find_packages

setup(
    name="my-mcp-server",
    version="1.0.0",
    description="A Python MCP server for data operations",
    author="Your Name",
    author_email="your.email@example.com",
    packages=find_packages(),
    install_requires=[
        "mcp>=1.0.0",
        "aiohttp>=3.8.0",
        "sqlite3",  # Built-in
    ],
    extras_require={
        "dev": [
            "pytest>=7.0.0",
            "pytest-asyncio>=0.21.0",
            "black>=22.0.0",
            "flake8>=4.0.0",
        ]
    },
    entry_points={
        "console_scripts": [
            "my-mcp-server=src.server:main",
        ],
    },
    python_requires=">=3.8",
)

2. Docker Deployment

Create a Docker container for your server:

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 source code
COPY src/ ./src/

# Set environment variables
ENV PYTHONPATH=/app
ENV DB_PATH=/data/database.db

# Create data directory
RUN mkdir -p /data

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

Best Practices

1. Code Organization

• Separate tools into logical modules
• Use dependency injection for external services
• Implement proper logging throughout your server
• Follow Python naming conventions and PEP 8

2. Performance Optimization

• Use connection pooling for database connections
• Implement caching for frequently accessed data
• Use async/await properly for I/O operations
• Monitor memory usage and implement cleanup

3. Security Considerations

• Validate all input parameters
• Use parameterized queries to prevent SQL injection
• Implement proper authentication and authorization
• Sanitize data before returning it to clients

Next Steps

Now that you have a basic understanding of building MCP servers with Python, you can:

• Explore the official Python SDK documentation
• Join the MCP Discord community
• Check out real-world examples in our showcases
• Contribute to the open-source MCP ecosystem