引言：MCP协议与极简实现的必要性

MCP（Minimal Communication Protocol）是一种轻量级应用层协议，旨在通过极简设计实现高效的跨进程通信。在物联网、微服务架构等场景中，开发者常面临协议实现复杂度高、开发周期长的痛点。本文通过Python标准库，演示如何用几行核心代码实现MCP服务端与客户端通信，重点解决协议设计、消息编解码及通信流程三大核心问题。

一、MCP协议设计：极简通信规则

MCP协议采用请求-响应模式，消息格式定义为：

[消息长度(4字节)][消息类型(1字节)][消息体(N字节)]

• 消息长度：4字节无符号整数，表示消息类型+消息体的总字节数
• 消息类型：1字节标识符（0x01=请求，0x02=响应）
• 消息体：可变长度二进制数据

此设计通过固定长度头+可变长度体的方式，兼顾效率与灵活性。Python的struct模块可完美处理这种二进制协议的编解码。

二、服务端实现：5行核心代码解析

import socket, struct
def mcp_server():
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.bind(('0.0.0.0', 9999))
        s.listen()
        conn, addr = s.accept()
        with conn:
            # 接收消息头（4字节长度+1字节类型）
            header = conn.recv(5)
            msg_len, msg_type = struct.unpack('!IB', header)
            # 接收消息体
            body = conn.recv(msg_len - 1)  # 减去已收的1字节类型
            print(f"Received: Type={msg_type}, Body={body}")
            # 构造响应（类型0x02）
            response = struct.pack('!IB', 5, 0x02) + b"OK"
            conn.sendall(response)

关键点解析：

1. socket初始化：使用TCP协议创建监听套接字
2. 消息头解析：struct.unpack('!IB', header)按网络字节序解析5字节头
3. 动态体接收：根据消息头中的长度字段动态接收剩余数据
4. 响应构造：struct.pack('!IB', 5, 0x02)打包响应头，后接消息体

此实现仅需5行核心逻辑代码，即可完成协议解析与响应。

三、客户端实现：4行核心代码解析

import socket, struct
def mcp_client():
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.connect(('127.0.0.1', 9999))
        # 构造请求（类型0x01，体"TEST"）
        request = struct.pack('!IB', 8, 0x01) + b"TEST"
        s.sendall(request)
        # 接收响应头
        header = s.recv(5)
        msg_len, msg_type = struct.unpack('!IB', header)
        # 接收响应体
        body = s.recv(msg_len - 1)
        print(f"Response: Type={msg_type}, Body={body}")

关键点解析：

1. 请求构造：struct.pack('!IB', 8, 0x01)打包请求头（总长度8字节，类型0x01）
2. 分步接收：先收5字节头，再根据头中长度收剩余体
3. 类型安全：通过msg_type字段区分请求/响应

客户端实现仅需4行核心逻辑，即可完成完整通信流程。

四、协议扩展与优化建议

1. 消息类型扩展

可通过扩展msg_type字段实现多命令支持：

CMD_ECHO = 0x01
CMD_AUTH = 0x02
# 服务端处理分支
if msg_type == CMD_ECHO:
    response = struct.pack('!IB', 5, CMD_ECHO) + body

2. 错误处理增强

添加超时与重试机制：

from socket import timeout
try:
    s.settimeout(2.0)
    header = s.recv(5)
except timeout:
    print("Connection timeout")

3. 性能优化

• 使用socket.makefile()缓冲IO
• 对大消息体采用分块传输
• 实现连接池管理

五、完整示例与测试

服务端完整代码

import socket, struct
def mcp_server():
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        s.bind(('0.0.0.0', 9999))
        s.listen()
        print("Server listening on port 9999")
        conn, addr = s.accept()
        with conn:
            try:
                header = conn.recv(5)
                if len(header) != 5:
                    raise ValueError("Invalid header")
                msg_len, msg_type = struct.unpack('!IB', header)
                body = conn.recv(msg_len - 1)
                print(f"Received from {addr}: Type={msg_type}, Body={body.decode()}")
                response = struct.pack('!IB', 5, 0x02) + b"OK"
                conn.sendall(response)
            except Exception as e:
                print(f"Error: {e}")

客户端完整代码

import socket, struct
def mcp_client(message):
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.connect(('127.0.0.1', 9999))
        request = struct.pack('!IB', len(message)+5, 0x01) + message.encode()
        s.sendall(request)
        header = s.recv(5)
        if len(header) != 5:
            raise ValueError("Invalid response header")
        msg_len, msg_type = struct.unpack('!IB', header)
        body = s.recv(msg_len - 1)
        print(f"Server response: {body.decode()}")
# 测试
if __name__ == "__main__":
    import threading
    # 启动服务端线程
    server_thread = threading.Thread(target=mcp_server)
    server_thread.daemon = True
    server_thread.start()
    # 客户端发送测试消息
    mcp_client("Hello MCP")

六、应用场景与价值

1. 物联网设备通信：轻量级协议适合资源受限设备
2. 微服务间调用：替代HTTP实现内部服务高效通信
3. 教学演示：快速展示网络编程核心概念
4. 协议验证：快速实现协议原型进行功能测试

结论：极简代码的深层价值

本文通过10行核心代码（服务端5行+客户端4行+协议定义1行）实现了完整的MCP通信，展示了：

1. 协议设计的核心原则：固定头+可变体
2. Python标准库的强大能力：socket+struct组合
3. 网络编程的本质：字节流处理与状态管理

这种极简实现不仅降低了开发门槛，更为复杂协议设计提供了可扩展的基础框架。开发者可在此基础上快速构建生产级通信系统。