Building MCP Servers

1# Install FastMCP
2pip install fastmcp
3
4# Or with uv (recommended for faster installs)
5uv add fastmcp

1from fastmcp import FastMCP
2
3# Create server instance
4mcp = FastMCP("My First MCP Server")
5
6# Define a simple tool
7@mcp.tool()
8def greet(name: str) -> str:
9    """Greet someone by name.
10
11    Args:
12        name: The name of the person to greet
13    """
14    return f"Hello, {name}!"
15
16# Define another tool with multiple parameters
17@mcp.tool()
18def calculate(operation: str, a: float, b: float) -> float:
19    """Perform a mathematical calculation.
20
21    Args:
22        operation: The operation to perform (add, subtract, multiply, divide)
23        a: First number
24        b: Second number
25    """
26    operations = {
27        "add": a + b,
28        "subtract": a - b,
29        "multiply": a * b,
30        "divide": a / b if b != 0 else float("inf")
31    }
32    return operations.get(operation, 0)
33
34# Run the server
35if __name__ == "__main__":
36    mcp.run()

1# Run directly
2python basic_server.py
3
4# Or run with fastmcp CLI for development
5fastmcp dev basic_server.py
6
7# Test with MCP Inspector
8fastmcp dev basic_server.py --with-inspector

1{
2  "mcpServers": {
3    "my-first-server": {
4      "command": "python",
5      "args": ["/path/to/basic_server.py"],
6      "env": {}
7    }
8  }
9}

1# Initialize project
2npm init -y
3
4# Install MCP SDK
5npm install @modelcontextprotocol/sdk
6
7# Install TypeScript dependencies
8npm install -D typescript @types/node tsx

1import { Server } from "@modelcontextprotocol/sdk/server/index.js";
2import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
3import {
4  CallToolRequestSchema,
5  ListToolsRequestSchema,
6} from "@modelcontextprotocol/sdk/types.js";
7
8// Create server instance
9const server = new Server(
10  {
11    name: "my-typescript-server",
12    version: "1.0.0",
13  },
14  {
15    capabilities: {
16      tools: {},
17    },
18  }
19);
20
21// Define available tools
22const TOOLS = [
23  {
24    name: "greet",
25    description: "Greet someone by name",
26    inputSchema: {
27      type: "object" as const,
28      properties: {
29        name: {
30          type: "string",
31          description: "The name of the person to greet",
32        },
33      },
34      required: ["name"],
35    },
36  },
37  {
38    name: "calculate",
39    description: "Perform a mathematical calculation",
40    inputSchema: {
41      type: "object" as const,
42      properties: {
43        operation: {
44          type: "string",
45          enum: ["add", "subtract", "multiply", "divide"],
46          description: "The operation to perform",
47        },
48        a: { type: "number", description: "First number" },
49        b: { type: "number", description: "Second number" },
50      },
51      required: ["operation", "a", "b"],
52    },
53  },
54];
55
56// Handle tool listing
57server.setRequestHandler(ListToolsRequestSchema, async () => ({
58  tools: TOOLS,
59}));
60
61// Handle tool execution
62server.setRequestHandler(CallToolRequestSchema, async (request) => {
63  const { name, arguments: args } = request.params;
64
65  switch (name) {
66    case "greet": {
67      const { name: personName } = args as { name: string };
68      return {
69        content: [
70          {
71            type: "text",
72            text: `Hello, ${personName}!`,
73          },
74        ],
75      };
76    }
77
78    case "calculate": {
79      const { operation, a, b } = args as {
80        operation: string;
81        a: number;
82        b: number;
83      };
84
85      let result: number;
86      switch (operation) {
87        case "add": result = a + b; break;
88        case "subtract": result = a - b; break;
89        case "multiply": result = a * b; break;
90        case "divide": result = b !== 0 ? a / b : Infinity; break;
91        default: throw new Error(`Unknown operation: ${operation}`);
92      }
93
94      return {
95        content: [
96          {
97            type: "text",
98            text: `Result: ${result}`,
99          },
100        ],
101      };
102    }
103
104    default:
105      throw new Error(`Unknown tool: ${name}`);
106  }
107});
108
109// Start the server
110async function main() {
111  const transport = new StdioServerTransport();
112  await server.connect(transport);
113  console.error("MCP Server running on stdio");
114}
115
116main().catch(console.error);

1# Run with tsx (recommended for development)
2npx tsx src/server.ts
3
4# Or compile and run
5npx tsc
6node dist/server.js

