1. 项目概述：从理论到实战，亲手解剖MCP协议的数据包

最近，Model Context Protocol (MCP) 在AI和LLM工具集成领域火得一塌糊涂。各种架构图、白皮书和概念解析文章满天飞，仿佛人人都成了协议专家。但作为一个在软件工程一线摸爬滚打多年的老手，我总觉得少了点什么。大家都在高谈阔论“双向通信”、“能力协商”、“上下文管理”这些宏大概念，却没人愿意掀开引擎盖，给你看看里面到底是怎么转的。这就像有人滔滔不绝地给你讲解汽车的空气动力学原理，却从不让你看一眼发动机的内部构造。对于工程师来说，这种隔靴搔痒的感觉实在难受。

所以，我决定干点“脏活累水”。这篇文章，我们不谈哲学，只看数据包。我们将彻底抛开那些精美的示意图和抽象的理论，直接深入到比特和字节的层面，亲手构建一个最原始的MCP客户端，并通过标准输入输出与一个真实的MCP服务器对话。我们的目标只有一个：亲眼看看在网络上流动的，究竟是怎样的JSON。通过这个“外科手术”式的实践，你不仅能透彻理解MCP的核心机制，更能获得一种“我能从头构建它”的底气和自信。无论你是想深度集成AI能力到现有系统，还是仅仅好奇协议底层如何工作，这篇从零开始的实战指南都将为你提供最直接的洞察。

2. 核心思路拆解：为什么选择“裸”协议分析？

在开始动手之前，我们先明确一下这次探索的独特视角和价值。市面上大多数MCP教程都基于官方SDK或高级框架，这当然能快速上手，但也像用自动挡学开车——你知道了怎么走，但不知道离合器、变速箱和发动机是如何协同工作的。

2.1 超越SDK：理解协议的“第一性原理”

使用官方SDK（比如JavaScript或Python的MCP库）无疑是最便捷的方式。它们封装了连接管理、消息序列化、错误处理等繁琐细节，让你可以专注于业务逻辑。然而，这种便利性也带来了一层抽象，屏蔽了协议最本质的通信模式。当出现网络问题、版本不兼容或需要深度定制时，这层抽象就可能变成黑箱，让你无从下手。

我们的方法截然不同：我们将仅使用Python标准库的 subprocess 和 json 模块，手动构造每一个JSON-RPC消息，并通过管道发送。这相当于我们直接在与协议的“语法”对话，跳过了所有“编译器”和“运行时”的中间层。这样做有几个不可替代的好处：

1. 绝对透明 ：每一个字节的发送和接收都完全在你的控制之下，没有任何魔法。你可以清晰地看到请求是如何构建的，响应是如何解析的。
2. 深度理解 ：通过手动实现最基本的握手和工具调用流程，你会对MCP的状态机、错误处理边界和设计哲学有肌肉记忆般的理解。
3. 调试利器 ：当你使用高级SDK遇到诡异问题时，拥有底层协议知识能让你快速定位问题是在应用层、SDK层还是协议层。你可以自己写一个小脚本，模拟客户端行为，直接测试服务器是否正常。
4. 不受限制 ：你不受任何特定SDK版本、设计模式或依赖项的束缚。你可以用任何语言、在任何环境中实现MCP客户端，只要它能启动子进程、读写JSON和管道。

2.2 聚焦核心：工具调用是MCP的“杀手锏”

MCP协议规范涵盖了不少内容：初始化、资源（Resources）、提示词（Prompts）、工具（Tools）等。但对于绝大多数实际应用场景——尤其是让LLM能够调用外部功能—— 工具调用（Tool Calling） 是绝对的核心和起点。

我们可以暂时忽略资源管理和提示词模板这些高级特性，将注意力完全集中在工具调用的生命周期上。这样一来，整个复杂的协议就被简化成了一个极其清晰的对话模型：

1. 客户端 ：“你好，我支持MCP协议，版本是X，这是我的基本信息。”（ initialize ）
2. 服务器 ：“你好，我是XX服务器，版本是Y，我支持这些能力。”（ initialized 响应）
3. 客户端 ：“好的，我准备好了，开始吧。”（ notifications/initialized ）
4. 客户端 ：“你都能做什么？”（ tools/list ）
5. 服务器 ：“我能做A、B、C这几件事。”（ tools/list 响应）
6. 客户端 ：“请帮我做A，这是参数。”（ tools/call ）
7. 服务器 ：“做完了，这是结果。”（ tools/call 响应）

看，只需要理解这 七种消息 的交换，你就掌握了MCP 80%以上的实用价值。我们的实验也将严格遵循这个精简流程。

2.3 实验环境与目标服务器

为了进行这次解剖，我们需要一个“标本”——一个简单、稳定、开源的MCP服务器。官方提供的 weather.py 示例服务器是绝佳的选择。它提供了两个工具： get_forecast （获取天气预报）和 get_alerts （获取天气警报）。功能具体，代码清晰，且完全符合MCP规范。

