大模型开发实战：（八）在 Dify 中使用 MCP 协议

pip install mcp httpx

import httpx
from mcp.server.fastmcp import FastMCP

# 初始化 FastMCP 服务器
mcp = FastMCP("Weather Service", host="0.0.0.0", port=8000)

asyncdef _get_lat_long(city_name: str):
    """
    内部辅助函数：使用 Open-Meteo Geocoding API 将城市名称转换为经纬度。
    """
    url = "https://geocoding-api.open-meteo.com/v1/search"
    params = {"name": city_name, "count": 1, "language": "zh", "format": "json"}
    
    asyncwith httpx.AsyncClient() as client:
        response = await client.get(url, params=params)
        data = response.json()
        
    ifnot data.get("results"):
        returnNone
        
    location = data["results"][0]
    return {
        "name": location["name"],
        "latitude": location["latitude"],
        "longitude": location["longitude"],
        "timezone": location.get("timezone", "UTC")
    }

@mcp.tool()
asyncdef get_weather(city_name: str) -> str:
    """
    获取指定城市的当前天气情况。
    输入城市名称（如 'Beijing', 'San Francisco', '上海'），返回温度、风速等信息。
    """
    # 1. 首先获取经纬度
    location = await _get_lat_long(city_name)
    ifnot location:
        returnf"错误：未找到城市 '{city_name}'。请尝试使用更具体的名称或英文名称。"

    # 2. 构建 Open-Meteo API 请求
    url = "https://api.open-meteo.com/v1/forecast"
    params = {
        "latitude": location["latitude"],
        "longitude": location["longitude"],
        "current": ["temperature_2m", "relative_humidity_2m", "apparent_temperature", "weather_code", "wind_speed_10m"],
        "timezone": location["timezone"]
    }

    asyncwith httpx.AsyncClient() as client:
        response = await client.get(url, params=params)
        weather_data = response.json()

    # 3. 解析并返回数据
    current = weather_data.get("current", {})
    units = weather_data.get("current_units", {})
    
    # 将天气代码 (WMO Code) 转换为文字描述 (简化版)
    weather_code = current.get("weather_code")
    weather_desc = "未知"
    if weather_code == 0: weather_desc = "晴朗"
    elif weather_code in [1, 2, 3]: weather_desc = "多云/阴"
    elif weather_code in [45, 48]: weather_desc = "有雾"
    elif51 <= weather_code <= 67: weather_desc = "有雨"
    elif71 <= weather_code <= 77: weather_desc = "有雪"
    elif weather_code >= 80: weather_desc = "雷雨/阵雨"

    return (
        f"--- {location['name']} 天气报告 ---\n"
        f"天气状况: {weather_desc}\n"
        f"当前温度: {current.get('temperature_2m')} {units.get('temperature_2m')}\n"
        f"体感温度: {current.get('apparent_temperature')} {units.get('apparent_temperature')}\n"
        f"相对湿度: {current.get('relative_humidity_2m')} {units.get('relative_humidity_2m')}\n"
        f"风速: {current.get('wind_speed_10m')} {units.get('wind_speed_10m')}"
    )

if __name__ == "__main__":
    # 运行 MCP 服务器
    mcp.run(transport="sse")

# Role
你是一位专业且贴心的出行规划专家。你擅长分析实时气象数据，提供包含穿衣、交通、安全及活动建议在内的全方位出行方案。

# Constraints
1. 语气专业、友好且富有亲和力。
2. 必须基于 MCP 工具返回的真实数据进行建议，严禁编造天气数值。
3. 输出应条理清晰，多使用列清单，避免大段文字。

# Goals
4. 数据调用：主动调用获取天气的 MCP 工具，获取用户指定目标的实时天气。
5. 多维分析：不仅报告气温，更要分析该天气对出行的具体影响。
6. 精准建议：给出极具参考价值的穿衣指南、交通方式推荐以及针对性提醒（如带伞、涂防晒霜或调整户外行程）。

# Role
你是一位资深的翻译专家，擅长将中文翻译为英文。用户的输入是一段中文，请将其翻译成最地道、最专业、最简洁的英文。直接输出翻译后的英文，不要包含任何解释或废话。

# Input
{{#1769390459829.chinese#}}