【初学者指南】使用 MCP 协议快速开发智能天气查询应用

在本文中，我将详细介绍如何使用 MCP（Model Context Protocol）协议开发一个智能天气查询应用。这个示例非常适合初学者，通过简单易懂的步骤，你将学习如何将大型语言模型（LLM）与专用的天气查询服务相结合，让用户能够使用自然语言查询天气信息。

什么是 MCP？

MCP（Model Context Protocol）是一个开放标准，旨在让大型语言模型（LLM）能够安全、可靠地与外部工具和服务进行交互。通过 MCP，我们可以让 AI 模型调用特定的函数或服务，从而扩展其能力范围。

在我们的天气查询应用中，MCP 将作为 AI 模型与天气 API 之间的桥梁，使模型能够根据用户的自然语言查询获取实时天气数据。这种方式极大地简化了开发流程，使初学者也能快速构建功能强大的 AI 应用。

环境准备

安装 uv 包管理工具

首先，我们需要安装 uv 包管理工具，它是一个快速、可靠的 Python 包管理器：

curl -LsSf https://astral.sh/uv/install.sh | sh

安装完成后，我们可以使用 uv 来初始化项目并创建虚拟环境：

uv init weather-mcp-example  # 初始化项目
uv venv                      # 创建虚拟环境
source .venv/bin/activate    # 激活虚拟环境（Linux/macOS）
# 或在 Windows 上使用：
# .venv\Scripts\activate

初学者提示：虚拟环境可以帮助你隔离项目依赖，避免与系统其他 Python 项目产生冲突。

安装依赖包

接下来，我们安装项目所需的依赖包：

uv add mcp httpx openai python-dotenv

这些包的作用如下：

mcp: MCP 协议的 Python 实现，是我们项目的核心
httpx: 现代化的异步 HTTP 客户端，用于调用天气 API
openai: OpenAI API 的官方 Python 客户端，用于与 LLM 交互
python-dotenv: 用于从 .env 文件加载环境变量，保护 API 密钥

配置环境变量

创建一个 .env 文件，用于存储 API 密钥和其他配置：

BASE_URL="https://api.deepseek.com"  # 或其他 API 提供商
MODEL=deepseek-chat                  # 或其他模型名称
OPENAI_API_KEY="your-openai-api-key"
OPENWEATHER_API_KEY="your-openweather-api-key"

初学者提示：请确保将 your-openai-api-key 和 your-openweather-api-key 替换为你自己的 API 密钥。你可以从 OpenAI 或 deepseek 和 OpenWeather 获取这些密钥。

实现 MCP 服务器

我们首先来实现 MCP 服务器，它将提供天气查询功能。以下是 server.py 的完整代码：

import json
import httpx
import os
import sys
from typing import Any
from mcp.server.fastmcp import FastMCP
from dotenv import load_dotenv

# 加载 .env 文件，确保 API Key 受到保护
load_dotenv()

# 初始化 MCP 服务器
mcp = FastMCP("WeatherServer")

# OpenWeather API 配置
OPENWEATHER_API_BASE = "https://api.openweathermap.org/data/2.5/weather"
API_KEY = os.getenv("OPENWEATHER_API_KEY")  # 从 .env 文件中读取 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, "units": "metric", "lang": "zh_cn"}
    headers = {"User-Agent": USER_AGENT}

    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(
                OPENWEATHER_API_BASE, params=params, headers=headers, timeout=30.0
            )
            response.raise_for_status()
            # 使用stderr输出日志，这样不会干扰MCP通信
            # print(f"\n\nAPI Response: {response.json()}", file=sys.stderr, flush=True)
            return response.json()  # 返回字典类型
        except httpx.HTTPStatusError as e:
            print(
                f"\n\nHTTP Error: {e.response.status_code}", file=sys.stderr, flush=True
            )
            return {"error": f"HTTP 错误: {e.response.status_code}"}
        except Exception as e:
            print(f"\n\nRequest Failed: {str(e)}", file=sys.stderr, flush=True)
            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']}"

    # 提取数据时做容错处理
    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 可能为空列表，因此用 [0] 前先提供默认字典
    weather_list = data.get("weather", [{}])
    description = weather_list[0].get("description", "未知")

    return (
        f"🌍 {city}, {country}\n"
        f"🌡 温度: {temp}°C\n"
        f"💧 湿度: {humidity}%\n"
        f"🌬 风速: {wind_speed} m/s\n"
        f"🌤 天气: {description}\n"
    )


@mcp.tool()
async def query_weather(city: str) -> str:
    """
    输入指定城市的英文名称，返回今日天气查询结果。
    :param city: 城市名称（需使用英文）
    :return: 格式化后的天气信息
    """
    data = await fetch_weather(city)
    return format_weather(data)
    # return data


