Building MCP Servers

Building MCP Servers

Building MCP servers is where the protocol moves from theory to practice. This lesson covers how to build MCP servers using both the Python FastMCP SDK and the TypeScript SDK, including how to expose tools, resources, and prompts, and how to configure different transports for development and production.

FastMCP: Python MCP Server SDK

FastMCP is the official Python SDK for building MCP servers. It provides a high-level, decorator-based API that makes it straightforward to expose tools, resources, and prompts. FastMCP handles all the protocol details -- JSON-RPC serialization, capability negotiation, transport management -- so you can focus on your server's logic.

Installing FastMCP

pip install fastmcp

Your First MCP Server

from fastmcp import FastMCP

# Create a server instance
mcp = FastMCP("WeatherServer")


@mcp.tool()
def get_weather(city: str, units: str = "celsius") -> str:
    """Get the current weather for a city.

    Args:
        city: The city name (e.g., San Francisco, London, Tokyo)
        units: Temperature units - either celsius or fahrenheit

    Returns:
        A string describing current weather conditions.
    """
    # Your actual weather API logic here
    weather_data = fetch_weather_api(city, units)
    return f"Weather in {city}: {weather_data['temp']}{units[0].upper()}, {weather_data['conditions']}"


@mcp.tool()
def get_forecast(city: str, days: int = 5) -> str:
    """Get the weather forecast for a city.

    Args:
        city: The city name
        days: Number of days to forecast (1-10)

    Returns:
        Multi-day forecast as a formatted string.
    """
    forecast = fetch_forecast_api(city, days)
    return format_forecast(forecast)


if __name__ == "__main__":
    mcp.run()  # Defaults to stdio transport

How FastMCP Generates Schemas

FastMCP uses Python's type hints and docstrings to automatically generate the JSON Schema that MCP clients receive. Understanding this mapping is important:

• str maps to "type": "string"
• int maps to "type": "integer"
• float maps to "type": "number"
• bool maps to "type": "boolean"
• list[str] maps to "type": "array", "items": {"type": "string"}
• Parameters without defaults become required
• Parameters with defaults become optional with their default values
• The docstring's Args section provides parameter descriptions

Exposing Resources

Resources provide read access to data. FastMCP supports both static resources and dynamic resource templates with URI parameters.

from fastmcp import FastMCP
import json

mcp = FastMCP("DataServer")


@mcp.resource("config://app/settings")
def get_app_settings() -> str:
    """Return the current application settings as JSON."""
    settings = load_settings()
    return json.dumps(settings, indent=2)


@mcp.resource("db://users/{user_id}/profile")
def get_user_profile(user_id: str) -> str:
    """Return profile data for a specific user.

    Args:
        user_id: The unique user identifier
    """
    profile = db.get_user(user_id)
    return json.dumps(profile.to_dict())


@mcp.resource("file://logs/{date}")
def get_logs_for_date(date: str) -> str:
    """Return application logs for a specific date.

    Args:
        date: Date in YYYY-MM-DD format
    """
    logs = read_log_file(date)
    return logs

Exposing Prompts

Prompts are reusable prompt templates that clients can discover and use. They are ideal for standardizing common interactions.

from fastmcp import FastMCP

mcp = FastMCP("CodeAssistant")


@mcp.prompt()
def review_code(code: str, language: str = "python") -> str:
    """Generate a thorough code review.

    Args:
        code: The code to review
        language: Programming language of the code
    """
    return f"""Please review the following {language} code thoroughly.
Focus on:
- Correctness and potential bugs
- Performance issues
- Security vulnerabilities
- Code style and readability
- Suggestions for improvement

Code to review:
```{language}
{code}
```"""


@mcp.prompt()
def explain_error(error_message: str, context: str = "") -> str:
    """Help debug an error message.

    Args:
        error_message: The error message to explain
        context: Additional context about where the error occurred
    """
    prompt = f"Please explain this error and suggest how to fix it:Error: {error_message}"
    if context:
        prompt += f"Context: {context}"
    return prompt

