Go back

How to create an MCP (Model Context Protocol) server with FastAPI

The Model Context Protocol (MCP) is an open standard developed by Anthropic that enables AI models to securely access external tools and resources in a standardized way. In this tutorial, you’ll learn how to create your own MCP server using FastAPI.

Programming code on screen
Photo by Pexels

Table of contents

Open Table of contents

What is MCP?

MCP (Model Context Protocol) is a protocol that establishes a standard interface between client applications (like Claude Desktop) and servers that provide context and tools. Think of it as a bridge that allows AI models to access:

Prerequisites

Before getting started, make sure you have:

Installing Dependencies

First, install the necessary dependencies:

pip install fastapi uvicorn python-multipart pydantic

For this tutorial, we’ll also need some additional libraries:

pip install httpx aiofiles python-json-logger

Project Structure

Let’s create the following file structure:

mcp-fastapi-server/
├── main.py
├── models/
│   ├── __init__.py
│   └── mcp_models.py
├── handlers/
│   ├── __init__.py
│   ├── tools.py
│   └── resources.py
├── utils/
│   ├── __init__.py
│   └── helpers.py
└── requirements.txt

Defining MCP Models

First, let’s create the data models that our MCP server will use:

# models/mcp_models.py
from pydantic import BaseModel, Field
from typing import Any, Dict, List, Optional, Union
from enum import Enum

class MCPVersion(str, Enum):
    V1 = "1.0"

class ToolParameter(BaseModel):
    type: str
    description: str
    required: Optional[bool] = False

class Tool(BaseModel):
    name: str
    description: str
    parameters: Dict[str, ToolParameter]

class Resource(BaseModel):
    uri: str
    name: str
    description: str
    mimeType: Optional[str] = None

class MCPRequest(BaseModel):
    jsonrpc: str = "2.0"
    id: Union[str, int]
    method: str
    params: Optional[Dict[str, Any]] = None

class MCPResponse(BaseModel):
    jsonrpc: str = "2.0"
    id: Union[str, int]
    result: Optional[Dict[str, Any]] = None
    error: Optional[Dict[str, Any]] = None

class InitializeParams(BaseModel):
    protocolVersion: str
    capabilities: Dict[str, Any]
    clientInfo: Dict[str, str]

class ServerCapabilities(BaseModel):
    tools: Optional[Dict[str, Any]] = None
    resources: Optional[Dict[str, Any]] = None
    prompts: Optional[Dict[str, Any]] = None

Implementing Tool Handlers

Now, let’s create handlers that will manage the available tools:

# handlers/tools.py
import httpx
import json
from typing import Dict, Any, List
from datetime import datetime

class ToolHandler:
    def __init__(self):
        self.tools = {
            "get_weather": self.get_weather,
            "calculate": self.calculate,
            "web_search": self.web_search,
            "file_operations": self.file_operations
        }
    
    async def get_weather(self, params: Dict[str, Any]) -> Dict[str, Any]:
        """Gets weather information for a city."""
        city = params.get("city", "")
        if not city:
            return {"error": "City is required"}
        
        # Simulate a weather API call
        # In a real case, you would use an API like OpenWeatherMap
        weather_data = {
            "city": city,
            "temperature": 22,
            "condition": "Sunny",
            "humidity": 65,
            "timestamp": datetime.now().isoformat()
        }
        
        return {
            "success": True,
            "data": weather_data
        }
    
    async def calculate(self, params: Dict[str, Any]) -> Dict[str, Any]:
        """Performs basic mathematical calculations."""
        expression = params.get("expression", "")
        if not expression:
            return {"error": "Mathematical expression is required"}
        
        try:
            # Safe evaluation of mathematical expressions
            # NOTE: In production, use a safer library like ast.literal_eval
            allowed_chars = set("0123456789+-*/(). ")
            if all(c in allowed_chars for c in expression):
                result = eval(expression)
                return {
                    "success": True,
                    "expression": expression,
                    "result": result
                }
            else:
                return {"error": "Expression contains forbidden characters"}
        except Exception as e:
            return {"error": f"Calculation error: {str(e)}"}
    
    async def web_search(self, params: Dict[str, Any]) -> Dict[str, Any]:
        """Performs simulated web searches."""
        query = params.get("query", "")
        if not query:
            return {"error": "Search query is required"}
        
        # Simulation of search results
        results = [
            {
                "title": f"Result 1 for '{query}'",
                "url": "https://example.com/1",
                "snippet": f"Relevant information about {query}..."
            },
            {
                "title": f"Result 2 for '{query}'",
                "url": "https://example.com/2", 
                "snippet": f"More details about {query}..."
            }
        ]
        
        return {
            "success": True,
            "query": query,
            "results": results
        }
    
    async def file_operations(self, params: Dict[str, Any]) -> Dict[str, Any]:
        """Handles basic file operations."""
        operation = params.get("operation", "")
        filename = params.get("filename", "")
        
        if operation == "list":
            import os
            try:
                files = os.listdir(".")
                return {
                    "success": True,
                    "operation": "list",
                    "files": files
                }
            except Exception as e:
                return {"error": f"Error listing files: {str(e)}"}
        
        elif operation == "read" and filename:
            try:
                with open(filename, 'r', encoding='utf-8') as f:
                    content = f.read()
                return {
                    "success": True,
                    "operation": "read",
                    "filename": filename,
                    "content": content[:1000]  # Limit to 1000 characters
                }
            except Exception as e:
                return {"error": f"Error reading file: {str(e)}"}
        
        else:
            return {"error": "Unsupported operation or missing parameters"}

    async def execute_tool(self, tool_name: str, params: Dict[str, Any]) -> Dict[str, Any]:
        """Executes a specific tool."""
        if tool_name not in self.tools:
            return {"error": f"Tool '{tool_name}' not found"}
        
        try:
            return await self.tools[tool_name](params)
        except Exception as e:
            return {"error": f"Error executing tool: {str(e)}"}