我们的实验目标非常明确：

1. 手动启动这个天气服务器进程。
2. 不使用任何MCP库，仅用标准库与其建立通信。
3. 完整走通上述7步对话，并捕获每一个环节在“线上”（即stdin/stdout管道）传输的原始JSON数据。
4. 分析这些数据包的结构、字段含义和设计意图。

注意 ：选择 stdin/stdout 作为传输层是MCP的常见实践，尤其对于本地集成的场景。它简单、高效，且避免了网络端口、认证等复杂性。虽然有人诟病其健壮性，但对于进程间通信（IPC）和快速原型来说，它是无可替代的利器。我们的实验正是基于这种模式。

3. 实战准备：搭建你的数字解剖台

理论铺垫完毕，现在进入实战环节。请确保你有一个可用的Python环境（3.7以上），我们将从零开始搭建实验环境。

3.1 获取并安装天气服务器

首先，我们需要获取MCP天气服务器的代码。官方示例存放在GitHub上。

# 创建一个专门的工作目录
mkdir mcp-wire-dissection && cd mcp-wire-dissection

# 下载天气服务器示例代码
curl -O https://raw.githubusercontent.com/modelcontextprotocol/quickstart-resources/main/weather-server-python/weather.py

# 查看文件，确认下载成功
head -20 weather.py

这个 weather.py 文件就是一个完整的MCP服务器实现。它使用了 mcp 这个Python SDK。因此，我们需要安装其依赖。推荐使用现代Python包管理工具 uv ，它比传统的 pip 更快、更可靠。

# 安装 uv (如果尚未安装)
# 在MacOS/Linux上：
curl -LsSf https://astral.sh/uv/install.sh | sh

# 初始化项目并安装依赖
uv init .
uv add mcp

实操心得 ：使用 uv 而不是 pip 和 venv 能大幅提升依赖管理效率。 uv 不仅安装极快，还能生成精确的锁文件，确保环境一致性。对于这类一次性实验项目，它能帮你省去很多配置虚拟环境的麻烦。

3.2 理解服务器启动方式

查看 weather.py 的末尾，通常你会看到类似这样的代码：

if __name__ == "__main__":
    # 使用 mcp 库的 `run` 函数启动服务器
    mcp.run(server)

这意味着我们可以直接用 python weather.py 来启动它。但更常见的做法是使用 uv run ，它能确保在正确的、隔离的依赖环境中运行脚本。这也是我们后续在客户端代码中启动子进程的方式。

你可以先手动测试一下服务器是否能正常运行：

uv run weather.py

如果一切正常，服务器会启动并等待标准输入上的连接。此时它不会输出任何内容（因为还没有客户端连接），你可以按 Ctrl+C 终止它。这个“安静”的启动是正常的，MCP服务器通常以这种“沉默的守护进程”模式运行。

4. 核心环节一：建立连接与初始化握手

现在，让我们开始编写我们的“外科医生”——一个纯手工的MCP客户端。我们将创建一个新的Python脚本 bare_mcp_client.py 。

4.1 启动服务器子进程

我们使用 subprocess.Popen 来启动服务器，并捕获其标准输入、输出和错误流。这是实现进程间通信的基础。

# bare_mcp_client.py
import subprocess
import json
from pprint import pprint
import time

def start_mcp_server(server_script_path):
    """
    启动MCP服务器作为子进程。
    返回一个Popen对象，通过其stdin/stdout进行通信。
    """
    # 关键参数解析：
    # - stdin=subprocess.PIPE: 我们可以向进程的stdin写入数据（我们的请求）
    # - stdout=subprocess.PIPE: 我们可以从进程的stdout读取数据（服务器的响应）
    # - stderr=subprocess.DEVNULL: 将服务器的错误输出重定向到空设备，避免干扰我们的主控制台。
    #    在实际调试时，你可以改为 stderr=subprocess.PIPE 来捕获错误日志。
    # - `["uv", "run", ...]`: 使用uv在正确的环境中运行脚本。
    process = subprocess.Popen(
        ["uv", "run", server_script_path],
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.DEVNULL,  # 或 subprocess.PIPE 用于调试
        text=False,  # 我们处理字节，而不是字符串，以获得更精确的控制
        bufsize=1,   # 行缓冲，确保每条消息能及时发送
    )
    print(f"[INFO] 服务器进程已启动，PID: {process.pid}")
    return process

注意事项 ：将 stderr 重定向到 DEVNULL 在实验阶段是干净的，但生产环境或调试时，一定要捕获并处理 stderr 。服务器可能会将重要的启动错误或运行时日志输出到标准错误流。

4.2 定义消息收发函数

MCP over stdio 的约定是 newline-delimited JSON (NDJSON) 。即每条完整的JSON消息独占一行，以换行符 \n 分隔。这是许多流式JSON协议（如JSON-RPC over HTTP/1.1的某些实现）的常见模式。

