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

从装饰器到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( nameweather_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, paramsparams ) 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 --reloadInspector主要功能面板面板名称功能描述使用场景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/docs6.3 与Claude Desktop集成在Claude配置文件中添加agents: weather: endpoint: http://localhost:8234 tools: - get_weather prompts: - weather_inquiry7. 高级技巧与性能优化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(maxsize100, ttltimedelta(minutes30)) 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的装饰器系统让每个功能模块都能独立开发测试最终无缝集成到统一的服务框架中。