码农之家

专注优质代码开发,为软件行业发展贡献力量

MCP应用——天气查询AI助手

有时候会遇到这些问题:

  • 想用AI模型(比如DeepSeek、OpenAI)去调用一些外部API(比如天气查询、GitHub仓库管理),但不知道如何让AI和这些服务交互?
  • 想让AI自动帮你完成一些任务,比如查天气、爬取网页、管理文件,但不知道怎么让AI“学会”调用这些工具?
有了MCP,学习了解之后。便可以让AI模型可以轻松调用各种外部工具(Tools),就像人类使用USB接口一样方便。

1. MCP Server 开发:天气查询服务

1.1 环境准备

首先,搭建开发环境。

安装 uv(现代化的Python包管理工具)
pip install uv

或者用PowerShell安装(Windows用户推荐):
powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"

初始化项目 & 创建虚拟环境
uv init
uv venv .venv
.venv\Scripts\activate  # 激活虚拟环境(Windows)

1.2 编写天气查询Server

写一个 MCP Server,让它能调用OpenWeatherMap的API查询天气。从这里openweathermap.org/ 去拿到自己的OpenWeather API Key。

核心代码(weather.py):
import json
import httpx
from typing import Any
from mcp.server.fastmcp import FastMCP
# 初始化 MCP 服务器,命名为"WeatherServer"
# 嗯,这里相当于创建了一个天气服务的后台程序
mcp = FastMCP("WeatherServer")
# OpenWeather API 的配置信息
# 注意:这个API_KEY是示例,你需要自己去OpenWeather官网申请一个哦!
OPENWEATHER_API_BASE = "https://api.openweathermap.org/data/2.5/weather"
API_KEY = "xxx"  # 请替换为你自己的 OpenWeather API Key
USER_AGENT = "weather-app/1.0"  # 这个用来标识我们的应用
async def fetch_weather(city: str) -> dict[str, Any] | None:
    """
    从 OpenWeather API 获取天气信息。
    :param city: 城市名称(需使用英文,如 Beijing)
    :return: 天气数据字典;若出错返回包含 error 信息的字典
    """
    # 准备请求参数
    params = {
        "q": city,          # 城市名
        "appid": API_KEY,   # API密钥
        "units": "metric",  # 使用公制单位(摄氏度)
        "lang": "zh_cn"     # 返回中文描述
    }
    headers = {"User-Agent": USER_AGENT}  # 设置请求头
    
    # 使用httpx的异步客户端发送请求
    async with httpx.AsyncClient() as client:
        try:
            # 发送GET请求,设置30秒超时
            response = await client.get(
                OPENWEATHER_API_BASE,
                params=params,
                headers=headers,
                timeout=30.0
            )
            response.raise_for_status()  # 如果响应不成功会抛出异常
            return response.json()  # 返回解析后的JSON数据
        except httpx.HTTPStatusError as e:
            # 处理HTTP错误(比如404,401等)
            return {"error": f"HTTP 错误: {e.response.status_code}"}
        except Exception as e:
            # 处理其他异常(比如网络问题)
            return {"error": f"请求失败: {str(e)}"}
def format_weather(data: dict[str, Any] | str) -> str:
    """
    将天气数据格式化为易读文本。
    :param data: 天气数据(可以是字典或 JSON 字符串)
    :return: 格式化后的天气信息字符串
    """
    # 如果传入的是字符串,先尝试转换成字典
    if isinstance(data, str):
        try:
            data = json.loads(data)
        except Exception as e:
            return f"无法解析天气数据: {e}"
    
    # 如果数据中包含错误信息,直接返回错误提示
    if "error" in data:
        return f"⚠ {data['error']}"
    
    # 从返回数据中提取天气信息,使用get方法避免KeyError
    city = data.get("name", "未知")  # 城市名
    country = data.get("sys", {}).get("country", "未知")  # 国家
    temp = data.get("main", {}).get("temp", "N/A")  # 温度
    humidity = data.get("main", {}).get("humidity", "N/A")  # 湿度
    wind_speed = data.get("wind", {}).get("speed", "N/A")  # 风速
    
    # weather字段是个列表,可能为空,所以先给个默认值
    weather_list = data.get("weather", [{}])
    description = weather_list[0].get("description", "未知")  # 天气描述
    
    # 返回格式化的字符串,用了emoji让输出更直观
    return (
        f"🌍 {city}, {country}\n"
        f"🌡 温度: {temp}°C\n"
        f"💧 湿度: {humidity}%\n"
        f"🌬 风速: {wind_speed} m/s\n"
        f"⛅ 天气: {description}\n"
    )