def send_mcp_message(process, message_dict):
    """
    向MCP服务器进程发送一条JSON-RPC消息。
    消息被序列化为JSON字符串，末尾添加换行符，然后编码为字节流写入stdin。
    """
    # 1. 将字典转换为JSON字符串
    json_str = json.dumps(message_dict)
    # 2. 添加NDJSON要求的换行符
    json_line = json_str + '\n'
    # 3. 编码为字节并写入进程的stdin
    process.stdin.write(json_line.encode('utf-8'))
    # 4. 刷新缓冲区，确保数据立即发送
    process.stdin.flush()
    print(f"[SENT] {json_str[:200]}...")  # 打印前200个字符便于观察

def read_mcp_message(process):
    """
    从MCP服务器进程的stdout读取一条JSON-RPC消息。
    读取一行，解码为字符串，然后解析为Python字典。
    这是一个阻塞调用，会等待服务器输出新的一行。
    """
    # 读取一行字节数据
    line_bytes = process.stdout.readline()
    if not line_bytes:
        # 如果读到空字节，可能意味着管道已关闭或进程结束
        return None
    # 解码为UTF-8字符串，并去除末尾的换行符
    line_str = line_bytes.decode('utf-8').rstrip('\n')
    if line_str:
        print(f"[RECV] 原始行长度: {len(line_str)} 字符")
        # 解析JSON
        message = json.loads(line_str)
        return message
    return None

4.3 执行初始化握手

现在，让我们发起第一次对话。根据MCP规范，连接建立后的第一条消息必须是 initialize 请求。

def perform_handshake(process, client_name="BareMCPClient", client_version="0.1.0"):
    """
    执行MCP协议初始化握手。
    返回服务器返回的初始化结果。
    """
    # 步骤1: 客户端发送 initialize 请求
    init_request = {
        "jsonrpc": "2.0",
        "id": 1,  # 请求ID，用于匹配响应
        "method": "initialize",
        "params": {
            "protocolVersion": "2025-03-26",  # 必须与服务器兼容的协议版本
            "capabilities": {},  # 客户端声明自己支持的能力（这里为空）
            "clientInfo": {
                "name": client_name,
                "version": client_version
            }
        }
    }
    print("\n=== 步骤 1: 发送 initialize 请求 ===")
    send_mcp_message(process, init_request)

    # 步骤2: 读取服务器的 initialize 响应
    print("\n=== 等待服务器 initialize 响应 ===")
    init_response = read_mcp_message(process)
    if not init_response:
        raise ConnectionError("服务器未响应 initialize 请求")
    print("服务器初始化响应:")
    pprint(init_response, depth=3)  # 限制打印深度，避免信息过载

    # 步骤3: 客户端发送 initialized 通知
    # 这是一个“通知”（notification），没有id字段，服务器不回复。
    initialized_notification = {
        "jsonrpc": "2.0",
        "method": "notifications/initialized",
        "params": {}  # 参数为空对象
    }
    print("\n=== 步骤 2: 发送 initialized 通知 ===")
    send_mcp_message(process, initialized_notification)
    # 发送通知后，没有响应需要读取

    return init_response

让我们逐行分析这个握手过程：

1. initialize 请求 ：

   • jsonrpc : "2.0": 声明我们使用JSON-RPC 2.0协议。这是MCP的底层传输协议。
   • id : 1: 每个请求需要一个唯一ID，服务器会在对应的响应中带回相同的ID。这对于异步通信至关重要。
   • method : "initialize": 指定要调用的方法。
   • params : 包含握手参数。
     • protocolVersion : 指定我们希望使用的MCP协议版本。 必须与服务器支持的版本匹配 ，否则握手会失败。
     • capabilities : 客户端向服务器宣告自己支持哪些MCP扩展功能（如资源订阅、提示词变更通知等）。我们这里留空，表示只支持基础功能。
     • clientInfo : 客户端的身份信息，方便服务器日志记录和识别。

2. 服务器 initialize 响应 ：

   • 服务器会返回一个同样包含 id : 1 的JSON-RPC响应。
   • result 字段中包含服务器的信息，如 serverInfo （名称、版本）和 capabilities （服务器支持的能力）。这是我们第一次了解服务器特性的机会。

3. notifications/initialized 通知 ：

   • 这是一个 通知 ，而不是请求。JSON-RPC中，通知的 id 字段为 null 或直接省略。它用于告知服务器“客户端已准备就绪，可以开始正式工作”。
   • 服务器收到后不会发送任何响应。这完成了握手的最后一步。

运行这部分代码，你会看到类似以下的输出（具体内容因服务器版本而异）：

[INFO] 服务器进程已启动，PID: 12345

