一. 什么是MCP

MCP（Model Context Protocol 模型上下文协议）是一个用于将 AI 应用程序连接到外部系统的开源标准。它通过标准化大语言模型与外部工具的交互方式，解决了AI 智能体应用开发中的痛点，正迅速成为 AI 领域的核心基础设施标准。Anthropic 于2024年底首次提出该协议，随后被 OpenAI、阿里、腾讯、字节跳动等全球顶尖科技企业所采纳，使其迅速成为 AI 基础设施的关键支柱。

MCP 的核心原理是将互联网服务（高德、谷歌）或本地操作系统 API（文件系统、数据库、终端）封装成 AI 智能体能够理解和使用的工具，让 AI 智能体能够自由地调用这些工具实现复杂的业务逻辑和功能。

二. MCP的核心架构

MCP 采用客户端-服务器架构，其中 MCP 主机（例如 Claude Code 或 Claude Desktop 这样的 AI 应用程序）会建立与一个或多个 MCP 服务器的连接。MCP 主机通过为每个 MCP 服务器创建一个 MCP 客户端来实现这一点。每个 MCP 客户端都与其对应的 MCP 服务器保持专用连接。

• MCP Host：协调并管理一个或多个 MCP Client的 AI 应用。
• MCP Client：一个维护与 MCP Server连接并从 MCP Server获取上下文以供 MCP Host 使用的组件。
• MCP Server：为 MCP 客户端提供上下文的程序。
  
  MCP 主要分为以下两层
• Data layer：定义基于 JSON-RPC 的 Client-Server 通信协议，涵盖生命周期管理及核心原语如 tools、resources、prompts 和 notifications。
• Transport layer：定义了 Client 和 Server 之间是如何建立连接、如何把消息打包发送，以及授权认证的。

三. MCP的核心原语

MCP 原语是 MCP 中最重要的概念。它们定义了客户端和服务器可以相互提供的内容。这些原语规定了可以与 AI 应用共享的上下文信息类型以及可以操作执行的范围。MCP 定义了服务器可暴露的三种核心原语。

• Tool（工具）：AI 应用可调用的可执行函数，用于执行各类操作（如文件操作、API 调用、数据库查询）。
• Resources（资源）：为 AI 应用提供上下文信息的数据源（如文件内容、数据库记录、API 响应）。
• Prompts（提示词）：借助提示词，服务器可定义可复用的模板与工作流，供客户端便捷地向用户和 LLM 展示，从而高效实现常见 LLM 交互的标准化与复用。

四. MCP的通信模式

MCP 的通信模式主要分为基于本地进程通信的 STDIO 和基于网络的 HTTP 传输。最新的协议版本使用 Streamable HTTP 取代了之前的 HTTP + SSE 方式，当前 MCP 官方推荐的传输方式，尤其适合生产环境和云服务。

五. 快速入门

服务器开发
引入包并初始化 MCP 服务

from typing import Any

import httpx
from mcp.server.fastmcp import FastMCP

# uv add "mcp[cli]" httpx
# Python版本 >= 3.10
# Python MCP SDK >= 1.2.0
# 初始化FastMCP服务器
mcp = FastMCP("weather")


# 定义常量
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响应"""
    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"""
