Python 神器 FastMCP:构建你的第一个 MCP 服务器

一行代码，就能让模型调用工具

你是否遇到过这样的场景？

好不容易训练了一个聪明的AI模型，却发现它“眼高手低”——能分析问题，却不会执行任务；能给出建议，却无法调用真实API；能理解需求，却连最简单的天气查询都做不到。

这不怪AI。大语言模型本质上是“思考者”，而不是“执行者”。它们擅长理解和生成文本，但缺乏与现实世界交互的“手”和“脚”。

今天，我要介绍一个改变游戏规则的技术：MCP（Model Context Protocol），以及它的Python利器——FastMCP。只需几行代码，你就能为AI模型装上“手脚”，让它从“纸上谈兵”变成“实干家”。

一、MCP：AI的“标准接口”

1.1 什么是MCP？

MCP（Model Context Protocol） 是一个标准化协议，它定义了AI客户端（如IDE插件、桌面应用、Web界面）如何安全地调用工具和获取资源。

简单来说，MCP就是：

“AI代理的REST API——但更智能、更标准化”

想象一下，如果没有MCP，每个AI应用都要自己造一套插件系统，就像每个电器都要用不同的插座一样混乱。MCP提供了统一的标准接口，让AI能够：

• 发现可用工具（像浏览应用商店）
• 调用工具并传递结构化参数（像函数调用）
• 接收结构化结果（像API响应）

1.2 为什么选择FastMCP？

在Python生态中，FastMCP 是最优雅的MCP实现之一。它把底层的协议细节封装成简洁的装饰器API，让你可以：

# 就这么简单！
@mcp.tool
defget_weather(city: str) -> dict:
"""查询城市天气"""
# 你的业务逻辑
return weather_data

FastMCP为你自动处理：

• 工具注册和发现
• 参数验证和类型安全
• 传输层（stdio/HTTP）
• 生产环境就绪

二、手把手：构建你的第一个MCP服务器

2.1 环境准备：5分钟搞定

首先，创建一个干净的Python环境：

# 创建虚拟环境（推荐）
python -m venv mcp-env

# 激活环境
# macOS/Linux
source mcp-env/bin/activate
# Windows
mcp-env\Scripts\activate

# 安装FastMCP
pip install fastmcp requests

检查安装是否成功：

python -c "import fastmcp; print(f'FastMCP版本: {fastmcp.__version__}')"

2.2 第一个工具：让AI会查天气

创建文件 weather_server.py：

from fastmcp import FastMCP

# 初始化MCP服务器，给它起个名字
mcp = FastMCP("Weather-Server")

@mcp.tool
defget_weather(city: str) -> dict:
"""
    查询指定城市的当前天气情况

    参数:
        city: 城市名称（如"北京"、"Tokyo"）

    返回:
        包含天气信息的字典
    """
# 这里是模拟数据，实际可以连接真实API
    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:
return {
"city": city,
"temperature_c": weather_database[city_key]["temp"],
"condition": weather_database[city_key]["condition"],
"humidity_percent": weather_database[city_key]["humidity"],
"source": "模拟数据"
        }
else:
# 默认返回
return {
"city": city,
"temperature_c": 20,
"condition": "未知",
"humidity_percent": 50,
"note": "城市不在数据库中，返回默认值",
"source": "模拟数据"
        }

@mcp.tool
defcelsius_to_fahrenheit(celsius: float) -> dict:
"""
    将摄氏度转换为华氏度

    参数:
        celsius: 摄氏度温度值

    返回:
        转换后的华氏度温度
    """
    fahrenheit = (celsius * 9/5) + 32
return {
"celsius": celsius,
"fahrenheit": round(fahrenheit, 1),
"formula": "℉ = (℃ × 9/5) + 32"
    }

if __name__ == "__main__":
# 使用stdio传输，适合本地开发和AI代理框架
    print("🌤️ 天气MCP服务器启动中...")
    mcp.run(transport="stdio")

关键点解析：

1. FastMCP("Weather-Server") 创建服务器实例
2. @mcp.tool 装饰器将普通函数变成MCP工具
3. 函数文档字符串会自动成为工具描述
4. 参数类型提示（city: str）会被自动转换为JSON Schema

2.3 测试客户端：眼见为实