=== 步骤 1: 发送 initialize 请求 ===
[SENT] {"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": "2025-03-26", "capabilities": {}, "clientInfo": {"name": "BareMCPClient", "version": "0.1.0"}}}...

=== 等待服务器 initialize 响应 ===
[RECV] 原始行长度: 266 字符
服务器初始化响应:
{'id': 1,
 'jsonrpc': '2.0',
 'result': {'capabilities': {'experimental': {},
                             'prompts': {'listChanged': False},
                             'resources': {'listChanged': False,
                                           'subscribe': False},
                             'tools': {'listChanged': False}},
           'protocolVersion': '2025-03-26',
           'serverInfo': {'name': 'weather', 'version': '1.9.4'}}}

=== 步骤 2: 发送 initialized 通知 ===
[SENT] {"jsonrpc": "2.0", "method": "notifications/initialized", "params": {}}...

恭喜！ 你已经成功完成了MCP连接握手。你看到了服务器返回的原始JSON，知道了它的名字、版本，以及它声明不支持动态工具列表变更 ( "listChanged": False )。这就是“线上”流动的第一个关键数据包。

5. 核心环节二：发现与调用工具

握手成功后，客户端和服务器就进入了工作状态。接下来，客户端需要知道服务器能做什么。

5.1 查询可用工具列表

我们发送 tools/list 请求来获取服务器暴露的所有工具。

def list_available_tools(process):
    """
    向服务器请求可用的工具列表。
    """
    list_tools_request = {
        "jsonrpc": "2.0",
        "id": 2,  # 新的请求ID
        "method": "tools/list",
        "params": {}  # 此请求通常不需要额外参数
    }
    print("\n=== 步骤 3: 查询工具列表 (tools/list) ===")
    send_mcp_message(process, list_tools_request)

    print("\n=== 等待工具列表响应 ===")
    list_response = read_mcp_message(process)
    if not list_response:
        raise ConnectionError("服务器未响应 tools/list 请求")

    # 检查是否有错误
    if "error" in list_response:
        print(f"查询工具列表失败: {list_response['error']}")
        return None

    tools = list_response.get("result", {}).get("tools", [])
    print(f"发现 {len(tools)} 个工具:")
    for tool in tools:
        print(f"  - 名称: {tool['name']}")
        print(f"    描述: {tool.get('description', 'N/A')[:100]}...")  # 截断长描述
        # 打印输入参数模式
        schema = tool.get('inputSchema', {})
        req = schema.get('required', [])
        props = schema.get('properties', {})
        if req or props:
            print(f"    输入参数: {list(props.keys())} (必填: {req})")
    print()
    return tools

运行后，你会看到服务器返回的详细工具列表。以天气服务器为例，输出可能如下：

=== 步骤 3: 查询工具列表 (tools/list) ===
[SENT] {"jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {}}...

=== 等待工具列表响应 ===
[RECV] 原始行长度: 732 字符
发现 2 个工具:
  - 名称: get_alerts
    描述: Get weather alerts for a US state. Args: state: Two-letter US state code (e.g. CA, NY)...
    输入参数: ['state'] (必填: ['state'])
  - 名称: get_forecast
    描述: Get weather forecast for a location. Args: latitude: Latitude of the location longitude:...
    输入参数: ['latitude', 'longitude'] (必填: ['latitude', 'longitude'])

关键字段解析 ：

• name : 工具的标识符，在调用时必须精确匹配。
• description : 给LLM看的自然语言描述。 这是工具能否被正确调用的关键 。LLM根据这段描述来决定是否以及如何使用这个工具。
• inputSchema : 一个遵循JSON Schema的对象，严格定义了调用此工具所需的参数名称、类型和是否必需。这是实现强类型接口和自动验证的基础。

实操心得 ： description 字段的撰写是一门艺术。它需要足够清晰，让LLM能准确理解工具的用途；又不能过于冗长，以免占用过多上下文窗口。一个好的描述应包含：1) 工具的核心功能；2) 每个参数的含义和示例；3) 可能返回的内容。这是你作为服务器开发者与LLM“沟通”的主要渠道。

5.2 构造并发送工具调用请求

现在，我们知道了服务器有 get_alerts 工具，它需要一个 state 参数（美国州代码）。让我们调用它来获取德克萨斯州（TX）的天气警报。

