How to Build Your First MCP Server in Python

The Model Context Protocol (MCP) is an open standard designed to help AI systems maintain context throughout a conversation. It provides a consistent way for AI applications to manage context, making it easier to build reliable AI systems with persistent memory.

What is Model Context Protocol (MCP) and what problem it solves?

In this blog, I will show you how to build MCP server from the scratch.

Prerequisites

Before starting:

Basic understanding of HTTP/WebSockets
Familiarity with a programming language (I’ll use Python for examples)
Knowledge of JSON structures
Development environment set up

Step 1: Understanding MCP Architecture

An MCP server is responsible for:

Storing conversation history
Managing context windows
Providing a standardized API for clients to retrieve/update context

Step 2: Setting Up Your Project

# Create project structure
mkdir mcp-server
cd mcp-server
mkdir -p src/models src/storage src/api tests

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

# Install dependencies
pip install fastapi uvicorn pydantic websockets

Step 3: Define Core Data Models

# src/models/context.py
from pydantic import BaseModel
from typing import List, Optional, Dict, Any
from datetime import datetime

class Message(BaseModel):
    role: str
    content: str
    timestamp: datetime = datetime.now()
    metadata: Optional[Dict[str, Any]] = None

class Context(BaseModel):
    id: str
    messages: List[Message] = []
    metadata: Optional[Dict[str, Any]] = None
    created_at: datetime = datetime.now()
    updated_at: datetime = datetime.now()

Step 4: Implement Storage Layer

# src/storage/context_store.py
from typing import Dict, Optional
from src.models.context import Context
import uuid

class ContextStore:
    def __init__(self):
        self.contexts: Dict[str, Context] = {}
    
    def create_context(self) -> Context:
        context_id = str(uuid.uuid4())
        context = Context(id=context_id)
        self.contexts[context_id] = context
        return context
    
    def get_context(self, context_id: str) -> Optional[Context]:
        return self.contexts.get(context_id)
    
    def update_context(self, context: Context) -> Context:
        self.contexts[context.id] = context
        return context
    
    def delete_context(self, context_id: str) -> bool:
        if context_id in self.contexts:
            del self.contexts[context_id]
            return True
        return False

Step 5: Create API Endpoints

# src/api/routes.py
from fastapi import APIRouter, HTTPException
from src.models.context import Context, Message
from src.storage.context_store import ContextStore
from typing import List

router = APIRouter()
context_store = ContextStore()

@router.post("/contexts", response_model=Context)
async def create_context():
    return context_store.create_context()

@router.get("/contexts/{context_id}", response_model=Context)
async def get_context(context_id: str):
    context = context_store.get_context(context_id)
    if not context:
        raise HTTPException(status_code=404, detail="Context not found")
    return context

@router.post("/contexts/{context_id}/messages", response_model=Context)
async def add_message(context_id: str, message: Message):
    context = context_store.get_context(context_id)
    if not context:
        raise HTTPException(status_code=404, detail="Context not found")
    
    context.messages.append(message)
    return context_store.update_context(context)

Step 6: Set Up Main Application

# src/main.py
from fastapi import FastAPI
from src.api.routes import router

app = FastAPI(title="MCP Server")
app.include_router(router, prefix="/api/v1")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run("src.main:app", host="0.0.0.0", port=8000, reload=True)

Step 7: Implement WebSocket Support

# src/api/websocket.py
from fastapi import APIRouter, WebSocket, WebSocketDisconnect
from src.storage.context_store import ContextStore
import json

ws_router = APIRouter()
context_store = ContextStore()

@ws_router.websocket("/ws/{context_id}")
async def websocket_endpoint(websocket: WebSocket, context_id: str):
    await websocket.accept()
    
    try:
        context = context_store.get_context(context_id)
        if not context:
            await websocket.close(code=1008, reason="Context not found")
            return
            
        # Send initial context
        await websocket.send_text(context.json())
        
        # Listen for updates
        while True:
            data = await websocket.receive_text()
            message_data = json.loads(data)
            
            # Process message and update context
            # ...
            
            # Send updated context back
            context = context_store.get_context(context_id)
            await websocket.send_text(context.json())
            
    except WebSocketDisconnect:
        pass

Step 8: Adding MCP-Specific Features

# src/models/mcp_extensions.py
from pydantic import BaseModel
from typing import List, Optional, Dict, Any

class MCPContextWindow(BaseModel):
    max_tokens: int = 4096
    current_tokens: int = 0
    
class MCPContext(BaseModel):
    window: MCPContextWindow
    capabilities: List[str] = ["basic", "structured_data"]
    version: str = "1.0"

Step 9: Implement Context Window Management

python# src/services/token_counter.py
def count_tokens(text: str) -> int:
    # This is a simplified implementation
    # You should use a proper tokenizer like tiktoken
    return len(text.split())

# src/services/context_manager.py
from src.models.context import Context, Message
from src.services.token_counter import count_tokens

class ContextManager:
    def __init__(self, max_tokens: int = 4096):
        self.max_tokens = max_tokens
    
    def trim_context(self, context: Context) -> Context:
        """Trim context to fit within token window."""
        total_tokens = 0
        retained_messages = []
        
        # Start from most recent messages
        for message in reversed(context.messages):
            message_tokens = count_tokens(message.content)
            if total_tokens + message_tokens <= self.max_tokens:
                retained_messages.insert(0, message)
                total_tokens += message_tokens
            else:
                break
                
        context.messages = retained_messages
        return context