if __name__ == "__main__":
    # 以标准 I/O 方式运行 MCP 服务器
    mcp.run(transport="stdio")

代码解析：

初始化 MCP 服务器：使用 FastMCP 类创建一个名为 "WeatherServer" 的 MCP 服务器。
天气数据获取：fetch_weather 函数使用 httpx 库异步请求 OpenWeather API，获取指定城市的天气数据。
数据格式化：format_weather 函数将 API 返回的 JSON 数据转换为易读的文本格式，包括城市、温度、湿度、风速和天气描述。
MCP 工具注册：使用 @mcp.tool() 装饰器将 query_weather 函数注册为 MCP 工具，这样客户端就可以调用它。
错误处理：代码包含完善的错误处理机制，确保在 API 请求失败时能够返回有用的错误信息。

初学者提示：MCP 工具本质上是一个函数，通过装饰器 @mcp.tool() 注册后，可以被 LLM 调用。这种方式使 AI 能够访问外部服务和数据。

实现 MCP 客户端

接下来，我们实现 MCP 客户端，它将连接 OpenAI API 和我们的 MCP 服务器。以下是 client-3.py 的完整代码：

import asyncio
import os
import json
import sys
from typing import Optional
from contextlib import AsyncExitStack
from openai import OpenAI
from dotenv import load_dotenv
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

# 加载 .env 文件，确保 API Key 受到保护
load_dotenv()

class MCPClient:
    def __init__(self):
        """初始化 MCP 客户端"""
        self.exit_stack = AsyncExitStack()
        self.openai_api_key = os.getenv("OPENAI_API_KEY")  # 读取 OpenAI API Key
        self.base_url = os.getenv("BASE_URL")  # 读取 BASE URL
        self.model = os.getenv("MODEL")  # 读取 model
        
        if not self.openai_api_key:
            raise ValueError("❌ 未找到 OpenAI API Key，请在 .env 文件中设置OPENAI_API_KEY")
        
        # 创建OpenAI client
        self.client = OpenAI(api_key=self.openai_api_key, base_url=self.base_url)
        self.session: Optional[ClientSession] = None
        self.exit_stack = AsyncExitStack()
    
    async def connect_to_server(self, server_script_path: str):
        """连接到 MCP 服务器并列出可用工具"""
        is_python = server_script_path.endswith('.py')
        is_js = server_script_path.endswith('.js')
        if not (is_python or is_js):
            raise ValueError("服务器脚本必须是 .py 或 .js 文件")
        
        command = "python" if is_python else "node"
        server_params = StdioServerParameters(
            command=command,
            args=[server_script_path],
            env=None
        )
        
        # 启动 MCP 服务器并建立通信
        stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))
        self.stdio, self.write = stdio_transport
        self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))
        await self.session.initialize()
        
        # 列出 MCP 服务器上的工具
        response = await self.session.list_tools()
        tools = response.tools
        print("\n已连接到服务器，支持以下工具:", [tool.name for tool in tools])
    
    async def process_query(self, query: str) -> str:
        """
        使用大模型处理查询并调用可用的 MCP 工具 (Function Calling)
        """
        messages = [{"role": "user", "content": query}]
        response = await self.session.list_tools()
        available_tools = [{
            "type": "function",
            "function": {
                "name": tool.name,
                "description": tool.description,
                "input_schema": tool.inputSchema
            }
        } for tool in response.tools]
        
        # print(available_tools)
        response = self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            tools=available_tools
        )
        
        # 处理返回的内容
        content = response.choices[0]
        if content.finish_reason == "tool_calls":
            # 如何是需要使用工具，就解析工具
            tool_call = content.message.tool_calls[0]
            tool_name = tool_call.function.name
            tool_args = json.loads(tool_call.function.arguments)
            
            # 执行工具
            result = await self.session.call_tool(tool_name, tool_args)
            print(f"\n\n[Calling tool {tool_name} with args {tool_args}]\n\n")
            
            # 将模型返回的调用哪个工具数据和工具执行完成后的数据都存入messages中
            messages.append(content.message.model_dump())
            messages.append({
                "role": "tool",
                "content": result.content[0].text,
                "tool_call_id": tool_call.id,
            })
            
            # 将上面的结果再返回给大模型用于生产最终的结果
            response = self.client.chat.completions.create(
                model=self.model,
                messages=messages,
            )
            return response.choices[0].message.content
            
        return content.message.content
    
    async def chat_loop(self):
        """运行交互式聊天循环"""
        print("\n🤖 MCP 客户端已启动！输入 'quit' 退出")
        while True:
            try:
                query = input("\n你: ").strip()
                if query.lower() == 'quit':
                    break
                response = await self.process_query(query)  # 发送用户输入到 OpenAI API
                print(f"\n🤖 OpenAI: {response}")
            except Exception as e:
                print(f"\n⚠️ 发生错误: {str(e)}")
    
    async def cleanup(self):
        """清理资源"""
        await self.exit_stack.aclose()