def call_mcp_tool(process, tool_name, arguments, request_id=3):
    """
    调用指定的MCP工具。
    """
    call_request = {
        "jsonrpc": "2.0",
        "id": request_id,
        "method": "tools/call",
        "params": {
            "name": tool_name,  # 必须与 tools/list 返回的 name 完全一致
            "arguments": arguments  # 参数必须符合 inputSchema 的定义
        }
    }
    print(f"\n=== 步骤 4: 调用工具 '{tool_name}' ===")
    print(f"    参数: {arguments}")
    send_mcp_message(process, call_request)

    print("\n=== 等待工具调用响应 ===")
    call_response = read_mcp_message(process)
    if not call_response:
        raise ConnectionError(f"服务器未响应 tools/call 请求 (id: {request_id})")

    # 检查响应中是否包含错误
    if "error" in call_response:
        error = call_response["error"]
        print(f"[ERROR] 工具调用失败!")
        print(f"        错误码: {error.get('code')}")
        print(f"        消息: {error.get('message')}")
        if "data" in error:
            print(f"        详情: {error.get('data')}")
        return None

    # 解析成功的结果
    result = call_response.get("result", {})
    is_error = result.get("isError", False)
    content = result.get("content", [])

    print(f"调用结果: isError={is_error}")
    print(f"返回了 {len(content)} 个内容块")

    for i, block in enumerate(content):
        block_type = block.get("type", "unknown")
        print(f"\n--- 内容块 {i+1} (类型: {block_type}) ---")
        if block_type == "text":
            text = block.get("text", "")
            # 对于长文本，只预览开头和结尾
            if len(text) > 500:
                print(text[:250] + "...\n...\n" + text[-250:])
            else:
                print(text)
        elif block_type == "image":
            # 如果是图片，可能包含数据URI或引用
            print(f"图片数据 (已省略): {block.get('data', 'N/A')[:100]}...")
        # 可以处理其他类型，如 audio, video, resource 等
    return result

现在，在主函数中调用它：

# 在主流程中，握手和获取工具列表之后...
tools = list_available_tools(process)
if tools:
    # 假设我们调用 get_alerts
    tool_to_call = "get_alerts"
    # 确保工具存在
    if any(t['name'] == tool_to_call for t in tools):
        # 准备参数
        tool_args = {"state": "TX"}  # 获取德克萨斯州的警报
        result = call_mcp_tool(process, tool_to_call, tool_args, request_id=3)
        if result:
            print("\n✅ 工具调用成功！")
        else:
            print("\n❌ 工具调用失败或返回错误。")
    else:
        print(f"\n工具 '{tool_to_call}' 不在可用列表中。")

运行这段代码，你将见证奇迹（或者说，一次标准的API调用）。客户端发送一个紧凑的请求，服务器返回可能非常庞大的天气警报数据。

=== 步骤 4: 调用工具 'get_alerts' ===
    参数: {'state': 'TX'}
[SENT] {"jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": {"name": "get_alerts", "arguments": {"state": "TX"}}}...

=== 等待工具调用响应 ===
[RECV] 原始行长度: 51305 字符
调用结果: isError=False
返回了 1 个内容块

--- 内容块 1 (类型: text) ---
Event: Flood Advisory
Area: Hidalgo, TX; Willacy, TX
Severity: Minor
Description: * WHAT...Flooding caused by excessive rainfall continues.
...
Instructions: Turn around, don't drown when encountering flooded roads...
The next statement will be issued Tuesday morning at 830 AM CDT.

✅ 工具调用成功！

响应结构深度解析 ： 你收到的响应是一个标准的JSON-RPC成功响应。核心在于 result 字段：

• isError : false 表示调用成功。如果是 true ，则 content 中可能包含错误信息。
• content : 一个数组，包含一个或多个“内容块”。每个块有 type 和具体数据。
  • type: "text" : 这是最常见的类型，数据在 text 字段中，是一个字符串。 注意 ：这个字符串内部可以包含任何结构化的文本，比如我们看到的格式化天气警报。LLM擅长从这种半结构化的自然语言文本中提取信息。
  • 其他类型：MCP规范还支持 image 、 audio 、 video 等类型，用于多模态输出。 resource 类型则用于引用服务器管理的资源（如文件）。

关键洞察 ：MCP工具调用的响应设计非常通用。 content 数组允许服务器返回多种类型、多个部分的结果。 text 类型的灵活性极高，它可以是纯文本、JSON字符串、CSV、HTML，甚至是自定义的标记语言。这种设计将“数据呈现”的复杂性留给了服务器和客户端（或LLM），协议本身只负责传输。这也是为什么你很少在工具定义中看到 outputSchema ——输出格式是隐含在 content 的类型和结构中的。

6. 核心环节三：完整流程整合与错误处理

我们已经拆解了每一个步骤。现在，让我们把它们组合成一个完整、健壮的客户端脚本，并加入必要的错误处理和资源清理。

6.1 完整的客户端脚本

# bare_mcp_client.py (完整版)
import subprocess
import json
import sys
from pprint import pprint

