# How to Build MCP Servers with FastMCP for Python: A Complete Implementation Guide > Build production-ready MCP servers with FastMCP for Python. Learn how to implement this powerful framework for automatic input validation, error handling, and multiple transport options. - Repository: [Composio/awesome-claude-skills](https://github.com/composiohq/awesome-claude-skills) - Tags: how-to-guide - Published: 2026-07-24 --- **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](https://github.com/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](https://github.com/ComposioHQ/awesome-claude-skills/blob/master/mcp-builder/reference/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`](https://github.com/ComposioHQ/awesome-claude-skills/blob/main/mcp-builder/reference/python_mcp_server.md), the pattern is to initialize the server with a descriptive name following the `{service}_mcp` convention. ```python 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. ```python @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`](https://github.com/ComposioHQ/awesome-claude-skills/blob/main/mcp-builder/reference/python_mcp_server.md) emphasizes using `Field` constraints for min/max lengths, ranges, and custom validators. ```python 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. ```python 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. ```python 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. ```python 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 ```python 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: ```python #!/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`](https://github.com/ComposioHQ/awesome-claude-skills/blob/main/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`](https://github.com/ComposioHQ/awesome-claude-skills/blob/main/mcp-builder/SKILL.md)** – The skill definition that exposes the MCP builder functionality to Claude's skill marketplace. - **[`mcp-builder/scripts/evaluation.py`](https://github.com/ComposioHQ/awesome-claude-skills/blob/main/mcp-builder/scripts/evaluation.py)** – A validation harness for testing your MCP server implementation against sample evaluation files. - **[`mcp-builder/scripts/connections.py`](https://github.com/ComposioHQ/awesome-claude-skills/blob/main/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`](https://github.com/ComposioHQ/awesome-claude-skills/blob/main/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.