Creating the FastAPI Server

Now let’s implement the main server with FastAPI:

# main.py
from fastapi import FastAPI, HTTPException, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
import json
import logging
from typing import Dict, Any

from models.mcp_models import (
    MCPRequest, MCPResponse, Tool, ToolParameter,
    InitializeParams, ServerCapabilities
)
from handlers.tools import ToolHandler

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

app = FastAPI(
    title="MCP FastAPI Server",
    description="An MCP server implemented with FastAPI",
    version="1.0.0"
)

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

# Handler initialization
tool_handler = ToolHandler()

# Server state
server_state = {
    "initialized": False,
    "client_info": None
}

# Available tools definition
AVAILABLE_TOOLS = [
    Tool(
        name="get_weather",
        description="Gets weather information for a specific city",
        parameters={
            "city": ToolParameter(
                type="string",
                description="City name",
                required=True
            )
        }
    ),
    Tool(
        name="calculate",
        description="Performs basic mathematical calculations",
        parameters={
            "expression": ToolParameter(
                type="string",
                description="Mathematical expression to evaluate",
                required=True
            )
        }
    ),
    Tool(
        name="web_search",
        description="Performs web searches",
        parameters={
            "query": ToolParameter(
                type="string",
                description="Search term",
                required=True
            )
        }
    ),
    Tool(
        name="file_operations",
        description="Basic file operations",
        parameters={
            "operation": ToolParameter(
                type="string",
                description="Type of operation (list, read)",
                required=True
            ),
            "filename": ToolParameter(
                type="string",
                description="File name (for read operation)",
                required=False
            )
        }
    )
]

@app.get("/")
async def root():
    """Server health endpoint."""
    return {
        "message": "MCP FastAPI Server is running",
        "version": "1.0.0",
        "status": "healthy"
    }

@app.post("/mcp")
async def handle_mcp_request(request: MCPRequest):
    """Handles all MCP requests."""
    logger.info(f"Received MCP request: {request.method}")
    
    try:
        if request.method == "initialize":
            return await handle_initialize(request)
        elif request.method == "tools/list":
            return await handle_list_tools(request)
        elif request.method == "tools/call":
            return await handle_call_tool(request)
        elif request.method == "resources/list":
            return await handle_list_resources(request)
        else:
            return MCPResponse(
                id=request.id,
                error={
                    "code": -32601,
                    "message": f"Method not found: {request.method}"
                }
            )
    except Exception as e:
        logger.error(f"Error processing request: {str(e)}")
        return MCPResponse(
            id=request.id,
            error={
                "code": -32603,
                "message": f"Internal server error: {str(e)}"
            }
        )

async def handle_initialize(request: MCPRequest) -> MCPResponse:
    """Handles MCP server initialization."""
    params = InitializeParams(**request.params)
    
    server_state["initialized"] = True
    server_state["client_info"] = params.clientInfo
    
    capabilities = ServerCapabilities(
        tools={
            "listChanged": True
        },
        resources={
            "subscribe": True,
            "listChanged": True
        }
    )
    
    logger.info(f"Server initialized for client: {params.clientInfo}")
    
    return MCPResponse(
        id=request.id,
        result={
            "protocolVersion": "1.0",
            "capabilities": capabilities.dict(),
            "serverInfo": {
                "name": "MCP FastAPI Server",
                "version": "1.0.0"
            }
        }
    )

async def handle_list_tools(request: MCPRequest) -> MCPResponse:
    """Lists all available tools."""
    tools_data = [tool.dict() for tool in AVAILABLE_TOOLS]
    
    return MCPResponse(
        id=request.id,
        result={
            "tools": tools_data
        }
    )