class BareMCPClient:
    def __init__(self, server_script_path):
        self.server_path = server_script_path
        self.process = None
        self.next_request_id = 1  # 用于生成唯一的请求ID

    def start(self):
        """启动MCP服务器进程"""
        print(f"[启动] 正在启动MCP服务器: {self.server_path}")
        try:
            self.process = subprocess.Popen(
                ["uv", "run", self.server_path],
                stdin=subprocess.PIPE,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,  # 改为PIPE以捕获错误
                text=False,
                bufsize=1,
            )
            # 启动后，先读一下stderr，看是否有立即报错
            self._drain_stderr()
            print(f"[启动] 服务器进程已启动 (PID: {self.process.pid})")
            return True
        except FileNotFoundError as e:
            print(f"[错误] 无法找到服务器脚本或uv命令: {e}")
            print(f"      请确保: 1) 脚本路径正确; 2) uv已安装; 3) 依赖已安装 (uv add mcp)")
            return False
        except Exception as e:
            print(f"[错误] 启动服务器时发生未知错误: {e}")
            return False

    def _drain_stderr(self):
        """非阻塞地读取并打印服务器的错误输出"""
        import select
        if self.process and self.process.stderr:
            # 检查stderr是否有内容可读（非阻塞）
            rlist, _, _ = select.select([self.process.stderr], [], [], 0.1)
            if rlist:
                err_output = self.process.stderr.read().decode('utf-8', errors='ignore')
                if err_output:
                    print(f"[服务器STDERR] {err_output.strip()}")

    def send(self, method, params=None, is_notification=False):
        """发送JSON-RPC请求或通知"""
        if not self.process or self.process.poll() is not None:
            raise ConnectionError("服务器进程未运行或已终止")

        request = {
            "jsonrpc": "2.0",
            "method": method,
        }

        if not is_notification:
            request["id"] = self.next_request_id
            self.next_request_id += 1

        if params:
            request["params"] = params

        json_line = json.dumps(request) + '\n'
        try:
            self.process.stdin.write(json_line.encode('utf-8'))
            self.process.stdin.flush()
            print(f"[发送] {method} (ID: {request.get('id', 'N/A-通知')})")
            # 打印精简的请求内容
            if params:
                preview = str(params)[:100]
                if len(str(params)) > 100:
                    preview += "..."
                print(f"      参数: {preview}")
            return request.get("id")  # 返回请求ID，用于匹配响应
        except BrokenPipeError:
            print(f"[错误] 向服务器写入失败，管道已断开")
            self.stop()
            raise ConnectionError("与服务器的连接已断开")

    def receive(self):
        """从服务器接收一条JSON-RPC消息"""
        if not self.process:
            return None

        # 先检查一下stderr，看是否有错误输出
        self._drain_stderr()

        try:
            line_bytes = self.process.stdout.readline()
            if not line_bytes:
                print("[接收] 读到空字节，服务器可能已关闭输出流")
                return None
            line_str = line_bytes.decode('utf-8', errors='replace').rstrip('\n')
            if not line_str:
                return None
            print(f"[接收] 原始数据长度: {len(line_str)} 字符")
            return json.loads(line_str)
        except json.JSONDecodeError as e:
            print(f"[错误] 解析服务器响应JSON失败: {e}")
            print(f"      原始行: {line_str[:200]}...")
            return None
        except Exception as e:
            print(f"[错误] 读取服务器响应时发生未知错误: {e}")
            return None

    def call_and_wait(self, method, params=None):
        """发送请求并等待响应，返回响应字典"""
        req_id = self.send(method, params, is_notification=False)
        if req_id is None:
            return None

        # 循环读取，直到找到对应ID的响应，或遇到错误/通知
        while True:
            response = self.receive()
            if response is None:
                print(f"[错误] 在等待响应 {req_id} 时连接断开")
                return None

            # 如果是通知（无id），打印并继续等
            if "id" not in response:
                print(f"[接收] 收到通知: {response.get('method', 'unknown')}")
                continue

            # 找到对应ID的响应
            if response.get("id") == req_id:
                return response

            # ID不匹配，可能是之前的响应未处理或服务器乱序（MCP通常不会）
            print(f"[警告] 收到不匹配的响应ID: 期望 {req_id}, 收到 {response.get('id')}")

    def stop(self):
        """停止服务器进程"""
        if self.process:
            print("[停止] 正在终止服务器进程...")
            try:
                self.process.terminate()  # 发送SIGTERM
                self.process.wait(timeout=5)  # 等待进程结束
                print("[停止] 服务器进程已终止")
            except subprocess.TimeoutExpired:
                print("[警告] 进程未正常终止，强制结束")
                self.process.kill()  # 发送SIGKILL
                self.process.wait()
            finally:
                self.process = None