async def main():
    if len(sys.argv) < 2:
        print("Usage: python client.py <path_to_server_script>")
        sys.exit(1)
    
    client = MCPClient()
    try:
        await client.connect_to_server(sys.argv[1])
        await client.chat_loop()
    finally:
        await client.cleanup()

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

代码解析：

初始化客户端：MCPClient 类负责初始化 OpenAI 客户端和 MCP 会话。
连接服务器：connect_to_server 方法启动 MCP 服务器进程并建立通信通道。
处理查询：process_query 方法是核心部分，它执行以下步骤：
- 将用户查询发送给 OpenAI API
- 提供可用的 MCP 工具信息
- 如果模型决定调用工具，则执行工具调用
- 将工具调用结果返回给模型，生成最终回复
交互式聊天：chat_loop 方法提供一个简单的命令行界面，让用户可以输入查询并查看回复。
资源清理：cleanup 方法确保在程序退出时正确关闭所有资源。

初学者提示：理解 process_query 方法是掌握 MCP 工作原理的关键。它展示了 LLM 如何决定调用工具，以及如何处理工具返回的结果。

运行应用

现在，我们可以运行我们的天气查询应用了：

uv run client-3.py server.py

这个命令会启动客户端，并让客户端连接到服务器。然后，你可以输入自然语言查询，例如：

"北京今天天气怎么样?"
"伦敦现在的温度是多少?"
"东京的湿度如何?"

系统会自动识别你的查询意图，调用适当的天气 API，并返回格式化的天气信息。

初学者提示：如果遇到错误，请检查你的 API 密钥是否正确设置，以及网络连接是否正常。

工作原理

让我们来看看整个系统是如何工作的：

用户输入查询：用户在命令行中输入自然语言查询。
OpenAI 处理查询：查询被发送到 OpenAI API，模型分析查询内容。
工具调用决策：如果模型认为需要查询天气，它会返回一个工具调用请求。
执行工具调用：客户端通过 MCP 协议调用服务器上的 query_weather 工具。
获取天气数据：服务器从 OpenWeather API 获取天气数据。
格式化结果：服务器将天气数据格式化为易读的文本。
返回结果：格式化的天气信息被返回给 OpenAI 模型。
生成最终回复：模型根据天气信息生成自然语言回复。
显示回复：最终回复显示给用户。

在 Cherry Studio 中集成使用

如果你想在 Cherry Studio 中使用这个应用。

MCP配置：

聊天框勾选MCP后操作即可。

常见问题解答

1. 为什么我的天气查询返回错误？

可能的原因包括：

API 密钥不正确或已过期
城市名称拼写错误或不存在
网络连接问题
OpenWeather API 服务暂时不可用

2. 如何添加更多天气功能？

你可以扩展 server.py 文件，添加更多的 MCP 工具，例如天气预报、空气质量指数等。只需创建新的函数并使用 @mcp.tool() 装饰器注册即可。

3. 如何支持更多语言？

你可以修改 fetch_weather 函数中的 lang 参数，OpenWeather API 支持多种语言。此外，你还可以使用 LLM 的翻译能力来处理不同语言的查询。

总结

在这篇文章中，我们学习了如何使用 MCP 协议开发一个智能天气查询应用。通过将大型语言模型与专用的天气查询服务相结合，我们创建了一个能够理解自然语言查询并提供实时天气信息的应用。

这个项目展示了 MCP 协议的强大功能，它允许 AI 模型安全、可靠地与外部工具和服务进行交互。通过这种方式，我们可以大大扩展 AI 模型的能力范围，使其能够访问实时数据和执行特定任务。

你可以在 GitHub 仓库中找到完整的代码。

希望这篇文章对你有所帮助，祝你在 MCP 开发之旅中取得成功！

什么是 MCP？

环境准备

安装 uv 包管理工具

安装依赖包

配置环境变量

实现 MCP 服务器

代码解析：

实现 MCP 客户端

代码解析：

运行应用

工作原理

在 Cherry Studio 中集成使用

常见问题解答

1. 为什么我的天气查询返回错误？

2. 如何添加更多天气功能？

3. 如何支持更多语言？

总结

推荐阅读

基于 Docker 的 PostgreSQL 数据库部署与运维指南

Nginx 目录穿越漏洞 复现及修复

mysql删除重复数据，只保留一条

评论

Nginx 目录穿越漏洞复现及修复