MCP Registry 与 2026 年 AI 工具发现生态：从碎片化到标准化的范式转移

from mcp.server.fastmcp import FastMCP
from typing import Optional
import httpx

# 初始化 MCP Server
mcp = FastMCP(
    "weather-service",
    version="1.0.0",
    description="查询全球城市实时天气信息的 MCP 服务器"
)

# 模拟天气数据源（实际可接入 OpenWeatherMap API）
WEATHER_DATA = {
    "beijing": {"temp": 22, "condition": "多云", "humidity": 45, "wind": "东南风 3级"},
    "shanghai": {"temp": 25, "condition": "晴", "humidity": 60, "wind": "东风 2级"},
    "tokyo": {"temp": 18, "condition": "小雨", "humidity": 75, "wind": "北风 4级"},
    "san francisco": {"temp": 15, "condition": "雾", "humidity": 80, "wind": "西风 5级"},
    "london": {"temp": 12, "condition": "阴", "humidity": 70, "wind": "西南风 3级"},
}

@mcp.tool()
async def get_weather(
    city: str,
    units: str = "celsius"
) -> str:
    """查询指定城市的实时天气。

    Args:
        city: 城市名称（英文，如 beijing, tokyo）
        units: 温度单位，celsius 或 fahrenheit
    """
    city_lower = city.lower().strip()

    if city_lower not in WEATHER_DATA:
        # 返回所有可用城市列表
        available = ", ".join(WEATHER_DATA.keys())
        return f"❌ 未找到城市 '{city}' 的天气数据。\n可用城市: {available}"

    weather = WEATHER_DATA[city_lower]
    temp = weather["temp"]

    if units == "fahrenheit":
        temp = temp * 9 / 5 + 32
        unit_label = "°F"
    else:
        unit_label = "°C"

    return (
        f"🌤️ {city.title()} 实时天气\n"
        f"📍 温度: {temp}{unit_label}\n"
        f"☁️ 天气: {weather['condition']}\n"
        f"💧 湿度: {weather['humidity']}%\n"
        f"🌬️ 风力: {weather['wind']}"
    )

@mcp.tool()
async def compare_weather(cities: list[str]) -> str:
    """比较多个城市的天气情况。

    Args:
        cities: 城市名称列表，最多 5 个城市
    """
    if len(cities) > 5:
        return "⚠️ 最多只能比较 5 个城市"

    results = []
    for city in cities:
        city_lower = city.lower().strip()
        if city_lower in WEATHER_DATA:
            w = WEATHER_DATA[city_lower]
            results.append(
                f"  • {city.title()}: {w['temp']}°C, {w['condition']}, "
                f"湿度 {w['humidity']}%"
            )
        else:
            results.append(f"  • {city}: 数据不可用")

    return "🌍 多城市天气对比\n" + "\n".join(results)

# 启动 Server
if __name__ == "__main__":
    mcp.run()

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

async def main():
    # 配置 MCP Server 连接参数
    server_params = StdioServerParameters(
        command="python",
        args=["weather-mcp-server.py"],
    )

    # 建立 stdio 连接
    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_weather",
                arguments={"city": "beijing", "units": "celsius"}
            )
            print(f"\n🌤️ 查询结果:\n{result.content[0].text}")

            # 调用比较工具
            result = await session.call_tool(
                "compare_weather",
                arguments={"cities": ["beijing", "tokyo", "london"]}
            )
            print(f"\n🌍 对比结果:\n{result.content[0].text}")

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

"""
多 MCP Server 协同的 AI Agent 示例
展示如何通过 MCP 协议协调多个外部工具
"""
import asyncio
from dataclasses import dataclass
from typing import Protocol

# 定义工具接口协议
class Tool(Protocol):
    async def execute(self, **kwargs) -> str: ...

@dataclass
class MCPToolAdapter:
    """MCP 工具适配器——统一所有 MCP Server 的调用接口"""
    name: str
    description: str
    session: object  # MCP ClientSession

    async def execute(self, **kwargs) -> str:
        """通过 MCP 协议调用远程工具"""
        result = await self.session.call_tool(self.name, kwargs)
        return result.content[0].text

class WeatherEmailAgent:
    """基于天气数据自动发送邮件的智能 Agent"""

    def __init__(self):
        self.tools: dict[str, MCPToolAdapter] = {}
        self.task_history: list[dict] = []

    def register_tool(self, tool: MCPToolAdapter):
        """注册 MCP 工具到 Agent"""
        self.tools[tool.name] = tool
        print(f"📦 已注册工具: {tool.name} - {tool.description}")

    def get_available_tools(self) -> str:
        """列出所有可用工具"""
        return "\n".join(
            f"  • {name}: {tool.description}"
            for name, tool in self.tools.items()
        )

    async def handle_task(self, task_description: str) -> str:
        """
        处理用户任务——根据描述自主决定调用哪些工具
        这里简化为直接路由，实际应用中可接入 LLM 进行意图理解
        """
        self.task_history.append({
            "task": task_description,
            "status": "processing"
        })

        task_lower = task_description.lower()

        # 简单的任务路由逻辑
        if "天气" in task_lower or "weather" in task_lower:
            if "compare" in task_lower or "对比" in task_lower:
                result = await self.tools["compare_weather"].execute(
                    cities=["beijing", "tokyo", "london"]
                )
            else:
                # 提取城市名称（简化处理）
                city = "beijing"  # 默认值
                result = await self.tools["get_weather"].execute(city=city)
        else:
            result = f"⚠️ 无法处理任务: {task_description}\n可用工具:\n{self.get_available_tools()}"

        self.task_history[-1]["status"] = "completed"
        self.task_history[-1]["result"] = result
        return result

# 使用示例
async def main():
    agent = WeatherEmailAgent()

    # 模拟注册 MCP 工具（实际环境中通过 Registry 发现）
    print("🤖 初始化 AI Agent...")
    print()

    # 注意：实际使用时需要通过 MCP ClientSession 连接
    # 这里仅展示 Agent 架构设计
    print("📋 可用工具:")
    print("  • get_weather: 查询城市实时天气")
    print("  • compare_weather: 多城市天气对比")
    print()

    # 模拟任务处理
    tasks = [
        "查询北京天气",
        "对比北京、东京和伦敦的天气",
    ]

    for task in tasks:
        print(f"📥 收到任务: {task}")
        # result = await agent.handle_task(task)
        # print(f"📤 结果: {result}")
        print()

    print(f"📊 任务统计: 已处理 {len(agent.task_history)} 个任务")

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