def main():
    client = BareMCPClient("weather.py")

    if not client.start():
        sys.exit(1)

    try:
        # 1. 初始化握手
        print("\n" + "="*50)
        print("阶段 1: 初始化握手")
        print("="*50)
        init_response = client.call_and_wait("initialize", {
            "protocolVersion": "2025-03-26",
            "capabilities": {},
            "clientInfo": {"name": "解剖客户端", "version": "1.0"}
        })
        if not init_response or "error" in init_response:
            print("初始化失败，退出。")
            return
        print("初始化成功。服务器信息:")
        pprint(init_response.get("result", {}).get("serverInfo"))

        # 2. 发送 initialized 通知
        print("\n" + "="*50)
        print("阶段 2: 发送就绪通知")
        print("="*50)
        client.send("notifications/initialized", {}, is_notification=True)
        print("已通知服务器客户端准备就绪。")

        # 3. 列出工具
        print("\n" + "="*50)
        print("阶段 3: 查询可用工具")
        print("="*50)
        list_response = client.call_and_wait("tools/list", {})
        if not list_response or "error" in list_response:
            print("获取工具列表失败，退出。")
            return

        tools = list_response.get("result", {}).get("tools", [])
        print(f"发现 {len(tools)} 个工具:")
        for t in tools:
            print(f"  - {t['name']}: {t.get('description', '无描述')[:80]}...")

        if not tools:
            print("没有可用工具，退出。")
            return

        # 4. 调用一个工具 (例如 get_alerts)
        print("\n" + "="*50)
        print("阶段 4: 调用工具")
        print("="*50)
        # 选择第一个工具，或按名称选择
        selected_tool = tools[0]  # 例如 get_alerts
        tool_name = selected_tool["name"]

        # 根据工具定义构造参数（这里硬编码，实际应由用户或逻辑决定）
        if tool_name == "get_alerts":
            arguments = {"state": "CA"}  # 加州
        elif tool_name == "get_forecast":
            arguments = {"latitude": 37.7749, "longitude": -122.4194}  # 旧金山
        else:
            print(f"未知工具 {tool_name}，使用空参数测试。")
            arguments = {}

        print(f"调用工具: {tool_name}，参数: {arguments}")
        call_response = client.call_and_wait("tools/call", {
            "name": tool_name,
            "arguments": arguments
        })

        if not call_response:
            print("工具调用无响应。")
        elif "error" in call_response:
            print(f"工具调用返回错误: {call_response['error']}")
        else:
            result = call_response.get("result", {})
            print(f"调用成功! isError: {result.get('isError')}")
            content = result.get("content", [])
            for i, block in enumerate(content):
                print(f"\n--- 内容块 {i+1}/{len(content)} ---")
                if block.get("type") == "text":
                    text = block.get("text", "")
                    # 显示前500字符作为预览
                    preview = text[:500]
                    if len(text) > 500:
                        preview += "..."
                    print(preview)
                else:
                    print(f"类型: {block.get('type')}, 数据: {str(block)[:200]}...")

        print("\n" + "="*50)
        print("所有操作完成。")
        print("="*50)

    except KeyboardInterrupt:
        print("\n[用户中断]")
    except Exception as e:
        print(f"\n[未处理的异常] {e}")
        import traceback
        traceback.print_exc()
    finally:
        client.stop()

if __name__ == "__main__":
    main()

这个完整的客户端类封装了连接、通信、错误处理和资源管理的所有细节。它更加健壮，包含了超时处理、错误流监控和更清晰的日志。

6.2 关键错误处理与边界情况

在实际操作中，你会遇到各种问题。以下是一些常见场景及处理思路：

1. 服务器启动失败 ：

   • 表现 ： Popen 抛出异常（如 FileNotFoundError ）或进程立即退出。
   • 排查 ：检查脚本路径、 uv 命令是否在PATH中、Python环境是否正确、依赖 ( mcp ) 是否已安装。我们的 _drain_stderr 方法会捕获并打印服务器的启动错误。

2. 初始化握手失败 ：

   • 表现 ：服务器对 initialize 请求返回一个包含 error 字段的JSON-RPC响应。
   • 常见原因 ： protocolVersion 不兼容、 capabilities 格式错误。
   • 处理 ：解析 error 对象，根据 code 和 message 调整客户端请求。

3. 工具调用参数错误 ：

   • 表现 ： tools/call 响应中 isError: true 或直接返回JSON-RPC错误。
   • 常见原因 ：参数缺失、参数类型不匹配、参数值不符合约束（如州代码不是两个字母）。
   • 处理 ：严格遵循 tools/list 返回的 inputSchema 构建参数。实现参数验证逻辑。

4. 服务器无响应或超时 ：

   • 表现 ： readline() 阻塞或长时间无返回。
   • 处理 ：在生产环境中，需要为 readline() 设置超时（例如使用 select 或线程）。我们的简单示例是阻塞的，对于可靠系统需要改进。

5. 协议版本升级 ：

   • 注意 ：MCP协议仍在快速发展中。 protocolVersion 字段是关键。如果未来版本不兼容，你需要根据服务器响应的版本或错误信息来适配客户端。

避坑技巧 ：在开发调试阶段， 不要 将 stderr 重定向到 DEVNULL 。服务器的错误日志是诊断问题的第一手资料。许多奇怪的连接问题，比如导入错误、权限问题，都会打印到标准错误流。

7. 协议深度解析与扩展思考

