FastMCP Foundations

Table of Contents

• Module 1: Introduction to FastMCP
  • What Is FastMCP?
  • Background on the Model Context Protocol
  • The MxN Problem and Why MCP Matters
  • MCP Architecture
  • Pros and Cons of FastMCP
  • Setting Up FastMCP and Running a Simple Server
  • FastMCP Cloud
  • MCP and the Agentic AI Ecosystem
  • Alternatives to MCP
  • Practical Use Cases of MCP
• Module 2: FastMCP Essentials
  • Comparing FastMCP to Standard MCP Implementations
  • Connecting to an MCP Server in VS Code
  • Connecting an MCP Server to ChatGPT
  • Connecting an MCP Server to Claude Desktop
  • Building a Password Generator MCP Server
  • Building a Weather MCP Server
  • Building a Crypto MCP Server
  • Building a Document and Knowledge Base MCP Server
  • Chaining Tools with a Crypto MCP
• Module 3: Metrics, Observability, and Production Readiness for FastMCP
  • Structured Message Handling
  • Using Middleware in FastMCP
  • Extending FastMCP with Plugins
  • State and Persistence
  • Integrations in FastMCP
  • Typed Configuration with fastmcp.config
  • Developing Custom Middleware Patterns
  • Debug Mode and Message Tracing
  • Security and Policy Enforcement
• Summary

Module 1: Introduction to FastMCP

What Is FastMCP?

FastMCP is an open-source project written in Python that allows developers to build servers and clients for the Model Context Protocol (MCP) — an open-source standard that exposes data and functions to applications built around large language models (LLMs).

A brief history of the project:

• FastMCP was created by Jeremiah Lowin, an entrepreneur who also founded the workflow orchestration startup Prefect and was previously a member of the Apache Airflow Project Management Committee (PMC).
• Version 1 of FastMCP was released in April 2025, quickly followed by version 2.0 a couple of weeks later.
• The project receives frequent updates, and its developer ecosystem is vibrant: nearly 100 contributors, roughly 17,000 GitHub stars, and about 1,100 forks — accumulated within roughly five months.

FastMCP’s popularity stems from how it reduces the boilerplate normally required to build raw MCP servers: server setup, protocol handlers, content types, and error management are all abstracted away. It targets a high level of abstraction, in a way comparable to how Express.js simplifies building web servers in Node.js — a lightweight framework that hides underlying protocol complexity.

Background on the Model Context Protocol

MCP was created by Anthropic, the company behind the Claude family of LLMs. Anthropic announced MCP in a blog post in late November 2024. Initial attention was limited, but developers soon began experimenting with it and recognized that MCP was a significant step forward for building generative and agentic AI applications. Interest then spiked, MCP became one of the most discussed topics in the AI world, and it gained rapid adoption from major players such as OpenAI, Google, and Microsoft — companies that are normally fierce competitors, which signaled how important a shared standard like MCP had become to the growth of the AI ecosystem.

Reference material:

• MCP documentation: https://modelcontextprotocol.io/docs/getting-started/intro

The MxN Problem and Why MCP Matters

To understand why MCP exists, it helps to look at the traditional way of building LLM-based applications and the integration cost this creates — commonly called the MxN problem.

Consider building an AI system for a software team. It needs to pull updates from GitHub, check calendars and messages in Slack, and search a company’s knowledge base. If the system also needs to work across three different LLMs, each LLM requires its own custom connector to each tool. With 3 models and 3 tools, that means 3 × 3 = 9 custom connectors to build and maintain.

MCP flips this multiplication into addition. Each model needs only one connector to MCP, and each tool needs only one connector to MCP. The formula becomes M + N — 3 models plus 3 tools yields 6 connections instead of 9. The savings look modest at small scale, but they become enormous once a system needs dozens of models and hundreds of tools.

flowchart TB
    subgraph Without_MCP["Without MCP — M x N connectors"]
        direction LR
        M1[Model 1] --> T1[Tool: GitHub]
        M1 --> T2[Tool: Slack]
        M1 --> T3[Tool: Knowledge Base]
        M2[Model 2] --> T1
        M2 --> T2
        M2 --> T3
        M3[Model 3] --> T1
        M3 --> T2
        M3 --> T3
    end

    subgraph With_MCP["With MCP — M + N connectors"]
        direction LR
        N1[Model 1] --> MCP((MCP Layer))
        N2[Model 2] --> MCP
        N3[Model 3] --> MCP
        MCP --> S1[Tool: GitHub]
        MCP --> S2[Tool: Slack]
        MCP --> S3[Tool: Knowledge Base]
    end

MCP acts as a standardized layer that sits between models and tools. It also removes a limitation of earlier tool/function-calling systems, which typically ran only against cloud/server-side APIs: MCP allows LLMs and agents to interact with both cloud services and local machines through the same protocol. Because a growing ecosystem of pre-built MCP servers already exists for popular apps and data sources, developers can plug into existing integrations instead of reinventing connectors from scratch, leading to faster development with fewer bugs and less ongoing maintenance.

MCP is especially relevant to agentic AI systems — applications where models go beyond single-turn chatbot answers and instead plan, reason, and take multi-step actions using tools and data sources. Example: an e-commerce support scenario where a customer writes, “My package never arrived, but I was billed.”

• A traditional chatbot would likely give a generic reply and escalate to a human agent.
• An agentic system instead: interprets the customer’s intent, builds a plan, queries the shipping provider’s API, pulls order history from a database, checks billing records (all via standardized MCP interfaces rather than bespoke connectors), determines that the shipment was lost, automatically issues a refund, creates a replacement order, and sends the customer a personalized update — autonomously, without manual intervention.

MCP Architecture

A traditional LLM-based application is simple: a user interacts with a host (an application such as a chatbot, which could be web-based or run locally), and the host talks directly to the underlying LLM. The limitation of this model is that responses are constrained to what the model was trained on.

MCP extends this with a client–host–server architecture, which provides modularity and composability: instead of one monolithic system trying to do everything, functionality is broken into smaller, focused parts that can be reused, extended, or swapped independently.

flowchart LR
    User((User)) --> Host[MCP Host / Application]
    Host --> LLM[Large Language Model]
    Host --> Client[MCP Client]
    Client -->|1:1 dedicated channel| ServerA[MCP Server A<br/>e.g. local files]
    Client -->|1:1 dedicated channel| ServerB[MCP Server B<br/>e.g. database]
    Client -->|1:1 dedicated channel| ServerC[MCP Server C<br/>e.g. external API]

Key components:

• Host — the application the user interacts with (for example, ChatGPT, Claude, or an IDE such as VS Code).
• MCP Client — embedded inside the host; acts as a bridge that connects to MCP servers over dedicated, one-to-one channels. A single client can maintain multiple simultaneous connections to different servers, pulling in capabilities from multiple domains without conflicts.
• MCP Server — a lightweight program that exposes a focused set of capabilities (for example, accessing local files, retrieving database records, or calling external APIs).

Each server can expose three types of primitives — the fundamental building blocks of MCP functionality:

In practice, tools receive the most attention because they let LLMs take programmatic action rather than just generate text.

MCP currently supports three transport mechanisms for client/server communication:

Together, MCP clients, hosts, servers, tools, and resources form the foundation of what is considered an “AI agent.” By standardizing how these pieces interact, MCP makes AI systems more modular, composable, and scalable.

Pros and Cons of FastMCP

Common use cases that fit FastMCP’s strengths:

• Prototypes — spin up something realistic-looking without days of setup, useful for pitches.
• Hackathons — eliminates setup overhead so teams can focus on their solution under time pressure.
• Minimum Viable Products (MVPs) — validate an idea quickly rather than building a fully hardened production system.
• Students and learners — a low-friction entry point into MCP and agentic AI without deep protocol-level detail.

Setting Up FastMCP and Running a Simple Server

FastMCP documentation is available at https://gofastmcp.com/getting-started/welcome.

Although pip install fastmcp works, FastMCP recommends installing with uv, a fast, modern Python package installer designed as an alternative to pip, Poetry, or pip-tools:

uv pip install fastmcp

Verify the installation:

fastmcp version

FastMCP follows semantic versioning, but breaking changes can appear in minor version bumps because the project moves quickly to track changes in the underlying MCP protocol. For production environments, pin an exact version:

fastmcp==2.11.0

Demo: A Minimal “Hello World” MCP Server

The demo uses two files: hello.py (the server) and client.py (the client).

hello.py sets up the MCP server, imports FastMCP, names the server, and defines a tool called say_hello that takes a name parameter and returns a greeting string. The server is started with the HTTP transport on port 8000, with a startup print statement for visibility in the terminal.

from fastmcp import FastMCP

mcp = FastMCP("Hello World Server")

@mcp.tool
def say_hello(name: str) -> str:
    return f"Hello, {name}!"

if __name__ == "__main__":
    print("Starting MCP server at http://localhost:8000/mcp")
    mcp.run(transport="http", port=8000)

Run the server:

python hello.py

This prints a FastMCP startup banner showing the server name, transport, and URL — the server is now live and waiting for requests.

client.py imports the Client class from FastMCP, connects to the running server’s URL, calls the say_hello tool with name="World", and prints the result:

import asyncio
from fastmcp import Client

async def main():
    async with Client("http://127.0.0.1:8000/mcp") as c:
        result = await c.call_tool("say_hello", {"name": "World"})
        print(result)

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

Run the client:

python client.py

The response returns as JSON, containing content and metadata about the tool call, including the text Hello, World!.

sequenceDiagram
    participant C as client.py
    participant S as hello.py (MCP Server, HTTP :8000)

    Note over S: python hello.py<br/>Server starts and listens
    C->>S: python client.py<br/>call_tool("say_hello", {"name": "World"})
    S->>S: Execute say_hello("World")
    S-->>C: JSON result: { content: "Hello, World!", metadata: {...} }

FastMCP Cloud

FastMCP is also available as a hosted cloud offering, with several benefits:

• No server configuration — connect a repo or use a FastMCP template; works with Claude web, Claude Desktop, Cursor, or any MCP client.
• Scale — a cloud foundation that can process millions of requests, with usage-based billing and no separate server costs.
• Security — built-in OAuth with no extra configuration, plus support for bringing your own identity provider.
• Git-native CI/CD workflows.
• MCP-native analytics — visibility into actual request/response pairs, tool usage tracking, and user behavior understanding.
• ChatMCP — a ChatGPT-like AI assistant built into the platform.

MCP and the Agentic AI Ecosystem

On top of MCP, a growing set of frameworks and platforms make it faster, more flexible, and more accessible to build agent-based systems.

Development frameworks:

• LangChain and LangGraph — building blocks for agent development: chaining model calls, managing state, and orchestrating complex workflows. If MCP is “the plumbing,” LangChain/LangGraph are the scaffolding used to construct agent architectures.
• CrewAI — focused on multi-agent collaboration, where teams of agents with distinct roles work together, similar to a project team composed entirely of AI.
• AutoGen — specializes in automated multi-agent chat systems, where agents converse with each other, share reasoning, and dynamically solve problems with minimal constant human prompting.

Low-code / no-code platforms:

• Zapier and n8n (an open-source automation tool) — allow visual drag-and-drop composition of workflows that integrate agents into everyday business processes like customer support, data processing, or IT automation, without heavy coding.

It helps to think of the agentic AI ecosystem as a layered stack:

flowchart TB
    subgraph Platforms["Platforms (no-code / low-code)"]
        Zapier[Zapier]
        N8N[n8n]
    end
    subgraph Frameworks["Development Frameworks"]
        LangChain[LangChain / LangGraph]
        CrewAI[CrewAI]
        AutoGen[AutoGen]
    end
    subgraph Protocols["Protocols"]
        MCP2[MCP]
        UTCP2[UTCP]
        A2A2[A2A]
    end

    Platforms --> Frameworks --> Protocols

No layer is inherently “better” than another — no-code/low-code tools suit simpler systems, while deeper customization and flexibility require going further down the stack toward the protocol layer.

Alternatives to MCP

Several projects aim to simplify agent integration, enhance security, or improve production performance, either alongside MCP or as an alternative to it.

Practical Use Cases of MCP

Real-world organizations have adopted MCP to simplify agentic AI architectures:

Module 2: FastMCP Essentials

Comparing FastMCP to Standard MCP Implementations

Standard (raw) MCP workflow:

• Every tool needs a schema describing its inputs and outputs — typically hand-written, verbose JSON.
• The developer is responsible for managing the transport layer, parsing messages, and formatting results in the correct MCP wire format.
• Responses must match the declared schema exactly, and error serialization must be defined manually.
• The developer must also handle transport protocol selection (stdio, HTTP, etc.), schema validation, and protocol compliance — all layered on top of the actual business logic.

This is comparable to writing raw HTTP servers before frameworks like Flask or Express existed: doable, and sometimes necessary for deep customization or complex environments, but heavy.

FastMCP workflow:

• Python decorators replace hand-written JSON schemas. FastMCP inspects the decorated function and automatically generates the schema and handles parameter parsing.
• Whatever the function returns is automatically serialized into MCP-compatible JSON — no hand-crafted responses required.
• A built-in server runner means the MCP server is immediately ready to connect to VS Code, ChatGPT, Claude Desktop, or any other MCP-aware client.

# Conceptual illustration of the FastMCP decorator-based workflow
@mcp.tool
def my_function(param: str) -> dict:
    """FastMCP infers schema from type hints and this docstring."""
    return {"result": param}

FastMCP removes the heavy lifting of schema generation, serialization, and transport management, letting developers focus on writing useful tools quickly.

Connecting to an MCP Server in VS Code

Connecting an MCP server to VS Code requires GitHub Copilot (a generative AI chatbot integrated into the editor) and a GitHub account (the free tier is sufficient).

Setup steps:

1. Install the GitHub Copilot extension from the Extensions view, then connect a GitHub account.
2. Open the Copilot chat interface; note the mode selector (Agent, Ask, Edit). Agent mode is required for MCP tool usage, since the agent can decide to call an MCP tool instead of just responding from the underlying model.
3. Click the tools icon in the chat interface to see available tools — built-in ones plus any installed MCP servers.

There are three main ways to connect an MCP server to VS Code:

Example .vscode/mcp.json for an HTTP-based remote server (context7, no API key needed for the simple demo):

{
  "servers": {
    "context7": {
      "type": "http",
      "url": "https://mcp.context7.com/mcp"
    }
  }
}

Example .vscode/mcp.json for a local stdio-based server, launched via npx:

{
  "servers": {
    "context7": {
      "type": "stdio",
      "command": "npx",
      "args": ["-y", "@upstash/context7-mcp"]
    }
  }
}

• type — the transport (http or stdio).
• url — for HTTP servers, the endpoint VS Code will send requests to.
• headers — extra HTTP headers, such as an API key, when required.
• command / args — for stdio servers, the executable and command-line arguments used to launch the process (e.g., npx -y <package> fetches and runs a package without a global install).

Once configured, click Start in the mcp.json editor to launch the server, then use the tools icon in Copilot chat to confirm the server is active (a checkmark indicates it’s running). A prompt referencing the tool’s purpose (for example, asking about a framework feature “using context7”) will trigger a permission prompt before the tool is invoked — grant permission (“Allow”) to let Copilot call the tool.

context7 in particular helps address a common LLM limitation: models are trained as of a fixed cutoff date, which is a problem when working against fast-moving frameworks. context7 supplies up-to-date documentation context to the LLM.

Occasionally an MCP tool may not activate on the first attempt due to the inherently probabilistic nature of LLM reasoning — simply retry the prompt if this happens.

Connecting an MCP Server to ChatGPT

Using MCP with ChatGPT currently requires a paid ChatGPT plan.

Using a built-in connector (e.g., Gmail):

1. Open profile menu → Settings → Connectors to see connected applications (e.g., Canva, GitHub, Gmail, Google Calendar, Google Drive).
2. Select a connector (e.g., Gmail) to see available actions such as get_profile, get_recent_emails, read_email.
3. Start a new conversation, select + to enable Agent mode, then enable the desired source (e.g., Gmail) under Sources.
4. Submit a prompt such as “Please check my email.” Approve the sensitive-data access prompt (“I understand”). ChatGPT then searches the connected account (this can take roughly a minute) and returns results such as sender, subject, received time, and a snippet — accessed through MCP under the hood.

Using an external/custom MCP server (e.g., a Microsoft Docs MCP endpoint):

1. Copy the target MCP server’s HTTP endpoint URL.
2. In ChatGPT: profile → Settings → Connectors → Advanced settings → enable Developer mode → Create.
3. Provide a name (e.g., “Microsoft Docs MCP”), optionally a description, and paste in the MCP endpoint URL.
4. Authentication can be configured for secured servers; for a basic/public server it can be skipped.
5. Acknowledge the disclaimer that OpenAI has not verified the third-party MCP server, then confirm creation.
6. Submit a prompt such as, “Give me the Azure CLI commands to create an Azure Container App with a managed identity. Search Microsoft docs.” ChatGPT accesses the MCP server and returns a response with code snippets and clickable references back to the source documentation.

Connecting an MCP Server to Claude Desktop

Prerequisites: install Claude Desktop and install Node.js.

Demo using a simple time MCP server (a Python-based tool invoked via uvx, part of the uv package):

{
  "mcpServers": {
    "time": {
      "command": "uvx",
      "args": ["-q", "mcp-server-time", "--local-timezone=America/New_York"]
    }
  }
}

Steps:

1. In Claude Desktop: Settings → Developer → Edit Config. This opens the config JSON file in the system’s default editor.
2. Paste the server configuration above and save the file.
3. Restart Claude Desktop to activate the MCP server.
4. Return to Settings → Developer to confirm the local MCP server is listed as installed.
5. Start a new conversation, click the tools icon to confirm the time tool is active.
6. Submit a prompt such as, “Please show me the current time.” Claude finds and invokes the MCP tool and returns the current time (in this example, Eastern Time).

sequenceDiagram
    participant U as User
    participant CD as Claude Desktop (Host + Client)
    participant TMCP as time MCP Server (uvx mcp-server-time)

    U->>CD: "Please show me the current time."
    CD->>TMCP: Invoke time tool (stdio via uvx)
    TMCP-->>CD: Current time (America/New_York)
    CD-->>U: Displays current time

Building a Password Generator MCP Server

This demo builds a password generator MCP server using the @mcp.tool decorator, parameters, and structured results.

Key imports and their purpose:

• from __future__ import annotations — delays evaluation of type hints, enabling forward references and faster type checking.
• dataclass — a simple way to define data containers without hand-written constructors.
• Dict — type hints for dictionaries, clarifying expected key/value types.
• secrets — provides cryptographically secure random values, important for generating unpredictable passwords.
• math — supplies math functions (used here for entropy calculation).
• FastMCP — the framework class used to stand up the MCP server.

# password_server.py
from __future__ import annotations
from dataclasses import dataclass
from typing import Dict
import secrets
import math
from fastmcp import FastMCP

# Initialize the MCP server
mcp = FastMCP("Password Generator")

# Character sets used to build passwords
UPPER = "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
LOWER = "abcdefghijklmnopqrstuvwxyz"
DIGITS = "0123456789"
SYMBOLS = "!@#$%^&*()-_=+[]{};:,.<>?/\\|~`"
AMBIGUOUS = set("O0oIl1|")  # characters that look alike, e.g. 0 and O

def _filtered(chars: str, exclude_ambiguous: bool) -> str:
    """Optionally filter out ambiguous characters."""
    if not exclude_ambiguous:
        return chars
    return "".join(c for c in chars if c not in AMBIGUOUS)

def _entropy_bits(pool_size: int, length: int) -> float:
    """Estimate entropy in bits based on pool size and password length."""
    return length * math.log2(pool_size) if pool_size > 0 else 0.0

@dataclass
class PasswordResult:
    password: str
    length: int
    pool_size: int
    entropy_bits: float
    required_each_class: bool
    options: Dict[str, bool]

@mcp.tool
def generate_password(
    length: int = 16,
    include_upper: bool = True,
    include_lower: bool = True,
    include_digits: bool = True,
    include_symbols: bool = True,
    exclude_ambiguous: bool = True,
    require_each_class: bool = True,
) -> PasswordResult:
    """
    Generate a strong random password with configurable options.

    Args:
      length: password length (min 4, max 256)
      include_upper: include A-Z
      include_lower: include a-z
      include_digits: include 0-9
      include_symbols: include symbols
      exclude_ambiguous: remove ambiguous characters (like O/0, l/1)
      require_each_class: ensure at least one char from each selected class

    Returns:
      PasswordResult: password + metadata including estimated entropy (bits).
    """
    if length < 4 or length > 256:
        raise ValueError("length must be between 4 and 256")

    pools = []
    if include_upper:   pools.append(_filtered(UPPER, exclude_ambiguous))
    if include_lower:   pools.append(_filtered(LOWER, exclude_ambiguous))
    if include_digits:  pools.append(_filtered(DIGITS, exclude_ambiguous))
    if include_symbols: pools.append(_filtered(SYMBOLS, exclude_ambiguous))

    # Remove empty pools (e.g., if all chars were filtered out)
    pools = [p for p in pools if p]
    if not pools:
        raise ValueError("No characters available. Enable at least one class or disable ambiguity filter.")

    pool = "".join(pools)

    # Ensure at least one character from each required class
    password_chars = []
    if require_each_class:
        if length < len(pools):
            raise ValueError(f"length {length} too short for {len(pools)} required classes.")
        for p in pools:
            password_chars.append(secrets.choice(p))

    # Fill the rest of the password from the combined pool
    while len(password_chars) < length:
        password_chars.append(secrets.choice(pool))

    # Shuffle to avoid predictable positions (e.g., required classes always first)
    for i in range(len(password_chars) - 1, 0, -1):
        j = secrets.randbelow(i + 1)
        password_chars[i], password_chars[j] = password_chars[j], password_chars[i]

    password = "".join(password_chars)
    entropy = _entropy_bits(len(pool), length)

    return PasswordResult(
        password=password,
        length=length,
        pool_size=len(pool),
        entropy_bits=round(entropy, 2),
        required_each_class=require_each_class,
        options={
            "include_upper": include_upper,
            "include_lower": include_lower,
            "include_digits": include_digits,
            "include_symbols": include_symbols,
            "exclude_ambiguous": exclude_ambiguous,
        },
    )

