1. 从协议理解到代码落地为什么MCP值得深究最近在梳理一些跨进程通信和微服务架构的底层实现MCPMessage Channel Protocol这个协议反复被提及。它不像HTTP或gRPC那样广为人知但在一些特定的、对消息传输的可靠性、顺序性和连接管理有严苛要求的内部系统中却扮演着核心角色。很多朋友在初次接触MCP时会觉得协议文档读起来有点抽象各种状态机、确认机制看着头大更别提自己动手实现一个简单的客户端或服务端了。这正是我写这篇笔记的原因——光看协议规范不够必须得把代码敲出来在调试中观察每一个字节的流动才能真正理解其设计精髓。所谓MCP你可以把它想象成在两个应用之间建立的一条“有保障的专用数据管道”。这条管道不仅负责搬运数据消息更重要的是它自己有一套严谨的“交通规则”如何建立连接握手、如何确保每个数据包不丢不重确认与重传、如何优雅地处理管道临时中断或永久关闭连接状态管理。我们常说的TCP是操作系统内核提供的通用、可靠的字节流管道而MCP则是在应用层基于TCP或其他可靠传输层自己再定义了一套面向“消息”的、更贴近业务逻辑的通信契约。理解并实现它对于构建健壮的分布式系统、自定义网关或中间件是项非常扎实的基本功。在这篇笔记里我不会重复协议的基础定义而是直接带大家撸起袖子通过一个可运行的代码示例拆解MCP核心的传输过程。我们将实现一个简化但功能完整的MCP协议栈涵盖连接建立、消息封装与解析、确认机制以及连接保活。你会看到协议字段如何映射到代码结构状态机如何驱动程序行为以及在实际编码中会遇到哪些“坑”。无论你是想深入了解某个使用了MCP的现有系统还是为自己的项目设计通信层希望这篇结合代码的剖析能给你带来直接、实用的参考。2. 示例项目的顶层设计模块划分与职责边界在开始写具体代码之前我们先花点时间设计整个示例项目的结构。一个清晰的架构能让我们在实现复杂状态逻辑时保持头脑清醒。我们的目标是实现一个双向通信的MCP协议示例包含客户端Initiator和服务端Acceptor角色。为了聚焦于协议本身我们使用本地回环地址127.0.0.1和TCP socket进行模拟。整个项目将分为以下几个核心模块协议常量与报文定义模块这里定义了所有MCP协议中用到的常量如版本号、消息类型连接请求、连接确认、数据消息、心跳、断开请求等、错误码。同时定义核心的协议报文Frame的数据结构包括头部Header和载荷Payload。编解码器模块这是协议栈的“翻译官”。负责将内存中的Frame对象序列化成可以在网络上传输的字节流编码以及将接收到的字节流反序列化回Frame对象解码。这里会严格遵循MCP协议的二进制格式规范。连接状态机模块这是MCP协议逻辑的核心。它定义了连接可能处于的各种状态如CLOSED,CONNECTING,ESTABLISHED,DISCONNECTING等以及在不同状态下接收到不同类型报文或发生特定事件如超时时应如何转移到下一个状态并执行相应动作如发送确认报文、通知应用层等。会话管理模块它封装了一个完整的MCP连接会话。内部会持有一个状态机实例、一个网络传输通道、发送/接收缓冲区、定时器用于心跳、超时重传等。它向上对应用层提供简单的sendMessage()和onMessage()回调接口向下管理编解码、状态变迁和网络I/O。客户端与服务端启动模块这是示例的入口。它们负责创建网络监听器或发起连接将建立的TCP socket包装进我们实现的MCP会话管理中并演示基本的消息发送与接收流程。为什么这么设计因为MCP协议的本质是一个状态驱动的、基于消息的契约。将状态机独立出来可以使协议逻辑与网络I/O、业务处理解耦让每一部分的职责单一便于测试和维护。编解码器独立则能确保协议格式的严格性一旦格式需要调整影响范围也最小。注意为了简化示例我们暂不考虑窗口流控、消息分片、加密等高级特性只实现最核心的可靠传输机制。同时错误处理会相对简单在实际工业级实现中需要更完备的异常处理和资源管理。3. 协议报文的结构化定义与编解码实现任何协议的第一步都是定义好“对话的语言”——即报文格式。MCP通常采用二进制协议以保证效率和紧凑性。让我们先定义我们的协议帧Frame。一个基本的MCP帧可以分为固定长度的头部Header和可变长度的载荷Payload。头部Header设计我们设计一个12字节的固定头部包含以下字段Magic Number (2字节)协议魔数用于快速识别和校验例如0x4D43(即 MC 的ASCII)。Version (1字节)协议版本主版本号。Type (1字节)消息类型。例如0x01连接请求 CONNECT0x02连接确认 CONNACK0x03数据消息 DATA0x04确认 ACK0x05心跳 PING0x06心跳回复 PONG0x07断开请求 DISCONNECT。Flags (1字节)标志位可用于表示压缩、加密、是否为请求/响应等。本例暂简单处理。Sequence ID (4字节)序列号用于标识消息和实现确认机制。对于需要确认的消息如DATA接收方需回复带有相同Sequence ID的ACK。Payload Length (4字节)载荷数据的长度单位字节。最大长度可根据需要设定。载荷Payload设计对于不同类型的消息载荷内容不同CONNECT: 可包含客户端标识、协议版本协商信息等。CONNACK: 可包含连接状态成功/失败、服务端支持的最大消息长度等。DATA: 就是应用层要传输的实际二进制数据。ACK: 通常为空或可包含一些元信息。PING/PONG: 通常为空。DISCONNECT: 可包含断开原因码。在代码中我们会用一个类来表征这个帧# protocol.py import struct from enum import IntEnum from dataclasses import dataclass from typing import Optional class FrameType(IntEnum): CONNECT 0x01 CONNACK 0x02 DATA 0x03 ACK 0x04 PING 0x05 PONG 0x06 DISCONNECT 0x07 dataclass class McpHeader: magic: int 0x4D43 # MC version: int 1 type: FrameType FrameType.DATA flags: int 0 seq_id: int 0 payload_len: int 0 # 将头部对象打包成12字节的二进制数据 def pack(self) - bytes: # struct格式: H B B B I I - 大端序 2字节无符号短整型3个1字节无符号字符2个4字节无符号整型 return struct.pack(H B B B I I, self.magic, self.version, self.type, self.flags, self.seq_id, self.payload_len) # 从12字节二进制数据解析出头部对象 classmethod def unpack(cls, data: bytes) - McpHeader: if len(data) ! 12: raise ValueError(fHeader length must be 12 bytes, got {len(data)}) magic, version, ftype, flags, seq_id, payload_len struct.unpack(H B B B I I, data) return cls(magic, version, FrameType(ftype), flags, seq_id, payload_len) dataclass class McpFrame: header: McpHeader payload: Optional[bytes] None def to_bytes(self) - bytes: 将整个Frame序列化为字节流用于发送 header_bytes self.header.pack() if self.payload: if len(self.payload) ! self.header.payload_len: # 这是一个安全校验确保头部长度字段与实际载荷一致 raise ValueError(fPayload length mismatch. Header: {self.header.payload_len}, Actual: {len(self.payload)}) return header_bytes self.payload else: if self.header.payload_len ! 0: raise ValueError(fHeader indicates payload length {self.header.payload_len}, but payload is None) return header_bytes classmethod def from_bytes(cls, data: bytes) - McpFrame: 从字节流解析出一个完整的Frame。注意调用者需确保data至少包含完整的头部。 if len(data) 12: raise ValueError(fData too short for a header, need at least 12 bytes, got {len(data)}) header McpHeader.unpack(data[:12]) total_frame_len 12 header.payload_len if len(data) total_frame_len: raise ValueError(fIncomplete frame. Need {total_frame_len} bytes, got {len(data)}) payload data[12:total_frame_len] if header.payload_len 0 else None return cls(header, payload)编解码器的关键细节与避坑点字节序Endianness我们使用了大端序网络字节序。这是网络协议的标准做法确保不同架构的机器能正确解析。如果你在本地测试的两端都是x86小端机但用了大端序编码而解码用了小端序数字会完全错乱。务必在协议文档和代码中明确规定并统一使用网络字节序。长度字段的校验Header中的payload_len必须与实际的payload长度严格一致。我们在to_bytes方法中做了校验这是一个防御性编程的好习惯能及早发现程序逻辑错误避免发送错误格式的报文导致对端解析崩溃。帧的完整性判断from_bytes方法体现了协议解析的一个经典模式——“偷看”头部。我们先解析出固定的12字节头部从中得到载荷长度然后才能判断接收缓冲区中是否已经有一个完整帧的数据。如果不够说明TCP流中数据还未收全需要等待更多数据到达。这是处理基于TCP的流式协议时必须实现的“拆包”逻辑。常见的错误是假设一次recv调用就能拿到一个完整帧这在网络波动或消息较大时几乎必然出错。枚举类型的使用使用IntEnum定义消息类型比直接用数字常量更安全、可读性更好。在解析时通过FrameType(ftype)进行转换如果收到非法的类型值会抛出ValueError便于错误处理。4. 连接状态机的核心逻辑与实现MCP协议连接的生命周期由状态机驱动。这是协议行为正确性的保证。我们定义一个简化的状态机包含以下状态CLOSED: 初始状态或最终状态。连接不存在。CONNECTING: 客户端已发送CONNECT请求等待CONNACK响应。ESTABLISHED: 连接已成功建立可以正常收发数据消息。DISCONNECTING: 已发送或收到DISCONNECT请求正在等待连接彻底关闭。ERROR: 发生错误如协议解析错误、超时等连接异常终止。状态迁移由事件触发主要事件包括send_connect,send_data,send_disconnect应用层主动发起的动作。frame_received从网络接收到一个完整的MCP帧。timer_expired某个定时器超时如连接超时、心跳超时、ACK等待超时。下面我们用代码勾勒这个状态机的骨架# state_machine.py from enum import Enum from typing import Optional, Callable, Any from protocol import McpFrame, FrameType import time class ConnectionState(Enum): CLOSED CLOSED CONNECTING CONNECTING ESTABLISHED ESTABLISHED DISCONNECTING DISCONNECTING ERROR ERROR class McpStateMachine: def __init__(self, on_state_change: Optional[Callable[[ConnectionState], None]] None): self.state ConnectionState.CLOSED self.on_state_change on_state_change # 用于跟踪已发送但未确认的消息key为seq_id, value为(发送时间, 重试次数, frame) self.pending_acks {} self.next_seq_id 1 # 序列号生成器 # 定时器相关简化处理实际需用调度器 self.connect_timer None self.heartbeat_timer None def _transition_to(self, new_state: ConnectionState): 状态转移并触发回调 if self.state ! new_state: old_state self.state self.state new_state if self.on_state_change: self.on_state_change(old_state, new_state) def handle_event(self, event: str, frame: Optional[McpFrame] None, **kwargs) - Optional[McpFrame]: 处理事件返回一个需要发送的Frame如果有的话。 这是状态机的核心分发逻辑。 if self.state ConnectionState.CLOSED: return self._handle_closed(event, frame, **kwargs) elif self.state ConnectionState.CONNECTING: return self._handle_connecting(event, frame, **kwargs) elif self.state ConnectionState.ESTABLISHED: return self._handle_established(event, frame, **kwargs) elif self.state ConnectionState.DISCONNECTING: return self._handle_disconnecting(event, frame, **kwargs) # ERROR状态通常不处理事件等待重置或销毁 return None def _handle_closed(self, event: str, frame: Optional[McpFrame], **kwargs): if event send_connect: # 应用层请求建立连接 connect_frame McpFrame( headerMcpHeader(typeFrameType.CONNECT, seq_idself._get_next_seq_id()), payloadbclient_info # 示例载荷 ) self._transition_to(ConnectionState.CONNECTING) # 启动连接超时定时器示例实际需定时器模块 self.connect_timer time.time() 10 # 10秒超时 return connect_frame return None def _handle_connecting(self, event: str, frame: Optional[McpFrame], **kwargs): if event frame_received and frame and frame.header.type FrameType.CONNACK: # 收到连接确认 # 检查CONNACK的载荷确认连接是否被接受简化处理假设成功 self._transition_to(ConnectionState.ESTABLISHED) # 取消连接超时定时器 self.connect_timer None # 启动心跳定时器 self._start_heartbeat() # 不需要立即回复帧 return None elif event timer_expired and kwargs.get(timer_id) connect: # 连接超时 self._transition_to(ConnectionState.ERROR) return None # 在CONNECTING状态下收到其他类型的帧可能是协议错误 elif event frame_received: self._transition_to(ConnectionState.ERROR) return None def _handle_established(self, event: str, frame: Optional[McpFrame], **kwargs): if event send_data: data_payload kwargs.get(data) if not data_payload: return None data_frame McpFrame( headerMcpHeader(typeFrameType.DATA, seq_idself._get_next_seq_id(), payload_lenlen(data_payload)), payloaddata_payload ) # 将消息加入等待确认队列 self.pending_acks[data_frame.header.seq_id] (time.time(), 0, data_frame) # 启动该消息的ACK超时定时器简化 return data_frame elif event frame_received and frame: if frame.header.type FrameType.DATA: # 收到数据消息需要回复ACK ack_frame McpFrame( headerMcpHeader(typeFrameType.ACK, seq_idframe.header.seq_id) ) # 将数据传递给应用层通过回调或其他机制此处略 # self.on_data_received(frame.payload) return ack_frame elif frame.header.type FrameType.ACK: # 收到ACK从等待队列中移除对应消息 seq_id frame.header.seq_id if seq_id in self.pending_acks: del self.pending_acks[seq_id] # 可选触发应用层消息发送成功的回调 elif frame.header.type FrameType.PING: # 回复PONG return McpFrame(headerMcpHeader(typeFrameType.PONG)) elif frame.header.type FrameType.DISCONNECT: # 对端请求断开 self._transition_to(ConnectionState.DISCONNECTING) # 可以回复一个DISCONNECT可选然后关闭连接 return McpFrame(headerMcpHeader(typeFrameType.DISCONNECT)) elif event send_disconnect: # 应用层请求断开 self._transition_to(ConnectionState.DISCONNECTING) return McpFrame(headerMcpHeader(typeFrameType.DISCONNECT)) elif event timer_expired and kwargs.get(timer_id) heartbeat: # 心跳超时发送PING self._start_heartbeat() # 重置心跳定时器 return McpFrame(headerMcpHeader(typeFrameType.PING)) return None def _handle_disconnecting(self, event: str, frame: Optional[McpFrame], **kwargs): # 在DISCONNECTING状态主要等待底层传输关闭或进行一些清理工作 # 收到对端的DISCONNECT如果之前是自己发起的或超时后可以转移到CLOSED if event frame_received and frame and frame.header.type FrameType.DISCONNECT: self._transition_to(ConnectionState.CLOSED) elif event connection_closed: self._transition_to(ConnectionState.CLOSED) return None def _get_next_seq_id(self) - int: seq self.next_seq_id self.next_seq_id 1 # 处理回绕简单示例 if self.next_seq_id 0xFFFFFFFF: self.next_seq_id 1 return seq def _start_heartbeat(self): # 简化设置一个下次心跳触发的时间点 self.heartbeat_timer time.time() 30 # 30秒后发下一次心跳状态机实现的难点与心得状态的纯粹性状态机只负责管理状态和根据事件决定要发送的协议帧。它不应该直接操作网络socket、启动线程或调用复杂的业务逻辑。它通过返回需要发送的McpFrame对象以及通过回调如on_state_change来影响外部世界。这种设计使得状态机本身易于单元测试。定时器的整合在实际系统中定时器管理是个麻烦事。上面的示例用简单的未来时间点来模拟。在真实实现中你需要一个集中的定时器调度器能够添加、取消定时器并在超时时向状态机发送timer_expired事件。定时器ID需要与特定任务如等待某seq_id的ACK、心跳关联。未确认消息的重传pending_acks字典是可靠传输的关键。当状态机在ESTABLISHED状态下收到timer_expired事件且定时器ID对应某个seq_id时就需要从pending_acks中取出该消息帧进行重传并增加重试计数。超过最大重试次数后应将连接置为ERROR状态。这部分逻辑在上例中省略了但它是必须实现的。并发与线程安全如果网络I/O和状态机驱动运行在不同的线程那么对状态机handle_event的调用必须是线程安全的通常需要加锁。一个更清晰的架构是使用单线程的事件循环如asyncio将所有事件网络收到数据、定时器超时、应用层发送请求都放入一个队列由事件循环顺序取出并调用状态机处理。5. 会话管理粘合协议栈与网络I/O状态机和编解码器准备好了我们需要一个“会话”来把它们和真实的网络连接结合起来并管理整个生命周期。这个会话类McpSession是给应用层使用的直接接口。# session.py import socket import threading import queue import time from typing import Optional, Callable from protocol import McpFrame, McpHeader, FrameType from state_machine import McpStateMachine, ConnectionState class McpSession: def __init__(self, sock: socket.socket, is_initiator: bool): self.sock sock self.is_initiator is_initiator self.state_machine McpStateMachine(on_state_changeself._on_state_change) self.send_queue queue.Queue() # 用于存放待发送的Frame self.receive_buffer bytearray() # 接收数据的缓冲区 self.running False self.recv_thread: Optional[threading.Thread] None self.send_thread: Optional[threading.Thread] None self.on_message: Optional[Callable[[bytes], None]] None # 应用层消息回调 def start(self): 启动会话开始处理I/O self.running True # 启动接收线程 self.recv_thread threading.Thread(targetself._recv_loop, daemonTrue) self.recv_thread.start() # 启动发送线程 self.send_thread threading.Thread(targetself._send_loop, daemonTrue) self.send_thread.start() # 如果是发起方客户端主动发送CONNECT if self.is_initiator: connect_frame self.state_machine.handle_event(send_connect) if connect_frame: self.send_queue.put(connect_frame) def send_data(self, data: bytes): 应用层调用此方法发送数据 if self.state_machine.state ! ConnectionState.ESTABLISHED: raise ConnectionError(Connection is not established) frame_to_send self.state_machine.handle_event(send_data, datadata) if frame_to_send: self.send_queue.put(frame_to_send) def disconnect(self): 应用层请求断开连接 frame_to_send self.state_machine.handle_event(send_disconnect) if frame_to_send: self.send_queue.put(frame_to_send) # 设置标志让循环退出 self.running False def _on_state_change(self, old_state: ConnectionState, new_state: ConnectionState): 状态变化回调 print(f[Session] State changed: {old_state} - {new_state}) if new_state ConnectionState.CLOSED or new_state ConnectionState.ERROR: self.running False self.sock.close() def _recv_loop(self): 接收线程循环从socket读数据拼装完整帧交给状态机处理 while self.running: try: # 设置超时以便能响应running标志的变化 self.sock.settimeout(1.0) data self.sock.recv(4096) if not data: # 对端关闭连接 self.state_machine.handle_event(connection_closed) break self.receive_buffer.extend(data) self._process_buffer() except socket.timeout: continue # 超时是正常的继续循环检查running标志 except (ConnectionError, OSError) as e: print(f[RecvLoop] Socket error: {e}) self.state_machine.handle_event(connection_closed) break def _process_buffer(self): 处理接收缓冲区尝试解析出完整帧 while len(self.receive_buffer) 12: # 至少有一个头部的长度 try: # 尝试解析头部获取完整帧长度 header McpHeader.unpack(self.receive_buffer[:12]) total_frame_len 12 header.payload_len if len(self.receive_buffer) total_frame_len: # 缓冲区数据还不够一个完整帧等待下次接收 break # 提取完整帧数据 frame_data bytes(self.receive_buffer[:total_frame_len]) # 从缓冲区移除已处理的数据 del self.receive_buffer[:total_frame_len] # 解码成Frame对象 frame McpFrame.from_bytes(frame_data) # 将接收到的帧交给状态机处理 frame_to_send self.state_machine.handle_event(frame_received, frameframe) # 如果状态机要求发送回复帧如ACK, PONG放入发送队列 if frame_to_send: self.send_queue.put(frame_to_send) # 如果是数据消息传递给应用层回调 if frame.header.type FrameType.DATA and self.on_message: self.on_message(frame.payload) except ValueError as e: # 解析错误协议格式非法转移到错误状态 print(f[ProcessBuffer] Protocol error: {e}) self.state_machine.handle_event(protocol_error) break def _send_loop(self): 发送线程循环从队列取Frame编码后通过socket发送 while self.running: try: frame self.send_queue.get(timeout1.0) frame_bytes frame.to_bytes() # 在实际项目中这里需要考虑TCP粘包问题吗 # 不需要因为to_bytes()已经将整个帧头载荷打包成一个完整的字节流。 # TCP是流式协议保证顺序但不保证消息边界。我们的“消息边界”就是通过“长度字段”自己定义的。 # 接收方_process_buffer中的“拆包”逻辑正是为了解决这个问题。 # 所以这里直接sendall即可。 self.sock.sendall(frame_bytes) except queue.Empty: continue # 队列为空继续循环 except (ConnectionError, OSError) as e: print(f[SendLoop] Socket error: {e}) self.state_machine.handle_event(connection_closed) break会话管理中的核心陷阱与解决方案TCP粘包/半包问题这是网络编程新手最常见的坑。_process_buffer方法是解决这个问题的经典模式。我们永远不能假设一次recv调用返回的数据就是一个完整的应用层报文MCP帧。我们必须维护一个应用层缓冲区不断累积数据并尝试从缓冲区头部解析出完整的帧。McpHeader中的payload_len字段就是我们判断“完整性”的关键。永远基于“长度字段”来拆包而不是依赖特殊分隔符或固定次数recv。双工通信与线程设计我们使用了独立的接收和发送线程。这是一个简单清晰的模型但引入了线程同步问题。这里send_queue是线程安全的queue.Queue用于协调状态机线程主线程或事件线程和发送线程。接收线程将收到的帧交给状态机状态机可能产生需要发送的帧如ACK这个帧也需要放入send_queue。要小心避免死锁和竞态条件。对于更高效的实现可以考虑使用asyncio单线程异步模型。资源清理当连接关闭或出错时必须确保socket被正确关闭线程被正确终止。示例中通过running标志和sock.settimeout来让线程优雅退出。在实际代码中还需要加入join线程等待其结束。_on_state_change回调在进入CLOSED或ERROR状态时关闭socket并停止循环这是一个集中清理的好地方。超时与重传的管理示例中简化了定时器。在真实场景中你需要一个更精细的定时器管理器。例如对于pending_acks中的每条消息都应该有一个独立的超时定时器。当定时器触发时向状态机发送一个携带seq_id的timer_expired事件。状态机检查该消息是否已被确认若未确认则重传。重传次数过多则断开连接。心跳定时器同理。6. 从零构建客户端与服务端的启动与交互演示最后我们编写客户端和服务端的启动脚本将以上所有模块串联起来进行一个简单的演示。服务端Acceptor代码# server.py import socket import threading from session import McpSession def on_server_message(data: bytes): print(f[Server] Received application data: {data.decode(utf-8, errorsignore)}) # 这里可以处理业务逻辑然后回复客户端示例中省略 def handle_client_connection(client_sock: socket.socket, client_addr: tuple): print(f[Server] New connection from {client_addr}) # 服务端是被动接受连接的一方is_initiatorFalse session McpSession(client_sock, is_initiatorFalse) session.on_message on_server_message session.start() # 模拟等待一段时间然后主动发送一条消息给客户端 import time time.sleep(2) if session.state_machine.state ConnectionState.ESTABLISHED: session.send_data(bHello from Server!) # 保持连接一段时间观察心跳等 time.sleep(10) session.disconnect() print(f[Server] Connection with {client_addr} closed.) def start_server(host127.0.0.1, port9999): server_sock socket.socket(socket.AF_INET, socket.SOCK_STREAM) server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server_sock.bind((host, port)) server_sock.listen(5) print(f[Server] Listening on {host}:{port}) try: while True: client_sock, client_addr server_sock.accept() # 为每个客户端连接创建一个新线程处理生产环境建议使用线程池 client_thread threading.Thread(targethandle_client_connection, args(client_sock, client_addr)) client_thread.daemon True client_thread.start() except KeyboardInterrupt: print(\n[Server] Shutting down...) finally: server_sock.close() if __name__ __main__: start_server()客户端Initiator代码# client.py import socket import time from session import McpSession def on_client_message(data: bytes): print(f[Client] Received application data: {data.decode(utf-8, errorsignore)}) def start_client(host127.0.0.1, port9999): sock socket.socket(socket.AF_INET, socket.SOCK_STREAM) try: sock.connect((host, port)) print(f[Client] Connected to {host}:{port}) # 客户端是发起方is_initiatorTrue session McpSession(sock, is_initiatorTrue) session.on_message on_client_message session.start() # 等待连接建立 time.sleep(1) if session.state_machine.state ! ConnectionState.ESTABLISHED: print([Client] Failed to establish MCP connection.) return # 连接建立后发送几条测试消息 for i in range(3): msg fTest message {i1} from client.encode(utf-8) print(f[Client] Sending: {msg.decode()}) session.send_data(msg) time.sleep(1.5) # 间隔发送便于观察 # 等待接收服务端的消息并保持连接以观察心跳 print([Client] Waiting for server messages and heartbeat...) time.sleep(15) # 主动断开连接 session.disconnect() print([Client] Disconnected.) except ConnectionRefusedError: print(f[Client] Could not connect to {host}:{port}) except Exception as e: print(f[Client] Error: {e}) finally: sock.close() if __name__ __main__: start_client()运行与观察首先在终端运行python server.py。然后在另一个终端运行python client.py。观察两个终端的输出。你应该能看到类似以下的过程客户端打印Connected to 127.0.0.1:9999。服务端打印New connection from (127.0.0.1, xxxxx)。双方打印状态变化[Session] State changed: CLOSED - CONNECTING然后- ESTABLISHED。客户端发送三条测试消息。服务端收到消息后打印[Server] Received application data: Test message 1...并在2秒后向客户端发送Hello from Server!。客户端收到服务端消息并打印。在连接空闲期间你会看到状态机定时示例中约30秒触发PING/PONG的日志需要在代码中添加相应打印。最后服务端或客户端断开连接状态变为DISCONNECTING-CLOSED。演示中的关键点验证连接握手客户端发送CONNECT服务端回复CONNACK状态机迁移到ESTABLISHED。可靠数据传输客户端发送的DATA消息带有递增的seq_id。服务端收到后应回复对应的ACK。你可以在_handle_established的ACK处理部分添加日志以确认这一机制工作正常。心跳保活在ESTABLISHED状态空闲一段时间后状态机应触发timer_expired心跳事件发送PING对端回复PONG。有序关闭调用disconnect()会发送DISCONNECT帧状态机进入DISCONNECTING收到对端的DISCONNECT或底层连接关闭事件后进入CLOSED状态。通过这个从协议定义到代码实现再到实际运行的完整示例MCP协议中那些抽象的概念——连接状态、序列号、确认、心跳——都变成了可以观察、可以调试的具体代码逻辑。在实现过程中最深的体会是协议设计决定了代码结构。一个定义良好的状态机是复杂协议实现的基石它能将杂乱的网络事件和业务请求梳理得井井有条。而缓冲区管理和定时器处理则是实现层面最容易出bug的地方需要格外小心。这个示例虽然简化但已经勾勒出了一个可靠应用层协议栈的核心骨架在此基础上增加流量控制、多路复用、加密等特性思路都是相通的。下次当你再阅读其他协议的RFC文档时不妨尝试用这种“状态机编解码会话管理”的框架去理解它并思考如何用代码实现相信会有更深的领悟。 从协议到代码：手把手实现MCP消息通道协议栈
1. 从协议理解到代码落地为什么MCP值得深究最近在梳理一些跨进程通信和微服务架构的底层实现MCPMessage Channel Protocol这个协议反复被提及。它不像HTTP或gRPC那样广为人知但在一些特定的、对消息传输的可靠性、顺序性和连接管理有严苛要求的内部系统中却扮演着核心角色。很多朋友在初次接触MCP时会觉得协议文档读起来有点抽象各种状态机、确认机制看着头大更别提自己动手实现一个简单的客户端或服务端了。这正是我写这篇笔记的原因——光看协议规范不够必须得把代码敲出来在调试中观察每一个字节的流动才能真正理解其设计精髓。所谓MCP你可以把它想象成在两个应用之间建立的一条“有保障的专用数据管道”。这条管道不仅负责搬运数据消息更重要的是它自己有一套严谨的“交通规则”如何建立连接握手、如何确保每个数据包不丢不重确认与重传、如何优雅地处理管道临时中断或永久关闭连接状态管理。我们常说的TCP是操作系统内核提供的通用、可靠的字节流管道而MCP则是在应用层基于TCP或其他可靠传输层自己再定义了一套面向“消息”的、更贴近业务逻辑的通信契约。理解并实现它对于构建健壮的分布式系统、自定义网关或中间件是项非常扎实的基本功。在这篇笔记里我不会重复协议的基础定义而是直接带大家撸起袖子通过一个可运行的代码示例拆解MCP核心的传输过程。我们将实现一个简化但功能完整的MCP协议栈涵盖连接建立、消息封装与解析、确认机制以及连接保活。你会看到协议字段如何映射到代码结构状态机如何驱动程序行为以及在实际编码中会遇到哪些“坑”。无论你是想深入了解某个使用了MCP的现有系统还是为自己的项目设计通信层希望这篇结合代码的剖析能给你带来直接、实用的参考。2. 示例项目的顶层设计模块划分与职责边界在开始写具体代码之前我们先花点时间设计整个示例项目的结构。一个清晰的架构能让我们在实现复杂状态逻辑时保持头脑清醒。我们的目标是实现一个双向通信的MCP协议示例包含客户端Initiator和服务端Acceptor角色。为了聚焦于协议本身我们使用本地回环地址127.0.0.1和TCP socket进行模拟。整个项目将分为以下几个核心模块协议常量与报文定义模块这里定义了所有MCP协议中用到的常量如版本号、消息类型连接请求、连接确认、数据消息、心跳、断开请求等、错误码。同时定义核心的协议报文Frame的数据结构包括头部Header和载荷Payload。编解码器模块这是协议栈的“翻译官”。负责将内存中的Frame对象序列化成可以在网络上传输的字节流编码以及将接收到的字节流反序列化回Frame对象解码。这里会严格遵循MCP协议的二进制格式规范。连接状态机模块这是MCP协议逻辑的核心。它定义了连接可能处于的各种状态如CLOSED,CONNECTING,ESTABLISHED,DISCONNECTING等以及在不同状态下接收到不同类型报文或发生特定事件如超时时应如何转移到下一个状态并执行相应动作如发送确认报文、通知应用层等。会话管理模块它封装了一个完整的MCP连接会话。内部会持有一个状态机实例、一个网络传输通道、发送/接收缓冲区、定时器用于心跳、超时重传等。它向上对应用层提供简单的sendMessage()和onMessage()回调接口向下管理编解码、状态变迁和网络I/O。客户端与服务端启动模块这是示例的入口。它们负责创建网络监听器或发起连接将建立的TCP socket包装进我们实现的MCP会话管理中并演示基本的消息发送与接收流程。为什么这么设计因为MCP协议的本质是一个状态驱动的、基于消息的契约。将状态机独立出来可以使协议逻辑与网络I/O、业务处理解耦让每一部分的职责单一便于测试和维护。编解码器独立则能确保协议格式的严格性一旦格式需要调整影响范围也最小。注意为了简化示例我们暂不考虑窗口流控、消息分片、加密等高级特性只实现最核心的可靠传输机制。同时错误处理会相对简单在实际工业级实现中需要更完备的异常处理和资源管理。3. 协议报文的结构化定义与编解码实现任何协议的第一步都是定义好“对话的语言”——即报文格式。MCP通常采用二进制协议以保证效率和紧凑性。让我们先定义我们的协议帧Frame。一个基本的MCP帧可以分为固定长度的头部Header和可变长度的载荷Payload。头部Header设计我们设计一个12字节的固定头部包含以下字段Magic Number (2字节)协议魔数用于快速识别和校验例如0x4D43(即 MC 的ASCII)。Version (1字节)协议版本主版本号。Type (1字节)消息类型。例如0x01连接请求 CONNECT0x02连接确认 CONNACK0x03数据消息 DATA0x04确认 ACK0x05心跳 PING0x06心跳回复 PONG0x07断开请求 DISCONNECT。Flags (1字节)标志位可用于表示压缩、加密、是否为请求/响应等。本例暂简单处理。Sequence ID (4字节)序列号用于标识消息和实现确认机制。对于需要确认的消息如DATA接收方需回复带有相同Sequence ID的ACK。Payload Length (4字节)载荷数据的长度单位字节。最大长度可根据需要设定。载荷Payload设计对于不同类型的消息载荷内容不同CONNECT: 可包含客户端标识、协议版本协商信息等。CONNACK: 可包含连接状态成功/失败、服务端支持的最大消息长度等。DATA: 就是应用层要传输的实际二进制数据。ACK: 通常为空或可包含一些元信息。PING/PONG: 通常为空。DISCONNECT: 可包含断开原因码。在代码中我们会用一个类来表征这个帧# protocol.py import struct from enum import IntEnum from dataclasses import dataclass from typing import Optional class FrameType(IntEnum): CONNECT 0x01 CONNACK 0x02 DATA 0x03 ACK 0x04 PING 0x05 PONG 0x06 DISCONNECT 0x07 dataclass class McpHeader: magic: int 0x4D43 # MC version: int 1 type: FrameType FrameType.DATA flags: int 0 seq_id: int 0 payload_len: int 0 # 将头部对象打包成12字节的二进制数据 def pack(self) - bytes: # struct格式: H B B B I I - 大端序 2字节无符号短整型3个1字节无符号字符2个4字节无符号整型 return struct.pack(H B B B I I, self.magic, self.version, self.type, self.flags, self.seq_id, self.payload_len) # 从12字节二进制数据解析出头部对象 classmethod def unpack(cls, data: bytes) - McpHeader: if len(data) ! 12: raise ValueError(fHeader length must be 12 bytes, got {len(data)}) magic, version, ftype, flags, seq_id, payload_len struct.unpack(H B B B I I, data) return cls(magic, version, FrameType(ftype), flags, seq_id, payload_len) dataclass class McpFrame: header: McpHeader payload: Optional[bytes] None def to_bytes(self) - bytes: 将整个Frame序列化为字节流用于发送 header_bytes self.header.pack() if self.payload: if len(self.payload) ! self.header.payload_len: # 这是一个安全校验确保头部长度字段与实际载荷一致 raise ValueError(fPayload length mismatch. Header: {self.header.payload_len}, Actual: {len(self.payload)}) return header_bytes self.payload else: if self.header.payload_len ! 0: raise ValueError(fHeader indicates payload length {self.header.payload_len}, but payload is None) return header_bytes classmethod def from_bytes(cls, data: bytes) - McpFrame: 从字节流解析出一个完整的Frame。注意调用者需确保data至少包含完整的头部。 if len(data) 12: raise ValueError(fData too short for a header, need at least 12 bytes, got {len(data)}) header McpHeader.unpack(data[:12]) total_frame_len 12 header.payload_len if len(data) total_frame_len: raise ValueError(fIncomplete frame. Need {total_frame_len} bytes, got {len(data)}) payload data[12:total_frame_len] if header.payload_len 0 else None return cls(header, payload)编解码器的关键细节与避坑点字节序Endianness我们使用了大端序网络字节序。这是网络协议的标准做法确保不同架构的机器能正确解析。如果你在本地测试的两端都是x86小端机但用了大端序编码而解码用了小端序数字会完全错乱。务必在协议文档和代码中明确规定并统一使用网络字节序。长度字段的校验Header中的payload_len必须与实际的payload长度严格一致。我们在to_bytes方法中做了校验这是一个防御性编程的好习惯能及早发现程序逻辑错误避免发送错误格式的报文导致对端解析崩溃。帧的完整性判断from_bytes方法体现了协议解析的一个经典模式——“偷看”头部。我们先解析出固定的12字节头部从中得到载荷长度然后才能判断接收缓冲区中是否已经有一个完整帧的数据。如果不够说明TCP流中数据还未收全需要等待更多数据到达。这是处理基于TCP的流式协议时必须实现的“拆包”逻辑。常见的错误是假设一次recv调用就能拿到一个完整帧这在网络波动或消息较大时几乎必然出错。枚举类型的使用使用IntEnum定义消息类型比直接用数字常量更安全、可读性更好。在解析时通过FrameType(ftype)进行转换如果收到非法的类型值会抛出ValueError便于错误处理。4. 连接状态机的核心逻辑与实现MCP协议连接的生命周期由状态机驱动。这是协议行为正确性的保证。我们定义一个简化的状态机包含以下状态CLOSED: 初始状态或最终状态。连接不存在。CONNECTING: 客户端已发送CONNECT请求等待CONNACK响应。ESTABLISHED: 连接已成功建立可以正常收发数据消息。DISCONNECTING: 已发送或收到DISCONNECT请求正在等待连接彻底关闭。ERROR: 发生错误如协议解析错误、超时等连接异常终止。状态迁移由事件触发主要事件包括send_connect,send_data,send_disconnect应用层主动发起的动作。frame_received从网络接收到一个完整的MCP帧。timer_expired某个定时器超时如连接超时、心跳超时、ACK等待超时。下面我们用代码勾勒这个状态机的骨架# state_machine.py from enum import Enum from typing import Optional, Callable, Any from protocol import McpFrame, FrameType import time class ConnectionState(Enum): CLOSED CLOSED CONNECTING CONNECTING ESTABLISHED ESTABLISHED DISCONNECTING DISCONNECTING ERROR ERROR class McpStateMachine: def __init__(self, on_state_change: Optional[Callable[[ConnectionState], None]] None): self.state ConnectionState.CLOSED self.on_state_change on_state_change # 用于跟踪已发送但未确认的消息key为seq_id, value为(发送时间, 重试次数, frame) self.pending_acks {} self.next_seq_id 1 # 序列号生成器 # 定时器相关简化处理实际需用调度器 self.connect_timer None self.heartbeat_timer None def _transition_to(self, new_state: ConnectionState): 状态转移并触发回调 if self.state ! new_state: old_state self.state self.state new_state if self.on_state_change: self.on_state_change(old_state, new_state) def handle_event(self, event: str, frame: Optional[McpFrame] None, **kwargs) - Optional[McpFrame]: 处理事件返回一个需要发送的Frame如果有的话。 这是状态机的核心分发逻辑。 if self.state ConnectionState.CLOSED: return self._handle_closed(event, frame, **kwargs) elif self.state ConnectionState.CONNECTING: return self._handle_connecting(event, frame, **kwargs) elif self.state ConnectionState.ESTABLISHED: return self._handle_established(event, frame, **kwargs) elif self.state ConnectionState.DISCONNECTING: return self._handle_disconnecting(event, frame, **kwargs) # ERROR状态通常不处理事件等待重置或销毁 return None def _handle_closed(self, event: str, frame: Optional[McpFrame], **kwargs): if event send_connect: # 应用层请求建立连接 connect_frame McpFrame( headerMcpHeader(typeFrameType.CONNECT, seq_idself._get_next_seq_id()), payloadbclient_info # 示例载荷 ) self._transition_to(ConnectionState.CONNECTING) # 启动连接超时定时器示例实际需定时器模块 self.connect_timer time.time() 10 # 10秒超时 return connect_frame return None def _handle_connecting(self, event: str, frame: Optional[McpFrame], **kwargs): if event frame_received and frame and frame.header.type FrameType.CONNACK: # 收到连接确认 # 检查CONNACK的载荷确认连接是否被接受简化处理假设成功 self._transition_to(ConnectionState.ESTABLISHED) # 取消连接超时定时器 self.connect_timer None # 启动心跳定时器 self._start_heartbeat() # 不需要立即回复帧 return None elif event timer_expired and kwargs.get(timer_id) connect: # 连接超时 self._transition_to(ConnectionState.ERROR) return None # 在CONNECTING状态下收到其他类型的帧可能是协议错误 elif event frame_received: self._transition_to(ConnectionState.ERROR) return None def _handle_established(self, event: str, frame: Optional[McpFrame], **kwargs): if event send_data: data_payload kwargs.get(data) if not data_payload: return None data_frame McpFrame( headerMcpHeader(typeFrameType.DATA, seq_idself._get_next_seq_id(), payload_lenlen(data_payload)), payloaddata_payload ) # 将消息加入等待确认队列 self.pending_acks[data_frame.header.seq_id] (time.time(), 0, data_frame) # 启动该消息的ACK超时定时器简化 return data_frame elif event frame_received and frame: if frame.header.type FrameType.DATA: # 收到数据消息需要回复ACK ack_frame McpFrame( headerMcpHeader(typeFrameType.ACK, seq_idframe.header.seq_id) ) # 将数据传递给应用层通过回调或其他机制此处略 # self.on_data_received(frame.payload) return ack_frame elif frame.header.type FrameType.ACK: # 收到ACK从等待队列中移除对应消息 seq_id frame.header.seq_id if seq_id in self.pending_acks: del self.pending_acks[seq_id] # 可选触发应用层消息发送成功的回调 elif frame.header.type FrameType.PING: # 回复PONG return McpFrame(headerMcpHeader(typeFrameType.PONG)) elif frame.header.type FrameType.DISCONNECT: # 对端请求断开 self._transition_to(ConnectionState.DISCONNECTING) # 可以回复一个DISCONNECT可选然后关闭连接 return McpFrame(headerMcpHeader(typeFrameType.DISCONNECT)) elif event send_disconnect: # 应用层请求断开 self._transition_to(ConnectionState.DISCONNECTING) return McpFrame(headerMcpHeader(typeFrameType.DISCONNECT)) elif event timer_expired and kwargs.get(timer_id) heartbeat: # 心跳超时发送PING self._start_heartbeat() # 重置心跳定时器 return McpFrame(headerMcpHeader(typeFrameType.PING)) return None def _handle_disconnecting(self, event: str, frame: Optional[McpFrame], **kwargs): # 在DISCONNECTING状态主要等待底层传输关闭或进行一些清理工作 # 收到对端的DISCONNECT如果之前是自己发起的或超时后可以转移到CLOSED if event frame_received and frame and frame.header.type FrameType.DISCONNECT: self._transition_to(ConnectionState.CLOSED) elif event connection_closed: self._transition_to(ConnectionState.CLOSED) return None def _get_next_seq_id(self) - int: seq self.next_seq_id self.next_seq_id 1 # 处理回绕简单示例 if self.next_seq_id 0xFFFFFFFF: self.next_seq_id 1 return seq def _start_heartbeat(self): # 简化设置一个下次心跳触发的时间点 self.heartbeat_timer time.time() 30 # 30秒后发下一次心跳状态机实现的难点与心得状态的纯粹性状态机只负责管理状态和根据事件决定要发送的协议帧。它不应该直接操作网络socket、启动线程或调用复杂的业务逻辑。它通过返回需要发送的McpFrame对象以及通过回调如on_state_change来影响外部世界。这种设计使得状态机本身易于单元测试。定时器的整合在实际系统中定时器管理是个麻烦事。上面的示例用简单的未来时间点来模拟。在真实实现中你需要一个集中的定时器调度器能够添加、取消定时器并在超时时向状态机发送timer_expired事件。定时器ID需要与特定任务如等待某seq_id的ACK、心跳关联。未确认消息的重传pending_acks字典是可靠传输的关键。当状态机在ESTABLISHED状态下收到timer_expired事件且定时器ID对应某个seq_id时就需要从pending_acks中取出该消息帧进行重传并增加重试计数。超过最大重试次数后应将连接置为ERROR状态。这部分逻辑在上例中省略了但它是必须实现的。并发与线程安全如果网络I/O和状态机驱动运行在不同的线程那么对状态机handle_event的调用必须是线程安全的通常需要加锁。一个更清晰的架构是使用单线程的事件循环如asyncio将所有事件网络收到数据、定时器超时、应用层发送请求都放入一个队列由事件循环顺序取出并调用状态机处理。5. 会话管理粘合协议栈与网络I/O状态机和编解码器准备好了我们需要一个“会话”来把它们和真实的网络连接结合起来并管理整个生命周期。这个会话类McpSession是给应用层使用的直接接口。# session.py import socket import threading import queue import time from typing import Optional, Callable from protocol import McpFrame, McpHeader, FrameType from state_machine import McpStateMachine, ConnectionState class McpSession: def __init__(self, sock: socket.socket, is_initiator: bool): self.sock sock self.is_initiator is_initiator self.state_machine McpStateMachine(on_state_changeself._on_state_change) self.send_queue queue.Queue() # 用于存放待发送的Frame self.receive_buffer bytearray() # 接收数据的缓冲区 self.running False self.recv_thread: Optional[threading.Thread] None self.send_thread: Optional[threading.Thread] None self.on_message: Optional[Callable[[bytes], None]] None # 应用层消息回调 def start(self): 启动会话开始处理I/O self.running True # 启动接收线程 self.recv_thread threading.Thread(targetself._recv_loop, daemonTrue) self.recv_thread.start() # 启动发送线程 self.send_thread threading.Thread(targetself._send_loop, daemonTrue) self.send_thread.start() # 如果是发起方客户端主动发送CONNECT if self.is_initiator: connect_frame self.state_machine.handle_event(send_connect) if connect_frame: self.send_queue.put(connect_frame) def send_data(self, data: bytes): 应用层调用此方法发送数据 if self.state_machine.state ! ConnectionState.ESTABLISHED: raise ConnectionError(Connection is not established) frame_to_send self.state_machine.handle_event(send_data, datadata) if frame_to_send: self.send_queue.put(frame_to_send) def disconnect(self): 应用层请求断开连接 frame_to_send self.state_machine.handle_event(send_disconnect) if frame_to_send: self.send_queue.put(frame_to_send) # 设置标志让循环退出 self.running False def _on_state_change(self, old_state: ConnectionState, new_state: ConnectionState): 状态变化回调 print(f[Session] State changed: {old_state} - {new_state}) if new_state ConnectionState.CLOSED or new_state ConnectionState.ERROR: self.running False self.sock.close() def _recv_loop(self): 接收线程循环从socket读数据拼装完整帧交给状态机处理 while self.running: try: # 设置超时以便能响应running标志的变化 self.sock.settimeout(1.0) data self.sock.recv(4096) if not data: # 对端关闭连接 self.state_machine.handle_event(connection_closed) break self.receive_buffer.extend(data) self._process_buffer() except socket.timeout: continue # 超时是正常的继续循环检查running标志 except (ConnectionError, OSError) as e: print(f[RecvLoop] Socket error: {e}) self.state_machine.handle_event(connection_closed) break def _process_buffer(self): 处理接收缓冲区尝试解析出完整帧 while len(self.receive_buffer) 12: # 至少有一个头部的长度 try: # 尝试解析头部获取完整帧长度 header McpHeader.unpack(self.receive_buffer[:12]) total_frame_len 12 header.payload_len if len(self.receive_buffer) total_frame_len: # 缓冲区数据还不够一个完整帧等待下次接收 break # 提取完整帧数据 frame_data bytes(self.receive_buffer[:total_frame_len]) # 从缓冲区移除已处理的数据 del self.receive_buffer[:total_frame_len] # 解码成Frame对象 frame McpFrame.from_bytes(frame_data) # 将接收到的帧交给状态机处理 frame_to_send self.state_machine.handle_event(frame_received, frameframe) # 如果状态机要求发送回复帧如ACK, PONG放入发送队列 if frame_to_send: self.send_queue.put(frame_to_send) # 如果是数据消息传递给应用层回调 if frame.header.type FrameType.DATA and self.on_message: self.on_message(frame.payload) except ValueError as e: # 解析错误协议格式非法转移到错误状态 print(f[ProcessBuffer] Protocol error: {e}) self.state_machine.handle_event(protocol_error) break def _send_loop(self): 发送线程循环从队列取Frame编码后通过socket发送 while self.running: try: frame self.send_queue.get(timeout1.0) frame_bytes frame.to_bytes() # 在实际项目中这里需要考虑TCP粘包问题吗 # 不需要因为to_bytes()已经将整个帧头载荷打包成一个完整的字节流。 # TCP是流式协议保证顺序但不保证消息边界。我们的“消息边界”就是通过“长度字段”自己定义的。 # 接收方_process_buffer中的“拆包”逻辑正是为了解决这个问题。 # 所以这里直接sendall即可。 self.sock.sendall(frame_bytes) except queue.Empty: continue # 队列为空继续循环 except (ConnectionError, OSError) as e: print(f[SendLoop] Socket error: {e}) self.state_machine.handle_event(connection_closed) break会话管理中的核心陷阱与解决方案TCP粘包/半包问题这是网络编程新手最常见的坑。_process_buffer方法是解决这个问题的经典模式。我们永远不能假设一次recv调用返回的数据就是一个完整的应用层报文MCP帧。我们必须维护一个应用层缓冲区不断累积数据并尝试从缓冲区头部解析出完整的帧。McpHeader中的payload_len字段就是我们判断“完整性”的关键。永远基于“长度字段”来拆包而不是依赖特殊分隔符或固定次数recv。双工通信与线程设计我们使用了独立的接收和发送线程。这是一个简单清晰的模型但引入了线程同步问题。这里send_queue是线程安全的queue.Queue用于协调状态机线程主线程或事件线程和发送线程。接收线程将收到的帧交给状态机状态机可能产生需要发送的帧如ACK这个帧也需要放入send_queue。要小心避免死锁和竞态条件。对于更高效的实现可以考虑使用asyncio单线程异步模型。资源清理当连接关闭或出错时必须确保socket被正确关闭线程被正确终止。示例中通过running标志和sock.settimeout来让线程优雅退出。在实际代码中还需要加入join线程等待其结束。_on_state_change回调在进入CLOSED或ERROR状态时关闭socket并停止循环这是一个集中清理的好地方。超时与重传的管理示例中简化了定时器。在真实场景中你需要一个更精细的定时器管理器。例如对于pending_acks中的每条消息都应该有一个独立的超时定时器。当定时器触发时向状态机发送一个携带seq_id的timer_expired事件。状态机检查该消息是否已被确认若未确认则重传。重传次数过多则断开连接。心跳定时器同理。6. 从零构建客户端与服务端的启动与交互演示最后我们编写客户端和服务端的启动脚本将以上所有模块串联起来进行一个简单的演示。服务端Acceptor代码# server.py import socket import threading from session import McpSession def on_server_message(data: bytes): print(f[Server] Received application data: {data.decode(utf-8, errorsignore)}) # 这里可以处理业务逻辑然后回复客户端示例中省略 def handle_client_connection(client_sock: socket.socket, client_addr: tuple): print(f[Server] New connection from {client_addr}) # 服务端是被动接受连接的一方is_initiatorFalse session McpSession(client_sock, is_initiatorFalse) session.on_message on_server_message session.start() # 模拟等待一段时间然后主动发送一条消息给客户端 import time time.sleep(2) if session.state_machine.state ConnectionState.ESTABLISHED: session.send_data(bHello from Server!) # 保持连接一段时间观察心跳等 time.sleep(10) session.disconnect() print(f[Server] Connection with {client_addr} closed.) def start_server(host127.0.0.1, port9999): server_sock socket.socket(socket.AF_INET, socket.SOCK_STREAM) server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server_sock.bind((host, port)) server_sock.listen(5) print(f[Server] Listening on {host}:{port}) try: while True: client_sock, client_addr server_sock.accept() # 为每个客户端连接创建一个新线程处理生产环境建议使用线程池 client_thread threading.Thread(targethandle_client_connection, args(client_sock, client_addr)) client_thread.daemon True client_thread.start() except KeyboardInterrupt: print(\n[Server] Shutting down...) finally: server_sock.close() if __name__ __main__: start_server()客户端Initiator代码# client.py import socket import time from session import McpSession def on_client_message(data: bytes): print(f[Client] Received application data: {data.decode(utf-8, errorsignore)}) def start_client(host127.0.0.1, port9999): sock socket.socket(socket.AF_INET, socket.SOCK_STREAM) try: sock.connect((host, port)) print(f[Client] Connected to {host}:{port}) # 客户端是发起方is_initiatorTrue session McpSession(sock, is_initiatorTrue) session.on_message on_client_message session.start() # 等待连接建立 time.sleep(1) if session.state_machine.state ! ConnectionState.ESTABLISHED: print([Client] Failed to establish MCP connection.) return # 连接建立后发送几条测试消息 for i in range(3): msg fTest message {i1} from client.encode(utf-8) print(f[Client] Sending: {msg.decode()}) session.send_data(msg) time.sleep(1.5) # 间隔发送便于观察 # 等待接收服务端的消息并保持连接以观察心跳 print([Client] Waiting for server messages and heartbeat...) time.sleep(15) # 主动断开连接 session.disconnect() print([Client] Disconnected.) except ConnectionRefusedError: print(f[Client] Could not connect to {host}:{port}) except Exception as e: print(f[Client] Error: {e}) finally: sock.close() if __name__ __main__: start_client()运行与观察首先在终端运行python server.py。然后在另一个终端运行python client.py。观察两个终端的输出。你应该能看到类似以下的过程客户端打印Connected to 127.0.0.1:9999。服务端打印New connection from (127.0.0.1, xxxxx)。双方打印状态变化[Session] State changed: CLOSED - CONNECTING然后- ESTABLISHED。客户端发送三条测试消息。服务端收到消息后打印[Server] Received application data: Test message 1...并在2秒后向客户端发送Hello from Server!。客户端收到服务端消息并打印。在连接空闲期间你会看到状态机定时示例中约30秒触发PING/PONG的日志需要在代码中添加相应打印。最后服务端或客户端断开连接状态变为DISCONNECTING-CLOSED。演示中的关键点验证连接握手客户端发送CONNECT服务端回复CONNACK状态机迁移到ESTABLISHED。可靠数据传输客户端发送的DATA消息带有递增的seq_id。服务端收到后应回复对应的ACK。你可以在_handle_established的ACK处理部分添加日志以确认这一机制工作正常。心跳保活在ESTABLISHED状态空闲一段时间后状态机应触发timer_expired心跳事件发送PING对端回复PONG。有序关闭调用disconnect()会发送DISCONNECT帧状态机进入DISCONNECTING收到对端的DISCONNECT或底层连接关闭事件后进入CLOSED状态。通过这个从协议定义到代码实现再到实际运行的完整示例MCP协议中那些抽象的概念——连接状态、序列号、确认、心跳——都变成了可以观察、可以调试的具体代码逻辑。在实现过程中最深的体会是协议设计决定了代码结构。一个定义良好的状态机是复杂协议实现的基石它能将杂乱的网络事件和业务请求梳理得井井有条。而缓冲区管理和定时器处理则是实现层面最容易出bug的地方需要格外小心。这个示例虽然简化但已经勾勒出了一个可靠应用层协议栈的核心骨架在此基础上增加流量控制、多路复用、加密等特性思路都是相通的。下次当你再阅读其他协议的RFC文档时不妨尝试用这种“状态机编解码会话管理”的框架去理解它并思考如何用代码实现相信会有更深的领悟。