Build Your First MCP Server with FastMCP – A Python Power Tool

1. Model Context Protocol (MCP)

MCP is a standardized protocol that defines how AI clients (e.g., IDE plugins, desktop apps, web interfaces) can discover, invoke, and receive structured results from external tools. It functions as an "AI agent REST API" with richer type safety and tool discovery.

1.1 Core capabilities

Tool discovery (like browsing an app store)

Tool invocation with structured parameters (function‑call style)

Structured result handling (JSON‑like responses)

2. FastMCP – Python implementation

FastMCP wraps low‑level MCP details behind a concise decorator API, allowing ordinary Python functions to become MCP tools.

# Example decorator usage
@mcp.tool
def get_weather(city: str) -> dict:
    """Query the weather of a city"""
    # business logic here
    return weather_data

3. Build your first MCP server

3.1 Environment preparation (≈5 minutes)

# Create a virtual environment
python -m venv mcp-env
# Activate environment (macOS/Linux)
source mcp-env/bin/activate
# Windows
mcp-env\Scripts\activate

# Install FastMCP and requests
pip install fastmcp requests

# Verify installation
python -c "import fastmcp; print(f'FastMCP version: {fastmcp.__version__}')"

3.2 First tool – a weather query (simulated data)

from fastmcp import FastMCP

# Initialise MCP server
mcp = FastMCP("Weather-Server")

@mcp.tool
def get_weather(city: str) -> dict:
    """Query the current weather for a city (simulated data)"""
    weather_database = {
        "beijing": {"temp": 22, "condition": "晴朗", "humidity": 45},
        "shanghai": {"temp": 25, "condition": "多云", "humidity": 65},
        "tokyo": {"temp": 18, "condition": "小雨", "humidity": 70},
        "new york": {"temp": 15, "condition": "阴天", "humidity": 60},
    }
    city_key = city.lower().strip()
    if city_key in weather_database:
        data = weather_database[city_key]
        return {"city": city, "temperature_c": data["temp"], "condition": data["condition"], "humidity_percent": data["humidity"], "source": "模拟数据"}
    else:
        return {"city": city, "temperature_c": 20, "condition": "未知", "humidity_percent": 50, "note": "城市不在数据库中，返回默认值", "source": "模拟数据"}

@mcp.tool
def celsius_to_fahrenheit(celsius: float) -> dict:
    """Convert Celsius to Fahrenheit"""
    fahrenheit = (celsius * 9 / 5) + 32
    return {"celsius": celsius, "fahrenheit": round(fahrenheit, 1), "formula": "℉ = (℃ × 9/5) + 32"}

if __name__ == "__main__":
    print("🌤️ 天气MCP服务器启动中...")
    mcp.run(transport="stdio")

3.3 Test client

import asyncio
from fastmcp import Client