if __name__ == "__main__":
    # Start the MCP server (stdio transport by default)
    mcp.run()

The @mcp.tool decorator registers the function with the FastMCP server so it becomes callable from VS Code or any MCP-aware client. Inside the function: character pools are built from the selected classes; if require_each_class is set, one character from each chosen category is guaranteed first, then the rest of the password is filled from the combined pool; characters are shuffled using secrets to avoid predictable positions; finally, entropy (password strength in bits) is calculated and returned alongside the password.

Registering the server in VS Code: open .vscode/mcp.json, click Add Server, choose Command (stdio), enter the command python password_server.py, name it “Password MCP,” and select Trust.

Sample interactions through GitHub Copilot chat:

• Prompt: “Create a 32-character password with symbols included.” → returns a generated password plus its metadata (length, pool size, entropy).
• Prompt: “Generate a 10-character password, but exclude ambiguous characters like O and 0.” → returns a matching password with ambiguous characters excluded.

To decide when to invoke this tool, the LLM evaluates the function name and its docstring, matching keywords in the prompt (e.g., “password,” “generate,” “length”) against them — reinforcing the importance of writing clear, descriptive function names and docstrings for every MCP tool. Parameters not specified in the prompt fall back to their declared defaults (for example, length=16).

Building a Weather MCP Server

This demo builds a Weather MCP server exposing two tools — get_weather and get_forecast — that return JSON-like structured results (rather than raw data retrieved from a live weather API; the demo uses deterministic stubbed data, which is useful for testing before wiring up a real backend).

# weather_server.py
from __future__ import annotations

from dataclasses import dataclass, asdict
from datetime import datetime, timedelta, timezone
from typing import List, Literal, Dict

from fastmcp import FastMCP

# -----------------------------------------------------------------------------
# Types & Data Models
# -----------------------------------------------------------------------------
Units = Literal["imperial", "metric"]

@dataclass
class Location:
    city: str
    country: str

@dataclass
class WeatherNow:
    location: Location
    temperature: float
    humidity: int
    condition: str
    observed_at: str
    units: Units

@dataclass
class ForecastDay:
    date: str
    high: float
    low: float
    condition: str

@dataclass
class WeatherForecast:
    location: Location
    days: List[ForecastDay]
    units: Units

# -----------------------------------------------------------------------------
# Server Initialization
# -----------------------------------------------------------------------------
mcp = FastMCP("Weather MCP (Stubbed)")

# -----------------------------------------------------------------------------
# Helpers (Deterministic Stubs)
# -----------------------------------------------------------------------------
# A minimal, deterministic baseline "climate" per city for demo purposes.
# You can expand this or switch to a real API later without changing output shapes.
BASELINES: Dict[str, Dict[str, float]] = {
    "san diego": {"temp_f": 72.0, "humidity": 55},
    "seattle":   {"temp_f": 64.0, "humidity": 70},
    "austin":    {"temp_f": 90.0, "humidity": 60},
    "london":    {"temp_f": 68.0, "humidity": 65},
    "new york":  {"temp_f": 75.0, "humidity": 58},
}

CONDITION_CYCLE = ["Sunny", "Partly Cloudy", "Cloudy", "Rain"]

def normalize_city(city: str) -> str:
    return city.strip().lower()

def to_celsius(fahrenheit: float) -> float:
    return round((fahrenheit - 32) * 5.0 / 9.0, 1)

def maybe_convert_temp(temp_f: float, units: Units) -> float:
    return round(temp_f, 1) if units == "imperial" else to_celsius(temp_f)

def iso_now() -> str:
    return datetime.now(timezone.utc).replace(microsecond=0).isoformat()

def pick_country(city_norm: str) -> str:
    # Very simple heuristic for demo purposes; customize as needed.
    if city_norm in {"london"}:
        return "GB"
    return "US"

def baseline_for_city(city_norm: str) -> Dict[str, float]:
    return BASELINES.get(city_norm, {"temp_f": 70.0, "humidity": 55})

def rotated_condition(offset: int) -> str:
    return CONDITION_CYCLE[offset % len(CONDITION_CYCLE)]

# -----------------------------------------------------------------------------
# Tools
# -----------------------------------------------------------------------------
@mcp.tool()
def get_weather(city: str, units: Units = "imperial") -> WeatherNow:
    """
    Return stubbed current weather for a city in structured form.

    Args:
      city: City name (e.g., "San Diego")
      units: "imperial" or "metric"

    Returns:
      WeatherNow dataclass with location, temperature, humidity, condition, timestamp, and units.
    """
    city_norm = normalize_city(city)
    if units not in ("imperial", "metric"):
        raise ValueError("units must be 'imperial' or 'metric'")

    base = baseline_for_city(city_norm)
    cond_index = abs(hash(city_norm)) % len(CONDITION_CYCLE)
    condition = rotated_condition(cond_index)

    temp = maybe_convert_temp(base["temp_f"], units)
    humidity = int(base["humidity"])

    return WeatherNow(
        location=Location(city=city.title(), country=pick_country(city_norm)),
        temperature=temp,
        humidity=humidity,
        condition=condition,
        observed_at=iso_now(),
        units=units,
    )

@mcp.tool()
def get_forecast(city: str, days: int = 5, units: Units = "imperial") -> WeatherForecast:
    """
    Return a stubbed multi-day forecast for a city.

    Args:
      city: City name (e.g., "Austin")
      days: Number of days (1..7)
      units: "imperial" or "metric"

    Returns:
      WeatherForecast dataclass with a list of ForecastDay entries.
    """
    if not (1 <= days <= 7):
        raise ValueError("days must be between 1 and 7")
    if units not in ("imperial", "metric"):
        raise ValueError("units must be 'imperial' or 'metric'")

    city_norm = normalize_city(city)
    base = baseline_for_city(city_norm)
    base_temp_f = base["temp_f"]
    start_offset = abs(hash(city_norm)) % len(CONDITION_CYCLE)

    forecast_days: List[ForecastDay] = []
    today = datetime.now(timezone.utc).date()

    for i in range(days):
        # Cycle conditions predictably: Sunny -> Partly Cloudy -> Cloudy -> Rain -> ...
        condition = rotated_condition(start_offset + i)
        wave = ((i % 3) - 1) * 3  # -3, 0, +3 repeating for a readable high/low pattern
        high_f = base_temp_f + 5 + wave
        low_f = base_temp_f - 5 + wave

        forecast_days.append(
            ForecastDay(
                date=(today + timedelta(days=i)).isoformat(),
                high=maybe_convert_temp(high_f, units),
                low=maybe_convert_temp(low_f, units),
                condition=condition,
            )
        )

    return WeatherForecast(
        location=Location(city=city.title(), country=pick_country(city_norm)),
        days=forecast_days,
        units=units,
    )

# -----------------------------------------------------------------------------
# Entrypoint
# -----------------------------------------------------------------------------
if __name__ == "__main__":
    # Start the MCP server (stdio transport by default)
    mcp.run()

Registering the server in VS Code: .vscode/mcp.json → Add Server → Command (stdio) → command python weather_server.py, named “Weather MCP,” scoped to the workspace (not global), then Trust.

Sample interactions:

• Prompt: “Get the weather for San Diego in imperial units.” → returns stubbed current weather for one of the built-in cities (San Diego, Seattle, Austin, London, New York).
• Prompt: “Give me a 5-day forecast for Austin.” → returns a 5-day forecast.

When designing tools that return data like this, focus on consistency: use the same location representation and the correct units across all related tools. Once results have been validated end-to-end, the stubbed data-retrieval functions can be swapped for a real weather API without changing the tools’ output shapes.

Building a Crypto MCP Server

This demo creates an MCP server that retrieves Bitcoin’s current price using the CoinMarketCap API (which has a free tier).

Setup:

1. Sign up for a CoinMarketCap API key.
2. Store the key as an environment variable rather than hard-coding it, using a .env file:

# .env
COINMARKETCAP_API_KEY=your_api_key_here

3. Install the required packages:

pip install requests
pip install python-dotenv

Key imports used in the server: typing.Literal (to restrict currency values, e.g., USD/EUR), time (timestamps), requests (HTTP calls to the CoinMarketCap API), and dotenv (securely loading the API key from .env).

# btc_price_server.py
from __future__ import annotations

from fastmcp import FastMCP
from typing import Literal, Dict, Any
import os
import time
import requests

from dotenv import load_dotenv, find_dotenv

# Load environment variables from a .env file if present.
# Search from the current working directory upward; fall back to the script directory
# (helpful if the MCP host changes the working directory).
def _init_env() -> None:
    try:
        dotenv_path = find_dotenv(usecwd=True)
    except Exception:
        dotenv_path = ""
    loaded = False
    if dotenv_path:
        loaded = load_dotenv(dotenv_path)
    if not loaded:
        try:
            import pathlib
            script_env = pathlib.Path(__file__).resolve().parent / ".env"
            if script_env.exists():
                load_dotenv(script_env)
        except Exception:
            pass

_init_env()

mcp = FastMCP("Bitcoin Price MCP (CoinMarketCap)")

VsCurrency = Literal["USD", "EUR", "GBP", "CAD", "AUD", "CHF", "JPY", "INR"]

CMC_QUOTES_URL = "https://pro-api.coinmarketcap.com/v1/cryptocurrency/quotes/latest"
USER_AGENT = "btc-price-mcp/1.0"
CACHE_TTL_SECONDS = 10

# Simple in-memory cache to avoid hitting the API too often
_CACHE: Dict[str, Dict[str, Any]] = {}