Advanced FastMCP Features

Lifespan Management

Use the lifespan context manager to set up and tear down resources (database connections, API clients, etc.) that your tools need. This ensures clean startup and shutdown.

from contextlib import asynccontextmanager
from fastmcp import FastMCP, Context


@asynccontextmanager
async def app_lifespan(server: FastMCP):
    """Set up and tear down shared resources."""
    # Startup: initialize database connection
    db = await Database.connect("postgresql://localhost/mydb")
    try:
        yield {"db": db}  # Make db available via context
    finally:
        # Shutdown: close database connection
        await db.disconnect()


mcp = FastMCP("DatabaseTools", lifespan=app_lifespan)


@mcp.tool()
async def query_records(ctx: Context, table: str, limit: int = 10) -> str:
    """Query database records.

    Args:
        table: Table name to query
        limit: Maximum records to return
    """
    db = ctx.request_context.lifespan_context["db"]
    records = await db.fetch(f"SELECT * FROM {table} LIMIT {limit}")
    return json.dumps(records)

Context Object and Logging

The Context object gives tools access to server capabilities including logging, progress reporting, and resource access.

from fastmcp import FastMCP, Context

mcp = FastMCP("ProcessingServer")


@mcp.tool()
async def process_large_dataset(ctx: Context, dataset_id: str) -> str:
    """Process a large dataset with progress tracking.

    Args:
        dataset_id: ID of the dataset to process
    """
    data = load_dataset(dataset_id)
    total = len(data)

    results = []
    for i, record in enumerate(data):
        result = process_record(record)
        results.append(result)

        # Report progress to the client
        await ctx.report_progress(i + 1, total)

        # Log processing details
        if i % 100 == 0:
            await ctx.info(f"Processed {i + 1}/{total} records")

    await ctx.info(f"Processing complete: {len(results)} records processed")
    return json.dumps({"processed": len(results), "summary": summarize(results)})

Context Logging Levels

The Context object supports multiple logging levels for different types of messages:

• ctx.debug() -- Detailed diagnostic information
• ctx.info() -- General informational messages
• ctx.warning() -- Warning messages for potential issues
• ctx.error() -- Error messages for failures

TypeScript MCP SDK

The TypeScript SDK provides similar functionality for building MCP servers in JavaScript and TypeScript environments. It uses Zod for schema validation instead of Python type hints.

// TypeScript MCP server example
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";

const server = new McpServer({
  name: "WeatherServer",
  version: "1.0.0",
});

// Define a tool using Zod schemas
server.tool(
  "get_weather",
  "Get current weather for a city. Returns temperature and conditions.",
  {
    city: z.string().describe("City name, e.g. San Francisco"),
    units: z.enum(["celsius", "fahrenheit"]).optional().describe("Temperature units"),
  },
  async ({ city, units }) => {
    const weather = await fetchWeather(city, units || "celsius");
    return {
      content: [
        {
          type: "text",
          text: JSON.stringify(weather),
        },
      ],
    };
  }
);

// Define a resource
server.resource(
  "config://settings",
  "Application configuration settings",
  async (uri) => ({
    contents: [
      {
        uri: uri.href,
        mimeType: "application/json",
        text: JSON.stringify(getSettings()),
      },
    ],
  })
);

// Start the server with stdio transport
const transport = new StdioServerTransport();
await server.connect(transport);

Transport Configuration

stdio Transport

The default transport for local development. The server runs as a subprocess.

# Python: Running with stdio (default)
if __name__ == "__main__":
    mcp.run()  # Uses stdio by default

# Or explicitly:
if __name__ == "__main__":
    mcp.run(transport="stdio")

SSE Transport

For remote servers that need to be accessible over HTTP.

