借助MCP尝试创建自己的Agent（二）

借助MCP尝试创建自己的Agent（二）

• Python SDK构建个性化天气查询服务器
• 1. 环境配置
• 2. 构建天气服务器
• 3. 在Claude桌面版中配置天气查询服务器
• 4. 改进weather服务器代码

关于模型上下文协议（model context protocol, MCP）的相关概念可以参考上一篇博客：借助MCP尝试创建自己的Agent（一）

Python SDK构建个性化天气查询服务器

尝试在claude客户端获取天气信息

目前，很多未联网的大模型是不能获取某地当天的天气信息的，例如，当我询问claude桌面版Q： 深圳今天的天气怎么样？却得到了下面的回复?

如何使本地agent具有天气查询能力呢？MCP文档¹给出了详细的实例，我在这里复现一遍。

1. 环境配置

在获取天气信息之前，需要先配置uv环境, uv是一个类似于conda的python包或项目管理工具²

# Windows
powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"
# MacOS/Linux
curl -LsSf https://astral.sh/uv/install.sh | sh

我用的是windows系统，在PowerShell 安装完成后如图所示。

安装完成后，重启Powershell，并通过下面的命令创建访问天气的项目。

# windows
# Create a new directory for our project
uv init weather
cd weather
# Create virtual environment and activate it
uv venv
.venv\Scripts\activate
# Install dependencies
uv add mcp[cli] httpx
# Create our server file
new-item weather.py

# MacOS/Linux
# Create a new directory for our project
uv init weather
cd weather
# Create virtual environment and activate it
uv venv
source .venv/bin/activate
# Install dependencies
uv add "mcp[cli]" httpx
# Create our server file
touch weather.py

这里需要在当前目录下创建一个名为weather的文件，如果不想在当前目录下创建，可以先进入到其他目录中再进行操作，我的操作结果如下图所示：

到这里已经完成了基本环境的配置，下一步可以开始配置服务器了

2. 构建天气服务器

打开之前创建的weather.py文件，导入相应的包，并且设置天气查询服务的MCP服务器设置

from typing import Any
import httpx
from mcp.server.fastmcp import FastMCP

# Initialize FastMCP server
mcp = FastMCP("weather")

# Constants
NWS_API_BASE = "https://api.weather.gov"
USER_AGENT = "weather-app/1.0"

The FastMCP class uses Python type hints and docstrings to automatically generate tool definitions, making it easy to create and maintain MCP tools.

FastMCP 类使用 Python 类型提示和文档字符串自动生成工具定义，使得创建和维护 MCP 工具变得容易。

接着，添加帮助函数，来从国家天气服务（National Weather Service）API查询天气数据

async def make_nws_request(url: str) -> dict[str, Any] | None:
    """Make a request to the NWS API with proper error handling."""
    headers = {
        "User-Agent": USER_AGENT,
        "Accept": "application/geo+json"
    }
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(url, headers=headers, timeout=30.0)
            response.raise_for_status()
            return response.json()
        except Exception:
            return None

def format_alert(feature: dict) -> str:
    """Format an alert feature into a readable string."""
    props = feature["properties"]
    return f"""
Event: {props.get('event', 'Unknown')}
Area: {props.get('areaDesc', 'Unknown')}
Severity: {props.get('severity', 'Unknown')}
Description: {props.get('description', 'No description available')}
Instructions: {props.get('instruction', 'No specific instructions provided')}
"""

其次，添加工具执行处理模块来负责执行每个工具：

@mcp.tool()
async def get_alerts(state: str) -> str:
    """Get weather alerts for a US state.

    Args:
        state: Two-letter US state code (e.g. CA, NY)
    """
    url = f"{NWS_API_BASE}/alerts/active/area/{state}"
    data = await make_nws_request(url)

    if not data or "features" not in data:
        return "Unable to fetch alerts or no alerts found."

    if not data["features"]:
        return "No active alerts for this state."

    alerts = [format_alert(feature) for feature in data["features"]]
    return "\n---\n".join(alerts)

