Build production-ready MCP servers in 2026 using FastMCP, covering the latest protocol features, security best practices, and deployment strategies.

Building Custom MCP Servers in 2026 with FastMCP

Overview

The Model Context Protocol (MCP) has become the standard for connecting AI agents to external tools and data sources. With MCP 2.1 (released July 2026), FastMCP has emerged as the fastest and most developer-friendly way to build custom MCP servers in Python and TypeScript.

This tutorial covers building production-ready MCP servers using FastMCP, including the latest MCP 2.1 features, security best practices, and deployment strategies.

What's New in MCP 2.1

Feature	Description
Resource Versioning	ETag-based content negotiation for cached resources
Server-Initiated Subscriptions	Push-based event notification from server to client
Enhanced Tool Discovery	Dynamic tool metadata with parameter validation
Improved Streaming	SSE-based streaming for long-running operations

Installation

# Python
pip install fastmcp

# TypeScript
npm install @fastmcp/core

Building Your First MCP Server

1. Basic Server with FastMCP

from fastmcp import FastMCP

mcp = FastMCP(
    name="my-custom-server",
    version="1.0.0",
    description="A custom MCP server for your AI agents",
)

@mcp.tool()
async def get_weather(city: str, units: str = "celsius") -> str:
    """Get current weather for a city.
    
    Args:
        city: City name
        units: 'celsius' or 'fahrenheit'
    """
    # Your implementation here
    return f"Weather in {city}: 22°C, sunny"

@mcp.tool()
async def search_database(query: str, limit: int = 10) -> list[dict]:
    """Search the database for records.
    
    Args:
        query: Search query
        limit: Maximum number of results
    """
    # Your implementation here
    return [{"id": 1, "name": "Result 1"}]

if __name__ == "__main__":
    mcp.run(transport="stdio")  # Standard transport for MCP clients

2. TypeScript Version

import { FastMCP } from "@fastmcp/core";

const mcp = new FastMCP({
  name: "my-custom-server",
  version: "1.0.0",
  description: "A custom MCP server for your AI agents",
});

mcp.tool("getWeather", {
  description: "Get current weather for a city",
  parameters: {
    type: "object",
    properties: {
      city: { type: "string", description: "City name" },
      units: { 
        type: "string", 
        description: "'celsius' or 'fahrenheit'",
        default: "celsius"
      },
    },
  },
  handler: async ({ city, units }) => {
    return `Weather in ${city}: 22°C, sunny`;
  },
});

mcp.start({ transport: "stdio" });

Advanced Features

1. MCP 2.1: Server-Initiated Subscriptions

New in MCP 2.1, servers can push updates to clients:

import asyncio
from fastmcp import FastMCP, Subscription

mcp = FastMCP("real-time-server")

@mcp.tool()
async def subscribe_to_updates(channel: str) -> dict:
    """Subscribe to real-time updates on a channel.
    
    Args:
        channel: Channel name to subscribe to
    """
    sub = Subscription(channel)
    mcp.subscriptions.add(sub)
    return {"subscription_id": sub.id}

@mcp.tool()
async def unsubscribe(subscription_id: str) -> dict:
    """Unsubscribe from updates.
    
    Args:
        subscription_id: The subscription ID
    """
    mcp.subscriptions.remove(subscription_id)
    return {"status": "unsubscribed"}

# Push updates to subscribers
async def push_update(channel: str, data: dict):
    for sub in mcp.subscriptions.get(channel):
        await mcp.send_to_client(
            subscription_id=sub.id,
            payload=data,
        )

2. MCP 2.1: Resource Versioning

ETag-based caching for resources:

from fastmcp import FastMCP, Resource

mcp = FastMCP("versioned-server")

@mcp.resource("data://users")
async def get_users(etag: str | None = None) -> list[dict]:
    """Get all users with versioning support.
    
    Args:
        etag: Optional ETag for conditional requests
    """
    current_data = await fetch_users()
    current_etag = compute_etag(current_data)
    
    if etag and etag == current_etag:
        return mcp.NotModified()
    
    return {
        "data": current_data,
        "etag": current_etag,
    }

3. MCP 2.1: Enhanced Tool Discovery

Dynamic tool metadata with validation:

from fastmcp import FastMCP, ToolMetadata

mcp = FastMCP("dynamic-server")

@mcp.tool()
async def query_api(
    endpoint: str,
    method: str = "GET",
    headers: dict | None = None,
) -> dict:
    """Query external API with dynamic endpoint support.
    
    Args:
        endpoint: API endpoint path
        method: HTTP method (GET, POST, PUT, DELETE)
        headers: Optional headers
    """
    # Dynamic tool discovery - tools are discovered at runtime
    return {"status": "success"}

Security Best Practices

1. Input Validation

from fastmcp import FastMCP, ValidationError