通过亲手实现这个“裸”客户端，我们已经穿透了抽象层，直接触摸到了MCP协议的筋骨。基于此，我们可以进行更深入的思考。

7.1 MCP协议设计的精妙之处

1. 基于JSON-RPC 2.0 ：这是一个非常明智的选择。JSON-RPC是一个简单、成熟、语言无关的RPC协议。它定义了请求、响应、通知和错误的标准格式，MCP直接继承，省去了重新设计消息格式的麻烦，并获得了广泛的客户端库支持。
2. 传输层无关性 ：协议规范只定义了消息的语义（ initialize , tools/call 等），而不规定传输方式（stdio, WebSocket, HTTP等）。这使得MCP能灵活适应从本地脚本插件到云端微服务的各种场景。我们使用的 stdio 只是其中一种实现。
3. 能力协商机制 ： initialize 握手过程中的 capabilities 字段为未来扩展留下了空间。服务器和客户端可以声明自己支持哪些可选功能（如资源订阅、工具动态更新），从而实现优雅的降级和功能发现。
4. 工具描述的枢纽作用 ： tools/list 返回的 description 和 inputSchema 是连接LLM世界与代码世界的桥梁。 description 让LLM理解工具功能， inputSchema 确保了调用的类型安全。这种设计分离了“意图理解”和“接口契约”。

7.2 从“裸”客户端到生产级实现的差距

我们的实验客户端是教学和理解的绝佳工具，但距离生产环境还有距离：

1. 连接管理与心跳 ：生产客户端需要处理连接断开重连、心跳保活、服务器崩溃重启等场景。
2. 异步与并发 ：一个客户端可能同时发起多个工具调用，或者需要处理服务器主动推送的通知（如 tools/listChanged ）。这需要异步IO和更复杂的消息分发机制。
3. 安全性 ：通过 stdio 通信相对安全（本地进程），但如果扩展到网络传输（如WebSocket），则需要考虑认证、授权和加密（TLS）。
4. 更完整的协议实现 ：我们只实现了工具调用部分。完整的MCP客户端还需要处理资源（ resources/* ）和提示词（ prompts/* ）相关的消息。
5. 健壮的错误处理与重试 ：网络波动、服务器临时过载、参数无效等都需要有系统的重试和降级策略。

7.3 下一步探索方向

理解了底层协议，你可以：

• 实现其他语言的客户端 ：用Go、Rust、Java甚至Shell脚本实现一个MCP客户端，原理完全相同。
• 调试复杂的MCP集成 ：当使用Claude Desktop、Cursor等集成MCP的工具出现问题时，你可以编写一个最小化的测试客户端，直接与问题服务器对话，精准定位是服务器错误、协议不匹配还是SDK问题。
• 开发自定义MCP服务器 ：现在你清楚地知道一个MCP服务器需要监听什么消息、返回什么格式。你可以为你的内部系统、数据库或独特API编写MCP适配器，让LLM直接调用。
• 分析协议流量 ：在 stdio 通信中插入一个中间人代理，记录和分析所有往来消息，用于性能分析、审计或调试。

8. 总结与最终建议

这次深入线缆（wire）之下的探险应该已经驱散了MCP周围的大部分迷雾。它不是一个神秘莫测的“AI黑魔法”，而是一个设计精良、基于成熟标准的RPC协议，其核心目标是为LLM提供一个标准化、可扩展的“手和脚”。

我的核心体会是 ：理解任何协议或框架最有效的方式，就是 亲手实现一次最简版本 。当你自己处理了JSON的序列化、管道的读写、状态机的维护，那些抽象的概念瞬间就变得具体而清晰。你不再害怕文档中未提及的细节，因为你知道数据包就在那里，你可以看到它、修改它、创造它。

对于想要在项目中应用MCP的开发者，我的建议是：

1. 从官方SDK开始 ：对于大多数应用，直接使用 @modelcontextprotocol/sdk (JavaScript/TypeScript) 或 mcp (Python) 等官方SDK是最快、最稳的选择。
2. 但保留底层知识 ：将本文的实践作为你的“逃生舱”知识。当SDK行为异常或你需要实现一些非常规功能时，这份底层理解能帮你快速找到方向。
3. 关注工具描述的质量 ：花时间打磨你的 tools/list 返回的 description 字段。清晰、准确、包含示例的描述，能极大提升LLM调用工具的准确率。
4. 循序渐进 ：先从简单的工具调用开始，成功后再逐步尝试资源、提示词等高级特性。MCP的模块化设计允许你按需采纳功能。

最后，记住那句老话： “一切皆文件描述符” 。在Unix哲学里，进程间通信不过是对文件描述符的读写。MCP over stdio 正是这一哲学的优雅体现——将复杂的AI能力集成，简化成了对标准输入输出流的JSON格式化读写。当你掌握了这一点，你就掌握了连接智能与代码世界最基础的钥匙。