1from fastmcp import FastMCP
2from pathlib import Path
3import os
4
5mcp = FastMCP("Filesystem Server")
6
7# Define allowed directories for security
8ALLOWED_ROOTS = [Path.home() / "projects", Path("/tmp")]
9
10def is_path_allowed(path: Path) -> bool:
11    """Check if path is within allowed directories."""
12    path = path.resolve()
13    return any(
14        path.is_relative_to(root) for root in ALLOWED_ROOTS
15    )
16
17@mcp.tool()
18def read_file(path: str) -> str:
19    """Read contents of a file.
20
21    Args:
22        path: Path to the file to read
23    """
24    file_path = Path(path)
25
26    if not is_path_allowed(file_path):
27        raise ValueError(f"Access denied: {path}")
28
29    if not file_path.exists():
30        raise FileNotFoundError(f"File not found: {path}")
31
32    return file_path.read_text()
33
34@mcp.tool()
35def write_file(path: str, content: str) -> str:
36    """Write content to a file.
37
38    Args:
39        path: Path to the file to write
40        content: Content to write to the file
41    """
42    file_path = Path(path)
43
44    if not is_path_allowed(file_path):
45        raise ValueError(f"Access denied: {path}")
46
47    file_path.parent.mkdir(parents=True, exist_ok=True)
48    file_path.write_text(content)
49
50    return f"Successfully wrote {len(content)} bytes to {path}"
51
52@mcp.tool()
53def list_directory(path: str) -> list[dict]:
54    """List contents of a directory.
55
56    Args:
57        path: Path to the directory to list
58    """
59    dir_path = Path(path)
60
61    if not is_path_allowed(dir_path):
62        raise ValueError(f"Access denied: {path}")
63
64    if not dir_path.is_dir():
65        raise NotADirectoryError(f"Not a directory: {path}")
66
67    entries = []
68    for entry in dir_path.iterdir():
69        entries.append({
70            "name": entry.name,
71            "type": "directory" if entry.is_dir() else "file",
72            "size": entry.stat().st_size if entry.is_file() else None
73        })
74
75    return entries
76
77if __name__ == "__main__":
78    mcp.run()

1from fastmcp import FastMCP
2import httpx
3from typing import Optional
4
5mcp = FastMCP("HTTP API Server")
6
7@mcp.tool()
8async def fetch_url(
9    url: str,
10    method: str = "GET",
11    headers: Optional[dict] = None,
12    body: Optional[str] = None
13) -> dict:
14    """Make an HTTP request to a URL.
15
16    Args:
17        url: The URL to fetch
18        method: HTTP method (GET, POST, PUT, DELETE)
19        headers: Optional headers to include
20        body: Optional request body for POST/PUT
21    """
22    async with httpx.AsyncClient() as client:
23        response = await client.request(
24            method=method.upper(),
25            url=url,
26            headers=headers or {},
27            content=body
28        )
29
30        return {
31            "status_code": response.status_code,
32            "headers": dict(response.headers),
33            "body": response.text[:10000]  # Limit response size
34        }
35
36@mcp.tool()
37async def get_json_api(url: str) -> dict:
38    """Fetch JSON from an API endpoint.
39
40    Args:
41        url: The API endpoint URL
42    """
43    async with httpx.AsyncClient() as client:
44        response = await client.get(url)
45        response.raise_for_status()
46        return response.json()
47
48if __name__ == "__main__":
49    mcp.run()

1from fastmcp import FastMCP
2import sqlite3
3from contextlib import contextmanager
4from typing import Any
5
6mcp = FastMCP("SQLite Server")
7
8DATABASE_PATH = "app.db"
9
10@contextmanager
11def get_connection():
12    """Get a database connection with proper cleanup."""
13    conn = sqlite3.connect(DATABASE_PATH)
14    conn.row_factory = sqlite3.Row
15    try:
16        yield conn
17    finally:
18        conn.close()
19
20@mcp.tool()
21def query(sql: str, params: list[Any] = None) -> list[dict]:
22    """Execute a read-only SQL query.
23
24    Args:
25        sql: The SQL query to execute (SELECT only)
26        params: Optional query parameters
27    """
28    # Security: Only allow SELECT queries
29    sql_upper = sql.strip().upper()
30    if not sql_upper.startswith("SELECT"):
31        raise ValueError("Only SELECT queries are allowed")
32
33    with get_connection() as conn:
34        cursor = conn.execute(sql, params or [])
35        columns = [col[0] for col in cursor.description]
36        rows = cursor.fetchall()
37
38        return [dict(zip(columns, row)) for row in rows]
39
40@mcp.tool()
41def get_schema() -> list[dict]:
42    """Get the database schema (table names and columns)."""
43    with get_connection() as conn:
44        # Get all tables
45        cursor = conn.execute(
46            "SELECT name FROM sqlite_master WHERE type='table'"
47        )
48        tables = [row[0] for row in cursor.fetchall()]
49
50        schema = []
51        for table in tables:
52            cursor = conn.execute(f"PRAGMA table_info({table})")
53            columns = [
54                {
55                    "name": row[1],
56                    "type": row[2],
57                    "nullable": not row[3],
58                    "primary_key": bool(row[5])
59                }
60                for row in cursor.fetchall()
61            ]
62            schema.append({"table": table, "columns": columns})
63
64        return schema
65
66if __name__ == "__main__":
67    mcp.run()

