自己写一个mcp

python

准备工作

vscode

cline ( vscode 插件 )

python v3.12

pip v25

创建工程

uv init weather
cd weather
uv venv
.venv\Scripts\activate
uv add "mcp[cli]" httpx

编写 mcp

vscode 打开刚刚的创建的工程weather

根目录创建 weather.py, 写入以下代码：

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

# 初始化 FastMCP 服务 
mcp = FastMCP("weather", log_level="ERROR")


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


# NWS 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": "app"
  }
  async with httpx.AsyncClient() as client:
    try:
      response = await client.get(url, headers=headers)
      response.raise_for_status()
      return response.json()
    except httpx.RequestError as e:
      mcp.logger.error(f"Error making request to NWS API: {e}")
      return None


# Helper Functions
def to_celsius(f: int) -> int:
    return int((f - 32) * 5 / 9)

def format_wind_speed(speed_str: str) -> str:
    # Extract numbers from "13 mph" or "13 to 17 mph"
    nums = [int(x) for x in re.findall(r'\d+', speed_str)]
    if not nums:
        return ""
    
    # 1 mph = 1.60934 km/h
    kmh_nums = [int(n * 1.60934) for n in nums]
    if len(kmh_nums) >= 2:
        return f"{kmh_nums[0]}-{kmh_nums[1]} km/h"
    elif len(kmh_nums) == 1:
        return f"{kmh_nums[0]} km/h"
    return ""

def translate_wind_direction(direction: str) -> str:
    direction_map = {
        "N": "北风", "NNE": "北东北风", "NE": "东北风", "ENE": "东东北风",
        "E": "东风", "ESE": "东东南风", "SE": "东南风", "SSE": "南东南风",
        "S": "南风", "SSW": "南西南风", "SW": "西南风", "WSW": "西西南风",
        "W": "西风", "WNW": "西西北风", "NW": "西北风", "NNW": "北西北风"
    }
    return direction_map.get(direction, direction)

def translate_condition(condition: str) -> str:
    # A simple map for common conditions
    condition_map = {
        "Sunny": "晴天",
        "Mostly Sunny": "大部晴天",
        "Partly Sunny": "局部晴天",
        "Clear": "晴朗",
        "Mostly Clear": "大部晴朗",
        "Partly Cloudy": "局部多云",
        "Mostly Cloudy": "大部多云",
        "Cloudy": "多云",
        "Rain": "雨",
        "Showers": "阵雨",
        "Thunderstorms": "雷暴",
        "Snow": "雪",
        "Fog": "雾",
        "Chance Rain": "可能有雨",
        "Slight Chance Rain": "略有雨"
    }
    # Attempt partial matches or just return original if not found
    return condition_map.get(condition, condition)


# format alert
def format_alert(alert: dict[str, Any]) -> str:
  """
  Format an alert into a human-readable string.
  """
  return f"{alert['event']} - {alert['description']}" 


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

  Args:
    state: The US state to get alerts for.

  Returns:
    A string containing the alerts for the state.
  """
  
  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 retrieve alerts for state."

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

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



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

  Args:
    latitude: The latitude of the location.
    longitude: The longitude of the location.

  Returns:
    A string containing the forecast for 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 retrieve forecast for location."  

  forecast_url = points_data["properties"]["forecast"]  
  forecast_data = await make_nws_request(forecast_url)

  if not forecast_data:
    return "Unable to retrieve forecast for location."  

  # format the periods into a human-readable string
  periods = forecast_data["properties"]["periods"]
  forecast_lines = []
  
  # Limit to 5 periods, or fewer if not available
  for period in periods[:5]: 
    name = period['name'] # e.g. "Today", "Tonight"
    temp_f = period['temperature']
    temp_c = to_celsius(temp_f)
    wind_speed = period['windSpeed']
    wind_kmh = format_wind_speed(wind_speed)
    wind_dir = period['windDirection']
    wind_dir_cn = translate_wind_direction(wind_dir)
    short_forecast = period.get('shortForecast', period.get('detailedForecast', ''))
    condition_cn = translate_condition(short_forecast)
    
    # Format:
    # Period Name:
    # 温度: 59°F (约15°C)
    # 天气状况: 晴天
    # 风速: 13-17 mph西北风 (约21-27 km/h)
    
    period_text = f"""
{name}:
温度: {temp_f}°F (约{temp_c}°C)
天气状况: {condition_cn} ({short_forecast})
风速: {wind_speed}{wind_dir_cn} (约{wind_kmh})
    """.strip()
    
    forecast_lines.append(period_text)

  return "\n\n".join(forecast_lines)


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

配置 client

{
  "mcpServers": {
    "weather": {
      "disabled": false,
      "timeout": 60,
      "command": "uv",
      "args": [
        "--directory",
        "D:\\project\\mcp-learn\\weather",
        "run",
        "weather.py"
      ],
      "transportType": "stdio"
    }
  }
}

注意： args 的第二个路径是你的项目地址

验证