Event: {props.get("event", "Unknown")}
Area: {props.get("areaDesc", "Unknown")}
Severity: {props.get("severity", "Unknown")}
Description: {props.get("description", "No description available")}
Instructions: {props.get("instruction", "No specific instructions provided")}
"""


@mcp.tool()
async def get_alerts(state: str) -> str:
    """获取指定州的天气警报（使用两字母州代码如CA/NY）"""
    url = f"{NWS_API_BASE}/alerts/active/area/{state}"
    data = await make_nws_request(url)

    if not data:
        return "无法连接到天气服务，请稍后再试。"
    if "features" not in data:
        return "天气服务返回格式异常。"
    if not data["features"]:
        return f"✅ {state}州目前没有活动天气警报"

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


@mcp.tool()
async def get_forecast(latitude: float, longitude: float) -> str:
    """获取指定位置的天气预报

    Args:
        latitude: 位置的纬度
        longitude: 位置的经度
    """
    # 首先获取预报网格端点
    points_url = f"{NWS_API_BASE}/points/{latitude},{longitude}"
    points_data = await make_nws_request(points_url)

    if not points_data:
        return "无法获取位置信息，请检查经纬度是否正确。"

    # 获取预报URL
    forecast_url = points_data["properties"]["forecast"]
    forecast_data = await make_nws_request(forecast_url)

    if not forecast_data:
        return "无法获取详细预报信息。"

    # 格式化预报时段（显示接下来3个时段）
    periods = forecast_data["properties"]["periods"]
    forecasts = []
    for period in periods[:5]:  # Only show next 5 periods
        forecast = f"""
{period["name"]}:
Temperature: {period["temperature"]}°{period["temperatureUnit"]}
Wind: {period["windSpeed"]} {period["windDirection"]}
Forecast: {period["detailedForecast"]}
"""
        forecasts.append(forecast)

    return "\n---\n".join(forecasts)


def main():
    # 初始化并运行服务
    mcp.run(transport="stdio")


if __name__ == "__main__":
    main()

客户端开发

import asyncio
import os
import sys
from contextlib import AsyncExitStack
from pathlib import Path

from anthropic import Anthropic
from dotenv import load_dotenv
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

# uv add mcp anthropic python-dotenv
# 加载 .env 文件中的环境变量（如 API 密钥）
load_dotenv()

# Claude 模型配置常量
ANTHROPIC_MODEL = "claude-sonnet-4-5"


class MCPClient:
    """
    MCP (Model Context Protocol) 客户端

    这个客户端负责：
    1. 连接到 MCP 服务器（通过 stdio）
    2. 获取服务器提供的工具列表
    3. 与 Claude API 交互，让 Claude 决定何时调用工具
    4. 执行工具调用并将结果返回给 Claude
    """

    def __init__(self):
        # MCP 会话对象，用于与服务器通信
        self.session: ClientSession | None = None

        # AsyncExitStack 用于管理多个异步上下文管理器
        # 确保所有资源在退出时正确清理
        self.exit_stack = AsyncExitStack()

        # Anthropic 客户端（延迟初始化）
        self._anthropic: Anthropic | None = None

    @property
    def anthropic(self) -> Anthropic:
        """
        延迟初始化 Anthropic 客户端

        只有在实际需要调用 Claude API 时才创建客户端，
        避免在不需要时占用资源
        """
        if self._anthropic is None:
            self._anthropic = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
        return self._anthropic

    async def connect_to_server(self, server_script_path: str):
        """
        连接到 MCP 服务器

        Args:
            server_script_path: 服务器脚本的路径（.py 或 .js 文件）

        Raises:
            ValueError: 如果文件类型不是 .py 或 .js
        """
        # 根据文件扩展名判断服务器类型
        is_python = server_script_path.endswith(".py")
        is_js = server_script_path.endswith(".js")
        if not (is_python or is_js):
            raise ValueError("Server script must be a .py or .js file")

        # 根据服务器类型配置启动参数
        if is_python:
            # Python 服务器：使用 uv 运行
            # 获取脚本的绝对路径
            path = Path(server_script_path).resolve()
            # 配置 stdio 服务器参数
            server_params = StdioServerParameters(
                command="uv",  # 使用 uv 包管理器
                args=["--directory", str(path.parent), "run", path.name],  # 在脚本目录下运行
                env=None,  # 继承当前环境变量
            )
        else:
            # JavaScript 服务器：直接使用 node 运行
            server_params = StdioServerParameters(
                command="node",  # 使用 Node.js
                args=[server_script_path],
                env=None,
            )

        # 启动服务器并建立 stdio 通信通道
        # enter_async_context 会自动管理资源的生命周期
        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()

        # 获取服务器提供的工具列表并打印
        response = await self.session.list_tools()
        tools = response.tools
        print("\n已连接到服务器，可用工具:", [tool.name for tool in tools])

    async def process_query(self, query: str) -> str:
        """
        处理用户查询

        流程：
        1. 将用户查询发送给 Claude
        2. 如果 Claude 决定调用工具，执行工具并返回结果
        3. 将工具执行结果返回给 Claude 继续处理
        4. 返回最终的响应文本

        Args:
            query: 用户的查询文本

        Returns:
            Claude 的最终响应文本
        """
        # 初始化消息列表
        messages = [{"role": "user", "content": query}]

        # 获取服务器上可用的工具列表
        response = await self.session.list_tools()
        available_tools = [
            {
                "name": tool.name,
                "description": tool.description,
                "input_schema": tool.inputSchema,  # 工具的输入参数模式
            }
            for tool in response.tools
        ]

        # 第一次调用 Claude API
        # 传入工具列表，让 Claude 可以决定是否需要调用工具
        response = self.anthropic.messages.create(
            model=ANTHROPIC_MODEL,
            max_tokens=1000,
            messages=messages,
            tools=available_tools,  # 告诉 Claude 可用的工具
        )

        # 存储最终响应的文本片段
        final_text = []

        # 处理 Claude 的响应内容
        for content in response.content:
            # 情况1：Claude 返回普通文本
            if content.type == "text":
                final_text.append(content.text)

            # 情况2：Claude 决定调用工具
            elif content.type == "tool_use":
                tool_name = content.name
                tool_args = content.input

                # 执行工具调用
                # 通过 MCP 协议将工具调用请求发送给服务器
                result = await self.session.call_tool(tool_name, tool_args)
                final_text.append(f"[Calling tool {tool_name} with args {tool_args}]")

                # 将工具调用结果添加到对话历史中
                # 先添加 Claude 的工具调用响应（如果有文本）
                if hasattr(content, "text") and content.text:
                    messages.append({"role": "assistant", "content": content.text})

                # 添加工具执行结果作为用户消息
                # MCP 返回的 result.content 包含工具的执行结果
                messages.append({"role": "user", "content": result.content})

                # 第二次调用 Claude API
                # 将工具执行结果发送给 Claude，让其生成最终响应
                response = self.anthropic.messages.create(
                    model=ANTHROPIC_MODEL,
                    max_tokens=1000,
                    messages=messages,
                )

                # 添加 Claude 的最终响应
                final_text.append(response.content[0].text)

        # 将所有响应片段合并返回
        return "\n".join(final_text)

    async def chat_loop(self):
        """
        运行交互式聊天循环

        用户可以在命令行中输入查询，程序会实时响应
        输入 'quit' 退出聊天
        """
        print("\nMCP 客户端已启动！")
        print("请输入查询内容，输入 'quit' 退出。")

        while True:
            try:
                # 获取用户输入
                query = input("\nQuery: ").strip()

                # 退出条件
                if query.lower() == "quit":
                    break

                # 处理查询并打印响应
                response = await self.process_query(query)
                print("\n" + response)

            except Exception as e:
                print(f"\n错误: {str(e)}")

    async def cleanup(self):
        """
        清理资源

        关闭所有打开的连接和上下文管理器
        """
        await self.exit_stack.aclose()


async def main():
    """
    主函数

    流程：
    1. 检查命令行参数（必须提供服务器脚本路径）
    2. 创建 MCP 客户端
    3. 连接到指定的服务器
    4. 检查 API 密钥
    5. 启动交互式聊天循环
    """
    # 检查是否提供了服务器脚本路径
    if len(sys.argv) < 2:
        print("使用方法: python client.py <服务器脚本路径>")
        sys.exit(1)

    client = MCPClient()
    try:
        # 连接到 MCP 服务器
        await client.connect_to_server(sys.argv[1])

        # 检查 Anthropic API 密钥是否存在
        api_key = os.getenv("ANTHROPIC_API_KEY")
        if not api_key:
            print("\n未找到 ANTHROPIC_API_KEY。要使用 Claude 查询这些工具，请设置您的 API 密钥：")
            print("  export ANTHROPIC_API_KEY=您的-api-密钥")
            return

        # 启动聊天循环
        await client.chat_loop()
    finally:
        # 确保资源被清理（即使发生异常）
        await client.cleanup()


if __name__ == "__main__":
    # 运行异步主函数
    asyncio.run(main())

六. 总结

MCP 采用模块化设计与多协议兼容的架构，专为复杂 AI 应用场景打造。这一架构从底层保障了系统的高可靠性、易扩展性和强适应性，使 MCP 成为构建企业级 A I应用的理想之选，能够从容应对多样化业务场景下的技术挑战。