async def test_weather_server():
    """Test the MCP server"""
    client = Client("weather_server.py")
    async with client:
        print("🔧 可用的工具列表:")
        tools = await client.list_tools()
        for tool in tools:
            print(f"  • {tool.name}
    描述: {tool.description}
    参数: {[p.name for p in tool.inputSchema['properties'].values()]}")
        print("="*60)
        for city in ["Beijing", "Tokyo", "Paris"]:
            result = await client.call_tool("get_weather", {"city": city})
            print(f"{city}: {result['temperature_c']}°C, {result['condition']}, {result['humidity_percent']}%")
        print("="*60)
        for temp in [0, 20, 37.5, 100]:
            conv = await client.call_tool("celsius_to_fahrenheit", {"celsius": temp})
            print(f"{temp}°C = {conv['fahrenheit']}°F")

async def main():
    print("🚀 开始测试MCP服务器...")
    await test_weather_server()
    print("✅ 测试完成！")

if __name__ == "__main__":
    asyncio.run(main())

4. Upgrade – Connect a real weather API

4.1 Store the API key securely

# Set environment variable (do NOT hard‑code the key)
export OPENWEATHER_API_KEY="your_api_key_here"
# Verify
echo $OPENWEATHER_API_KEY

4.2 Extend the server to call OpenWeatherMap

import os, requests
from fastmcp import FastMCP

mcp = FastMCP("Real-Weather-Server")

OPENWEATHER_BASE_URL = "https://api.openweathermap.org/data/2.5/weather"
OPENWEATHER_API_KEY = os.getenv("OPENWEATHER_API_KEY")

@mcp.tool
def get_real_weather(city: str, country_code: str = None, units: str = "metric") -> dict:
    """Fetch real‑time weather from OpenWeatherMap"""
    if not OPENWEATHER_API_KEY:
        return {"error": "API密钥未配置", "solution": "请设置OPENWEATHER_API_KEY环境变量", "hint": 'export OPENWEATHER_API_KEY="your_key_here"'}
    query = f"{city},{country_code}" if country_code else city
    params = {"q": query, "appid": OPENWEATHER_API_KEY, "units": units, "lang": "zh_cn"}
    try:
        response = requests.get(OPENWEATHER_BASE_URL, params=params, timeout=10)
        response.raise_for_status()
        data = response.json()
        weather_info = {
            "city": data.get("name", city),
            "country": data.get("sys", {}).get("country", "未知"),
            "coordinates": {"lon": data["coord"]["lon"], "lat": data["coord"]["lat"]},
            "temperature": {
                "current": data["main"]["temp"],
                "feels_like": data["main"]["feels_like"],
                "min": data["main"]["temp_min"],
                "max": data["main"]["temp_max"]
            },
            "pressure_hpa": data["main"]["pressure"],
            "humidity_percent": data["main"]["humidity"],
            "weather": {"main": data["weather"][0]["main"], "description": data["weather"][0]["description"], "icon": data["weather"][0]["icon"]},
            "wind": {"speed": data["wind"]["speed"], "direction_deg": data["wind"].get("deg", 0), "gust": data["wind"].get("gust", 0)},
            "cloudiness_percent": data["clouds"]["all"],
            "visibility_meters": data.get("visibility", 10000),
            "timestamp": data["dt"],
            "timezone_offset": data["timezone"],
            "data_source": "OpenWeatherMap",
            "units": units
        }
        if units == "metric":
            weather_info["temperature_unit"] = "°C"
            weather_info["wind_speed_unit"] = "m/s"
        elif units == "imperial":
            weather_info["temperature_unit"] = "°F"
            weather_info["wind_speed_unit"] = "mph"
        return weather_info
    except requests.exceptions.HTTPError as e:
        if response.status_code == 401:
            return {"error": "API密钥无效", "status_code": 401}
        if response.status_code == 404:
            return {"error": f"找不到城市: {city}", "status_code": 404}
        return {"error": f"API请求失败: {str(e)}", "status_code": response.status_code}
    except requests.exceptions.RequestException as e:
        return {"error": f"网络错误: {str(e)}", "type": "network_error"}
    except (KeyError, IndexError) as e:
        return {"error": f"数据解析失败: {str(e)}", "type": "data_parsing_error"}

@mcp.tool
def format_weather_summary(weather_data: dict) -> str:
    """Turn raw weather data into a human‑readable summary"""
    if "error" in weather_data:
        return f"❌ 获取天气数据失败: {weather_data['error']}"
    try:
        location = f"{weather_data['city']}, {weather_data.get('country', '')}".strip(', ')
        temp = weather_data['temperature']['current']
        unit = weather_data.get('temperature_unit', '°C')
        condition = weather_data['weather']['description']
        humidity = weather_data['humidity_percent']
        summary = f"""
🌤️ {location} 天气报告
----------------------------
🌡️ 当前温度: {temp}{unit} (体感: {weather_data['temperature']['feels_like']}{unit})
🌪️ 天气状况: {condition}
💧 湿度: {humidity}%
☁️ 云量: {weather_data['cloudiness_percent']}%
🌬️ 风速: {weather_data['wind']['speed']}{weather_data.get('wind_speed_unit', 'm/s')}
📏 能见度: {weather_data['visibility_meters']}米
📊 气压: {weather_data['pressure_hpa']} hPa
📅 数据时间: {weather_data['timestamp']}
""".strip()
        return summary
    except KeyError as e:
        return f"⚠️ 数据格式错误: 缺少字段 {str(e)}"

if __name__ == "__main__":
    print("🌍 真实天气MCP服务器启动...")
    mcp.run(transport="stdio")

4.3 Test the real‑API server

import os, asyncio, sys
from fastmcp import Client
from fastmcp.client.transports import StdioTransport

async def test_real_weather():
    env = os.environ.copy()
    transport = StdioTransport(command=sys.executable,
                               args=["weather_server.py"],
                               env=env,
                               cwd=os.getcwd())
    client = Client(transport)
    async with client:
        tools = await client.list_tools()
        for tool in tools:
            print(f"✓ {tool.name}: {tool.description[:60]}...")
        cases = [
            {"city": "Beijing", "country_code": "CN"},
            {"city": "Tokyo", "country_code": "JP"},
            {"city": "London", "country_code": "GB"},
            {"city": "New York", "country_code": "US"},
        ]
        for case in cases:
            data = await client.call_tool("get_real_weather", case)
            if "error" in data:
                print(f"❌ 错误: {data['error']}")
                continue
            summary = await client.call_tool("format_weather_summary", {"weather_data": data})
            print(summary)
            print(f"温度范围: {data['temperature']['min']}°C ~ {data['temperature']['max']}°C")
            print(f"风向: {data['wind'].get('direction_deg', 'N/A')}°")
            print(f"坐标: ({data['coordinates']['lat']}, {data['coordinates']['lon']})")
        # Demonstrate imperial units
        imp = await client.call_tool("get_real_weather", {"city": "New York", "country_code": "US", "units": "imperial"})
        if "error" not in imp:
            print(f"纽约当前温度: {imp['temperature']['current']}°F")

async def main():
    await test_real_weather()
    print("✅ 测试完成！")

if __name__ == "__main__":
    asyncio.run(main())

5. Production deployment guide

5.1 HTTP transport (recommended for cloud)

from fastmcp import FastMCP
import uvicorn

mcp = FastMCP("Production-Server", port=8000)

@mcp.tool
def production_tool():
    """Production‑ready tool"""
    return {"status": "ready"}

if __name__ == "__main__":
    mcp.run(transport="http", host="0.0.0.0", port=8000)

5.2 Dockerised deployment

# Dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
ENV PORT=8000
ENV HOST=0.0.0.0
CMD ["python", "production_server.py"]

5.3 Security considerations

from fastmcp import FastMCP
from fastmcp.middleware import AuthMiddleware

mcp = FastMCP("Secure-Server")

@mcp.middleware
async def auth_middleware(call, next_fn):
    api_key = call.context.get("headers", {}).get("x-api-key")
    if not validate_api_key(api_key):
        raise PermissionError("Invalid API key")
    return await next_fn(call)

These steps allow FastMCP to run as a secure, containerised, HTTP‑based service consumable by any MCP‑compatible AI client.