def _get_cached(key: str) -> dict | None:
    entry = _CACHE.get(key)
    if not entry or (time.time() - entry["ts"] > CACHE_TTL_SECONDS):
        return None
    return entry["data"]

def _set_cached(key: str, data: dict) -> None:
    _CACHE[key] = {"ts": time.time(), "data": data}

def _fetch_cmc_btc_spot(convert: str) -> dict:
    api_key = os.getenv("COINMARKETCAP_API_KEY")
    if not api_key:
        return {"error": "missing_api_key", "details": "Set COINMARKETCAP_API_KEY in your environment."}

    headers = {
        "X-CMC_PRO_API_KEY": api_key,
        "Accept": "application/json",
        "User-Agent": USER_AGENT,
    }
    params = {
        "symbol": "BTC",
        "convert": convert,  # e.g., USD
    }
    try:
        resp = requests.get(CMC_QUOTES_URL, headers=headers, params=params, timeout=10)
    except requests.RequestException as e:
        return {"error": f"network_error: {e.__class__.__name__}: {e}"}

    if resp.status_code != 200:
        return {"error": f"http_{resp.status_code}", "details": resp.text[:400]}

    try:
        j = resp.json()
        quote = j["data"]["BTC"]["quote"][convert]
        price = float(quote["price"])
        pct_24h = float(quote.get("percent_change_24h", 0.0))
        ts = quote.get("last_updated")
    except Exception as e:
        return {"error": f"parse_error: {e.__class__.__name__}", "details": str(e)[:400], "body": resp.text[:400]}

    return {
        "symbol": "BTC",
        "vs_currency": convert,
        "price": price,
        "percent_change_24h": pct_24h,
        "last_updated": ts,
        "source": "coinmarketcap",
        "endpoint": "quotes/latest",
        "attribution": "Data from CoinMarketCap",
    }

@mcp.tool(
    name="get_btc_price",
    description="Get the current BTC price (CoinMarketCap) in a target fiat currency."
)
def get_btc_price(vs_currency: VsCurrency = "USD") -> dict:
    """
    Args:
        vs_currency: One of USD, EUR, GBP, CAD, AUD, CHF, JPY, INR (default USD)
    Returns:
        { symbol, vs_currency, price, percent_change_24h, last_updated, source, ... } or { error, ... }
    """
    cur = vs_currency.upper()
    cache_key = f"btc:{cur}"
    hit = _get_cached(cache_key)
    if hit:
        return hit

    out = _fetch_cmc_btc_spot(cur)
    if "error" not in out:
        _set_cached(cache_key, out)
    return out

# Optional resource form
@mcp.resource("price://btc/{vs_currency}")
def btc_price_resource(vs_currency: VsCurrency = "USD") -> dict:
    return get_btc_price(vs_currency)

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

Registering the server in VS Code: .vscode/mcp.json → Add Server → Command (stdio) → command python btc_price_server.py, named “Crypto MCP.”

Sample interaction: prompt “What is the current price of bitcoin?” returns the current price, the 24-hour percentage change (in this example, a 1.32% decrease), and the data source (CoinMarketCap) — all built from the tool’s structured JSON response.

Building a Document and Knowledge Base MCP Server

MCP exposes functionality in two complementary ways:

• Tools — action-oriented functions invoked on demand (e.g., “get the current Bitcoin price”).
• Resources — data-oriented, addressable via stable URI-like identifiers; a client can fetch a resource directly, browse resources like a directory, or subscribe to updates if supported.

This demo builds an MCP server that manages project documents and knowledge-base/helpdesk articles using both resources (for stable, discoverable read access) and tools (for listing and updating content).

# mcp_docs_server.py
from __future__ import annotations

from dataclasses import dataclass, asdict
from typing import Dict, List, Optional, Literal
from datetime import datetime, timezone

from fastmcp import FastMCP

mcp = FastMCP("Docs & KB MCP")

# -----------------------------------------------------------------------------
# Simple in-memory stores for demo purposes
# -----------------------------------------------------------------------------
@dataclass
class Document:
    uri: str            # e.g., doc://project/plan
    title: str
    body: str
    tags: List[str]
    space: Literal["project", "helpdesk"]
    key: str            # slug or id
    last_updated: str   # ISO8601

# project docs by slug
PROJECT_DOCS: Dict[str, Document] = {}
# helpdesk articles by id (string for simplicity)
HELP_ARTICLES: Dict[str, Document] = {}

def _now_iso() -> str:
    return datetime.now(timezone.utc).isoformat()

def _mk_project_uri(slug: str) -> str:
    return f"doc://project/{slug}"

def _mk_help_uri(aid: str) -> str:
    return f"help://article/{aid}"

# Seed a couple of examples
PROJECT_DOCS["plan"] = Document(
    uri=_mk_project_uri("plan"),
    title="Project Plan",
    body="## Objectives\n- Ship MVP in 4 weeks\n- Define success metrics\n\n## Milestones\n1. Spec\n2. Build\n3. Test\n4. Launch",
    tags=["project", "plan", "mvp"],
    space="project",
    key="plan",
    last_updated=_now_iso(),
)

HELP_ARTICLES["123"] = Document(
    uri=_mk_help_uri("123"),
    title="How to Reset Your Password",
    body="1) Click 'Forgot Password'\n2) Check your email\n3) Follow the link to reset",
    tags=["help", "password", "account"],
    space="helpdesk",
    key="123",
    last_updated=_now_iso(),
)

# -----------------------------------------------------------------------------
# Resources (URI-addressable)
# -----------------------------------------------------------------------------

# Project documents: doc://project/{slug}
@mcp.resource("doc://project/{slug}")
def get_project_doc(slug: str) -> dict:
    """
    Fetch a project document by slug as a resource.
    Returns a JSON-like structure with title, body, tags, last_updated, and uri.
    """
    doc = PROJECT_DOCS.get(slug)
    if not doc:
        return {"error": "not_found", "details": f"project doc '{slug}' not found"}
    return asdict(doc)

# Helpdesk articles: help://article/{aid}
@mcp.resource("help://article/{aid}")
def get_help_article(aid: str) -> dict:
    """
    Fetch a helpdesk article by id as a resource.
    """
    doc = HELP_ARTICLES.get(aid)
    if not doc:
        return {"error": "not_found", "details": f"help article '{aid}' not found"}
    return asdict(doc)

# -----------------------------------------------------------------------------
# Tools (manage content & discover resources)
# -----------------------------------------------------------------------------

@mcp.tool(
    name="list_resources",
    description="List available document URIs for browsing."
)
def list_resources(kind: Literal["all", "project", "helpdesk"] = "all") -> dict:
    """
    Returns a list of URIs (and minimal metadata) to discover what's available.
    """
    items = []

    def add_entry(d: Document):
        items.append({
            "uri": d.uri,
            "title": d.title,
            "tags": d.tags,
            "last_updated": d.last_updated,
        })

    if kind in ("all", "project"):
        for d in PROJECT_DOCS.values():
            add_entry(d)

    if kind in ("all", "helpdesk"):
        for d in HELP_ARTICLES.values():
            add_entry(d)

    return {"count": len(items), "items": items}

@mcp.tool(
    name="upsert_project_doc",
    description="Create or update a project doc at doc://project/{slug}."
)
def upsert_project_doc(slug: str, title: str, body: str, tags: Optional[List[str]] = None) -> dict:
    """
    Create or update a project document. Returns the stored document record.
    """
    tags = tags or []
    uri = _mk_project_uri(slug)
    doc = Document(
        uri=uri,
        title=title,
        body=body,
        tags=tags,
        space="project",
        key=slug,
        last_updated=_now_iso(),
    )
    PROJECT_DOCS[slug] = doc
    return {"status": "ok", "document": asdict(doc)}

@mcp.tool(
    name="upsert_help_article",
    description="Create or update a help article at help://article/{id}."
)
def upsert_help_article(aid: str, title: str, body: str, tags: Optional[List[str]] = None) -> dict:
    """
    Create or update a help article. Returns the stored document record.
    """
    tags = tags or []
    uri = _mk_help_uri(aid)
    doc = Document(
        uri=uri,
        title=title,
        body=body,
        tags=tags,
        space="helpdesk",
        key=aid,
        last_updated=_now_iso(),
    )
    HELP_ARTICLES[aid] = doc
    return {"status": "ok", "document": asdict(doc)}

# -----------------------------------------------------------------------------
# Server entrypoint
# -----------------------------------------------------------------------------
if __name__ == "__main__":
    # Best for GitHub Copilot MCP (Developer Mode)
    mcp.run(transport="stdio")
    # Or, to expose HTTP (optional):
    # mcp.run(transport="streamable-http", host="127.0.0.1", port=8787)

Registering the server in VS Code: .vscode/mcp.json → Add Server → Command (stdio) → command python mcp_docs_server.py, named “Docs MCP Server.”

Sample interactions:

• “Show me doc://project/plan” → returns the project plan document.
• “Retrieve help://article/123” → returns the password-reset helpdesk article.
• “List all available documents.” → returns both seeded documents via the list_resources tool.
• “Show me only the helpdesk articles.” → returns just the helpdesk article via the kind="helpdesk" filter.

As a rule of thumb, resources and tools are usually paired in the same server: resources provide stable, discoverable access to data, while tools handle the actions and management around that data (creating, updating, listing).

Chaining Tools with a Crypto MCP

This demo extends the crypto server to compute a simple moving average (SMA) for Bitcoin and classify market sentiment as bullish, neutral, or bearish — introducing the concept of chaining tools together.

Server-level setup: VsCurrency restricts allowed fiat currencies; CMC_QUOTES_URL is the CoinMarketCap endpoint; USER_AGENT identifies the client to CoinMarketCap; CACHE_TTL_SECONDS = 10 caches responses briefly. In-memory state includes _CACHE (recent responses keyed by currency), _HISTORY (a bounded list of historical price points per currency, capped at _HISTORY_MAX = 1000 to bound memory), and _LAST_SPOT (the most recently fetched spot price per currency).

# btc_trend_server.py
from __future__ import annotations

import os
import time
import random
from typing import Dict, Any, List, Literal

import requests
from fastmcp import FastMCP
from dotenv import load_dotenv, find_dotenv

# Env init (mirrors btc_price_server.py pattern)
def _init_env() -> None:
    try:
        dotenv_path = find_dotenv(usecwd=True)
    except Exception:
        dotenv_path = ""
    loaded = False
    if dotenv_path:
        loaded = load_dotenv(dotenv_path)
    if not loaded:
        try:
            import pathlib
            script_env = pathlib.Path(__file__).resolve().parent / ".env"
            if script_env.exists():
                load_dotenv(script_env)
        except Exception:
            pass

_init_env()