1from fastmcp import FastMCP
2from pathlib import Path
3
4mcp = FastMCP("Resource Server")
5
6# Static resource from string
7@mcp.resource("config://app/settings")
8def get_app_settings() -> str:
9    """Application configuration settings."""
10    return """
11    {
12        "app_name": "MyApp",
13        "version": "1.0.0",
14        "debug": false,
15        "max_connections": 100
16    }
17    """
18
19# Dynamic resource from file
20@mcp.resource("file://readme")
21def get_readme() -> str:
22    """Project README file."""
23    readme_path = Path("README.md")
24    if readme_path.exists():
25        return readme_path.read_text()
26    return "README not found"
27
28# Resource with dynamic URI
29@mcp.resource("logs://{date}")
30def get_logs(date: str) -> str:
31    """Get logs for a specific date.
32
33    Args:
34        date: Date in YYYY-MM-DD format
35    """
36    log_path = Path(f"logs/{date}.log")
37    if log_path.exists():
38        return log_path.read_text()
39    return f"No logs found for {date}"

1import { Server } from "@modelcontextprotocol/sdk/server/index.js";
2import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
3import {
4  ListResourcesRequestSchema,
5  ReadResourceRequestSchema,
6} from "@modelcontextprotocol/sdk/types.js";
7import * as fs from "fs/promises";
8import * as path from "path";
9
10const server = new Server(
11  { name: "resource-server", version: "1.0.0" },
12  { capabilities: { resources: {} } }
13);
14
15const DOCS_DIR = "./docs";
16
17// List all available resources
18server.setRequestHandler(ListResourcesRequestSchema, async () => {
19  const files = await fs.readdir(DOCS_DIR);
20
21  const resources = files
22    .filter(f => f.endsWith(".md"))
23    .map(f => ({
24      uri: `docs://${f}`,
25      name: f.replace(".md", ""),
26      description: `Documentation: ${f}`,
27      mimeType: "text/markdown",
28    }));
29
30  return { resources };
31});
32
33// Read a specific resource
34server.setRequestHandler(ReadResourceRequestSchema, async (request) => {
35  const { uri } = request.params;
36
37  // Parse the URI
38  const match = uri.match(/^docs:\/\/(.+)$/);
39  if (!match) {
40    throw new Error(`Invalid resource URI: ${uri}`);
41  }
42
43  const filename = match[1];
44  const filePath = path.join(DOCS_DIR, filename);
45
46  try {
47    const content = await fs.readFile(filePath, "utf-8");
48    return {
49      contents: [
50        {
51          uri,
52          mimeType: "text/markdown",
53          text: content,
54        },
55      ],
56    };
57  } catch (error) {
58    throw new Error(`Resource not found: ${uri}`);
59  }
60});
61
62async function main() {
63  const transport = new StdioServerTransport();
64  await server.connect(transport);
65}
66
67main();

