mcp-server-fastmcp-python

# FastMCP Server Implementation FastMCP is a Python framework that simplifies MCP (Model Context Protocol) server development. Loading this skill makes you capable of building fully-featured MCP servers with Pydantic-validated tools, dynamic resources, prompt templates, and production-grade error handling using the FastMCP v3.x/4.0.0b1 SDK with Python 3.10+. ## TL;DR Checklist - [ ] Install: `pip install fastmcp pydantic` - [ ] Create server with `@mcp.tool` and `@mcp.resource` decorators - [ ] Validate inputs with Pydantic models, not raw strings - [ ] Use `McpError` for structured error responses, never raise bare `Exception` - [ ] Choose transport: stdio (local CLI tools), HTTP/Streamable (integrations) - [ ] Never `print()` on stdio transport — use logging module - [ ] Test with `mcp dev` CLI before deploying to production --- ## When to Use Use this skill when: - Building a new MCP server from scratch in Python - Integrating existing Python functions as MCP tools - Exposing APIs or resources through the MCP protocol - Creating reusable prompt templates for LLM clients - Implementing production-grade tool registration with validation - Debugging or fixing errors in an existing FastMCP server - Choosing between stdio vs HTTP transport for your server --- ## When NOT to Use Avoid this skill for: - Building MCP clients (use `mcp-client-integration` instead) - Designing tool interfaces (use `mcp-tool-design-patterns` instead) - Security and authorization policies (use `mcp-security-authorization` instead) - General MCP protocol questions without Python context - Non-Python MCP servers (JavaScript/Rust/Go have different SDKs) --- ## Core Workflow 1. **Define Server Class and Transport** — Create a FastMCP server instance with transport selection (stdio for local, HTTP/Streamable for remote). **Checkpoint:** Verify `mcp` object is properly initialized. 2. **Design Tool Signatures** — Sketch tool names, input parameters as Pydantic models, and expected outputs. Define constraints (required fields, ranges, validation rules) in Pydantic `Field()` definitions. **Checkpoint:** Validate each tool solves ONE specific problem. 3. **Register Tools with Decorators** — Use `@mcp.tool()` decorator with docstring describing behavior. Input validation is automatic via Pydantic. **Checkpoint:** Verify all tool parameters are type-hinted. 4. **Implement Resource Management** — Register static URIs (read-only data) and dynamic URIs (computed on-demand) with `@mcp.resource()` decorators. Define content MIME types. **Checkpoint:** Test resource retrieval with sample URIs. 5. **Add Prompt Templates** — Create reusable prompt templates with `@mcp.prompt()` decorator for common LLM use cases. Include template arguments as Pydantic models. **Checkpoint:** Verify all templates are self-documenting. 6. **Implement Error Handling** — Catch errors from tool execution, convert to `McpError` with proper error codes, never expose raw tracebacks. **Checkpoint:** Test error paths; verify all errors return diagnostic messages. 7. **Run and Validate Server** — Start server with `mcp.run()`, test with `mcp dev` CLI, verify tools/resources/prompts are discoverable. **Checkpoint:** Confirm server starts without stderr pollution on stdio transport. --- ## Implementation Patterns ### Pattern 1: Tool Registration with Pydantic Validation **Problem:** Tool inputs must be validated before execution. Raw string parameters are unsafe. **Solution:** Use Pydantic models for all tool inputs. FastMCP automatically converts JSON to Pydantic instances and validates constraints. ```python from fastmcp import FastMCP from pydantic import BaseModel, Field from typing import Optional # Initialize server with stdio transport (for local CLI usage) mcp = FastMCP(name="weather-service", version="1.0.0") # Define input schema as Pydantic model class WeatherQuery(BaseModel): """Request schema for weather lookup.""" location: str = Field( ..., description="City name or coordinates (e.g., 'San Francisco' or '37.7749,-122.4194')", min_length=1, max_length=200, ) units: str = Field( default="celsius", description="Temperature units: celsius, fahrenheit, kelvin", pattern="^(celsius|fahrenheit|kelvin)$", ) forecast_days: int = Field( default=1, description="Number of forecast days (1-14)", ge=1, le=14, ) @mcp.tool() def get_weather(query: WeatherQuery) -> dict: """Fetch current weather and forecast for a location. Args: query: WeatherQuery containing location, units, and forecast days. Returns: Dictionary with current conditions and forecast array. Raises: McpError: If location is not found or API fails. """ # At this point, query is already validated by Pydantic # location is guaranteed to be 1-200 characters # units matches the regex pattern # forecast_days is between 1-14 try: # Simulate API call if query.location.lower() == "invalid": from mcp.types import McpError, ErrorCode raise McpError( code=ErrorCode.INVALID_PARAMS, message=f"Location '{query.location}' not found in weather database", ) current = { "location": query.location, "temperature": 22.5, "units": query.units, "conditions": "Partly cloudy", "humidity": 65, } forecast = [ {"day": i + 1, "high": 24 + i, "low": 18 - i} for i in range(query.forecast_days) ] return { "current": current, "forecast": forecast, } except Exception as e: # Convert any internal error to McpError from mcp.types import McpError, ErrorCode raise McpError( code=ErrorCode.INTERNAL_ERROR, message=f"Weather service error: {str(e)}", ) if __name__ == "__main__": mcp.run() ``` **Key Points:** - Pydantic `Field()` provides automatic validation and documentation - `min_length`, `max_length`, `ge`, `le`, `pattern` enforce constraints - Type hints (`location: str`, `forecast_days: int`) are required - Docstrings appear in LLM client UIs - Errors are structured as `McpError` with proper error codes --- ### Pattern 2: Resource Management (Static + Dynamic URIs) **Problem:** Resources (read-only data) need both static lookup and dynamic generation. **Solution:** Use `@mcp.resource()` decorator for static URIs and `@mcp.resource_list()` for discovering available resources. ```python from fastmcp import FastMCP from mcp.types import McpError, ErrorCode from pydantic import BaseModel, Field from typing import Optional import json mcp = FastMCP(name="config-manager", version="1.0.0") # Static configuration resources CONFIG_STORE = { "app:production": {"database": "prod-db.example.com", "log_level": "INFO"}, "app:staging": {"database": "staging-db.example.com", "log_level": "DEBUG"}, "app:development": {"database": "localhost", "log_level": "TRACE"}, } @mcp.resource(uri_template="config://app/{env}") def get_app_config(env: str) -> str: """Retrieve application configuration for an environment. URI format: config://app/{env} Example: config://app/production """ if env not in CONFIG_STORE: raise McpError( code=ErrorCode.RESOURCE_NOT_FOUND, message=f"Environment '{env}' not found. Available: {', '.join(CONFIG_STORE.keys())}", ) return json.dumps(CONFIG_STORE[env], indent=2) @mcp.resource_list() async def list_configs() -> list[dict]: """List all available configuration resources.""" return [ { "uri": f"config://app/{env}", "name": f"Config for {env}", "mimeType": "application/json", } for env in CONFIG_STORE.keys() ] # Dynamic resource generation @mcp.resource(uri_template="stats://process/{pid}") def get_process_stats(pid: str) -> str: """Get runtime statistics for a process. URI format: stats://process/{pid} Example: stats://process/12345 """ try: pid_int = int(pid) except ValueError: raise McpError( code=ErrorCode.INVALID_PARAMS, message=f"Invalid PID: '{pid}' must be a number", ) # Simulate process stats lookup if pid_int < 1: raise McpError( code=ErrorCode.RESOURCE_NOT_FOUND, message=f"Process {pid_int} not running", ) stats = { "pid": pid_int, "memory_mb": 128.5 + pid_int, "cpu_percent": 25.3, "threads": 4, } return json.dumps(stats, indent=2) if __name__ == "__main__": mcp.run() ``` **Key Points:** - `uri_template` defines parameterized URIs (e.g., `config://app/{env}`) - Parameters are extracted automatically and passed as function arguments - `@mcp.resource_list()` returns discoverable resources for LLM clients - Always return `str` (JSON-serializable) from resource handlers - Use `McpError` for not-found or invalid resource cases --- ### Pattern 3: Complete Server (Tools + Resources + Prompts + Error Handling) **Problem:** Real-world servers need tools, resources, prompts, and robust error handling all together. **Solution:** Build a multi-capability server with all features integrated. ```python from fastmcp import FastMCP from mcp.types import McpError, ErrorCode from pydantic import BaseModel, Field from typing import Optional import json import logging from datetime import datetime # Configure logging (never print to stdout on stdio transport) logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" ) logger = logging.getLogger(__name__) # Initialize server mcp = FastMCP( name="data-analysis-server", version="1.0.0", ) # ============================================================================ # TOOLS # ============================================================================ class DatasetQuery(BaseModel): """Query schema for dataset analysis.""" dataset_id: str = Field( ..., description="Dataset identifier", min_length=1, max_length=50, pattern="^[a-zA-Z0-9_-]+$", ) operation: str = Field( default="summary", description="Analysis operation: summary, schema, sample, stats", pattern="^(summary|schema|sample|stats)$", ) limit: int = Field( default=10, description="Number of rows to return for sample operation", ge=1, le=1000, ) @mcp.tool() def analyze_dataset(query: DatasetQuery) -> dict: """Analyze a dataset and return statistics or samples. Supports multiple analysis operations: - summary: Basic metadata about the dataset - schema: Column names and types - sample: First N rows of data - stats: Statistical summary of numeric columns """ logger.info(f"Analyzing dataset: {query.dataset_id}, operation: {query.operation}") try: # Simulate dataset lookup datasets = { "sales-2024": { "rows": 50000, "columns": ["date", "product_id", "quantity", "price"], "types": ["timestamp", "string", "integer", "float"], }, "users-active": { "rows": 12500, "columns": ["user_id", "signup_date", "last_login", "country"], "types": ["string", "timestamp", "timestamp", "string"], }, } if query.dataset_id not in datasets: logger.warning(f"Dataset not found: {query.dataset_id}") raise McpError( code=ErrorCode.RESOURCE_NOT_FOUND, message=f"Dataset '{query.dataset_id}' not found. Available: {', '.join(datasets.keys())}", ) dataset_info = datasets[query.dataset_id] # Handle different operations if query.operation == "summary": result = { "dataset_id": query.dataset_id, "row_count": dataset_info["rows"], "column_count": len(dataset_info["columns"]), "columns": dataset_info["columns"], } elif query.operation == "schema": result = { "dataset_id": query.dataset_id, "schema": { col: col_type for col, col_type in zip(dataset_info["columns"], dataset_info["types"]) }, }