mcp = FastMCP("BTC Trend MCP (SMA Signal)")

VsCurrency = Literal["USD", "EUR", "GBP", "CAD", "AUD", "CHF", "JPY", "INR"]

CMC_QUOTES_URL = "https://pro-api.coinmarketcap.com/v1/cryptocurrency/quotes/latest"
USER_AGENT = "btc-trend-mcp/1.0"
CACHE_TTL_SECONDS = 10

# Simple in-memory price cache and history per currency
_CACHE: Dict[str, Dict[str, Any]] = {}
_HISTORY: Dict[str, List[Dict[str, float]]] = {}  # key: vs_currency -> list of {ts, price}
_HISTORY_MAX = 1000
_LAST_SPOT: Dict[str, Dict[str, Any]] = {}  # last fetched spot by vs_currency

def _get_cached(key: str) -> dict | None:
    entry = _CACHE.get(key)
    if not entry or (time.time() - entry["ts"] > CACHE_TTL_SECONDS):
        return None
    return entry["data"]

def _set_cached(key: str, data: dict) -> None:
    _CACHE[key] = {"ts": time.time(), "data": data}

def _fetch_cmc_btc_spot(convert: str) -> dict:
    api_key = os.getenv("COINMARKETCAP_API_KEY")
    if not api_key:
        return {"error": "missing_api_key", "details": "Set COINMARKETCAP_API_KEY in your environment."}

    headers = {
        "X-CMC_PRO_API_KEY": api_key,
        "Accept": "application/json",
        "User-Agent": USER_AGENT,
    }
    params = {"symbol": "BTC", "convert": convert}
    try:
        resp = requests.get(CMC_QUOTES_URL, headers=headers, params=params, timeout=10)
    except requests.RequestException as e:
        return {"error": f"network_error: {e.__class__.__name__}: {e}"}

    if resp.status_code != 200:
        return {"error": f"http_{resp.status_code}", "details": resp.text[:400]}

    try:
        j = resp.json()
        quote = j["data"]["BTC"]["quote"][convert]
        price = float(quote["price"])
        pct_24h = float(quote.get("percent_change_24h", 0.0))
        ts = quote.get("last_updated")
    except Exception as e:
        return {"error": f"parse_error: {e.__class__.__name__}", "details": str(e)[:400], "body": resp.text[:400]}

    return {
        "symbol": "BTC",
        "vs_currency": convert,
        "price": price,
        "percent_change_24h": pct_24h,
        "last_updated": ts,
        "source": "coinmarketcap",
        "endpoint": "quotes/latest",
        "attribution": "Data from CoinMarketCap",
    }

def _append_history(vs: str, price: float, ts: float | None = None) -> None:
    ts = ts or time.time()
    hist = _HISTORY.setdefault(vs, [])
    hist.append({"ts": ts, "price": float(price)})
    if len(hist) > _HISTORY_MAX:
        del hist[: len(hist) - _HISTORY_MAX]

def _ensure_history_for_windows(
    vs: str,
    target_points: int,
    current_price: float,
    pct_24h_bias: float,
    simulate_if_needed: bool = True,
) -> int:
    """Ensure at least target_points exist in history for vs.

    Returns the number of synthetic points added (0 if none). Simulation uses a
    mild random walk around current_price, biased by pct_24h_bias, so demos work
    even before enough real history has accumulated.
    """
    hist = _HISTORY.setdefault(vs, [])
    have = len(hist)
    if have >= target_points or not simulate_if_needed:
        return 0

    to_add = target_points - have
    try:
        drift_per_min = (1.0 + pct_24h_bias / 100.0) ** (1.0 / 1440.0) - 1.0
    except Exception:
        drift_per_min = 0.0

    vol_bps = 5.0  # basis points of volatility per synthetic step
    vol = vol_bps / 10000.0

    base_price = hist[0]["price"] if hist else current_price
    synthetic: List[Dict[str, float]] = []
    now = time.time()
    for i in range(to_add, 0, -1):
        t = now - i * 60
        shock = random.gauss(mu=drift_per_min, sigma=vol)
        base_price *= (1.0 + shock)
        synthetic.append({"ts": t, "price": base_price})

    _HISTORY[vs] = synthetic + hist
    if len(_HISTORY[vs]) > _HISTORY_MAX:
        del _HISTORY[vs][: len(_HISTORY[vs]) - _HISTORY_MAX]
    return to_add

def _sma(values: List[float], window: int) -> float:
    if window <= 0 or window > len(values):
        raise ValueError("window must be in 1..len(values)")
    return float(sum(values[-window:]) / window)

def _signal_from_sma(short_sma: float, long_sma: float, threshold_bps: float) -> dict:
    if long_sma <= 0:
        return {"signal": "neutral", "ratio_bps": 0.0, "rationale": "Invalid long SMA"}
    ratio = (short_sma - long_sma) / long_sma
    ratio_bps = round(ratio * 10000.0, 2)
    if ratio_bps > threshold_bps:
        sig = "bullish"
    elif ratio_bps < -threshold_bps:
        sig = "bearish"
    else:
        sig = "neutral"
    rationale = (
        f"short SMA {short_sma:.2f} vs long SMA {long_sma:.2f}; "
        f"spread {ratio_bps:.2f} bps (threshold +/-{threshold_bps:.2f} bps)"
    )
    return {"signal": sig, "ratio_bps": ratio_bps, "rationale": rationale}

# --- Explicit multi-tool chaining: two distinct, separately-callable tools ---

@mcp.tool(
    name="fetch_btc_spot",
    description=(
        "Fetch the current BTC spot price (CoinMarketCap) and update in-memory history. "
        "Call this first in a chain before computing trend."
    ),
)
def fetch_btc_spot(vs_currency: VsCurrency = "USD") -> dict:
    """
    Fetch BTC spot price in the specified fiat currency, cache it briefly, and
    append the price to the in-memory history for subsequent trend calculations.

    Returns the spot payload or an error dict.
    """
    vs = vs_currency.upper()
    cache_key = f"btc:{vs}"
    hit = _get_cached(cache_key)
    if hit is None:
        spot = _fetch_cmc_btc_spot(vs)
        if "error" in spot:
            return spot
        _set_cached(cache_key, spot)
    else:
        spot = hit

    try:
        price = float(spot["price"])
    except Exception:
        return {"error": "invalid_spot_payload", "details": "Missing price from spot response."}

    _append_history(vs, price)
    _LAST_SPOT[vs] = spot
    return spot

@mcp.tool(
    name="compute_btc_trend_signal",
    description=(
        "Compute BTC SMA trend from in-memory price history. "
        "Chain by calling fetch_btc_spot first to update history with the latest price."
    ),
)
def compute_btc_trend_signal(
    vs_currency: VsCurrency = "USD",
    lookback_minutes: int = 60,
    short_window: int = 5,
    long_window: int = 20,
    neutral_threshold_bps: float = 25.0,
    simulate_if_needed: bool = True,
) -> dict:
    """
    Compute a simple SMA-based trend signal for BTC using already-fetched price history.

    This tool does NOT fetch data itself; it assumes fetch_btc_spot has already run.

    Returns:
        { symbol, vs_currency, price, last_updated, sma_short, sma_long, signal,
          ratio_bps, threshold_bps, points_used, source, attribution, note? }
        or { error, ... }
    """
    if short_window <= 0 or long_window <= 0:
        return {"error": "invalid_window", "details": "Windows must be positive"}
    if short_window >= long_window:
        return {"error": "invalid_window_order", "details": "short_window must be < long_window"}
    lookback_minutes = max(5, min(int(lookback_minutes), 1440))

    vs = vs_currency.upper()

    # Require that fetch_btc_spot has already populated the latest spot price
    spot = _LAST_SPOT.get(vs)
    if not spot:
        return {
            "error": "missing_spot",
            "details": "No recent spot price in memory. Call fetch_btc_spot first, then compute_btc_trend_signal.",
        }

    try:
        price = float(spot["price"])
    except Exception:
        return {"error": "invalid_spot_payload", "details": "Missing price from stored spot response."}
    pct_24h = float(spot.get("percent_change_24h", 0.0))

    target_points = max(long_window, min(lookback_minutes, _HISTORY_MAX))
    added = _ensure_history_for_windows(vs, target_points, price, pct_24h, simulate_if_needed)

    hist = _HISTORY.get(vs, [])
    now = time.time()
    cutoff_ts = now - lookback_minutes * 60
    recent = [p for p in hist if p["ts"] >= cutoff_ts]
    if len(recent) < long_window:
        return {
            "error": "insufficient_history",
            "details": f"Need at least {long_window} points, have {len(recent)}. Set simulate_if_needed=true or wait to accumulate.",
        }

    prices = [p["price"] for p in recent]

    try:
        sma_short = _sma(prices, short_window)
        sma_long = _sma(prices, long_window)
    except ValueError as e:
        return {"error": "sma_error", "details": str(e)}

    sig = _signal_from_sma(sma_short, sma_long, neutral_threshold_bps)

    return {
        "symbol": "BTC",
        "vs_currency": vs,
        "price": price,
        "last_updated": spot.get("last_updated"),
        "sma_short": round(sma_short, 2),
        "sma_long": round(sma_long, 2),
        "signal": sig["signal"],
        "ratio_bps": sig["ratio_bps"],
        "threshold_bps": float(neutral_threshold_bps),
        "points_used": len(prices),
        "source": spot.get("source", "coinmarketcap"),
        "attribution": spot.get("attribution", "Data from CoinMarketCap"),
        "endpoint": spot.get("endpoint", "quotes/latest"),
        "note": f"simulated {added} synthetic points to fill history" if added > 0 else "",
    }

# --- Implicit internal chaining: a convenience wrapper that runs both steps ---

@mcp.tool(
    name="get_btc_trend_signal",
    description=(
        "Convenience wrapper: fetch latest BTC spot and then compute the SMA trend signal. "
        "For explicit chaining, call fetch_btc_spot then compute_btc_trend_signal."
    ),
)
def get_btc_trend_signal(
    vs_currency: VsCurrency = "USD",
    lookback_minutes: int = 60,
    short_window: int = 5,
    long_window: int = 20,
    neutral_threshold_bps: float = 25.0,
    simulate_if_needed: bool = True,
) -> dict:
    spot = fetch_btc_spot(vs_currency=vs_currency)
    if "error" in spot:
        return spot
    return compute_btc_trend_signal(
        vs_currency=vs_currency,
        lookback_minutes=lookback_minutes,
        short_window=short_window,
        long_window=long_window,
        neutral_threshold_bps=neutral_threshold_bps,
        simulate_if_needed=simulate_if_needed,
    )

