How to build an MCP server in Python

1# Install
2pip install mcp
3
4# server.py
5from mcp.server.fastmcp import FastMCP
6import sys
7
8# Create server instance
9mcp = FastMCP("my-python-server")
10
11# All logging goes to stderr (stdout is reserved for MCP protocol)
12print("Server initializing...", file=sys.stderr)

1# Python
2import json
3import httpx
4
5@mcp.tool()
6async def get_weather(city: str, units: str = "celsius") -> str:
7    """Get current weather for a city.
8
9    Args:
10        city: City name, e.g. San Francisco
11        units: Temperature units (celsius or fahrenheit)
12    """
13    async with httpx.AsyncClient() as client:
14        response = await client.get(
15            "https://api.weather.example/v1/current",
16            params={"city": city, "units": units},
17        )
18        response.raise_for_status()
19        data = response.json()
20        return f"{city}: {data['temperature']}° {units}, {data['condition']}"
21
22@mcp.tool()
23async def calculate(expression: str) -> str:
24    """Evaluate a mathematical expression safely.
25
26    Args:
27        expression: Math expression like '2 + 3 * 4'
28    """
29    # Safe eval for basic math
30    allowed = set("0123456789+-*/.() ")
31    if not all(c in allowed for c in expression):
32        raise ValueError("Error: Only basic math operators allowed")
33    result = eval(expression)  # safe after character validation
34    return str(result)

1# Python
2import os
3
4@mcp.resource("config://app")
5async def get_config() -> str:
6    """Application configuration."""
7    config = {
8        "app_name": "My Python App",
9        "version": "1.0.0",
10        "debug": os.getenv("DEBUG", "false") == "true",
11    }
12    return json.dumps(config, indent=2)
13
14@mcp.resource("file://project/{path}")
15async def read_project_file(path: str) -> str:
16    """Read a project source file."""
17    # Security: prevent directory traversal
18    safe_path = os.path.normpath(path)
19    if safe_path.startswith("..") or os.path.isabs(safe_path):
20        raise ValueError("Access denied: invalid path")
21
22    full_path = os.path.join(os.getcwd(), safe_path)
23    with open(full_path, "r") as f:
24        return f.read()

1# Python
2@mcp.prompt()
3async def review_code(code: str, language: str = "python") -> str:
4    """Generate a structured code review.
5
6    Args:
7        code: The code to review
8        language: Programming language of the code
9    """
10    return f"""Review this {language} code for bugs, security issues, and improvements:
11
12```{language}
13{code}
14```
15
16Provide feedback in this format:
171. Critical issues (bugs, security)
182. Performance improvements
193. Style and readability suggestions"""
20
21@mcp.prompt()
22async def debug_error(error: str, context: str | None = None) -> str:
23    """Help debug an error with structured analysis.
24
25    Args:
26        error: The error message or stack trace
27        context: Additional context about when the error occurs
28    """
29    ctx = f"\nContext: {context}" if context else ""
30    return f"""Debug this error:{ctx}
31
32{error}
33
34Analyze:
351. Root cause
362. Step-by-step fix
373. Prevention strategy"""

1# Python — error handling in tools
2@mcp.tool()
3async def query_database(sql: str) -> str:
4    """Run a read-only SQL query.
5
6    Args:
7        sql: SELECT query to execute
8    """
9    if not sql.strip().upper().startswith("SELECT"):
10        raise ValueError("Only SELECT queries are allowed")
11
12    try:
13        # Using asyncpg for async PostgreSQL
14        import asyncpg
15        conn = await asyncpg.connect(os.getenv("DATABASE_URL"))
16        try:
17            rows = await conn.fetch(sql)
18            return json.dumps([dict(r) for r in rows], indent=2, default=str)
19        finally:
20            await conn.close()
21    except asyncpg.PostgresError as e:
22        raise ValueError(f"Database error: {e}")
23
24# Run the server
25if __name__ == "__main__":
26    mcp.run(transport="stdio")
27
28# Or for HTTP:
29# mcp.run(transport="streamable-http", host="0.0.0.0", port=3000)