How to Build an MCP Server with Python SDK
Build production-ready Model Context Protocol servers in Python using FastMCP and the official SDK with custom tools, resources, and schemas.
Integrating large language models with enterprise backend systems historically suffered from fragmented integration patterns. Engineering teams wrote custom tool wrappers, hand-rolled JSON schema formatters, and maintained brittle synchronization logic for every model release. The Model Context Protocol (MCP) establishes an open, vendor-neutral specification (protocol version 2024-11-05) that decouples model orchestration from backend execution.
In this guide, we walk through building a production-ready MCP server in Python using FastMCP and the official Anthropic MCP SDK. We will implement strongly typed operational tools backed by Pydantic validation, static and dynamic URI resources, and automated workflow prompts. We will also examine how Python FastMCP compares against the TypeScript SDK and verify the entire system with an automated client test suite.
All code from this tutorial is available in the ZeroLabs Claude Recipes Monorepo.
Table of Contents
- Understanding MCP Core Primitives: Tools, Resources, and Prompts
- Architecture of FastMCP vs Low-Level Protocol Handlers
- Prerequisites and Environment Setup
- Step 1: Initializing the FastMCP Server
- Step 2: Defining Strongly Typed Tools with Pydantic
- Step 3: Emitting Logs and Progress with Context Injection
- Step 4: Exposing Static and Dynamic Resources
- Step 5: Registering Workflow Prompts
- Step 6: Executing Over Standard I/O (stdio)
- Step 7: Verifying with an Automated Client Test Suite
- Python FastMCP vs TypeScript MCP SDK Comparison
- Connecting FastMCP to Claude Desktop
- Production Hardening and Error Handling Guardrails
- Frequently Asked Questions
Understanding MCP Core Primitives: Tools, Resources, and Prompts
The Model Context Protocol specification governs client-server communication using structured JSON-RPC 2.0 messages. MCP divides external capabilities into three distinct primitives:
- Tools (
tools/list,tools/call): Executable functions that Claude or another LLM can invoke to perform side effects, query databases, or execute computational tasks. Tools accept structured parameters defined by JSON Schema and return content blocks (text, images, or resource references) to the model. - Resources (
resources/list,resources/read): Read-only context providers that expose documents, application configurations, database records, or telemetry streams. Resources are addressed via uniform resource identifiers (URIs) such ascluster://topology/overviewor parameterized URI templates such asplaybook://incident/{incident_type}. Unlike tools, resources do not trigger system side effects. - Prompts (
prompts/list,prompts/get): Server-managed prompt templates designed to steer user-facing interactions or guide multi-step analytical tasks. Prompts allow servers to expose recommended conversational recipes and system contexts to the host client.
For foundational API concepts on native tool declarations and schema validation, review our guides on How to Define and Register Tools with Claude and How to Implement Strict Tool Use and Error Recovery Loops.
Architecture of FastMCP vs Low-Level Protocol Handlers
Building an MCP server directly on the low-level mcp.server.lowlevel.Server requires writing explicit request routing tables, serializing JSON-RPC envelopes, and manually registering capability handlers (list_tools, call_tool, list_resources, read_resource).
FastMCP abstracts this boilerplate into high-level decorators while retaining complete protocol compliance:
┌────────────────────────────────────────────────────────────────────────┐│ MCP Client (Claude) │└───────────────────────────────────┬────────────────────────────────────┘ │ JSON-RPC 2.0 (stdio or SSE) ▼┌────────────────────────────────────────────────────────────────────────┐│ FastMCP Server Runtime ││ ││ ┌───────────────────────┐ ┌───────────────────┐ ┌──────────────────┐ ││ │ @mcp.tool() │ │ @mcp.resource() │ │ @mcp.prompt() │ ││ │ Pydantic v2 Parsing │ │ URI Template Route│ │ Template Parsing │ ││ │ Context Log Injection │ │ Read Serialization│ │ Message Builder │ ││ └───────────────────────┘ └───────────────────┘ └──────────────────┘ ││ ││ ┌──────────────────────────────────────────────────────────────────┐ ││ │ stdio Transport Stream │ ││ │ stdin (Incoming RPC Requests) / stdout (Outgoing RPC Responses) │ ││ └──────────────────────────────────────────────────────────────────┘ │└────────────────────────────────────────────────────────────────────────┘When using FastMCP:
- Function signatures and docstrings generate complete JSON Schema schemas for tool parameters and descriptions.
- Type hints enforce input boundaries. Pydantic models validate incoming arguments and reject malformed client inputs with descriptive errors.
- Decorator routing handles protocol handshakes, capability negotiation, and notification events without manual event loop management.
Prerequisites and Environment Setup
Before implementing the server, verify your local environment meets these requirements:
- Python 3.10 or higher.
- A functional virtual environment.
- The official
mcpPython SDK (mcp>=1.0.0,=2.0.0.
Set up the project directory and install the necessary dependencies:
mkdir -p mcp-python-servercd mcp-python-serverpython3 -m venv .venvsource .venv/bin/activatepip install "mcp>=1.0.0,=2.0.0" "httpx>=0.28.0"Verify that the SDK packages resolve properly:
python3 -c "import mcp; from mcp.server.fastmcp import FastMCP; print('MCP SDK ready!')"Step 1: Initializing the FastMCP Server
Create a new file named server.py. We begin by initializing the FastMCP application instance and defining our mock database state:
import jsonfrom typing import Optional, List, Dict, Anyfrom pydantic import BaseModel, Fieldfrom mcp.server.fastmcp import FastMCP, Contextimport mcp.types as types// Initialize FastMCP Server instancemcp = FastMCP( name="production-python-mcp-server", dependencies=["pydantic", "httpx"])// Simulated in-memory deployment stateDEPLOYMENTS_DB: Dict[str, Dict[str, Any]] = { "prod-api-1": { "id": "prod-api-1", "service": "billing-api", "environment": "production", "status": "healthy", "replicas": 4, "region": "us-east-1", "version": "v2.14.0", "error_rate_pct": 0.02 }, "staging-auth-2": { "id": "staging-auth-2", "service": "auth-service", "environment": "staging", "status": "degraded", "replicas": 2, "region": "us-west-2", "version": "v2.15.1-rc1", "error_rate_pct": 4.18 }}INCIDENT_PLAYBOOKS: Dict[str, str] = { "high-error-rate": "1. Verify canary metrics.\n2. Roll back container digest to stable.\n3. Flush query cache.\n4. Alert on-call platform engineer.", "memory-leak": "1. Collect heap dump.\n2. Scale horizontally.\n3. Cycle pod processes.\n4. Open priority investigation ticket."}The name parameter identifies your server during the initial protocol handshake. The optional dependencies list communicates required external packages to hosting runtimes.
Step 2: Defining Strongly Typed Tools with Pydantic
Tools expose operational actions to language models. While FastMCP supports native Python primitive arguments, wrapping complex parameters in Pydantic v2 BaseModel classes ensures strict type enforcement, numeric range validation, and comprehensive schema documentation.
Add the following Pydantic schemas and tool implementations to server.py:
// Pydantic Models for Input Validationclass DeploymentFilter(BaseModel): environment: Optional[str] = Field( default=None, description="Filter by environment: 'production', 'staging', or 'development'" ) status: Optional[str] = Field( default=None, description="Filter by status: 'healthy', 'degraded', or 'offline'" )class RestartDeploymentRequest(BaseModel): deployment_id: str = Field( ..., description="Unique deployment identifier, e.g. 'prod-api-1'" ) grace_period_seconds: int = Field( default=30, ge=5, le=300, description="Grace period in seconds for container teardown (5-300)" ) reason: str = Field( ..., min_length=5, max_length=200, description="Audit reason for triggering deployment restart" )Now attach the tool functions using the @mcp.tool() decorator:
// MCP Tools Registration@mcp.tool()async def query_deployments(filter_criteria: DeploymentFilter, ctx: Context) -> str: """Query infrastructure deployment records with structured filters. Inspects active microservice deployments and returns operational health status. """ await ctx.info(f"Filtering deployments with criteria: {filter_criteria.model_dump()}") results = [] for dep in DEPLOYMENTS_DB.values(): if filter_criteria.environment and dep["environment"] != filter_criteria.environment: continue if filter_criteria.status and dep["status"] != filter_criteria.status: continue results.append(dep) return json.dumps({ "total_matched": len(results), "deployments": results }, indent=2)@mcp.tool()async def restart_service(request: RestartDeploymentRequest, ctx: Context) -> str: """Trigger graceful restart for a target service deployment. Validates deployment existence, emits operational logs via Context, and updates state. """ await ctx.info(f"Received restart request for deployment '{request.deployment_id}'") if request.deployment_id not in DEPLOYMENTS_DB: await ctx.error(f"Deployment '{request.deployment_id}' not found in cluster registry") raise ValueError(f"Deployment '{request.deployment_id}' not found.") dep = DEPLOYMENTS_DB[request.deployment_id] dep["status"] = "healthy" dep["error_rate_pct"] = 0.00 await ctx.info(f"Deployment '{request.deployment_id}' restarted. Reason: {request.reason}") return json.dumps({ "status": "success", "action": "graceful_restart", "deployment_id": request.deployment_id, "grace_period_seconds": request.grace_period_seconds, "reason": request.reason, "current_state": dep }, indent=2)Notice how ge=5, le=300 on grace_period_seconds prevents models from passing values outside safe ranges. If a client passes an invalid value, FastMCP raises a validation error before the function body executes.
Step 3: Emitting Logs and Progress with Context Injection
When executing an MCP tool over standard input/output (stdio), writing to print() or sys.stdout corrupts the JSON-RPC communication stream.
FastMCP provides a Context injection mechanism. By adding a parameter of type Context to any @mcp.tool() function signature, FastMCP automatically injects the active request context:
@mcp.tool()async def long_running_task(task_id: str, ctx: Context) -> str: # 1. Structured logging (info, warning, error, debug) await ctx.info(f"Beginning execution for task {task_id}") # 2. Bidirectional progress reporting to client await ctx.report_progress(progress=25, total=100) # 3. Reading client session metadata request_id = ctx.request_id await ctx.info(f"Processing on RPC request ID: {request_id}") await ctx.report_progress(progress=100, total=100) return f"Task {task_id} completed successfully."The MCP client intercepts these log and progress notifications out of band, allowing the UI or agent orchestrator to display live updates without interfering with the tool return value.
Step 4: Exposing Static and Dynamic Resources
Resources expose read-only state to Claude. We declare resources using @mcp.resource(). FastMCP supports both static URIs and parameterized dynamic URI templates:
// MCP Resources (Static & Dynamic)@mcp.resource("cluster://topology/overview")def get_cluster_overview() -> str: """Read-only topology overview of all managed cluster nodes and services.""" return json.dumps({ "cluster_name": "us-east-core-production", "kubernetes_version": "1.31.2", "total_nodes": 12, "managed_deployments": list(DEPLOYMENTS_DB.keys()) }, indent=2)@mcp.resource("playbook://incident/{incident_type}")def get_incident_playbook(incident_type: str) -> str: """Retrieve procedural remediation steps for specific incident classes.""" if incident_type not in INCIDENT_PLAYBOOKS: return f"Unknown incident type: '{incident_type}'. Available: {list(INCIDENT_PLAYBOOKS.keys())}" return INCIDENT_PLAYBOOKS[incident_type]Key resource behaviors:
cluster://topology/overview: Appears inresources/listas a concrete resource that Claude can read before planning an operational task.playbook://incident/{incident_type}: Appears inresources/templates/list. When Claude encounters an incident involving high error rates, it queriesplaybook://incident/high-error-rateto retrieve targeted documentation.
Step 5: Registering Workflow Prompts
Prompts allow an MCP server to provide pre-engineered prompt workflows directly into the user interface or orchestrator.
Add the prompt registration to server.py:
// MCP Prompts Registration@mcp.prompt()def triage_service_alert(service_name: str, error_snippet: str) -> str: """Generate structured prompt instructions for triaging production alerts.""" return ( f"You are a Site Reliability Engineer triaging an alert on service '{service_name}'.\n\n" f"Alert Log Output:\n" f"```\n{error_snippet}\n```\n\n" f"Instructions:\n" f"1. Query deployments using 'query_deployments' to check current operational health.\n" f"2. Inspect the relevant incident playbook via 'playbook://incident/high-error-rate'.\n" f"3. Recommend remediation actions or trigger 'restart_service' if safe." )When an operator selects triage_service_alert in Claude Desktop or an automated incident pipeline, Claude receives the pre-structured context alongside tool calling recommendations.
Step 6: Executing Over Standard I/O (stdio)
To run the server, append the standard execution entrypoint:
// Server Entry Pointif __name__ == "__main__": # Launch server with standard I/O transport mcp.run(transport="stdio")When invoked via python server.py, FastMCP listens on stdin for JSON-RPC 2.0 messages and writes responses to stdout. All internal server logs are routed to stderr to protect transport integrity.
Step 7: Verifying with an Automated Client Test Suite
To verify that the MCP server initializes correctly and responds to protocol requests, create a client test script named python/client_test.py:
"""Automated Client Test for FastMCP Python Server over Stdio."""import sysimport osimport asyncioimport jsonfrom mcp import ClientSession, StdioServerParametersfrom mcp.client.stdio import stdio_clientasync def run_client_probe(): current_dir = os.path.dirname(os.path.abspath(__file__)) server_script = os.path.join(current_dir, "server.py") python_bin = sys.executable print(f"[*] Launching MCP Server: {python_bin} {server_script}") server_params = StdioServerParameters( command=python_bin, args=[server_script], env=os.environ.copy() ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: # 1. Initialize session handshake print("[*] Initializing MCP Session...") init_result = await session.initialize() print(f"[+] Connected to: {init_result.serverInfo.name} (Protocol: {init_result.protocolVersion})") # 2. Verify and call tools print("\n[Section 1: Testing Tools]") tools_response = await session.list_tools() tool_names = [t.name for t in tools_response.tools] print(f"[+] Tools registered: {tool_names}") assert "query_deployments" in tool_names assert "restart_service" in tool_names tool_call_res = await session.call_tool( "query_deployments", {"filter_criteria": {"environment": "production"}} ) data = json.loads(tool_call_res.content[0].text) print(f"[+] Matched deployments: {data['total_matched']}") assert data["total_matched"] == 1 # 3. Verify and read resources print("\n[Section 2: Testing Resources]") resources_response = await session.list_resources() resource_uris = [str(r.uri) for r in resources_response.resources] print(f"[+] Resources: {resource_uris}") cluster_res = await session.read_resource("cluster://topology/overview") print(f"[+] Resource content verified:\n{cluster_res.contents[0].text[:80]}...") # 4. Verify and retrieve prompts print("\n[Section 3: Testing Prompts]") prompts_response = await session.list_prompts() prompt_names = [p.name for p in prompts_response.prompts] print(f"[+] Prompts: {prompt_names}") assert "triage_service_alert" in prompt_names print("\n[SUCCESS] All FastMCP client probe tests passed cleanly!")if __name__ == "__main__": asyncio.run(run_client_probe())Execute the test suite directly from your terminal:
python3 python/client_test.pyExpected output:
[*] Launching MCP Server: /path/to/.venv/bin/python server.py[*] Initializing MCP Session...[+] Connected to: production-python-mcp-server (Protocol: 2024-11-05)[Section 1: Testing Tools][+] Tools registered: ['query_deployments', 'restart_service'][+] Matched deployments: 1[Section 2: Testing Resources][+] Resources: ['cluster://topology/overview'][+] Resource content verified:{ "cluster_name": "us-east-core-production", "kubernetes_version": "1.31.2"...[Section 3: Testing Prompts][+] Prompts: ['triage_service_alert'][SUCCESS] All FastMCP client probe tests passed cleanly!Our automated probe completes the full handshake, tool invocation, and resource fetch cycle in under 1,500 ms.
Python FastMCP vs TypeScript MCP SDK Comparison
Engineering teams often evaluate whether to implement MCP servers in Python or TypeScript. Here is how the two official SDKs compare:
| Dimension | Python SDK (FastMCP) | TypeScript SDK (@modelcontextprotocol/sdk) |
|---|---|---|
| API Paradigm | High-level declarative decorators (@mcp.tool()) | Object-oriented server class with handler callbacks |
| Schema Validation | Native Pydantic v2 models and Python type hints | Zod schema objects (z.object({})) |
| Context Access | Injected Context object in function parameters | Explicit request handler context argument |
| Transport Support | Standard I/O (stdio) and Server-Sent Events (SSE) | Standard I/O (stdio) and Server-Sent Events (SSE) |
| Best Used For | AI pipelines, data science workflows, Python services | Node.js web backends, browser tools, TypeScript stacks |
Equivalent TypeScript Implementation Pattern
In the TypeScript SDK, equivalent functionality requires registering tools via server.setRequestHandler(ListToolsRequestSchema, ...) or using the lightweight McpServer helper with Zod:
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";import { z } from "zod";const server = new McpServer({ name: "production-ts-mcp-server", version: "1.0.0"});// Defining tool with Zod validationserver.tool( "query_deployments", { environment: z.enum(["production", "staging", "development"]).optional(), status: z.enum(["healthy", "degraded", "offline"]).optional() }, async ({ environment, status }) => { return { content: [{ type: "text", text: JSON.stringify({ matched: 1 }) }] }; });const transport = new StdioServerTransport();await server.connect(transport);FastMCP provides a more succinct syntax for Python developers by leveraging native decorators and Pydantic models.
Connecting FastMCP to Claude Desktop
To test your FastMCP server directly within Claude Desktop, add your server configuration to the desktop configuration file:
- macOS:
~/Library/Application Support/Claude/claude_desktop_config.json - Linux:
~/.config/Claude/claude_desktop_config.json - Windows:
%APPDATA%\Claude\claude_desktop_config.json
Add the server entry:
{ "mcpServers": { "python-production-ops": { "command": "/absolute/path/to/mcp-python-server/.venv/bin/python", "args": [ "/absolute/path/to/mcp-python-server/python/server.py" ], "env": { "APP_ENV": "production" } } }}After restarting Claude Desktop, a hammer icon will appear in the lower right corner of the chat input box. Claude will automatically discover query_deployments, restart_service, cluster://topology/overview, and the prompt template.
Production Hardening and Error Handling Guardrails
When promoting an MCP server from local development to production infrastructure, enforce the following guardrails:
- Stdio Stream Isolation: Never allow third-party libraries or internal routines to write raw strings to standard output (
sys.stdout). A single strayprint()statement will corrupt the JSON-RPC framing and terminate the client session. Route all diagnostic output throughctx.info()or write explicitly tosys.stderr. - Schema Sanitization: Keep field descriptions concise and informative. Claude uses Pydantic field docstrings and descriptions to reason about when and how to invoke tools.
- Execution Timeouts: Wrap external HTTP requests and database calls in strict timeouts using
httpxorasyncio.wait_for. MCP clients enforce default response timeouts; hanging tool calls will degrade the user experience. - Execution Sandboxing: When tools execute system commands or file modifications, enforce strict sandboxing. Refer to our operational guide on How to Integrate and Sandbox the Claude Bash Tool for sandbox hardening patterns.
- Specification Conformance: Always verify that your server complies with the official Model Context Protocol Specification.
Frequently Asked Questions
- Can FastMCP run over HTTP or Server-Sent Events instead of Stdio?
Yes. FastMCP supports SSE transport mode by running
mcp.run(transport="sse"). In SSE mode, the server spawns an HTTP listener using Starlette and Uvicorn, allowing remote clients to connect over a network port.
- How does FastMCP handle tool exceptions?
When a tool function raises an uncaught exception (such as
ValueErrororKeyError), FastMCP captures the exception and returns a structured JSON-RPC error response withisError: true. The client receives the error message without the server crashing.
- Can I run multiple FastMCP servers in parallel within Claude Desktop?
Yes. The
claude_desktop_config.jsonconfiguration file accepts multiple entries undermcpServers. Claude Desktop spawns separate worker processes for each declared server and aggregates their tools, resources, and prompts into a unified context.