@mcp.tool()
async def get_forecast(latitude: float, longitude: float) -> str:
    """Get weather forecast for a location.

    Args:
        latitude: Latitude of the location
        longitude: Longitude of the location
    """
    # First get the forecast grid endpoint
    points_url = f"{NWS_API_BASE}/points/{latitude},{longitude}"
    points_data = await make_nws_request(points_url)

    if not points_data:
        return "Unable to fetch forecast data for this location."

    # Get the forecast URL from the points response
    forecast_url = points_data["properties"]["forecast"]
    forecast_data = await make_nws_request(forecast_url)

    if not forecast_data:
        return "Unable to fetch detailed forecast."

    # Format the periods into a readable forecast
    periods = forecast_data["properties"]["periods"]
    forecasts = []
    for period in periods[:5]:  # Only show next 5 periods
        forecast = f"""
{period['name']}:
Temperature: {period['temperature']}°{period['temperatureUnit']}
Wind: {period['windSpeed']} {period['windDirection']}
Forecast: {period['detailedForecast']}
"""
        forecasts.append(forecast)

    return "\n---\n".join(forecasts)

最后，实现运行该服务器，在创建的weather文件夹下通过uv run weather.py运行天气查询服务

if __name__ == "__main__":
    # Initialize and run the server
    mcp.run(transport='stdio')

3. 在Claude桌面版中配置天气查询服务器

找到并打开claude_desktop_config.json文件，将添加的weather服务加入到mcpServers中

{
    "mcpServers": {
        "weather": {
            "command": "uv",
            "args": [
                "--directory",
                "C:\\ABSOLUTE\\PATH\\TO\\PARENT\\FOLDER\\weather",
                "run",
                "weather.py"
            ]
        }
    }
}

重启claude 客户端

!!! 然而，重启之后出现了问题，发现配置失败。原因是访问位置不在美国，导致了天气访问的API出现了问题。挂美国的梯子也不行，所以对weather.py的代码进行改进

4. 改进weather服务器代码

第二天我重新按官方文档设置了一遍，目前可以查看美国各州的天气警告信息，但访问其他地方的天气时还会出现问题，估计是API的原因

这里的改进用了和风天气的API³，想尝试的小伙伴可以看一看，可以免费使用1000个token。根据和风天气返回数据的特点，删除了预测和警告信息查看功能，只保留了查询目前天气情况的功能，该变代码如下：

from typing import Any
import httpx
from mcp.server.fastmcp import FastMCP
import asyncio
# Initialize FastMCP server
mcp = FastMCP("weather")
# Constants
NWS_API_BASE = "https://devapi.qweather.com/v7/weather"
USER_AGENT = "weather-app/1.0"
key = "xxxx" # 输入你自己的key
async def make_nws_request(url: str) -> dict[str, Any] | None:
    """
        访问网站
    """
    headers = {
        "User-Agent": USER_AGENT,
        "Accept": "application/geo+json"
    }
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(url, headers=headers, timeout=30.0)
            response.raise_for_status()
            return response.json()
        except Exception:
            return None
@mcp.tool()
async def get_now(latitude: float, longitude: float) -> str:
    """Get weather forecast for a location.

    Args:
        latitude: Latitude of the location
        longitude: Longitude of the location
    """
    # First get the forecast grid endpoint
    points_url =  f"{NWS_API_BASE}/now?location={longitude},{latitude}&key={key}"
    points_data = await make_nws_request(points_url)

    if not points_data:
        return "Unable to fetch detailed forecast"
    now = points_data['now']
    weather_info = (
        f"观测时间: {now['obsTime']}\n"
        f"温度: {now['temp']}°C\n"
        f"体感温度: {now['feelsLike']}°C\n"
        f"天气: {now['text']}\n"
        f"风向: {now['windDir']}\n"
        f"风力等级: {now['windScale']}\n"
        f"风速: {now['windSpeed']}km/h\n"
        f"湿度: {now['humidity']}%\n"
        f"降水量: {now['precip']}mm\n"
        f"气压: {now['pressure']}hPa\n"
        f"能见度: {now['vis']}km"
    )
    return weather_info
if __name__ == "__main__":
    # Initialize and run the server
    mcp.run(transport='stdio')
    # weather_data = asyncio.run(get_now(22.5,114))
    # if weather_data:
    #     print(weather_data)

最后，成功通过claude获取了某个城市天气信息，这里不仅仅是添加了天气服务，而且窥探了通过python自定义服务的功能，这里需要仔细阅读官方给出了Python SDK网页内容，才能实现自己后续的开发。

今天的分享就到这里，欢迎各位大佬批评指正。我是会编程的加缪，目前正在学习LLMs相关知识，有问题欢迎与我讨论！