how-to-guide

How to Build MCP Servers with FastMCP for Python: A Complete Implementation Guide

July 24, 2026 composiohq/awesome-claude-skills ↗

FastMCP is the high-level Python framework from the official Model Context Protocol (MCP) SDK that lets you create production-ready MCP servers with automatic input validation, error handling, and multiple transport options.

This guide walks through building MCP servers using FastMCP based on the reference implementation in the ComposioHQ/awesome-claude-skills repository. Whether you are exposing internal APIs or wrapping third-party services, FastMCP streamlines tool registration, schema generation, and context management so you can focus on business logic.

What Is FastMCP?

FastMCP is the official high-level framework provided by the MCP Python SDK for creating Model-Context-Protocol servers. An MCP server exposes tools—functions that AI agents can invoke—with automatic handling of input validation, pagination, output formatting, and transport protocols. According to the python_mcp_server.md reference guide in the repository, FastMCP follows a convention where server names use the {service}_mcp format (e.g., github_mcp, slack_mcp).

Core Architecture Components

Building a robust FastMCP server requires understanding seven key architectural patterns demonstrated in the Composio reference implementation.

Server Initialization

Every FastMCP server starts with a single instance that encapsulates tool registration and transport configuration. In mcp-builder/reference/python_mcp_server.md, the pattern is to initialize the server with a descriptive name following the {service}_mcp convention.

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("example_mcp")  # Server name follows {service}_mcp convention

Tool Registration with Decorators

Tools are registered using the @mcp.tool decorator, which supplies machine-readable metadata through the annotations parameter. This decorator automatically generates OpenAPI-compatible schemas and registers the coroutine with the server instance.

@mcp.tool(
    name="example_search_users",
    annotations={
        "title": "Search Example Users",
        "readOnlyHint": True,
        "destructiveHint": False,
        "idempotentHint": True,
        "openWorldHint": True,
    },
)
async def example_search_users(params: UserSearchInput, ctx: Context | None = None) -> str:
    """Search for users in the Example system."""
    # Implementation details...

Input Validation via Pydantic

FastMCP uses Pydantic v2 models to define input schemas, automatically generating inputSchema entries for the MCP protocol and performing runtime validation. The reference implementation in mcp-builder/reference/python_mcp_server.md emphasizes using Field constraints for min/max lengths, ranges, and custom validators.

from pydantic import BaseModel, Field, ConfigDict, field_validator
from typing import Optional
from enum import Enum

class ResponseFormat(str, Enum):
    MARKDOWN = "markdown"
    JSON = "json"

