MCP架构：应用、客户端与服务端怎样配合

更新 2026-10-05

正在读取学习记录…

1. 先分清三个角色

MCP（Model Context Protocol）约定应用与外部工具和资料服务怎样交换消息。Host是用户使用的应用，管理会话、模型和权限；Client是应用里与某个服务通信的组件；Server提供工具、资源和模板。

例如编辑器是Host，它为笔记服务建立一个Client连接，笔记服务是Server。模型提出查笔记的动作，应用检查后由Client发送请求，Server查询并返回，再由应用交给模型组织答复。一个Host可以连接多个Server，各条连接分别处理。

Host内有Client，Client与Server交换消息，Server查询业务资料 — 从用户任务出发：应用选择动作，客户端发送消息，服务端返回资料。

2. 初始化怎样开始

按MCP2025-11-25版本：Client发initialize，Server返回版本与能力，Client发notifications/initialized，随后发送常规请求。生命周期规范给出这套顺序。

发出方	消息	作用
Client	initialize，编号1	协商版本和能力
Server	初始化响应，编号1	返回协商结果
Client	notifications/initialized	告知准备完成
Client	tools/list，编号2	询问工具目录
Server	目录响应，编号2	返回工具列表

请求与响应使用相同编号；通知没有编号，也不等待配对响应。新建trace_lab.py，用事件列表观察顺序。

每个 Event 对象保存一条消息的发送方 actor、接收方 target、动作名称 method 和请求编号 request_id。例如 Event("client", "server", "initialize", 1) 表示客户端发送编号1的初始化请求。

这里的 Event 是程序用来记录交互步骤的对象。initialize_result 和 tools/list_result 给两种响应起了便于区分的记录名称；真实JSON-RPC响应用 id 对应请求，把返回内容放在 result 字段。读执行记录时，把这些名称理解为“初始化的响应”和“工具目录的响应”。

字典通过 event["actor"] 读取字段；这里的对象通过 event.actor 读取属性。@dataclass 帮我们生成创建对象的方法，统一列出这四个属性；frozen=True 保持创建后的字段稳定。Optional[int] 表示编号可以是整数，也可以是 None（没有编号）。这份事件记录用于检查通信顺序，实际传输消息在MCP传输中展示。

先在 PowerShell 创建本篇的独立目录并进入；路径可换成你自己的非系统盘位置：

mkdir -Force "F:\学习\agentdepth-l2-mcp-architecture"
cd "F:\学习\agentdepth-l2-mcp-architecture"

将本篇的 trace_lab.py、test_trace_lab.py 保存到这个目录。保存全部文件后运行 dir *.py，核对上述文件名，再执行下文命令；新开窗口时先回到该目录，并按 Python 环境篇核对 python 的解释器路径。

"""沿客户端与服务端事件检查初始化顺序。"""

from dataclasses import dataclass
from typing import List, Optional, Dict


@dataclass(frozen=True)
class Event:
    actor: str
    target: str
    method: str
    request_id: Optional[int] = None


# 跟踪初始化和未完成请求，将响应与请求编号配对。
def check_trace(events: List[Event]) -> List[str]:
    """检查单客户端、单服务端的简化记录，返回违反交互规则的步骤。"""
    errors: List[str] = []
    initialized = False
    ready = False
    pending: Dict[int, str] = {}
    for position, event in enumerate(events, 1):
        if (event.actor, event.target) not in {("client", "server"), ("server", "client")}:
            errors.append(f"{position}:cross_boundary")
            continue
        if event.method == "initialize":
            if initialized or event.actor != "client" or event.request_id is None:
                errors.append(f"{position}:invalid_initialize")
            else:
                initialized = True
                pending[event.request_id] = "initialize_result"
        elif event.method == "initialize_result":
            if not initialized or event.actor != "server" or pending.get(event.request_id) != "initialize_result":
                errors.append(f"{position}:unmatched_initialize_result")
            else:
                del pending[event.request_id]
        elif event.method == "notifications/initialized":
            if event.actor != "client" or not initialized or pending:
                errors.append(f"{position}:initialized_too_early")
            else:
                ready = True
        elif event.method == "tools/list":
            if event.actor != "client" or not ready or event.request_id is None:
                errors.append(f"{position}:tools_before_ready")
            elif event.request_id in pending:
                errors.append(f"{position}:duplicate_request_id")
            else:
                pending[event.request_id] = "tools/list_result"
        elif event.method == "tools/list_result":
            if event.actor != "server" or pending.get(event.request_id) != "tools/list_result":
                errors.append(f"{position}:unmatched_result")
            else:
                del pending[event.request_id]
        else:
            errors.append(f"{position}:unsupported_teaching_event")
    return errors


