基于 MCP 协议的智能体落地示例：以天气预报工具为例

背景

在之前的文章中，我们介绍了 MCP（Model Context Protocol）的概念、架构及解决的问题，但缺少具体示例来进一步理解整套框架如何落地。

本文基于官方示例——获取天气预报，演示 MCP 落地的整条链路。大模型虽然能力强，但内容往往是非实时的，例如无法直接获取实时天气预报和严重天气警报。我们将使用 MCP 来解决这一问题。

构建一个服务器，提供两个工具：获取警报（get-alerts）和获取预报（get-forecast）。然后将该服务器连接到 MCP 主机（本例中为 Claude for Desktop）。

环境配置

1. 安装 uv

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

安装完成后会提示安装成功。

2. 安装依赖包

![安装依赖包截图]

3. 构建服务端代码

在 server.py 中构建相应的 get-alerts 和 get-forecast 工具：

from typing import Any
import asyncio
import httpx
from mcp.server.models import InitializationOptions
import mcp.types as types
from mcp.server import NotificationOptions, Server
import mcp.server.stdio

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

server = Server("weather")

@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
    """List available tools."""
    return [
        types.Tool(
            name="get-alerts",
            description="Get weather alerts for a state",
            inputSchema={
                "type": "object",
                "properties": {
                    "state": {"type": "string", "description": },
                },
                : [],
            },
        ),
        types.Tool(
            name=,
            description=,
            inputSchema={
                : ,
                : {
                    : {: , : },
                    : {: , : },
                },
                : [, ],
            },
        ),
    ]

  () -> [, ] | :
    
    headers = {: USER_AGENT, : }
    :
        response =  client.get(url, headers=headers, timeout=)
        response.raise_for_status()
         response.json()
     Exception:
         

 () -> :
    
    props = feature[]
     (
        
        
        
        
        
        
    )


  () -> [types.TextContent | types.ImageContent | types.EmbeddedResource]:
    
      arguments:
         ValueError()
    
     name == :
        state = arguments.get()
          state:
             ValueError()
        state = state.upper()
         (state) != :
             ValueError()
          httpx.AsyncClient()  client:
            alerts_url = 
            alerts_data =  make_nws_request(client, alerts_url)
              alerts_data:
                 [types.TextContent(=, text=)]
            features = alerts_data.get(, [])
              features:
                 [types.TextContent(=, text=)]
            formatted_alerts = [format_alert(feature)  feature  features[:]]
            alerts_text =  + .join(formatted_alerts)
             [types.TextContent(=, text=alerts_text)]
     name == :
        :
            latitude = (arguments.get())
            longitude = (arguments.get())
         (TypeError, ValueError):
             [types.TextContent(=, text=)]
          (- <= latitude <= )   (- <= longitude <= ):
             [types.TextContent(=, text=)]
          httpx.AsyncClient()  client:
            points_url = 
            points_data =  make_nws_request(client, points_url)
              points_data:
                 [types.TextContent(=, text=)]
            properties = points_data.get(, {})
            forecast_url = properties.get()
              forecast_url:
                 [types.TextContent(=, text=)]
            forecast_data =  make_nws_request(client, forecast_url)
              forecast_data:
                 [types.TextContent(=, text=)]
            periods = forecast_data.get(, {}).get(, [])
              periods:
                 [types.TextContent(=, text=)]
            formatted_forecast = []
             period  periods:
                forecast_text = (
                    
                    
                    
                    
                    
                )
                formatted_forecast.append(forecast_text)
            forecast_text =  + .join(formatted_forecast)
             [types.TextContent(=, text=forecast_text)]
    :
         ValueError()

  ():
      mcp.server.stdio.stdio_server()  (read_stream, write_stream):
         server.run(
            read_stream, write_stream,
            InitializationOptions(
                server_name=,
                server_version=,
                capabilities=server.get_capabilities(notification_options=NotificationOptions(), experimental_capabilities={}),
            ),
        )

 __name__ == :
    asyncio.run(main())