从装饰器到MCP协议：用FastMCP构建可复用的天气查询Agent

在当今AI应用开发领域，如何快速构建功能模块化、可复用的智能服务成为开发者关注的焦点。FastMCP作为新一代MCP协议实现框架，通过其独特的装饰器系统，为开发者提供了简洁高效的开发体验。本文将手把手指导您使用@mcp.tool()、@mcp.resource()和@mcp.prompt()三大核心装饰器，构建一个完整的天气查询Agent服务。

1. 环境准备与项目初始化

首先确保您的开发环境已安装Python 3.8+版本。推荐使用uv工具管理依赖：

uv add mcp

创建项目目录结构如下：

weather_agent/
├── server.py       # 主服务文件
├── config.py       # 配置文件
└── templates/      # 提示模板目录

在server.py中初始化FastMCP服务：

from mcp.server.fastmcp import FastMCP

mcp = FastMCP(
    name="weather_agent",
    description="多功能天气查询服务"
)

2. 工具函数开发与@mcp.tool()应用

@mcp.tool()装饰器将普通Python函数转化为MCP工具，这是构建Agent的基础能力单元。我们首先实现天气API查询功能：

import requests
from typing import Literal

WEATHER_API_KEY = "your_api_key"  # 实际使用时应从配置读取

@mcp.tool(
    title="实时天气查询",
    description="获取指定城市的实时天气数据"
)
def get_weather(
    city: str,
    unit: Literal["celsius", "fahrenheit"] = "celsius"
) -> dict:
    """
    调用第三方天气API获取实时数据
    参数:
        city: 城市名称(中文或拼音)
        unit: 温度单位(摄氏度/华氏度)
    返回:
        {
            "temperature": 当前温度,
            "condition": 天气状况,
            "humidity": 湿度百分比
        }
    """
    params = {
        "key": WEATHER_API_KEY,
        "location": city,
        "unit": unit[:1]
    }
    response = requests.get(
        "https://api.weather.com/v3/current",
        params=params
    )
    return {
        "temperature": response.json()["temp"],
        "condition": response.json()["condition"],
        "humidity": response.json()["humidity"]
    }

关键点说明：

• 类型注解(Literal等)会被自动转换为API参数约束
• 函数文档字符串将作为工具描述展示在MCP Inspector中
• 返回值的类型定义确保客户端能获得结构化数据

3. 资源管理与@mcp.resource()配置

@mcp.resource()提供静态资源和动态模板两种使用方式：

3.1 静态城市列表资源

@mcp.resource("resource://supported_cities")
def get_supported_cities() -> list:
    """返回服务支持的城市列表"""
    return [
        {"name": "北京", "code": "BJ"},
        {"name": "上海", "code": "SH"},
        # 其他城市数据...
    ]

3.2 动态天气详情模板

@mcp.resource("resource://{city}/details")
def get_city_details(city: str) -> dict:
    """生成城市天气详情页面"""
    weather = get_weather(city)  # 复用之前定义的tool
    return {
        "city": city,
        "current": weather,
        "forecast": get_forecast(city)  # 假设存在的预报函数
    }

资源URI的{city}参数会自动映射到函数参数，这种动态绑定机制极大简化了RESTful风格API的开发。

4. 对话模板与@mcp.prompt()设计

@mcp.prompt()装饰器用于定义标准化的对话模板，确保AI交互的一致性：

@mcp.prompt()
def weather_inquiry(
    city: str = "北京",
    detail_level: Literal["brief", "full"] = "brief"
) -> list[dict]:
    """
    生成天气查询的标准对话模板
    参数:
        city: 查询城市(默认北京)
        detail_level: 详情级别(brief只返回温度，full返回完整报告)
    """
    base_content = f"请提供{city}的天气信息"
    
    if detail_level == "full":
        return [
            {
                "role": "user",
                "content": base_content + "，包括温度、湿度、风速等完整数据"
            }
        ]
    else:
        return [
            {
                "role": "user",
                "content": base_content + "，只需返回当前温度"
            }
        ]

5. 服务调试与MCP Inspector使用

FastMCP内置的MCP Inspector提供了可视化调试界面。启动服务时添加开发模式参数：

mcp dev server.py --reload

Inspector主要功能面板：

面板名称	功能描述	使用场景
Tools	查看注册的工具列表	测试工具调用
Resources	静态/动态资源管理	预览资源内容
Prompts	对话模板调试	验证提示词效果

调试技巧：

• 在工具函数中添加print语句输出调试信息
• 使用try-except块捕获并记录异常
• 通过Inspector的"Raw JSON"视图检查数据格式

6. 客户端集成实践

完成服务开发后，可通过多种方式与客户端集成：

6.1 命令行调用示例

# 获取北京天气
response = mcp.call_tool("get_weather", {"city": "北京"})
print(response.json())

6.2 Web API集成

FastMCP自动生成OpenAPI文档，可通过Swagger UI访问：

http://localhost:8234/docs

6.3 与Claude Desktop集成

在Claude配置文件中添加：

agents:
  weather:
    endpoint: http://localhost:8234
    tools:
      - get_weather
    prompts:
      - weather_inquiry

7. 高级技巧与性能优化

7.1 异步工具开发

对于IO密集型操作，使用async/await提升并发性能：

@mcp.tool()
async def async_get_weather(city: str) -> dict:
    """异步版本天气查询"""
    async with aiohttp.ClientSession() as session:
        async with session.get(API_URL, params={"city": city}) as resp:
            return await resp.json()

7.2 缓存机制实现

通过@mcp.resource()实现自动缓存：

from datetime import timedelta
from functools import lru_cache

@mcp.resource("resource://cached_weather/{city}")
@lru_cache(maxsize=100, ttl=timedelta(minutes=30))
def get_cached_weather(city: str) -> dict:
    """带缓存的天气查询"""
    return get_weather(city)

7.3 错误处理最佳实践

@mcp.tool()
def safe_get_weather(city: str) -> dict:
    """带错误处理的天气查询"""
    try:
        return get_weather(city)
    except requests.exceptions.RequestException as e:
        return {
            "error": "天气服务暂不可用",
            "detail": str(e)
        }

8. 项目扩展与模块化

建议将大型项目拆分为多个模块：

weather_agent/
├── core/           # 核心功能
│   ├── tools.py    # 工具函数
│   └── models.py   # 数据模型
├── api/            # 接口定义
│   ├── v1.py       # API版本1
│   └── v2.py       # API版本2
└── main.py         # 服务入口

在main.py中整合各模块：

from core.tools import *
from api.v1 import api_router

mcp = FastMCP()
mcp.include_router(api_router)

通过这种模块化设计，天气查询Agent可以轻松扩展空气质量监测、灾害预警等新功能，同时保持代码的可维护性。FastMCP的装饰器系统让每个功能模块都能独立开发测试，最终无缝集成到统一的服务框架中。