if __name__ == "__main__":
    trace = [
        Event("client", "server", "initialize", 1),
        Event("server", "client", "initialize_result", 1),
        Event("client", "server", "notifications/initialized"),
        Event("client", "server", "tools/list", 2),
        Event("server", "client", "tools/list_result", 2),
    ]
    print("valid_trace", check_trace(trace))
    early = [Event("client", "server", "tools/list", 2)]
    print("early_tools", check_trace(early))

python trace_lab.py

输出：

valid_trace []
early_tools ['1:tools_before_ready']

valid_trace后面的空列表表示当前检查没有发现异常。early_tools把目录请求放在第一条，得到tools_before_ready，因为连接尚未初始化。

Event保存发出方、接收方、方法和编号。pending把请求编号映射到预期响应事件：初始化对应initialize_result，目录对应tools/list_result。响应需要编号与事件类型都匹配，随后从pending移除。ready保存准备状态，通知之后才允许目录请求。

把最后响应编号2改成9，会出现unmatched_result。恢复后，把准备通知移到初始化响应之前，会出现initialized_too_early。这两个修改分别检查编号和时序。

4. 角色与业务职责

组件	负责什么
Host	用户目标、模型、权限和展示
Client	连接、编码、发送和编号配对
Server	目录、能力和请求处理
业务工具	查询或写入具体系统

内存事件检查帮助理解顺序；传输篇将实际启动子进程交换JSON消息。协议能力说明连接支持什么，业务权限说明当前用户可对哪个对象做什么，应用把二者分别处理。多个服务同时连接时，全局任务编号可把不同连接中的动作关联起来。

总结：Host管理任务，Client收发消息，Server提供能力。继续读 MCP三种能力。

5. 验证消息与边界

测试文件保存到同一目录：

下面的测试检查初始化先后顺序，并确认响应与尚未完成的请求编号和方法匹配。

测试代码：test_trace_lab.py

# 核对MCP架构的正常结果和错误分支；断言失败时会显示对应的测试。
import unittest

from trace_lab import Event, check_trace


def ready():
    return [
        Event("client", "server", "initialize", 1),
        Event("server", "client", "initialize_result", 1),
        Event("client", "server", "notifications/initialized"),
    ]


class TraceTests(unittest.TestCase):
    def test_response_method_must_match_request(self):
        events = [Event("client","server","initialize",1),
                  Event("server","client","tools/list_result",1),
                  Event("client","server","notifications/initialized")]
        self.assertIn("2:unmatched_result",check_trace(events))
        self.assertIn("3:initialized_too_early",check_trace(events))
        events = ready() + [Event("client","server","tools/list",2),
                           Event("server","client","initialize_result",2)]
        self.assertIn("5:unmatched_initialize_result",check_trace(events))
    def test_normal_order(self):
        events = ready() + [Event("client", "server", "tools/list", 2),
                            Event("server", "client", "tools/list_result", 2)]
        self.assertEqual(check_trace(events), [])

    def test_tools_before_ready(self):
        self.assertIn("1:tools_before_ready", check_trace([Event("client", "server", "tools/list", 2)]))

    def test_result_id_must_match(self):
        events = ready() + [Event("client", "server", "tools/list", 2),
                            Event("server", "client", "tools/list_result", 3)]
        self.assertIn("5:unmatched_result", check_trace(events))

    def test_server_cannot_skip_client_boundary(self):
        self.assertIn("1:cross_boundary", check_trace([Event("server", "host", "tools/list_result", 1)]))


if __name__ == "__main__":
    unittest.main()

python -m unittest test_trace_lab -v

恢复实验中的改动后运行测试。失败时先查看对应参数、编号或权限条件，再核对程序分支。

课后练习展开习题收起习题

正在读取本地记录…

不调用模型，不执行代码。简答、代码和场景题仅保存答案并展示自查标准，不生成评分、徽章或认证。输入内容不会自动保存，请点击提交；不要填写个人信息或密钥。

练习 1 / 4 · v1.0.1

选择题 · 先判断边界

关于“MCP Host、Client 与 Server”，哪种做法更稳妥？

对应知识点：L2-mcp-architecture

选择一个答案A. 先明确输入、证据和停止条件B. 默认输出永远正确C. 用一次成功推断全部场景

练习 2 / 4 · v1.0.0

简答题 · 解释失败

写出一个失败输入并说明记录字段。

对应知识点：L2-mcp-architecture

你的回答

练习 3 / 4 · v1.0.0

代码题 · 边界测试

补充正常、边界、失败三组输入和预期结果。

对应知识点：L2-mcp-architecture

你的代码或伪代码

练习 4 / 4 · v1.0.0

场景题 · 结果未知

证据不足时如何设计状态和人工接管？

对应知识点：L2-mcp-architecture

你的回答