MCP（Model Context Protocol）从入门到实战：构建你的第一个 AI 工具服务器

# Python 3.10+
python3 --version
 
# 安装 MCP Python SDK
pip install mcp
 
# 如果需要调用 LLM 做测试，准备一个 API Key（OpenAI / Anthropic / 任意兼容接口）

mkdir mcp-weather-server
cd mcp-weather-server
python3 -m venv venv
source venv/bin/activate
pip install mcp httpx

from mcp.server import Server, NotificationOptions
from mcp.server.models import InitializationOptions
import mcp.server.stdio
import mcp.types as types
import httpx
import json
from typing import Any
 
# 创建 MCP 服务器实例
server = Server("weather-server")
 
# 定义一个模拟的天气数据源（实际项目中可替换为真实 API）
WEATHER_API_BASE = "https://api.open-meteo.com/v1"
 
@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
    """
    声明服务器提供的工具列表。
    MCP Client 会在连接时调用此方法进行工具发现。
    """
    return [
        types.Tool(
            name="get_current_weather",
            description="获取指定城市的实时天气信息",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称，支持中文（如：北京、上海、深圳）",
                    },
                    "units": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "温度单位",
                        "default": "celsius",
                    },
                },
                "required": ["city"],
            },
        ),
        types.Tool(
            name="get_forecast",
            description="获取指定城市未来三天的天气预报",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "城市名称"},
                    "days": {
                        "type": "integer",
                        "description": "预报天数（1-3）",
                        "minimum": 1,
                        "maximum": 3,
                        "default": 3,
                    },
                },
                "required": ["city"],
            },
        ),
    ]
 
@server.call_tool()
async def handle_call_tool(
    name: str, arguments: dict[str, Any] | None
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
    """
    处理工具调用请求。当 LLM 决定使用某个工具时，
    MCP Client 会调用此方法并将结果返回给 LLM。
    """
    if not arguments:
        raise ValueError("Missing arguments")
    
    if name == "get_current_weather":
        city = arguments["city"]
        units = arguments.get("units", "celsius")
        
        # 这里使用 Open-Meteo 免费 API（无需 API Key）
        # 实际项目中可以替换为任意天气服务
        async with httpx.AsyncClient() as client:
            # 先用地理编码查坐标
            geo_resp = await client.get(
                "https://geocoding-api.open-meteo.com/v1/search",
                params={"name": city, "count": 1, "language": "zh"},
            )
            geo_data = geo_resp.json()
            
            if not geo_data.get("results"):
                return [types.TextContent(
                    type="text",
                    text=f"未找到城市：{city}",
                )]
            
            lat = geo_data["results"][0]["latitude"]
            lon = geo_data["results"][0]["longitude"]
            city_name = geo_data["results"][0]["name"]
            
            # 查询天气
            temp_unit = "celsius" if units == "celsius" else "fahrenheit"
            weather_resp = await client.get(
                f"{WEATHER_API_BASE}/forecast",
                params={
                    "latitude": lat,
                    "longitude": lon,
                    "current_weather": True,
                    "temperature_unit": temp_unit,
                    "timezone": "auto",
                },
            )
            weather = weather_resp.json()["current_weather"]
            
            result = (
                f"📍 {city_name} 当前天气
"
                f"🌡️ 温度：{weather['temperature']}°{'C' if units == 'celsius' else 'F'}
"
                f"💨 风速：{weather['windspeed']} km/h
"
                f"🧭 风向：{weather['winddirection']}°
"
                f"🕐 更新时间：{weather['time']}"
            )
            return [types.TextContent(type="text", text=result)]
    
    elif name == "get_forecast":
        city = arguments["city"]
        days = arguments.get("days", 3)
        
        async with httpx.AsyncClient() as client:
            geo_resp = await client.get(
                "https://geocoding-api.open-meteo.com/v1/search",
                params={"name": city, "count": 1, "language": "zh"},
            )
            geo_data = geo_resp.json()
            
            if not geo_data.get("results"):
                return [types.TextContent(
                    type="text",
                    text=f"未找到城市：{city}",
                )]
            
            lat = geo_data["results"][0]["latitude"]
            lon = geo_data["results"][0]["longitude"]
            city_name = geo_data["results"][0]["name"]
            
            forecast_resp = await client.get(
                f"{WEATHER_API_BASE}/forecast",
                params={
                    "latitude": lat,
                    "longitude": lon,
                    "daily": ["temperature_2m_max", "temperature_2m_min", "weathercode"],
                    "forecast_days": days,
                    "timezone": "auto",
                },
            )
            daily = forecast_resp.json()["daily"]
            
            lines = [f"📍 {city_name} 未来 {days} 天预报"]
            for i in range(len(daily["time"])):
                lines.append(
                    f"
📅 {daily['time'][i]}: "
                    f"最高 {daily['temperature_2m_max'][i]}°C / "
                    f"最低 {daily['temperature_2m_min'][i]}°C"
                )
            
            return [types.TextContent(type="text", text="\n".join(lines))]
    
    else:
        raise ValueError(f"未知工具：{name}")
 
async def main():
    """启动 MCP 服务器"""
    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            InitializationOptions(
                server_name="weather-server",
                server_version="1.0.0",
                capabilities=server.get_capabilities(
                    notification_options=NotificationOptions(),
                    experimental_capabilities={},
                ),
            ),
        )
 
if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

python3 weather_server.py

{
  "mcpServers": {
    "weather-server": {
      "command": "python3",
      "args": ["/path/to/weather_server.py"]
    }
  }
}

import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
 
async def test():
    server_params = StdioServerParameters(
        command="python3",
        args=["weather_server.py"],
    )
    
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            
            # 列出可用工具
            tools = await session.list_tools()
            print("可用工具：")
            for tool in tools.tools:
                print(f"  - {tool.name}: {tool.description}")
            
            # 调用天气查询
            result = await session.call_tool(
                "get_current_weather",
                {"city": "北京"}
            )
            print("\n查询结果：")
            print(result.content[0].text)
 
asyncio.run(test())