1from fastmcp import FastMCP
2from fastmcp.prompts import Prompt, TextContent
3
4mcp = FastMCP("Prompt Server")
5
6@mcp.prompt()
7def code_review(
8    language: str,
9    code: str,
10    focus_areas: str = "bugs, performance, readability"
11) -> list[TextContent]:
12    """Generate a code review prompt.
13
14    Args:
15        language: Programming language of the code
16        code: The code to review
17        focus_areas: Comma-separated areas to focus on
18    """
19    return [
20        TextContent(
21            type="text",
22            text=f"""Please review the following {language} code.
23
24Focus on these areas: {focus_areas}
25
26Code to review:
27```{language}
28{code}
29```
30
31Provide specific, actionable feedback for each issue found.
32Format your response as:
331. Issue: [description]
34   Severity: [low/medium/high]
35   Suggestion: [how to fix]
36"""
37        )
38    ]
39
40@mcp.prompt()
41def explain_error(
42    error_message: str,
43    stack_trace: str = "",
44    context: str = ""
45) -> list[TextContent]:
46    """Generate a prompt to explain an error.
47
48    Args:
49        error_message: The error message
50        stack_trace: Optional stack trace
51        context: Optional additional context
52    """
53    prompt = f"""I encountered the following error:
54
55Error: {error_message}
56"""
57
58    if stack_trace:
59        prompt += f"""
60Stack trace:
61{stack_trace}
62"""
63
64    if context:
65        prompt += f"""
66Additional context:
67{context}
68"""
69
70    prompt += """
71Please explain:
721. What this error means
732. Common causes
743. How to fix it
754. How to prevent it in the future
76"""
77
78    return [TextContent(type="text", text=prompt)]
79
80@mcp.prompt()
81def generate_tests(
82    code: str,
83    test_framework: str = "pytest",
84    coverage_target: str = "edge cases and error conditions"
85) -> list[TextContent]:
86    """Generate a prompt for creating tests.
87
88    Args:
89        code: The code to write tests for
90        test_framework: Testing framework to use
91        coverage_target: What aspects to focus testing on
92    """
93    return [
94        TextContent(
95            type="text",
96            text=f"""Generate comprehensive tests for the following code using {test_framework}.
97
98Code to test:
99```python
100{code}
101```
102
103Focus on testing: {coverage_target}
104
105Requirements:
106- Write clear, descriptive test names
107- Include both positive and negative test cases
108- Mock external dependencies
109- Add comments explaining what each test verifies
110"""
111        )
112    ]

1# Install MCP Inspector globally
2npm install -g @modelcontextprotocol/inspector
3
4# Run your server with inspector
5npx mcp-inspector python my_server.py
6
7# Or for TypeScript
8npx mcp-inspector node dist/server.js

1import pytest
2from fastmcp.testing import MCPClient
3
4from my_server import mcp
5
6@pytest.fixture
7async def client():
8    """Create a test client for the MCP server."""
9    async with MCPClient(mcp) as client:
10        yield client
11
12@pytest.mark.asyncio
13async def test_greet_tool(client):
14    """Test the greet tool."""
15    result = await client.call_tool("greet", {"name": "Alice"})
16
17    assert "Hello, Alice!" in result.content[0].text
18
19@pytest.mark.asyncio
20async def test_calculate_tool(client):
21    """Test the calculate tool."""
22    result = await client.call_tool(
23        "calculate",
24        {"operation": "add", "a": 5, "b": 3}
25    )
26
27    assert "8" in result.content[0].text
28
29@pytest.mark.asyncio
30async def test_list_tools(client):
31    """Test that all expected tools are listed."""
32    tools = await client.list_tools()
33
34    tool_names = [t.name for t in tools]
35    assert "greet" in tool_names
36    assert "calculate" in tool_names
37
38@pytest.mark.asyncio
39async def test_invalid_operation(client):
40    """Test error handling for invalid operations."""
41    with pytest.raises(Exception):
42        await client.call_tool(
43            "calculate",
44            {"operation": "invalid", "a": 1, "b": 2}
45        )

1import subprocess
2import json
3import time
4
5def test_server_integration():
6    """Test server as actual subprocess."""
7
8    # Start server
9    proc = subprocess.Popen(
10        ["python", "my_server.py"],
11        stdin=subprocess.PIPE,
12        stdout=subprocess.PIPE,
13        stderr=subprocess.PIPE,
14        text=True
15    )
16
17    try:
18        # Send initialize request
19        init_request = {
20            "jsonrpc": "2.0",
21            "id": 1,
22            "method": "initialize",
23            "params": {
24                "protocolVersion": "2024-11-05",
25                "capabilities": {},
26                "clientInfo": {"name": "test", "version": "1.0"}
27            }
28        }
29        proc.stdin.write(json.dumps(init_request) + "\n")
30        proc.stdin.flush()
31
32        # Read response
33        response = json.loads(proc.stdout.readline())
34        assert response["id"] == 1
35        assert "result" in response
36
37        # Send tool list request
38        list_request = {
39            "jsonrpc": "2.0",
40            "id": 2,
41            "method": "tools/list"
42        }
43        proc.stdin.write(json.dumps(list_request) + "\n")
44        proc.stdin.flush()
45
46        response = json.loads(proc.stdout.readline())
47        assert "tools" in response["result"]
48
49    finally:
50        proc.terminate()
51        proc.wait()