Developing MCP Servers

Quick Start

from fastmcp import FastMCP

mcp = FastMCP("my-service")

@mcp.tool() def get_weather(city: str) -> str: """Get current weather for a city.

Args:
    city: Name of the city to get weather for
"""
return f"Weather in {city}: 72F, Sunny"

@mcp.resource("config://settings") def get_settings() -> str: """Expose application settings as a resource.""" return json.dumps({"theme": "dark", "language": "en"})

@mcp.prompt() def analyze_code(code: str, language: str = "python") -> str: """Generate a prompt for code analysis.""" return f"Analyze this {language} code:\n{language}\n{code}\n"

if name == "main": mcp.run()

Features

Feature Description Guide

Tool Definition Create callable tools with typed parameters Use @mcp.tool() decorator with docstrings

Resource Exposure Expose data resources for AI to read Use @mcp.resource() with URI patterns

Prompt Templates Define reusable prompt templates Use @mcp.prompt() for consistent prompts

Dynamic Resources Create parameterized resource URIs Use URI templates like "user://{id}/profile"

Progress Reporting Report progress for long operations Use ctx.report_progress() in async tools

Streaming Results Stream results for large outputs Use AsyncGenerator return type

Lifecycle Hooks Manage server startup and shutdown Use lifespan context manager

Middleware Add logging, rate limiting, auth Implement middleware functions

Error Handling Return structured error responses Use try/except with error codes

Testing Test tools and resources in isolation Use MCPTestClient for unit tests

Common Patterns

TypeScript MCP Server

import { Server } from "@modelcontextprotocol/sdk/server/index.js"; import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";

const server = new Server( { name: "my-server", version: "1.0.0" }, { capabilities: { tools: {}, resources: {} } } );

server.setRequestHandler(ListToolsRequestSchema, async () => ({ tools: [{ name: "search", description: "Search the database", inputSchema: { type: "object", properties: { query: { type: "string" } }, required: ["query"], }, }], }));

server.setRequestHandler(CallToolRequestSchema, async (request) => { const { name, arguments: args } = request.params; if (name === "search") { const results = await searchDatabase(args.query); return { content: [{ type: "text", text: JSON.stringify(results) }] }; } throw new Error(Unknown tool: ${name}); });

const transport = new StdioServerTransport(); await server.connect(transport);

Database Integration Server

from fastmcp import FastMCP import asyncpg

mcp = FastMCP("database-server") pool: asyncpg.Pool = None

@mcp.tool() async def query(sql: str, params: list = None) -> dict: """Execute read-only SQL query.""" if not sql.strip().upper().startswith("SELECT"): raise ValueError("Only SELECT queries allowed")

async with pool.acquire() as conn:
    rows = await conn.fetch(sql, *(params or []))
    return {
        "columns": list(rows[0].keys()) if rows else [],
        "rows": [dict(row) for row in rows],
        "count": len(rows),
    }

@mcp.resource("schema://tables") async def list_tables() -> str: """List all database tables.""" async with pool.acquire() as conn: tables = await conn.fetch( "SELECT table_name FROM information_schema.tables WHERE table_schema = 'public'" ) return json.dumps([t["table_name"] for t in tables])

Secure File System Server

from pathlib import Path

mcp = FastMCP("filesystem-server") WORKSPACE = Path(os.getenv("WORKSPACE", ".")).resolve()

def validate_path(path: str) -> Path: """Ensure path is within workspace.""" full_path = (WORKSPACE / path).resolve() if not str(full_path).startswith(str(WORKSPACE)): raise ValueError("Path outside workspace") return full_path

@mcp.tool() def read_file(path: str) -> str: """Read file contents safely.""" file_path = validate_path(path) return file_path.read_text()

@mcp.tool() def list_directory(path: str = ".") -> list[dict]: """List directory contents.""" dir_path = validate_path(path) return [{"name": f.name, "type": "dir" if f.is_dir() else "file"} for f in dir_path.iterdir()]

Testing MCP Servers

import pytest from fastmcp.testing import MCPTestClient

@pytest.fixture def client(): return MCPTestClient(mcp)

class TestMCPServer: async def test_tool_execution(self, client): result = await client.call_tool("get_weather", {"city": "Seattle"}) assert result.success assert "Seattle" in result.content

async def test_resource_access(self, client):
    result = await client.read_resource("config://settings")
    assert result.success
    data = json.loads(result.content)
    assert "theme" in data

Best Practices

Do Avoid

Document tools thoroughly with clear docstrings Vague or missing tool descriptions

Validate all inputs with type hints Trusting user input without validation

Return structured error responses Exposing internal error details

Use async for I/O-bound operations Blocking the event loop with sync I/O

Implement pagination for large results Returning unbounded data sets

Add rate limiting for resource-intensive tools Allowing unlimited API calls

Test tools with MCPTestClient Skipping unit tests for tools

Follow MCP specification strictly Deviating from protocol standards

Sanitize paths and SQL queries Allowing path traversal or SQL injection

Log tool calls for debugging Missing audit trail for operations

Related Skills

• python - Primary language for FastMCP

• typescript - MCP SDK for TypeScript

• api-architecture - API design patterns

References

• MCP Specification

• FastMCP Documentation

• MCP TypeScript SDK

• Claude MCP Guide