@mcp.resource("trend://btc/{vs_currency}")
def btc_trend_resource(vs_currency: VsCurrency = "USD") -> dict:
    # Resource wrapper using defaults; runs the full chain under the hood
    return get_btc_trend_signal(vs_currency=vs_currency)

if __name__ == "__main__":
    # stdio transport by default
    mcp.run()

There are two chaining styles demonstrated here:

Both approaches work because of stateful side effects: fetch_btc_spot records the latest price in memory, and compute_btc_trend_signal depends on that stored state.

sequenceDiagram
    participant U as User / LLM
    participant S1 as fetch_btc_spot
    participant Mem as In-memory cache & history
    participant S2 as compute_btc_trend_signal

    U->>S1: fetch_btc_spot(vs_currency="USD")
    S1->>Mem: append_history(price), set _LAST_SPOT
    S1-->>U: spot price payload
    U->>S2: compute_btc_trend_signal(vs_currency="USD")
    S2->>Mem: read _LAST_SPOT + _HISTORY
    S2-->>U: { signal: bullish/neutral/bearish, sma_short, sma_long, ... }

Registering the server in VS Code: .vscode/mcp.json → Add Server → Command (stdio) → command python btc_trend_server.py, named “Crypto MCP Chain.”

Sample interaction: prompt “What is the current price of Bitcoin and the sentiment?” triggers a call to fetch_btc_spot followed by a call to the sentiment-computing tool, returning a neutral signal in the example shown.

Module 3: Metrics, Observability, and Production Readiness for FastMCP

Structured Message Handling

Every tool call, response, and error in MCP is fundamentally a message moving between client and server. Loose or inconsistent messages create fragile systems — leading to errors, retries, debugging headaches, and even security issues. Structured message handling solves this by ensuring inputs and outputs follow explicit contracts. FastMCP already provides structured outputs via Python type hints; the focus here is enforcing structure on the input side, before a request even reaches the server.

Approaches and techniques:

• Wrap every message in a standard envelope. Use a Pydantic model that defines exactly which fields every message must carry: a version for compatibility, a unique msg_id for tracing/retries, a type to classify the message, and a payload containing the business data. This guarantees end-to-end traceability and allows safe upgrades without breaking older clients.

from pydantic import BaseModel
from typing import Any

class MessageEnvelope(BaseModel):
    version: str
    msg_id: str
    type: str
    payload: dict[str, Any]

• Validate early and fail fast. Parse raw JSON into the envelope model before it ever reaches the server’s business logic. If fields are missing or mistyped, the client catches the problem instantly, so malformed data never pollutes the server.
• Treat the payload as a first-class contract. Define strongly typed models for each tool’s payload rather than relying on freeform dictionaries. FastMCP automatically generates JSON schemas from these models and enforces validation, turning business logic into an enforced contract rather than “whatever JSON shows up.”
• Design for retries and correlation. Use msg_id as both a correlation ID and an idempotency key: clients generate a UUID per request, and servers log and skip duplicates when the same msg_id reappears. This guarantees safe retries (e.g., a customer is never double-charged) and enables end-to-end tracing of a workflow.
• Define a structured error model. Clients can branch logic based on an explicit error code (retry on timeout, halt on validation error) instead of parsing messy error strings, turning error handling into clean, programmatic logic.

Additional techniques worth adopting: publish schemas as resources so clients can fetch them dynamically; normalize error formats across all tools to simplify debugging; and propagate correlation IDs across chained tool calls to trace an entire workflow end-to-end.

Using Middleware in FastMCP

Middleware intercepts requests and responses as they flow through the server, enabling logging, security checks, or redaction of sensitive data before it reaches logs.

This demo builds middleware that logs every request/response and automatically masks sensitive fields (like passwords or tokens).

# middleware_server.py
from __future__ import annotations

import json
import logging
import sys
from typing import Any, Dict

from fastmcp import FastMCP
from fastmcp.server.middleware import Middleware, MiddlewareContext

# ---------------------------------------------------------------------
# Logging to a file + STDERR (safe choice for stdio transport)
# ---------------------------------------------------------------------
log = logging.getLogger("fastmcp.middleware")
log.setLevel(logging.INFO)
if not log.handlers:
    fh = logging.FileHandler("fastmcp_middleware.log", encoding="utf-8")
    eh = logging.StreamHandler(sys.stderr)  # IMPORTANT: stderr, not stdout
    fmt = logging.Formatter("[%(asctime)s] %(levelname)s %(message)s", "%Y-%m-%d %H:%M:%S")
    fh.setFormatter(fmt); eh.setFormatter(fmt)
    log.addHandler(fh); log.addHandler(eh)

def safe_json(obj: Any) -> str:
    try:
        return json.dumps(obj, default=str, ensure_ascii=False)
    except Exception:
        return f"<non-serializable:{type(obj).__name__}>"

def to_jsonable(obj: Any) -> Any:
    if hasattr(obj, "model_dump"):
        try:
            return obj.model_dump()
        except Exception:
            pass
    if isinstance(obj, (dict, list)):
        return obj
    return {"_type": type(obj).__name__}

def unwrap_toolresult(obj: Any) -> Any:
    for attr in ("result", "data", "value", "content", "output"):
        if hasattr(obj, attr):
            try:
                inner = getattr(obj, attr)
                if hasattr(inner, "model_dump"):
                    return inner.model_dump()
                return inner
            except Exception:
                pass
    if hasattr(obj, "model_dump"):
        try:
            return obj.model_dump()
        except Exception:
            pass
    return obj

def redact_content_parts_for_log(obj: Any, redact_fn):
    """
    For MCP 'content parts' (lists of parts), if a text part's text is JSON,
    parse -> redact -> re-serialize so logs show masked values.
    """
    def get_attr(o, name, default=None):
        return getattr(o, name, default) if not isinstance(o, dict) else o.get(name, default)

    if isinstance(obj, list) and all(
        (hasattr(p, "type") or (isinstance(p, dict) and "type" in p)) for p in obj
    ):
        view = []
        for p in obj:
            ptype = get_attr(p, "type")
            item = {"type": ptype}
            text_val = get_attr(p, "text")
            if isinstance(text_val, str):
                try:
                    parsed = json.loads(text_val)
                    red = redact_fn(parsed) if isinstance(parsed, (dict, list)) else parsed
                    item["text"] = json.dumps(red, ensure_ascii=False)
                except Exception:
                    item["text"] = text_val  # not JSON; log as-is
            else:
                item["preview"] = f"<{type(text_val).__name__}>"
            meta = get_attr(p, "meta")
            if meta:
                item["meta"] = meta
            view.append(item)
        return view
    return obj

class RedactingLoggingMiddleware(Middleware):
    SENSITIVE_KEYS = {
        "password", "api_key", "token", "secret", "authorization",
        "bearer", "access_token", "refresh_token",
    }

    @classmethod
    def _redact(cls, obj: Any) -> Any:
        if isinstance(obj, dict):
            out: Dict[str, Any] = {}
            for k, v in obj.items():
                out[k] = "***MASKED***" if k.lower() in cls.SENSITIVE_KEYS else cls._redact(v)
            return out
        if isinstance(obj, list):
            return [cls._redact(x) for x in obj]
        return obj

    async def on_message(self, context: MiddlewareContext, call_next):
        # ---- inbound ----
        try:
            payload = getattr(context.message, "model_dump", lambda: {})()
            log.info("-> %s from %s :: %s", context.method, context.source, safe_json(self._redact(payload)))
        except Exception as e:
            print(f"[middleware] inbound log failed: {e}", file=sys.stderr, flush=True)

        # ---- execute next handler ----
        result = await call_next(context)

        # ---- outbound ----
        try:
            unwrapped = unwrap_toolresult(result)
            prepared = redact_content_parts_for_log(unwrapped, self._redact)
            jsonable = to_jsonable(prepared)
            redacted = self._redact(jsonable) if isinstance(jsonable, (dict, list)) else jsonable
            log.info("<- %s :: %s", context.method, safe_json(redacted))
        except Exception as e:
            print(f"[middleware] outbound log failed: {e}", file=sys.stderr, flush=True)

        return result

    async def on_call_tool(self, context: MiddlewareContext, call_next):
        tool_name = getattr(getattr(context, "message", None), "name", "<unknown>")
        log.info("calling tool: %s", tool_name)
        return await call_next(context)

# ---------------------------------------------------------------------
# MCP Server + Sample Tool
# ---------------------------------------------------------------------
mcp = FastMCP("middleware-demo")
mcp.add_middleware(RedactingLoggingMiddleware())

@mcp.tool
async def echo_secret(message: str, password: str):
    """
    Echo a message and include intentionally sensitive-looking fields.
    NOTE: Middleware only redacts these fields in the logs; the client still
    receives the full (unredacted) response.
    """
    print(f"[echo_secret] message={message!r}, password={password!r}", file=sys.stderr, flush=True)
    return {"message": message, "password": password, "secret": "TOP_SECRET_VALUE"}

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

Registering the server in VS Code: .vscode/mcp.json → Add Server → Command (stdio) → command python middleware_server.py, named “MCP Middleware Server.”

Sample interaction: prompt “Use echo_secret with message=‘Hello FastMCP’ and password=‘supersecret’” returns the tool result to the client, but the resulting log file (fastmcp_middleware.log) shows the password field masked as ***MASKED*** — proving that redaction is applied to logs without altering what the client actually receives.

Middleware is broadly useful beyond logging: authentication, rate limiting, metrics collection, or even modifying responses before they’re returned to the client. Well-designed middleware makes MCP servers easier to monitor, safer to operate, and more production-ready.

Extending FastMCP with Plugins

A plug-in in FastMCP is a modular piece of functionality that can be dropped into an MCP server. Unlike tools written directly in the server file, plug-ins let you import and extend a server with pre-built logic — similar to installing a browser extension rather than rewriting the whole browser. This improves scalability, reusability, and maintainability. Plug-ins do not need to relate to the server’s original purpose (though grouping related plug-ins together is usually better practice in production).

This demo attaches two unrelated plug-ins — sentiment analysis and foreign-language translation — to a Weather MCP server.

# weather_plugins.py
from __future__ import annotations
from dataclasses import dataclass, asdict
from datetime import datetime, timedelta
from typing import List, Literal, Dict, Any
from fastmcp import FastMCP

Units = Literal["imperial", "metric"]

@dataclass
class Location:
    city: str
    country: str = "US"

@dataclass
class CurrentWeather:
    location: Location
    temperature: float
    humidity: int
    condition: str
    observed_at: str
    units: Units

@dataclass
class DailyForecast:
    date: str
    high: float
    low: float
    condition: str

@dataclass
class ForecastResponse:
    location: Location
    days: List[DailyForecast]
    units: Units