@mcp.tool()  # 这个装饰器把这个函数注册为MCP可调用的工具
async def query_weather(city: str) -> str:
    """
    输入指定城市的英文名称,返回今日天气查询结果。
    :param city: 城市名称(需使用英文)
    :return: 格式化后的天气信息
    """
    data = await fetch_weather(city)  # 先获取原始天气数据
    return format_weather(data)  # 然后格式化输出
if __name__ == "__main__":
    # 以标准 I/O 方式运行 MCP 服务器
    # 这样其他程序就可以通过标准输入输出和这个服务通信啦
    mcp.run(transport='stdio')

关键点解析:
  • @mcp.tool() 这个装饰器告诉MCP:“嘿,这个函数是一个可以被AI调用的工具!”
  • query_weather 函数会调用OpenWeather API,返回格式化后的天气数据。

1.3 在Trae里配置MCP Server

为了让AI能调用这个Server,在 Trae(一个支持MCP的AI平台)里配置:
{
  "mcpServers": {
    "weather": {
      "command": "uv",
      "args": ["run", "weather.py"]
    }
  }
}

2. MCP Client 开发:让AI学会调用天气查询

2.1 安装依赖

Client需要调用OpenAI(或者DeepSeek)的模型,首先装好依赖:
uv add openai python-dotenv

2.2 配置模型参数

在 .env 文件里写你的API信息:
BASE_URL=https://api.deepseek.com
MODEL=deepseek-chat
API_KEY=你的API_KEY

2.3 编写MCP Client

  1. 连接MCP Server
  2. 让AI模型自动决定什么时候调用天气查询

核心代码(client.py):
from openai import OpenAI
from mcp import ClientSession
client = OpenAI(api_key="你的API_KEY")
async def ask_ai(query):
    response = client.chat.completions.create(
        model="deepseek-chat",
        messages=[{"role": "user", "content": query}],
        tools=[{
            "type": "function",
            "function": {
                "name": "query_weather",
                "description": "查询指定城市的天气",
                "parameters": {"city": {"type": "string"}}
            }
        }]
    )
    return response.choices[0].message

当运行Client并询问"今天北京天气怎么样?"时,AI会返回类似这样的格式化天气信息:
🌍 Beijing, CN
🌡 温度: 25°C
💧 湿度: 65%
🌬 风速: 3.2 m/s
⛅ 天气: 晴

ai识别当你问“北京天气”,ai自动回发现要去调query_weather工具,然后就自动把北京转成英文Beijing作为参数。
然后就去调用OpenWeather的api去发送请求获取北京天气,收到的数据是:
{
  "name": "Beijing",
  "sys": {"country": "CN"},
  "main": {"temp": 25, "humidity": 65},
  "wind": {"speed": 3.2},
  "weather": [{"description": "晴"}]
}

3. 进阶:让MCP更强大

3.1 更多工具

除了天气查询,还能让AI调用:
  1. GitHub仓库管理(查代码、提交PR)
  2. Figma设计稿解析(让AI看懂UI设计)
  3. 本地文件管理(让AI帮你整理文件夹)

只要写好对应的 @mcp.tool(),AI就能学会调用!

3.2 调试技巧

如果AI调工具出错,可以用 MCP Inspector 调试:
npx @modelcontextprotocol/inspector uv run weather.py

luodong

0 评论数