async def handle_call_tool(request: MCPRequest) -> MCPResponse:
    """Executes a specific tool."""
    if not request.params:
        return MCPResponse(
            id=request.id,
            error={
                "code": -32602,
                "message": "Parameters required to call tool"
            }
        )
    
    tool_name = request.params.get("name")
    tool_params = request.params.get("arguments", {})
    
    if not tool_name:
        return MCPResponse(
            id=request.id,
            error={
                "code": -32602,
                "message": "Tool name is required"
            }
        )
    
    logger.info(f"Executing tool: {tool_name} with parameters: {tool_params}")
    
    result = await tool_handler.execute_tool(tool_name, tool_params)
    
    return MCPResponse(
        id=request.id,
        result={
            "content": [
                {
                    "type": "text",
                    "text": json.dumps(result, indent=2, ensure_ascii=False)
                }
            ]
        }
    )

async def handle_list_resources(request: MCPRequest) -> MCPResponse:
    """Lists available resources."""
    return MCPResponse(
        id=request.id,
        result={
            "resources": []
        }
    )

@app.get("/health")
async def health_check():
    """Health check endpoint."""
    return {
        "status": "healthy",
        "initialized": server_state["initialized"],
        "tools_count": len(AVAILABLE_TOOLS),
        "timestamp": "2025-08-10T15:30:00Z"
    }

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

Configuration and Execution

Create a requirements.txt file:

fastapi==0.104.1
uvicorn[standard]==0.24.0
pydantic==2.5.0
httpx==0.25.2
aiofiles==23.2.1
python-multipart==0.0.6
python-json-logger==2.0.7

To run the server:

# Install dependencies
pip install -r requirements.txt

# Run the server
python main.py

# Or using uvicorn directly
uvicorn main:app --host 0.0.0.0 --port 8000 --reload

Testing the Server

Once the server is running, you can test it using curl or tools like Postman:

1. Check server health

curl http://localhost:8000/health

2. Initialize the MCP server

curl -X POST http://localhost:8000/mcp \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "id": 1,
    "method": "initialize",
    "params": {
      "protocolVersion": "1.0",
      "capabilities": {},
      "clientInfo": {
        "name": "Test Client",
        "version": "1.0.0"
      }
    }
  }'

3. List available tools

curl -X POST http://localhost:8000/mcp \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "id": 2,
    "method": "tools/list"
  }'

4. Execute a tool

curl -X POST http://localhost:8000/mcp \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "id": 3,
    "method": "tools/call",
    "params": {
      "name": "calculate",
      "arguments": {
        "expression": "2 + 2 * 3"
      }
    }
  }'

Integration with Claude Desktop

To use your MCP server with Claude Desktop, you need to configure it in the configuration file:

On macOS:

# Edit the configuration
nano ~/Library/Application\ Support/Claude/claude_desktop_config.json

On Windows:

# File location
%APPDATA%\Claude\claude_desktop_config.json

Configuration file content:

{
  "mcpServers": {
    "fastapi-mcp": {
      "command": "python",
      "args": ["/path/to/your/project/main.py"],
      "env": {
        "PYTHONPATH": "/path/to/your/project"
      }
    }
  }
}

Advanced Features

Authentication and Security

For a production environment, consider adding:

from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from fastapi import Depends, HTTPException, status

security = HTTPBearer()

async def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
    """Verifies authentication token."""
    token = credentials.credentials
    # Implement your verification logic here
    if token != "your-secret-token":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid token",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return token

# Apply authentication to endpoints
@app.post("/mcp")
async def handle_mcp_request(
    request: MCPRequest, 
    token: str = Depends(verify_token)
):
    # Your logic here
    pass

Advanced Logging

import logging
from python_json_logger import jsonlogger

# Configure structured logging
logHandler = logging.StreamHandler()
formatter = jsonlogger.JsonFormatter()
logHandler.setFormatter(formatter)
logger = logging.getLogger()
logger.addHandler(logHandler)
logger.setLevel(logging.INFO)

Custom Error Handling

@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    logger.error(f"Unhandled error: {str(exc)}")
    return JSONResponse(
        status_code=500,
        content={
            "jsonrpc": "2.0",
            "error": {
                "code": -32603,
                "message": "Internal server error"
            }
        }
    )

Best Practices

  1. Input Validation: Always validate input parameters using Pydantic
  2. Error Handling: Implement robust error handling and logging
  3. Documentation: Use FastAPI’s auto-documentation features
  4. Testing: Write unit tests for your tools
  5. Security: Implement proper authentication and input validation
  6. Monitoring: Add metrics and health monitoring
  7. Rate Limiting: Implement rate limits to prevent abuse

Additional Resources

Conclusion

Creating an MCP server with FastAPI allows you to extend the capabilities of Claude and other AI models in a powerful and flexible way. This tutorial has shown you the fundamentals, but the possibilities are endless: from integrating databases to creating specialized tools for your specific domain.

The MCP protocol is designed to be extensible and secure, making it an excellent choice for building robust integrations with AI models. Experiment with different tools and see what you can create!

Did you find this tutorial helpful? Share your own MCP implementations in the comments!

Share this post on:
Share this post on LinkedInShare this post on FacebookShare this post on XShare this post via TelegramShare this post on PinterestShare this post via email
Previous Post
Deploying FastAPI Applications on AWS Lambda with Mangum - A Complete Guide