创建 test_client.py 来测试我们的服务器：

import asyncio
from fastmcp import Client

asyncdeftest_weather_server():
"""测试MCP服务器"""

# 创建客户端，指向我们的服务器文件
    client = Client("weather_server.py")

asyncwith client:
# 1. 查看所有可用工具
        print("🔧 可用的工具列表:")
        print("-" * 40)

        tools = await client.list_tools()
for tool in tools:
            print(f"  • {tool.name}")
            print(f"    描述: {tool.description}")
            print(f"    参数: {[p.name for p in tool.inputSchema['properties'].keys()]}")
            print()

        print("=" * 60)

# 2. 测试查询天气
        print("🌍 测试天气查询:")
        print("-" * 40)

        test_cities = ["Beijing", "Tokyo", "Paris"]

for city in test_cities:
            print(f"\n查询 {city} 的天气...")
            result = await client.call_tool("get_weather", {"city": city})

            print(f"  温度: {result['temperature_c']}°C")
            print(f"  天气: {result['condition']}")
            print(f"  湿度: {result['humidity_percent']}%")

        print("\n" + "=" * 60)

# 3. 测试温度转换
        print("🌡️ 测试温度转换:")
        print("-" * 40)

        test_temps = [0, 20, 37.5, 100]  # 水的冰点、室温、体温、沸点

for temp in test_temps:
            result = await client.call_tool(
"celsius_to_fahrenheit", 
                {"celsius": temp}
            )
            print(f"  {temp}°C = {result['fahrenheit']}°F")

asyncdefmain():
"""主函数"""
    print("🚀 开始测试MCP服务器...\n")
await test_weather_server()
    print("\n✅ 测试完成！")

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

运行测试：

python test_client.py

你会看到类似这样的输出：

🚀 开始测试MCP服务器...

🔧 可用的工具列表:
----------------------------------------
  • get_weather
    描述: 查询指定城市的当前天气情况...
    参数: ['city']

  • celsius_to_fahrenheit
    描述: 将摄氏度转换为华氏度...
    参数: ['celsius']

============================================================
🌍 测试天气查询:
----------------------------------------

查询 Beijing 的天气...
  温度: 22°C
  天气: 晴朗
  湿度: 45%

查询 Tokyo 的天气...
  温度: 18°C
  天气: 小雨
  湿度: 70%

查询 Paris 的天气...
  温度: 20°C
  天气: 未知
  湿度: 50%

============================================================
🌡️ 测试温度转换:
----------------------------------------
  0°C = 32.0°F
  20°C = 68.0°F
  37.5°C = 99.5°F
  100°C = 212.0°F

✅ 测试完成！

三、升级版：连接真实API

模拟数据不够过瘾？让我们连接真实的天气API！

3.1 获取OpenWeatherMap API密钥

1. 访问 OpenWeatherMap
2. 注册免费账户（每月1000次调用，足够学习使用）
3. 在控制台获取API Key

3.2 环境变量配置（安全最佳实践）

# 设置环境变量（不要硬编码密钥！）
export OPENWEATHER_API_KEY="your_api_key_here"

# 验证设置
echo$OPENWEATHER_API_KEY

3.3 升级服务器：连接真实天气数据

更新 weather_server.py：

import os
import requests
from typing import Optional
from fastmcp import FastMCP

# 初始化MCP服务器
mcp = FastMCP("Real-Weather-Server")

# OpenWeatherMap API配置
OPENWEATHER_BASE_URL = "https://api.openweathermap.org/data/2.5/weather"
OPENWEATHER_API_KEY = os.getenv("OPENWEATHER_API_KEY")

@mcp.tool
defget_real_weather(
    city: str, 
    country_code: Optional[str] = None,
    units: str = "metric"
) -> dict:
"""
    使用OpenWeatherMap API查询真实天气数据

    参数:
        city: 城市名称（如"Beijing"）
        country_code: 国家代码（可选，如"CN"）
        units: 温度单位 - "metric"(摄氏度), "imperial"(华氏度), "standard"(开尔文)

    返回:
        包含详细天气信息的字典
    """
# 安全检查
ifnot 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:
# 调用API
        response = requests.get(OPENWEATHER_BASE_URL, params=params, timeout=10)
        response.raise_for_status()  # 检查HTTP错误
        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}