class UserSearchInput(BaseModel):
    """Validated parameters for the search tool."""
    model_config = ConfigDict(str_strip_whitespace=True, validate_assignment=True)

    query: str = Field(..., description="Search string", min_length=2, max_length=200)
    limit: Optional[int] = Field(default=20, ge=1, le=100)
    offset: Optional[int] = Field(default=0, ge=0)
    response_format: ResponseFormat = Field(default=ResponseFormat.MARKDOWN)

    @field_validator('query')
    @classmethod
    def _no_blank(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("Query cannot be empty.")
        return v.strip()

Shared Utilities and Error Handling

Production servers should implement reusable utilities to keep code DRY. The reference implementation defines _make_api_request for async HTTP operations and _handle_api_error for consistent error formatting.

import httpx

API_BASE_URL = "https://api.example.com/v1"

async def _make_api_request(endpoint: str, method: str = "GET", **kwargs) -> dict:
    """Reusable async HTTP client with timeout handling."""
    async with httpx.AsyncClient() as client:
        resp = await client.request(
            method,
            f"{API_BASE_URL}/{endpoint}",
            timeout=30.0,
            **kwargs,
        )
        resp.raise_for_status()
        return resp.json()

def _handle_api_error(e: Exception) -> str:
    """Consistent, user-friendly error messages."""
    if isinstance(e, httpx.HTTPStatusError):
        if e.response.status_code == 404:
            return "Error: Resource not found."
        if e.response.status_code == 429:
            return "Error: Rate limit exceeded."
        return f"Error: API request failed (status {e.response.status_code})."
    if isinstance(e, httpx.TimeoutException):
        return "Error: Request timed out."
    return f"Error: Unexpected error – {type(e).__name__}."

Response Formats and Pagination

Tools should support both human-friendly Markdown and machine-readable JSON outputs, selectable via an enum. Pagination is handled through standard limit and offset parameters passed directly to the underlying API.

if params.response_format == ResponseFormat.MARKDOWN:
    lines = [f"# Results for `{params.query}`", ""]

    for user in users:
        lines.append(f"## {user['name']}")

    return "\n".join(lines)
else:
    return json.dumps({"users": users, "total": total}, indent=2)

Context Injection

FastMCP supports optional Context injection for advanced scenarios. By including ctx: Context | None = None in your tool signature, you gain access to progress reporting, logging, and interactive elicitations.

async def example_search_users(params: UserSearchInput, ctx: Context | None = None) -> str:
    if ctx:
        await ctx.report_progress(0, 100)
    # Tool implementation...

Transport Selection

FastMCP supports three transport mechanisms:

stdio (default): Standard input/output for local process communication
streamable_http: HTTP transport for web-based deployments
sse: Server-Sent Events for real-time streaming

if __name__ == "__main__":
    # Default stdio transport

    mcp.run()
    
    # Alternative: HTTP transport

    # mcp.run(transport="streamable_http", port=8000)

Complete FastMCP Server Implementation

Below is the full working example from the Composio repository, demonstrating all architectural components in a single file:

#!/usr/bin/env python3
"""
MCP Server for a fictional Example service.
Provides user search with pagination and selectable output format.
"""

from typing import Optional, List
from enum import Enum
import json
import httpx
from pydantic import BaseModel, Field, ConfigDict, field_validator
from mcp.server.fastmcp import FastMCP, Context

# ----------------------------------------------------------------------

# Server initialization

# ----------------------------------------------------------------------

mcp = FastMCP("example_mcp")

# ----------------------------------------------------------------------

# Constants & utilities

# ----------------------------------------------------------------------

API_BASE_URL = "https://api.example.com/v1"
CHARACTER_LIMIT = 25_000

async def _make_api_request(endpoint: str, method: str = "GET", **kwargs) -> dict:
    """Reusable async HTTP client."""
    async with httpx.AsyncClient() as client:
        resp = await client.request(
            method,
            f"{API_BASE_URL}/{endpoint}",
            timeout=30.0,
            **kwargs,
        )
        resp.raise_for_status()
        return resp.json()

def _handle_api_error(e: Exception) -> str:
    """Consistent error formatting."""
    if isinstance(e, httpx.HTTPStatusError):
        if e.response.status_code == 404:
            return "Error: Resource not found."
        if e.response.status_code == 403:
            return "Error: Permission denied."
        if e.response.status_code == 429:
            return "Error: Rate limit exceeded."
        return f"Error: API request failed (status {e.response.status_code})."
    if isinstance(e, httpx.TimeoutException):
        return "Error: Request timed out."
    return f"Error: Unexpected error – {type(e).__name__}."

# ----------------------------------------------------------------------

# Input models

# ----------------------------------------------------------------------

class ResponseFormat(str, Enum):
    MARKDOWN = "markdown"
    JSON = "json"

class UserSearchInput(BaseModel):
    """Validated search parameters."""
    model_config = ConfigDict(str_strip_whitespace=True, validate_assignment=True)

    query: str = Field(..., description="Search string", min_length=2, max_length=200)
    limit: Optional[int] = Field(default=20, ge=1, le=100)
    offset: Optional[int] = Field(default=0, ge=0)
    response_format: ResponseFormat = Field(default=ResponseFormat.MARKDOWN)

    @field_validator('query')
    @classmethod
    def _no_blank(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("Query cannot be empty.")
        return v.strip()

# ----------------------------------------------------------------------

# Tool definition

# ----------------------------------------------------------------------

@mcp.tool(
    name="example_search_users",
    annotations={
        "title": "Search Example Users",
        "readOnlyHint": True,
        "destructiveHint": False,
        "idempotentHint": True,
        "openWorldHint": True,
    },
)
async def example_search_users(params: UserSearchInput, ctx: Context | None = None) -> str:
    """Search for users with pagination and format selection."""
    try:
        data = await _make_api_request(
            "users/search",
            params={"q": params.query, "limit": params.limit, "offset": params.offset},
        )
        users = data.get("users", [])
        total = data.get("total", 0)

        if not users:
            return f"No users found matching '{params.query}'."

        if params.response_format == ResponseFormat.MARKDOWN:
            lines = [
                f"# User Search Results for `{params.query}`",

                "",
                f"Found {total} users (showing {len(users)}).",
                "",
            ]
            for u in users:
                lines.append(f"## {u['name']} ({u['id']})")

                lines.append(f"- **Email**: {u['email']}")
                if u.get("team"):
                    lines.append(f"- **Team**: {u['team']}")
                lines.append("")
            return "\n".join(lines)

        return json.dumps({
            "total": total,
            "count": len(users),
            "offset": params.offset,
            "users": users,
        }, indent=2)

    except Exception as exc:
        return _handle_api_error(exc)

# ----------------------------------------------------------------------

# Entry point

# ----------------------------------------------------------------------

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

Key Reference Files

The Composio repository contains several critical files for building and validating FastMCP servers:

mcp-builder/reference/python_mcp_server.md – The complete implementation guide, best-practice checklist, and extended code examples for building production MCP servers.
mcp-builder/SKILL.md – The skill definition that exposes the MCP builder functionality to Claude's skill marketplace.
mcp-builder/scripts/evaluation.py – A validation harness for testing your MCP server implementation against sample evaluation files.
mcp-builder/scripts/connections.py – Client-side connection examples showing how to connect to FastMCP servers via SSE, stdio, or HTTP transports.

Summary

FastMCP provides the high-level framework for building MCP servers in Python with minimal boilerplate.
Use @mcp.tool decorators with annotation dictionaries to register functions and supply machine-readable metadata.
Define input schemas using Pydantic v2 models with Field constraints for automatic validation and schema generation.
Implement shared utilities like _make_api_request and _handle_api_error to maintain consistent error handling across tools.
Support multiple response formats (Markdown and JSON) and pagination via standard limit/offset parameters.
Access advanced features like progress reporting through optional Context injection.
Choose the appropriate transport (stdio, streamable_http, or sse) based on your deployment environment.

Frequently Asked Questions

What is the difference between FastMCP and the low-level MCP SDK?

FastMCP is a high-level abstraction built on top of the official MCP Python SDK. While the low-level SDK requires manual handling of protocol messages, request routing, and schema generation, FastMCP automates these concerns through decorators and Pydantic integration. As implemented in mcp-builder/reference/python_mcp_server.md, FastMCP handles input validation, error serialization, and transport management automatically.

How do I handle authentication in a FastMCP server?

Authentication is typically handled within your shared utility functions, such as _make_api_request. You can inject API keys, OAuth tokens, or other credentials into the HTTP client headers before making requests. For sensitive credentials, use environment variables rather than hardcoding values, and consider implementing a configuration loader that validates required secrets at server startup.

Can FastMCP servers support multiple transports simultaneously?

No, a single FastMCP instance runs one transport at a time. You select the transport when calling mcp.run() by specifying the transport parameter (e.g., stdio, streamable_http, or sse). If you need to expose the same tools over multiple protocols simultaneously, you must run separate server instances or implement a proxy layer that routes between transports.

What are the size limits for MCP tool responses?

The reference implementation in the Composio repository suggests implementing a character limit (typically around 25,000 characters) to prevent oversized responses from overwhelming the context window. You should truncate large datasets or implement pagination parameters (limit and offset) to keep individual responses manageable. The _handle_api_error function should also catch timeout exceptions to prevent hanging on long-running operations.

Have a question about this repo?

These articles cover the highlights, but your codebase questions are specific. Give your agent direct access to the source. Share this with your agent to get started:

Share the following with your agent to get started:

curl -s "https://instagit.com/install.md"

Add to your MCP client configuration:

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}

Ask your agent:

"Use Instagit MCP to understand how composiohq/awesome-claude-skills works."

Works with

Claude Codex Cursor VS Code OpenClaw Any MCP Client

Maintain an open-source project? Get it listed too →