Step 10: Implement Full MCP Standard Compliance

# src/api/mcp_routes.py
from fastapi import APIRouter, HTTPException
from src.models.context import Context
from src.models.mcp_extensions import MCPContext, MCPContextWindow
from src.storage.context_store import ContextStore
from src.services.context_manager import ContextManager

mcp_router = APIRouter()
context_store = ContextStore()
context_manager = ContextManager()

@mcp_router.post("/mcp/contexts", response_model=Context)
async def create_mcp_context():
    context = context_store.create_context()
    # Add MCP-specific metadata
    context.metadata = {
        "mcp": MCPContext(
            window=MCPContextWindow(
                max_tokens=4096,
                current_tokens=0
            )
        ).dict()
    }
    return context_store.update_context(context)

@mcp_router.get("/mcp/contexts/{context_id}", response_model=Context)
async def get_mcp_context(context_id: str):
    context = context_store.get_context(context_id)
    if not context:
        raise HTTPException(status_code=404, detail="Context not found")
    return context

Step 11: Add Authentication and Security

# src/api/security.py
from fastapi import Depends, HTTPException, status
from fastapi.security import APIKeyHeader
from typing import Optional

API_KEY_NAME = "X-API-Key"
API_KEY = "your_secret_api_key"  # In production, use environment variables

api_key_header = APIKeyHeader(name=API_KEY_NAME)

async def get_api_key(api_key: str = Depends(api_key_header)):
    if api_key != API_KEY:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid API Key"
        )
    return api_key

Step 12: Testing Your MCP Server

# tests/test_context_api.py
import pytest
from fastapi.testclient import TestClient
from src.main import app

client = TestClient(app)

def test_create_context():
    response = client.post("/api/v1/contexts")
    assert response.status_code == 200
    data = response.json()
    assert "id" in data
    assert "messages" in data

def test_add_message():
    # Create context
    response = client.post("/api/v1/contexts")
    context_id = response.json()["id"]
    
    # Add message
    message = {"role": "user", "content": "Hello, world!"}
    response = client.post(f"/api/v1/contexts/{context_id}/messages", json=message)
    assert response.status_code == 200
    data = response.json()
    assert len(data["messages"]) == 1
    assert data["messages"][0]["content"] == "Hello, world!"

Step 13: Dockerizing Your MCP Server

Create a Dockerfile:

FROM python:3.9-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["uvicorn", "src.main:app", "--host", "0.0.0.0", "--port", "8000"]

Create a docker-compose.yml file:


services:
  mcp-server:
    build: .
    ports:
      - "8000:8000"
    environment:
      - API_KEY=your_secret_api_key

Step 14: Adding Persistence

For a production setup, you’ll want to persist contexts:

# src/storage/persistent_context_store.py
import json
import os
from typing import Dict, Optional
from src.models.context import Context

class PersistentContextStore:
    def __init__(self, storage_path: str = "data/contexts"):
        self.storage_path = storage_path
        os.makedirs(storage_path, exist_ok=True)
        self.contexts: Dict[str, Context] = self._load_contexts()
    
    def _load_contexts(self) -> Dict[str, Context]:
        contexts = {}
        for filename in os.listdir(self.storage_path):
            if filename.endswith(".json"):
                context_id = filename.replace(".json", "")
                with open(os.path.join(self.storage_path, filename), "r") as f:
                    context_data = json.load(f)
                    contexts[context_id] = Context(**context_data)
        return contexts
    
    def _save_context(self, context: Context):
        with open(os.path.join(self.storage_path, f"{context.id}.json"), "w") as f:
            f.write(context.json())
    
    # Implement the same methods as ContextStore, but add _save_context calls

Getting Started

Clone the repository

git clone https://github.com/ajeetraina/mcp-server

Install the required modules

pip install -r requirements.txt

Run the server

python -m src.main

OR use Docker Compose:

docker-compose up -d

The server will be available at http://localhost:8000 with API documentation at http://localhost:8000/docs

Test creating a context

 curl -X POST "http://localhost:8000/api/v1/contexts" \
  -H "X-API-Key: your_secret_api_key" \
  -H "Content-Type: application/json"
{"id":"a3e79579-a976-4dd8-ab77-c4dd08ad6d9a","messages":[],"metadata":null,"created_at":"2025-05-15T13:39:46.340408","updated_at":"2025-05-15T13:39:46.340410"}%

Great! I can see that you’ve successfully set up and started the MCP server. The JSON response you’re seeing is from the root endpoint (/) of the server, which confirms that everything is working correctly.

Since you likely didn’t set any API_KEY environment variable when starting the server, it’s using the literal default value “your_secret_api_key”. Your curl commands are including that exact header, so the authentication is working.

If you’re integrating this with an AI system, you’d typically:

Create a context at the start of a conversation
Add messages as the conversation progresses
Use the context to provide history to your AI model

Step 15: Scaling Considerations

For production deployments, consider:

Database Backend: Replace in-memory storage with PostgreSQL or MongoDB
Caching: Add Redis caching for frequently accessed contexts
Load Balancing: Set up multiple MCP server instances behind a load balancer
Monitoring: Add Prometheus metrics and Grafana dashboards

Conclusion

You now have a functional MCP server that implements the core features of the Model Context Protocol. This server can be extended with additional features like:

Enhanced context summarization
Multi-modal context support
Context versioning
Collaborative contexts

Remember that MCP is an evolving standard, so check the official documentation for the latest specifications.