elif response.status_code == 404:
return {"error": f"找不到城市: {city}", "status_code": 404}
else:
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
defget_air_pollution(lat: float, lon: float) -> dict:
"""
    获取指定坐标的空气污染指数

    参数:
        lat: 纬度 (-90 到 90)
        lon: 经度 (-180 到 180)

    返回:
        空气污染数据
    """
# 注意：空气污染API需要单独的调用
# 这里提供框架，实际需要OpenWeatherMap的air_pollution端点
return {
"coordinates": {"lat": lat, "lon": lon},
"note": "需要OpenWeatherMap的air_pollution API订阅",
"suggestion": "查看 https://openweathermap.org/api/air-pollution"
    }

@mcp.tool  
defformat_weather_summary(weather_data: dict) -> str:
"""
    将天气数据格式化为友好的人类可读摘要

    参数:
        weather_data: get_real_weather返回的天气数据

    返回:
        格式化的天气摘要字符串
    """
if"error"in weather_data:
returnf"❌ 获取天气数据失败: {weather_data['error']}"

try:
        city = weather_data["city"]
        country = weather_data.get("country", "")
        temp = weather_data["temperature"]["current"]
        temp_unit = weather_data.get("temperature_unit", "°C")
        condition = weather_data["weather"]["description"]
        humidity = weather_data["humidity_percent"]

        location = f"{city}, {country}"if country else city

        summary = f"""
🌤️ {location} 天气报告
----------------------------
🌡️ 当前温度: {temp}{temp_unit} (体感: {weather_data['temperature']['feels_like']}{temp_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:
returnf"⚠️ 数据格式错误: 缺少字段 {str(e)}"

if __name__ == "__main__":
    print("🌍 真实天气MCP服务器启动...")
    print(f"📋 可用工具: get_real_weather, get_air_pollution, format_weather_summary")
    mcp.run(transport="stdio")

3.4 升级测试客户端

更新 test_client.py：

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

asyncdeftest_real_weather():
"""测试真实天气API"""

# 准备环境变量（传递给服务器进程）
    env = os.environ.copy()

# 检查API密钥
if"OPENWEATHER_API_KEY"notin env:
        print("⚠️  警告: OPENWEATHER_API_KEY 环境变量未设置")
        print("   请运行: export OPENWEATHER_API_KEY='your_key_here'")
        print("   或直接在代码中设置: env['OPENWEATHER_API_KEY'] = 'your_key_here'")
return

# 创建传输层，显式传递环境变量
    transport = StdioTransport(
        command=sys.executable,  # 使用当前Python解释器
        args=["weather_server.py"],  # 服务器文件
        env=env,  # 传递环境变量
        cwd=os.getcwd()  # 当前工作目录
    )

    client = Client(transport)

asyncwith client:
        print("🔍 发现服务器工具...")
        tools = await client.list_tools()

for tool in tools:
            print(f"  ✓ {tool.name}: {tool.description[:60]}...")

        print("\n" + "="*60)
        print("🌤️ 测试真实天气查询")
        print("="*60)

# 测试几个主要城市
        test_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 test_cases:
            print(f"\n📍 查询 {case['city']}, {case['country_code']}...")

# 调用天气查询
            weather_data = await client.call_tool(
"get_real_weather",
                {"city": case["city"], "country_code": case["country_code"]}
            )

if"error"in weather_data:
                print(f"   ❌ 错误: {weather_data['error']}")
continue

# 格式化显示
            summary = await client.call_tool(
"format_weather_summary",
                {"weather_data": weather_data}
            )

            print(f"   ✅ 成功获取数据:")
            print("   " + "\n   ".join(summary.split("\n")))

# 显示一些关键数据
            print(f"   📊 详细数据:")
            print(f"     温度范围: {weather_data['temperature']['min']}°C ~ {weather_data['temperature']['max']}°C")
            print(f"     风向: {weather_data['wind'].get('direction_deg', 'N/A')}°")
            print(f"     坐标: ({weather_data['coordinates']['lat']}, {weather_data['coordinates']['lon']})")

        print("\n" + "="*60)
        print("🎯 高级功能测试")
        print("="*60)

# 测试华氏度单位
        print("\n🇺🇸 测试华氏度单位 (纽约)...")
        fahrenheit_data = await client.call_tool(
"get_real_weather",
            {"city": "New York", "country_code": "US", "units": "imperial"}
        )

if"error"notin fahrenheit_data:
            print(f"   🌡️ 纽约当前温度: {fahrenheit_data['temperature']['current']}°F")

# 测试空气污染（需要付费API）
        print("\n🌫️ 测试空气污染查询...")
        pollution_data = await client.call_tool(
"get_air_pollution",
            {"lat": 39.9042, "lon": 116.4074}  # 北京坐标
        )
        print(f"   {pollution_data.get('note', 'N/A')}")

asyncdefmain():
"""主函数"""
    print("🚀 MCP真实天气服务器测试")
    print("="*60)

await test_real_weather()

    print("\n" + "="*60)
    print("✅ 测试完成！")
    print("\n💡 提示: 现在你的AI助手可以调用这些工具了！")
    print("   例如在Claude Desktop、Cursor等支持MCP的工具中")

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

运行真实API测试：

# 确保已设置API密钥
export OPENWEATHER_API_KEY="your_actual_key_here"

# 运行测试
python test_client.py

四、MCP在实际系统中的应用场景

4.1 MCP的“用武之地”

MCP不仅限于天气查询，它在真实系统中有着广泛的应用：

# 企业级MCP服务器示例架构
@mcp.tool
defquery_database(sql: str, params: dict = None) -> list:
"""安全地查询企业数据库"""
# 实际实现会包含权限验证、SQL注入防护等
pass

@mcp.tool  
defcall_internal_api(endpoint: str, payload: dict) -> dict:
"""调用内部REST API"""
pass

@mcp.tool
defgenerate_report(data_source: str, format: str = "pdf") -> dict:
"""生成业务报告"""
pass

@mcp.tool
defsend_notification(channel: str, message: str, priority: str = "normal") -> dict:
"""发送通知到各种渠道"""
pass

4.2 典型应用场景

1. AI助手增强

• ChatGPT/Claude插件开发
• IDE智能编程助手
• 企业内部知识问答系统

2. 自动化工作流

   @mcp.tool
   defautomate_daily_report():
   """自动生成日报"""
   # 1. 从数据库拉取数据
   # 2. 分析关键指标
   # 3. 生成可视化图表
   # 4. 发送到Slack/邮件
   pass
   

3. 数据平台集成

• 连接Snowflake/Databricks
• 实时数据查询
• 特征工程管道

4. 运维与监控

   @mcp.tool
   defcheck_system_health() -> dict:
   """检查系统健康状态"""
   return {
   "cpu_usage": get_cpu_usage(),
   "memory_usage": get_memory_usage(),
   "service_status": check_services(),
   "alerts": get_active_alerts()
       }
   

五、生产环境部署指南

5.1 使用HTTP传输（生产推荐）

# production_server.py
from fastmcp import FastMCP
import uvicorn

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

@mcp.tool
defproduction_tool():
"""生产环境工具"""
return {"status": "ready"}

if __name__ == "__main__":
# HTTP模式，适合云部署
    mcp.run(transport="http", host="0.0.0.0", port=8000)

5.2 Docker化部署

# 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 安全性考虑

# 添加认证中间件
from fastmcp import FastMCP
from fastmcp.middleware import AuthMiddleware

mcp = FastMCP("Secure-Server")

# 添加API密钥验证
@mcp.middleware
asyncdefauth_middleware(call, next_fn):
    api_key = call.context.get("headers", {}).get("x-api-key")

ifnot validate_api_key(api_key):
raise PermissionError("Invalid API key")

returnawait next_fn(call)

写在最后

通过本文，你已经掌握了：

1. MCP的核心概念：AI代理的标准接口协议
2. FastMCP的基本使用：用装饰器快速构建工具
3. 真实API集成：连接OpenWeatherMap获取实时数据
4. 生产级考虑：环境变量、错误处理、部署方案

MCP的真正威力在于标准化和可组合性。一旦你建立了一套MCP工具，它们可以被任何兼容MCP的AI客户端使用——无论是Claude Desktop、Cursor，还是你自定义的AI应用。