# Python: Running with SSE transport
if __name__ == "__main__":
    mcp.run(transport="sse", host="0.0.0.0", port=8080)

Streamable HTTP Transport

For production deployments that need scalability.

# Python: Running with Streamable HTTP transport
if __name__ == "__main__":
    mcp.run(transport="streamable-http", host="0.0.0.0", port=8080)

Client Configuration

MCP clients (like Claude Desktop) discover servers through configuration files. Here is an example Claude Desktop configuration:

{
  "mcpServers": {
    "weather": {
      "command": "python",
      "args": ["weather_server.py"],
      "env": {
        "WEATHER_API_KEY": "your-api-key-here"
      }
    },
    "database": {
      "command": "npx",
      "args": ["-y", "@company/db-mcp-server"],
      "env": {
        "DB_CONNECTION_STRING": "postgresql://localhost/mydb"
      }
    },
    "remote-api": {
      "url": "https://api.example.com/mcp/sse",
      "headers": {
        "Authorization": "Bearer your-token-here"
      }
    }
  }
}

Composing MCP Servers

FastMCP supports composing multiple servers together, allowing you to build modular server architectures where each component handles a specific domain. This is a powerful pattern for organizing large tool sets.

from fastmcp import FastMCP
import json

# Create domain-specific sub-servers
user_server = FastMCP("UserTools")
order_server = FastMCP("OrderTools")

@user_server.tool()
def get_user(user_id: str) -> str:
    """Look up a user by ID."""
    return json.dumps(db.get_user(user_id))

@order_server.tool()
def get_order(order_id: str) -> str:
    """Look up an order by ID."""
    return json.dumps(db.get_order(order_id))

# Compose into a single server
main_server = FastMCP("MainServer")
main_server.mount("users", user_server)
main_server.mount("orders", order_server)

# Tools are now available as users_get_user and orders_get_order
if __name__ == "__main__":
    main_server.run()

Server composition is especially useful for:

• Organizing tools by domain (users, orders, analytics)
• Reusing server modules across different applications
• Testing individual server components in isolation
• Enabling different teams to own different server modules

Testing MCP Servers

FastMCP provides utilities for testing servers without needing a full client connection. The test client lets you call tools, list resources, and verify behavior programmatically.

import pytest
from fastmcp import FastMCP

mcp = FastMCP("TestServer")

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

@mcp.tool()
def divide_numbers(a: float, b: float) -> float:
    """Divide a by b."""
    if b == 0:
        raise ValueError("Cannot divide by zero")
    return a / b

# Test using the MCP test client
async def test_add_numbers():
    async with mcp.test_client() as client:
        result = await client.call_tool("add_numbers", {"a": 5, "b": 3})
        assert result[0].text == "8"

async def test_list_tools():
    async with mcp.test_client() as client:
        tools = await client.list_tools()
        tool_names = [t.name for t in tools]
        assert "add_numbers" in tool_names
        assert "divide_numbers" in tool_names

async def test_divide_by_zero():
    async with mcp.test_client() as client:
        with pytest.raises(Exception):
            await client.call_tool("divide_numbers", {"a": 10, "b": 0})

Error Handling in MCP Servers

MCP servers should handle errors gracefully and return meaningful error messages that help the model (and ultimately the user) understand what went wrong.

from fastmcp import FastMCP

mcp = FastMCP("RobustServer")


@mcp.tool()
def get_customer(customer_id: str) -> str:
    """Look up a customer by ID.

    Args:
        customer_id: Customer ID in format CUST-XXXXX
    """
    # Validate input format
    if not customer_id.startswith("CUST-"):
        raise ValueError(
            f"Invalid customer ID format: {customer_id}. "
            "Expected format: CUST-XXXXX"
        )

    customer = db.find_customer(customer_id)
    if customer is None:
        raise ValueError(
            f"No customer found with ID {customer_id}. "
            "Please verify the ID and try again."
        )

    return json.dumps(customer.to_dict())