# Create the FastMCP instance (decorate tools off this)
server = FastMCP("weather-with-plugins")

# --- helpers/stubs ---
_CONDITIONS = ["Sunny", "Partly Cloudy", "Cloudy", "Showers", "Thunderstorms"]

def _now_iso() -> str:
    return datetime.utcnow().isoformat(timespec="seconds") + "Z"

def _fake_temp(city: str, units: Units) -> float:
    base = 72.0 if units == "imperial" else 22.0
    tweak = (sum(ord(c) for c in city) % 7) - 3  # -3..+3
    return float(base + tweak)

def _fake_forecast(city: str, units: Units, days: int) -> List[DailyForecast]:
    start = datetime.utcnow().date()
    out: List[DailyForecast] = []
    base = 72.0 if units == "imperial" else 22.0
    for i in range(days):
        d = start + timedelta(days=i + 1)
        delta = (i % 5) - 2
        high = base + 5 + delta
        low = base - 3 + delta
        cond = _CONDITIONS[(i + (sum(ord(c) for c in city) % 5)) % len(_CONDITIONS)]
        out.append(DailyForecast(date=d.isoformat(), high=float(high), low=float(low), condition=cond))
    return out

# --- core tools (decorated off `server`) ---
@server.tool
def get_weather(city: str, units: Units = "imperial") -> Dict[str, Any]:
    """Get current weather (stubbed) for a city."""
    if units not in ("imperial", "metric"):
        raise ValueError('units must be "imperial" or "metric"')
    cw = CurrentWeather(
        location=Location(city=city),
        temperature=_fake_temp(city, units),
        humidity=55,
        condition="Sunny",
        observed_at=_now_iso(),
        units=units,
    )
    return asdict(cw)

@server.tool
def get_forecast(city: str, days: int = 5, units: Units = "imperial") -> Dict[str, Any]:
    """Get a multi-day forecast (stubbed)."""
    if not (1 <= days <= 7):
        raise ValueError("days must be between 1 and 7")
    if units not in ("imperial", "metric"):
        raise ValueError('units must be "imperial" or "metric"')
    resp = ForecastResponse(location=Location(city=city), days=_fake_forecast(city, units, days), units=units)
    return {"location": asdict(resp.location), "days": [asdict(d) for d in resp.days], "units": resp.units}

# --- plugin registration (each plugin exposes register(server)) ---
try:
    import sentiment_plugin
    sentiment_plugin.register(server)
except Exception as e:
    print(f"[weather-with-plugins] sentiment_plugin not loaded: {e}")

try:
    import translate_plugin
    translate_plugin.register(server)
except Exception as e:
    print(f"[weather-with-plugins] translate_plugin not loaded: {e}")

if __name__ == "__main__":
    server.run()

The sentiment plug-in:

# sentiment_plugin.py
from __future__ import annotations
from typing import Dict, Any
from fastmcp import FastMCP

def register(server: FastMCP) -> None:
    """Attach sentiment tools to the given FastMCP server."""

    @server.tool
    def analyze_sentiment(text: str) -> Dict[str, Any]:
        """Analyze text sentiment (simple rule-based demo)."""
        t = text.lower()
        pos = any(w in t for w in ["love", "great", "awesome", "excellent", "amazing"])
        neg = any(w in t for w in ["hate", "bad", "terrible", "awful", "crash", "bug"])
        if pos and neg:
            label = "mixed"
        elif pos:
            label = "positive"
        elif neg:
            label = "negative"
        else:
            label = "neutral"
        return {"sentiment": label, "input": text}

The translation plug-in:

# translate_plugin.py
from __future__ import annotations
from typing import Dict, Any, Literal
from fastmcp import FastMCP

Lang = Literal["en", "es", "fr", "de", "it"]

_DICT = {
    ("hello", "es"): "hola",
    ("hello", "fr"): "bonjour",
    ("hello", "de"): "hallo",
    ("hello", "it"): "ciao",
}

def _fake_translate(text: str, target: Lang) -> str:
    key = (text.strip().lower(), target)
    return _DICT.get(key, f"[{target}] {text}")

def register(server: FastMCP) -> None:
    """Attach translation tools to the given FastMCP server."""

    @server.tool
    def translate_text(text: str, target_lang: Lang = "en") -> Dict[str, Any]:
        """Translate text to target_lang (demo stub)."""
        if target_lang not in ("en", "es", "fr", "de", "it"):
            raise ValueError("Unsupported target_lang. Use: en, es, fr, de, it.")
        return {"input": text, "target_lang": target_lang, "translated": _fake_translate(text, target_lang)}

Each plug-in module exposes a register(server) function that the main server calls to attach its tools — this convention keeps the registration workflow consistent across plug-ins.

flowchart LR
    Main[weather_plugins.py<br/>FastMCP server] -->|register server| Sentiment[sentiment_plugin.py<br/>analyze_sentiment tool]
    Main -->|register server| Translate[translate_plugin.py<br/>translate_text tool]

Registering the server in VS Code: .vscode/mcp.json → Add Server → Command (stdio) → command python weather_plugins.py, named “Weather Plugin MCP.”

Sample interactions:

• “Analyze the sentiment of: ‘I love the new feature but don’t like the bugs.’” → returns mixed sentiment.
• “Translate ‘hello’ to Spanish.” → returns hola.

State and Persistence

By default, MCP servers are stateless — each request is processed fresh with no memory of previous calls, which is fine for simple tools but insufficient for many real-world applications (for example, a shopping cart that must persist items across multiple requests).

# Conceptual shopping-cart example illustrating in-memory state
cart: list[dict] = []  # global variable, persists while the server process is running

@mcp.tool
def add_item(name: str, quantity: int = 1) -> dict:
    """Add an item to the shopping cart."""
    cart.append({"name": name, "quantity": quantity})
    return {"status": "ok", "cart": cart}

@mcp.tool
def view_cart() -> dict:
    """View everything currently in the cart."""
    return {"cart": cart}

Because cart is a global variable, every tool call shares access to the same list — each call to add_item grows the cart, and view_cart reflects everything added so far.

The important caveat: this kind of state only persists while the server process is running. Restarting the server resets the global list and loses all data. True persistence — surviving restarts — requires writing state to a file, database, or other durable storage backend.

Example interaction:

• “Add an apple to my cart.”
• “Now add a banana too.”
• “Show me everything in my cart.”

Expected structured JSON output reflects both items added to the cart.

Integrations in FastMCP

FastMCP is designed to connect with a wider ecosystem of identity providers, AI assistants, APIs, and development environments — rather than being a completely standalone server. An integration is a connector between an MCP server and an external system, letting the server stay lightweight and focused on business logic while integrations handle common cross-cutting needs.

Authentication integrations remove the need to manage passwords or build custom login flows — users sign in with an existing account, and the server automatically recognizes their identity, relying on hardened, well-tested providers. This matters most when MCP tools interact with sensitive systems or personal data.

Authorization integrations let you define policies externally (e.g., “a regular employee can only view data, but a manager can also approve changes”) instead of hard-coding rules in the server, keeping access policies maintainable as the system grows.

AI platform integrations mean developers or business users can invoke MCP tools without leaving their existing environment (an IDE, a chat assistant) rather than learning a new interface.

OpenAPI integration provides reusability and portability: organizations with existing OpenAPI specs can generate MCP servers from them (or vice versa) instead of rebuilding integrations from scratch, and the resulting MCP server becomes easier for other systems to consume since OpenAPI is already a common language across APIs, documentation, and tooling.

Typed Configuration with fastmcp.config

Plain environment variables work for small projects, but real-world MCP servers often need structured, validated configuration. FastMCP addresses this with a fastmcp.config-style class, conceptually similar to a Pydantic model, that automatically reads from environment variables and validates them on startup.

Benefits: cleaner and more predictable configuration, automatic validation with defaults, and easier deployment across environments (local, staging, production).

# Example configuration model
from fastmcp import FastMCPBaseSettings  # illustrative base class

class AppConfig(FastMCPBaseSettings):
    openai_api_key: str
    database_url: str
    debug: bool = False

When the server starts, FastMCP reads environment variables with matching names (e.g., OPENAI_API_KEY, DATABASE_URL). If a required value is missing or invalid, startup fails immediately with a clear validation error — no silent bugs or guesswork.

Once defined, the configuration model can be injected directly into tools as a default parameter — a pattern known as runtime injection:

config = AppConfig()

@mcp.tool
def show_config_info(config: AppConfig = config) -> dict:
    """Expose non-secret configuration/environment info for diagnostics."""
    masked_key = config.openai_api_key[:4] + "..." if config.openai_api_key else None
    return {
        "environment": "production" if not config.debug else "development",
        "database_configured": bool(config.database_url),
        "api_key_prefix": masked_key,
    }

This keeps configuration flexible without hard-coding values into function bodies — useful for switching between test/production databases or toggling debug/verbose logging.

When deploying to platforms like Render, Vercel, or AWS, .env files are typically replaced with platform-managed secrets defined directly in the deployment dashboard, so each environment (development, staging, production) can have its own settings without any code changes. This keeps secrets out of Git and makes deployments repeatable, safe, and simple to update without redeploying code.

Starting the server and testing configuration:

uvx mcp run config_server.py

Prompt example: “Use show_config_info” returns a structured resource showing the current environment and a masked API key prefix — confirming that FastMCP successfully read and validated the environment variables.

Developing Custom Middleware Patterns

Beyond simple logging, custom middleware can implement authentication, rate limiting, and distributed tracing — turning middleware into a genuine control plane for every request.

Authentication middleware — verify the caller before allowing tool execution:

class ApiKeyAuthMiddleware(Middleware):
    EXPECTED_KEY = "shared-secret-value"

    async def on_message(self, context, call_next):
        headers = getattr(context, "headers", {}) or {}
        api_key = headers.get("X-API-Key")
        if api_key != self.EXPECTED_KEY:
            raise PermissionError("Invalid or missing X-API-Key header")
        return await call_next(context)

This can be extended to OAuth tokens, JWTs, or signed headers for more robust authentication — the pattern ensures an MCP server is not left as an open door on the network.

Rate limiting middleware — track request frequency per client and reject requests over a threshold:

import time
from collections import defaultdict

class RateLimitMiddleware(Middleware):
    MAX_REQUESTS = 5
    WINDOW_SECONDS = 60

    def __init__(self):
        self._hits: dict[str, list[float]] = defaultdict(list)

    async def on_message(self, context, call_next):
        client_id = getattr(context, "source", "unknown")
        now = time.time()
        window_start = now - self.WINDOW_SECONDS
        self._hits[client_id] = [t for t in self._hits[client_id] if t > window_start]

        if len(self._hits[client_id]) >= self.MAX_REQUESTS:
            raise RuntimeError(f"Rate limit exceeded for {client_id}")

        self._hits[client_id].append(now)
        return await call_next(context)