mcp = FastMCP("secure-server")

@mcp.tool()
async def create_user(name: str, email: str) -> dict:
    """Create a new user with validation.
    
    Args:
        name: User name (max 100 chars)
        email: User email (must be valid email)
    """
    if len(name) > 100:
        raise ValidationError("Name too long")
    
    if not "@" in email:
        raise ValidationError("Invalid email")
    
    return {"status": "created", "name": name}

2. Rate Limiting

import asyncio
from fastmcp import FastMCP, RateLimiter

mcp = FastMCP("rate-limited-server")
limiter = RateLimiter(max_requests=10, window_seconds=60)

@mcp.tool(rate_limit=limiter)
async def expensive_operation(data: dict) -> dict:
    """Perform expensive operation with rate limiting.
    
    Args:
        data: Input data
    """
    return {"status": "completed"}

3. Authentication

import os
from fastmcp import FastMCP, AuthConfig

mcp = FastMCP(
    "authenticated-server",
    auth=AuthConfig(
        type="api_key",
        header="Authorization",
        env_var="MCP_API_KEY",
    ),
)

@mcp.tool()
async def sensitive_operation(data: dict) -> dict:
    """Perform sensitive operation requiring authentication.
    
    Args:
        data: Input data
    """
    return {"status": "success"}

Deployment Strategies

1. Docker Deployment

FROM python:3.12-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .

CMD ["python", "server.py"]

# docker-compose.yml
version: "3.8"
services:
  mcp-server:
    build: .
    ports:
      - "3000:3000"
    environment:
      - MCP_API_KEY=${MCP_API_KEY}
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:3000/health"]
      interval: 30s
      timeout: 10s
      retries: 3

2. Cloud Deployment

# Cloud-run compatible
from fastmcp import FastMCP, CloudRunConfig

mcp = FastMCP(
    "cloud-server",
    cloud_run=CloudRunConfig(
        port=8080,
        health_check_path="/health",
        max_concurrent=100,
    ),
)

@mcp.tool()
async def cloud_operation(data: dict) -> dict:
    """Cloud-ready operation.
    
    Args:
        data: Input data
    """
    return {"status": "success"}

3. Self-Hosting

# Standalone server
fastmcp run server.py --transport stdio

# SSE server for web clients
fastmcp run server.py --transport sse --port 3000

# WebSocket server
fastmcp run server.py --transport websocket --port 3001

Testing Your MCP Server

1. Unit Tests

import pytest
from fastmcp import FastMCP

mcp = FastMCP("test-server")

@mcp.tool()
async def add(a: int, b: int) -> int:
    """Add two numbers."""
    return a + b

class TestAddTool:
    async def test_add_positive(self):
        result = await mcp.run_tool("add", {"a": 2, "b": 3})
        assert result == 5
    
    async def test_add_negative(self):
        result = await mcp.run_tool("add", {"a": -1, "b": -1})
        assert result == -2

2. Integration Tests

import pytest
from fastmcp import ClientSession

@pytest.fixture
async def mcp_client():
    async with ClientSession(transport="stdio", command=["python", "server.py"]) as client:
        yield client

class TestIntegration:
    async def test_tool_call(self, mcp_client):
        result = await mcp_client.call_tool("get_weather", {"city": "Tokyo"})
        assert "Weather" in result
    
    async def test_error_handling(self, mcp_client):
        with pytest.raises(Exception):
            await mcp_client.call_tool("get_weather", {"city": ""})

Debugging

1. Logging

import logging
from fastmcp import FastMCP

logging.basicConfig(level=logging.DEBUG)
mcp = FastMCP("debug-server", log_level="DEBUG")

@mcp.tool()
async def debug_tool(data: dict) -> dict:
    """Tool with detailed logging."""
    mcp.logger.info(f"Processing: {data}")
    return {"status": "success"}

2. MCP Inspector

Use the MCP Inspector to debug your server interactively:

# Run MCP Inspector
npx @modelcontextprotocol/inspector

# Connect to your server
python server.py

Pros

✅ FastMCP is the fastest MCP SDK for Python/TypeScript
✅ MCP 2.1 features: subscriptions, versioning, streaming
✅ Rich tool discovery with dynamic metadata
✅ Built-in security: validation, rate limiting, auth
✅ Multiple deployment options: Docker, cloud, self-hosted
✅ Excellent testing and debugging tooling

Cons

❌ FastMCP is relatively new with smaller ecosystem
❌ MCP 2.1 features not supported by all clients
❌ TypeScript support still maturing
❌ Limited examples compared to other MCP SDKs

When to Use

Building custom MCP servers for AI agents
Need MCP 2.1 features (subscriptions, versioning)
Want fast Python/TypeScript development
Building production-ready MCP servers
Need flexible deployment options