【智能体】用代码理解模型上下文协议（MCP）的最简单方法

import httpx  # 用于向气象API发起请求
from typing import Any
from mcp.server.fastmcp import FastMCP  # MCP服务器库

# 1. 初始化服务器
mcp = FastMCP("weather")  # 服务器命名为"weather"

# 外部API常量
NWS_API_BASE = "https://api.weather.gov"
USER_AGENT = "weather-app/1.0"


async def make_nws_request(url: str) -> dict[str, Any] | None:
    """向NWS API发起GET请求并返回JSON，出错时返回None"""
    headers = {
        "User-Agent": USER_AGENT,
        "Accept": "application/geo+json"
    }
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(url, headers=headers, timeout=30.0)
            response.raise_for_status()
            return response.json()
        except Exception:
            return None


def format_alert(feature: dict) -> str:
    """将天气警报特征转换为可读字符串"""
    props = feature["properties"]
    return (
        f"事件: {props.get('event', '未知')}\n"
        f"区域: {props.get('areaDesc', '未知')}\n"
        f"严重性: {props.get('severity', '未知')}\n"
        f"描述: {props.get('description', '无描述信息')}\n"
        f"应对指南: {props.get('instruction', '无应对指南')}"
    )


@mcp.tool()
async def get_alerts(state: str) -> str:
    """获取美国州级天气警报（使用两字母代码，如CA）"""
    url = f"{NWS_API_BASE}/alerts/active/area/{state}"
    data = await make_nws_request(url)
    if not data or "features" not in data:
        return "无法获取警报或未找到警报"
    if not data["features"]:
        return f"{state}当前无活跃警报"

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


@mcp.tool()
async def get_forecast(latitude: float, longitude: float) -> str:
    """获取坐标位置短期天气预报"""
    points_url = f"{NWS_API_BASE}/points/{latitude},{longitude}"
    points_data = await make_nws_request(points_url)
    if not points_data or "properties" not in points_data:
        return "无法获取该位置气象数据"

    forecast_url = points_data["properties"].get("forecast")
    if not forecast_url:
        return "未找到该位置预报URL"

    forecast_data = await make_nws_request(forecast_url)
    if not forecast_data or "properties" not in forecast_data:
        return "无法获取详细预报"

    periods = forecast_data["properties"].get("periods", [])
    if not periods:
        return "无可用预报时段"

    lines = []
    for period in periods[:5]:  # 仅取前5个时段
        lines.append(
            f"{period['name']}:\n"
            f"  温度: {period['temperature']}°{period['temperatureUnit']}\n"
            f"  风力: {period['windSpeed']} {period['windDirection']}\n"
            f"  预报: {period['detailedForecast']}"
        )
    return "\n\n".join(lines)


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

import os
import sys
import json
import asyncio
from typing import Optional
from dotenv import load_dotenv

load_dotenv()  # 从.env加载OPENAI_API_KEY

from openai import AsyncOpenAI
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from contextlib import AsyncExitStack


class MCPClient:
    def __init__(self):
        self.session: Optional[ClientSession] = None
        self.exit_stack = AsyncExitStack()
        # 使用异步OpenAI客户端
        self.openai = AsyncOpenAI(api_key=os.getenv("OPENAI_API_KEY"))
        self.model = "gpt-4o"  

    async def connect_to_server(self, server_script_path: str):
        """启动MCP服务器子进程并建立连接"""
        if server_script_path.endswith(".py"):
            command = "python"
        elif server_script_path.endswith(".js"):
            command = "node"
        else:
            raise ValueError("服务器脚本必须是.py或.js")

        server_params = StdioServerParameters(command=command, args=[server_script_path], env=None)
        stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))
        self.stdio, self.write = stdio_transport
        self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))
        await self.session.initialize()

        # 打印可用工具列表
        tool_list = await self.session.list_tools()
        print(f"已连接服务器。可用工具: {[t.name for t in tool_list.tools]}")

    async def _get_openai_functions(self):
        """将MCP工具转换为OpenAI函数调用格式"""
        tool_list = await self.session.list_tools()
        functions = []
        for tool in tool_list.tools:
            fn_schema = {
                "name": tool.name,
                "description": tool.description or tool.name,
                "parameters": {
                    "type": "object",
                    "properties": {}
                },
                "required": []
            }
            if tool.inputSchema and tool.inputSchema.get("type") == "object":
                fn_schema["parameters"] = tool.inputSchema
            functions.append(fn_schema)
        return functions

    async def process_user_message(self, user_message: str) -> str:
        if not self.session:
            raise RuntimeError("未找到MCP会话，请先调用connect_to_server")

        functions = await self._get_openai_functions()
        # 第一次GPT调用：提供用户消息+可用函数
        response = await self.openai.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": user_message}],
            functions=functions,
            function_call="auto"
        )
        assistant_message = response.choices[0].message

        if hasattr(assistant_message, "function_call") and assistant_message.function_call:
            fn_name = assistant_message.function_call.name
            fn_args_json = assistant_message.function_call.arguments
            try:
                fn_args = json.loads(fn_args_json) if fn_args_json else {}
            except json.JSONDecodeError:
                return "[错误] GPT生成非法JSON参数"

            print(f"\n[工具请求] GPT希望调用{fn_name}，参数：{fn_args}")
            tool_result = await self.session.call_tool(fn_name, fn_args)

            # 第二次GPT调用：提供工具执行结果
            followup = await self.openai.chat.completions.create(
                model=self.model,
                messages=[
                    {"role": "user", "content": user_message},
                    assistant_message.to_dict(),
                    {"role": "function", "name": fn_name, "content": tool_result.content}
                ]
            )
            return followup.choices[0].message.content
        else:
            # GPT不需要调用函数
            return assistant_message.content

    async def chat_loop(self):
        print("MCP + OpenAI 客户端。输入'quit'或'exit'退出")
        while True:
            user_inp = input("\n你: ")
            if user_inp.strip().lower() in ("quit", "exit"):
                break
            answer = await self.process_user_message(user_inp)
            print(f"\n助手: {answer}")

    async def cleanup(self):
        await self.exit_stack.aclose()


async def main():
    if len(sys.argv) < 2:
        print("用法: python client.py <服务器脚本路径>")
        sys.exit(1)
    server_path = sys.argv[1]

    client = MCPClient()
    try:
        await client.connect_to_server(server_path)
        await client.chat_loop()
    finally:
        await client.cleanup()


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

pip install "mcp[cli]" openai httpx python-dotenv

OPENAI_API_KEY=sk-12345...

python client.py weather.py