In production, an in-memory counter should be replaced with a persistent store such as Redis so rate limits are enforced consistently across multiple server instances.

Tracing middleware — attach a unique trace_id to each request’s lifecycle for debugging and observability:

import uuid

class TracingMiddleware(Middleware):
    async def on_message(self, context, call_next):
        trace_id = str(uuid.uuid4())
        context.trace_id = trace_id
        log.info("trace_id=%s method=%s", trace_id, context.method)
        result = await call_next(context)
        log.info("trace_id=%s completed", trace_id)
        return result

These trace_id values can be forwarded to external observability tools such as OpenTelemetry, Datadog, or CloudWatch for full distributed tracing.

Middleware layers run in the order they are registered, so combining patterns requires careful ordering — typically authenticate first, then check rate limits, then trace:

flowchart LR
    Req([Incoming Request]) --> Auth[Authentication Middleware]
    Auth --> Rate[Rate Limiting Middleware]
    Rate --> Trace[Tracing Middleware]
    Trace --> Tool[Tool Execution]
    Tool --> Resp([Response])

On platforms like Render or Vercel, this composed middleware stack lets you handle production concerns — securing endpoints with tokens or OAuth, applying per-user or per-team quotas, adding distributed tracing headers — without touching core business logic. It is clean, reusable, and easy to test.

Debug Mode and Message Tracing

Every FastMCP server supports a debug flag. When enabled, the server outputs every request, response, and internal event, including full stack traces for failed tool calls — printing each message envelope (payloads, schemas, IDs) to the console. This is especially useful when developing a new tool and encountering schema mismatches or silent timeouts (for example, “tool not found” or “invalid schema” errors).

Because middleware runs in a stack, debugging sometimes means figuring out which layer is intercepting a message. FastMCP lets you inspect the currently attached middleware chain:

@mcp.tool
def get_middleware_chain() -> dict:
    """Return the currently attached middleware chain for debugging."""
    return {"middleware": [type(m).__name__ for m in mcp.middleware]}

FastMCP also automatically exposes metadata about registered tools — their input/output schemas can be inspected at runtime without restarting the server, which helps debug cases where a plug-in wasn’t properly registered or a schema differs from what’s expected:

@mcp.tool
def list_registered_tools() -> dict:
    """Return a JSON summary of every registered tool, including plug-ins."""
    return {
        "tools": [
            {"name": t.name, "description": t.description, "schema": t.input_schema}
            for t in mcp.tools
        ]
    }

During execution, each message carries a context (trace IDs, auth data, middleware-injected fields). Printing or inspecting this live context from within a chain is invaluable when troubleshooting missing trace IDs, API keys, or user-scoped data. It can also help to store message envelopes to a file, then reload and rerun them later to isolate and replay a specific failed request safely, observe middleware effects, and confirm a fix without generating new live traffic.

Observability with custom metrics. FastMCP does not ship metrics out of the box, but adding counters and latency tracking is straightforward:

import time
from collections import defaultdict

metrics = defaultdict(lambda: {"calls": 0, "errors": 0, "avg_latency": 0.0})

@mcp.middleware
async def metrics_collector(context, call_next):
    tool_name = getattr(context, "method", "unknown")
    start = time.time()
    try:
        result = await call_next(context)
        return result
    finally:
        elapsed = time.time() - start
        m = metrics[tool_name]
        m["calls"] += 1
        m["avg_latency"] = ((m["avg_latency"] * (m["calls"] - 1)) + elapsed) / m["calls"]

@mcp.tool
def get_metrics() -> dict:
    """Expose collected per-tool metrics: call counts, error counts, and average latency."""
    return dict(metrics)

• metrics is a defaultdict that automatically creates a record per tool with calls, errors (a placeholder for future error tracking), and avg_latency (a running average of call duration).
• metrics_collector wraps every request, recording the start time, awaiting the tool’s execution, then updating the tool’s call count and running average latency once complete.
• get_metrics exposes the collected stats as a queryable MCP tool.

Full distributed tracing with OpenTelemetry. For visibility across distributed systems, traces can be exported to observability backends such as Grafana, Datadog, or Honeycomb:

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

tracer_provider = TracerProvider()
tracer_provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
trace.set_tracer_provider(tracer_provider)
tracer = trace.get_tracer("fastmcp-server")

@mcp.middleware
async def tracing_middleware(context, call_next):
    with tracer.start_as_current_span(context.method) as span:
        span.set_attribute("msg_id", getattr(context, "msg_id", "unknown"))
        return await call_next(context)

• TracerProvider creates a manager for all traces in the application; get_tracer retrieves a tracer instance used to start spans, where a span represents one operation (e.g., a single MCP tool call).
• OTLPSpanExporter sends trace data to an OpenTelemetry collector or endpoint, usually via gRPC; BatchSpanProcessor batches spans efficiently instead of sending each one individually.
• tracing_middleware wraps every incoming request in a trace span using tracer.start_as_current_span, marking the start/end of the tool’s execution and attaching custom attributes such as msg_id. When the tool finishes, the span automatically closes and is exported — making every MCP tool call a traceable event that can be visualized for latency, error rates, and request flow in an observability platform.

Security and Policy Enforcement

MCP acts as a bridge between LLMs and real-world tools — powerful, but potentially dangerous if misused. Common threat categories to account for:

These risks are not a reason to avoid FastMCP — they are a reason to plan for security from day one. FastMCP already includes several built-in security layers, especially in its cloud offering:

• OAuth integrated by default — use FastMCP’s own identity provider or plug in an external one (e.g., Okta, Azure Active Directory) without writing custom authentication code.
• Secure multi-tenant cloud environment — isolated requests, encrypted credentials, and MCP-native analytics tracking which tools are used, by whom, and how often (important for enterprise governance).
• Schema-based validation — strict input/output formats for every tool, so malformed or malicious payloads are automatically rejected.

Vulnerabilities that teams often overlook:

Recommended steps to strengthen a FastMCP security posture:

1. Manage secrets carefully. Use vaults such as AWS Secrets Manager, HashiCorp Vault, or Azure Key Vault to store API keys and tokens securely; never hard-code credentials in server code or configuration files.
2. Enforce authentication and authorization. Every MCP server should verify requests via an API key, OAuth token, or JWT; integrate a policy enforcement service such as Permit.io or Eunomia for granular roles (which users/agents can call which tools or access which datasets).
3. Protect transports. stdio may be acceptable for development, but production should always use Streamable HTTP with HTTPS enabled to ensure data in transit is encrypted and verified.
4. Implement auditing and observability. Log every request, tool call, and error with metadata such as user ID, timestamp, and response time; aggregate via telemetry tools like OpenTelemetry or built-in FastMCP analytics; use dashboards to monitor unusual activity or spikes in failed authentication attempts.
5. Mask and redact sensitive data. Never return raw personally identifiable information (PII) or confidential business data in responses; apply redaction or anonymization before data leaves the server — especially important for compliance with privacy regulations such as GDPR and HIPAA.
6. Follow secure development hygiene. Regularly review dependencies, run security scans, and rotate API keys; treat FastMCP middleware and plug-ins as part of the application’s attack surface, because they are.

flowchart TB
    Client([Client / LLM Request]) --> TLS[Streamable HTTP + HTTPS]
    TLS --> AuthN[Authentication<br/>API Key / OAuth / JWT]
    AuthN --> AuthZ[Authorization / Policy Engine<br/>Permit.io, Eunomia, RBAC]
    AuthZ --> Validate[Schema Validation<br/>Reject malformed/malicious input]
    Validate --> Tool[Tool Execution]
    Tool --> Redact[Redact Sensitive Data<br/>before logging / response]
    Redact --> Audit[Audit Log + Observability<br/>OpenTelemetry / Analytics]
    Audit --> Resp([Response to Client])

Summary

FastMCP is a high-level, opinionated Python framework built on top of the Model Context Protocol (MCP) — the open standard, created by Anthropic, that lets LLMs and agentic AI systems interact with tools, data, and services through a modular client–host–server architecture. By replacing MCP’s boilerplate (manual schemas, transport handling, and serialization) with Python decorators and automatic schema generation, FastMCP dramatically lowers the barrier to building, connecting, and operating MCP servers, while still allowing a path toward production-grade concerns like middleware, typed configuration, observability, and security.

Key principles covered across this course:

• MCP solves the MxN integration problem, turning per-model-per-tool custom connectors into a shared M + N layer.
• MCP’s client–host–server architecture exposes three primitives — tools (model-invoked actions), resources (application-consumed structured/addressable data), and prompts (user-controlled templates) — over one of three transports: stdio (local/dev), SSE (legacy), or Streamable HTTP (production).
• FastMCP trades some flexibility and enterprise-grade support maturity for dramatically faster development, at the cost of being a young, fast-moving, opinionated dependency.
• MCP servers can be connected into real hosts — VS Code (via GitHub Copilot, using the Microsoft MCP directory, remote HTTP, or local stdio configuration), ChatGPT (via built-in or custom connectors), and Claude Desktop (via a local JSON config) — using consistent patterns.
• FastMCP tools can be composed through explicit chaining (separate tools relying on shared stored state) or implicit chaining (a wrapper tool that calls other tools internally).
• Production readiness depends on structured message handling, middleware (logging, redaction, authentication, rate limiting, tracing), plug-ins for modular extension, typed configuration, state/persistence strategies, debug/tracing tooling, and a defense-in-depth approach to security (authentication, authorization, schema validation, transport encryption, secrets management, and auditing).
• Real-world adoption (Walmart, Block, TrueFoundry, Visa) shows MCP being used to consolidate agent sprawl, enforce governance, and connect AI agents securely to enterprise systems at scale.

Quick Reference

Checklist Before Deploying a FastMCP Server to Production

• Pin an exact FastMCP version (e.g., fastmcp==2.11.0) rather than tracking latest.
• Use Streamable HTTP with HTTPS instead of stdio for networked deployments.
• Require authentication (API key, OAuth, or JWT) on every server.
• Apply schema validation to all tool inputs; whitelist only the tools actually needed.
• Add middleware for logging, redaction of sensitive fields, and rate limiting.
• Store secrets in a vault or platform-managed secret store — never hard-code them.
• Add tracing/metrics (custom counters or OpenTelemetry) for observability.
• Define a typed configuration model instead of relying on ad hoc environment variable reads.
• Document and test error handling using a structured error model.
• Review dependencies and rotate credentials on a regular cadence.

Search Terms

fastmcp · foundations · ai · agents · orchestration · artificial · intelligence · generative · mcp · server · connecting · crypto · message · middleware · production