pip install "mcp[cli]" httpx

from mcp.server.fastmcp import FastMCP, Context
import httpx
import json
from typing import Dict, Any, List, Optional

# 创建一个 MCP 服务器实例
mcp = FastMCP("WeatherServer")

# 模拟的天气 API 密钥（实际应用中应该从环境变量获取）
WEATHER_API_KEY = "demo_key"

@mcp.tool()
async def get_current_weather(city: str, ctx: Context) -> Dict[str, Any]:
    await ctx.info(f"Getting current weather for {city}")
    try:
        # 模拟 API 调用
        result = {
            "city": city,
            "temperature": 23.5,
            "feels_like": 24.2,
            "humidity": 65,
            "pressure": 1012,
            "weather": "多云",
            "wind_speed": 3.6,
            "wind_direction": 120
        }

        await ctx.info(f"Got current weather for {city}")
        return result

    except Exception as e:
        await ctx.error(f"Error getting weather data: {str(e)}")
        return {"error": str(e)}


@mcp.tool()
async def get_weather_forecast(city: str, days: int = 3, ctx: Context) -> List[Dict[str, Any]]:
    await ctx.info(f"Getting weather forecast for {city} for {days} days")
    try:
        # 模拟 API 调用
        forecast = []
        import datetime
        import random
        today = datetime.datetime.now()

        weather_types = ["晴", "多云", "阴", "小雨", "中雨"]

        for i in range(days):
            day = today + datetime.timedelta(days=i)
            forecast.append({
                "date": day.strftime("%Y-%m-%d"),
                "temperature_high": round(20 + random.uniform(-5, 5), 1),
                "temperature_low": round(15 + random.uniform(-5, 5), 1),
                "weather": random.choice(weather_types),
                "humidity": random.randint(40, 90),
                "wind_speed": round(random.uniform(1.0, 5.0), 1)
            })

        await ctx.info(f"Got weather forecast for {city} for {days} days")
        return forecast

    except Exception as e:
        await ctx.error(f"Error getting weather forecast: {str(e)}")
        return [{"error": str(e)}]

@mcp.resource("cities://list")
def get_supported_cities() -> List[str]:
    """获取支持的城市列表"""
    return [
        "北京", "上海", "广州", "深圳", "成都",
        "杭州", "武汉", "西安", "南京", "重庆"
    ]


@mcp.resource("cities://{city}/info")
def get_city_info(city: str) -> Dict[str, Any]:
    """获取城市基本信息"""
    # 这里使用模拟数据
    city_info = {
        "北京": {"country": "中国", "province": "北京", "population": "2100万", "timezone": "UTC+8"},
        "上海": {"country": "中国", "province": "上海", "population": "2400万", "timezone": "UTC+8"},
        "广州": {"country": "中国", "province": "广东", "population": "1500万", "timezone": "UTC+8"}
        # 其他城市...
    }

    if city in city_info:
        return city_info[city]
    else:
        return {"error": "未找到该城市信息"}

@mcp.prompt()
def get_weather_help() -> str:
    """创建一个天气查询的帮助提示"""
    return """
    # 天气查询帮助

    我可以帮你查询城市的天气情况，你可以尝试：

    1. 查询当前天气：例如"请告诉我北京现在的天气"
    2. 查询天气预报：例如"请告诉我未来3天上海的天气预报"

    目前支持的城市包括：北京、上海、广州、深圳、成都、杭州、武汉、西安、南京、重庆。
    """

if __name__ == "__main__":
    mcp.run()

python weather_server.py

mcp run weather_server.py

mcp dev weather_server.py

import asyncio
import json
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def simple_client():
    """Client for weather server"""
    server_params = StdioServerParameters(
        command="python",
        args=["weather_server.py"]
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            print("✅ Connected to weather server")

            result = await session.call_tool(
                "get_current_weather",
                arguments={"city": "Beijing"}
            )
            weather_data = json.loads(result.content[0].text)
            print(f"🌤️ Beijing weather: {weather_data['weather']} {weather_data['temperature']}°C")

            forecast = await session.call_tool(
                "get_weather_forecast",
                arguments={"city": "Shanghai", "days": 3}
            )
            forecast_data = json.loads(forecast.content[0].text)
            print(f"📅 Shanghai 3-day forecast: {len(forecast_data)} days data")

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