Demo MCP

好的，这是一个使用 MCP 协议的简单客户端和服务器文件示例： **服务器 (server.py):** ```python import socket import struct # MCP 协议常量 MCP_MAGIC = 0x4D4350 # "MCP" 的 ASCII 码 MCP_VERSION = 1 MCP_STATUS_OK = 0 MCP_STATUS_ERROR = 1 def handle_client(conn, addr): """处理客户端连接.""" print(f"连接来自: {addr}") try: # 1. 读取头部 (魔数, 版本, 命令) header_data = conn.recv(8) # 4 字节魔数 + 1 字节版本 + 3 字节命令 if len(header_data) != 8: print("头部读取失败或连接关闭") return magic, version, command = struct.unpack("!I B 3s", header_data) if magic != MCP_MAGIC: print("无效的魔数") return if version != MCP_VERSION: print("不支持的版本") return command = command.decode('utf-8').strip('\x00') # 解码命令并移除填充的空字节 print(f"收到命令: {command}") # 2. 处理命令 if command == "PING": # 响应 PING 命令 response_data = "PONG".encode('utf-8') response_length = len(response_data) # 构建响应头部 (状态, 数据长度) response_header = struct.pack("!B I", MCP_STATUS_OK, response_length) # 发送响应 conn.sendall(response_header + response_data) print("发送 PONG 响应") elif command == "ECHO": # 读取数据长度 length_data = conn.recv(4) if len(length_data) != 4: print("数据长度读取失败") return data_length = struct.unpack("!I", length_data)[0] # 读取数据 data = conn.recv(data_length) if len(data) != data_length: print("数据读取失败") return received_message = data.decode('utf-8') print(f"收到消息: {received_message}") # 构建响应 (ECHO 命令返回相同的数据) response_data = received_message.encode('utf-8') response_length = len(response_data) # 构建响应头部 (状态, 数据长度) response_header = struct.pack("!B I", MCP_STATUS_OK, response_length) # 发送响应 conn.sendall(response_header + response_data) print("发送 ECHO 响应") else: print(f"未知命令: {command}") # 构建错误响应 error_message = "未知命令".encode('utf-8') response_length = len(error_message) response_header = struct.pack("!B I", MCP_STATUS_ERROR, response_length) conn.sendall(response_header + error_message) except Exception as e: print(f"处理客户端时发生错误: {e}") finally: conn.close() print("连接关闭") def main(): """主函数.""" host = "127.0.0.1" # 监听所有接口 port = 12345 # 使用一个未使用的端口 server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server_socket.bind((host, port)) server_socket.listen(5) # 允许 5 个连接排队 print(f"服务器监听在 {host}:{port}") try: while True: conn, addr = server_socket.accept() handle_client(conn, addr) except KeyboardInterrupt: print("服务器关闭") finally: server_socket.close() if __name__ == "__main__": main() ``` **客户端 (client.py):** ```python import socket import struct # MCP 协议常量 (与服务器相同) MCP_MAGIC = 0x4D4350 MCP_VERSION = 1 MCP_STATUS_OK = 0 MCP_STATUS_ERROR = 1 def send_mcp_message(host, port, command, data=None): """发送 MCP 消息并接收响应.""" try: client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) client_socket.connect((host, port)) # 1. 构建头部 command_bytes = command.encode('utf-8') header = struct.pack("!I B 3s", MCP_MAGIC, MCP_VERSION, command_bytes) # 2. 发送头部 client_socket.sendall(header) # 3. 发送数据 (如果存在) if data: data_bytes = data.encode('utf-8') data_length = len(data_bytes) length_header = struct.pack("!I", data_length) client_socket.sendall(length_header + data_bytes) # 4. 接收响应头部 (状态, 数据长度) response_header = client_socket.recv(5) # 1 字节状态 + 4 字节长度 if len(response_header) != 5: print("响应头部读取失败") return None status, data_length = struct.unpack("!B I", response_header) # 5. 接收响应数据 response_data = client_socket.recv(data_length) if len(response_data) != data_length: print("响应数据读取失败") return None response_message = response_data.decode('utf-8') if status == MCP_STATUS_OK: print(f"服务器响应: {response_message}") return response_message else: print(f"服务器错误: {response_message}") return None except Exception as e: print(f"发生错误: {e}") return None finally: client_socket.close() def main(): """主函数.""" host = "127.0.0.1" port = 12345 # 发送 PING 命令 send_mcp_message(host, port, "PING") # 发送 ECHO 命令 message = "Hello, MCP Server!" send_mcp_message(host, port, "ECHO", message) # 发送未知命令 send_mcp_message(host, port, "UNKNOWN") if __name__ == "__main__": main() ``` **如何运行:** 1. **保存:** 将代码分别保存为 `server.py` 和 `client.py`。 2. **运行服务器:** 在终端中运行 `python server.py`。 3. **运行客户端:** 在另一个终端中运行 `python client.py`。 **代码解释:** * **MCP 协议:** 这个例子定义了一个简单的 MCP 协议，包含： * **魔数 (Magic Number):** `0x4D4350` (ASCII 码 "MCP")，用于验证消息是否为 MCP 消息。 * **版本 (Version):** `1`，用于支持协议的未来版本。 * **命令 (Command):** 一个 3 字节的字符串，用于指定服务器要执行的操作。 * **状态 (Status):** 一个字节，表示操作是否成功 (`MCP_STATUS_OK` 或 `MCP_STATUS_ERROR`)。 * **数据长度 (Data Length):** 一个 4 字节的整数，表示数据的长度。 * **数据 (Data):** 实际的数据负载。 * **`struct` 模块:** 用于将数据打包成二进制格式，以便通过网络发送。 `!I` 表示网络字节序 (大端) 的无符号整数，`!B` 表示网络字节序的无符号字节。 * **服务器 (`server.py`):** * 监听指定的端口。 * 接受客户端连接。 * 读取 MCP 头部 (魔数, 版本, 命令)。 * 根据命令执行相应的操作 (PING, ECHO)。 * 构建并发送响应。 * 处理错误情况。 * **客户端 (`client.py`):** * 连接到服务器。 * 构建 MCP 头部和数据 (如果需要)。 * 发送消息。 * 接收响应。 * 解析响应。 * 打印结果。 **关键点:** * **字节序 (Endianness):** 使用 `!` 指定网络字节序 (大端) 以确保不同架构的机器可以正确地解释数据。 * **错误处理:** 包含基本的错误处理，例如检查魔数、版本和数据长度。 * **命令处理:** 服务器根据接收到的命令执行不同的操作。 * **数据打包和解包:** 使用 `struct` 模块将数据打包成二进制格式，以便通过网络发送，并在接收时解包。 * **编码:** 使用 UTF-8 编码来处理字符串数据。 **改进方向:** * **更健壮的错误处理:** 添加更详细的错误处理和日志记录。 * **线程/异步处理:** 使用线程或异步编程来处理多个客户端连接。 * **更复杂的命令:** 添加更多的命令和功能。 * **加密:** 使用 SSL/TLS 加密连接以提高安全性。 * **数据验证:** 对接收到的数据进行验证，以防止恶意攻击。 这个例子提供了一个使用 MCP 协议进行通信的基本框架。 你可以根据自己的需要进行修改和扩展。 **中文翻译:** 好的，这是一个使用 MCP 协议的简单客户端和服务器文件示例： **服务器 (server.py):** ```python import socket import struct # MCP 协议常量 MCP_MAGIC = 0x4D4350 # "MCP" 的 ASCII 码 MCP_VERSION = 1 MCP_STATUS_OK = 0 MCP_STATUS_ERROR = 1 def handle_client(conn, addr): """处理客户端连接.""" print(f"连接来自: {addr}") try: # 1. 读取头部 (魔数, 版本, 命令) header_data = conn.recv(8) # 4 字节魔数 + 1 字节版本 + 3 字节命令 if len(header_data) != 8: print("头部读取失败或连接关闭") return magic, version, command = struct.unpack("!I B 3s", header_data) if magic != MCP_MAGIC: print("无效的魔数") return if version != MCP_VERSION: print("不支持的版本") return command = command.decode('utf-8').strip('\x00') # 解码命令并移除填充的空字节 print(f"收到命令: {command}") # 2. 处理命令 if command == "PING": # 响应 PING 命令 response_data = "PONG".encode('utf-8') response_length = len(response_data) # 构建响应头部 (状态, 数据长度) response_header = struct.pack("!B I", MCP_STATUS_OK, response_length) # 发送响应 conn.sendall(response_header + response_data) print("发送 PONG 响应") elif command == "ECHO": # 读取数据长度 length_data = conn.recv(4) if len(length_data) != 4: print("数据长度读取失败") return data_length = struct.unpack("!I", length_data)[0] # 读取数据 data = conn.recv(data_length) if len(data) != data_length: print("数据读取失败") return received_message = data.decode('utf-8') print(f"收到消息: {received_message}") # 构建响应 (ECHO 命令返回相同的数据) response_data = received_message.encode('utf-8') response_length = len(response_data) # 构建响应头部 (状态, 数据长度) response_header = struct.pack("!B I", MCP_STATUS_OK, response_length) # 发送响应 conn.sendall(response_header + response_data) print("发送 ECHO 响应") else: print(f"未知命令: {command}") # 构建错误响应 error_message = "未知命令".encode('utf-8') response_length = len(error_message) response_header = struct.pack("!B I", MCP_STATUS_ERROR, response_length) conn.sendall(response_header + error_message) except Exception as e: print(f"处理客户端时发生错误: {e}") finally: conn.close() print("连接关闭") def main(): """主函数.""" host = "127.0.0.1" # 监听所有接口 port = 12345 # 使用一个未使用的端口 server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server_socket.bind((host, port)) server_socket.listen(5) # 允许 5 个连接排队 print(f"服务器监听在 {host}:{port}") try: while True: conn, addr = server_socket.accept() handle_client(conn, addr) except KeyboardInterrupt: print("服务器关闭") finally: server_socket.close() if __name__ == "__main__": main() ``` **客户端 (client.py):** ```python import socket import struct # MCP 协议常量 (与服务器相同) MCP_MAGIC = 0x4D4350 MCP_VERSION = 1 MCP_STATUS_OK = 0 MCP_STATUS_ERROR = 1 def send_mcp_message(host, port, command, data=None): """发送 MCP 消息并接收响应.""" try: client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) client_socket.connect((host, port)) # 1. 构建头部 command_bytes = command.encode('utf-8') header = struct.pack("!I B 3s", MCP_MAGIC, MCP_VERSION, command_bytes) # 2. 发送头部 client_socket.sendall(header) # 3. 发送数据 (如果存在) if data: data_bytes = data.encode('utf-8') data_length = len(data_bytes) length_header = struct.pack("!I", data_length) client_socket.sendall(length_header + data_bytes) # 4. 接收响应头部 (状态, 数据长度) response_header = client_socket.recv(5) # 1 字节状态 + 4 字节长度 if len(response_header) != 5: print("响应头部读取失败") return None status, data_length = struct.unpack("!B I", response_header) # 5. 接收响应数据 response_data = client_socket.recv(data_length) if len(response_data) != data_length: print("响应数据读取失败") return None response_message = response_data.decode('utf-8') if status == MCP_STATUS_OK: print(f"服务器响应: {response_message}") return response_message else: print(f"服务器错误: {response_message}") return None except Exception as e: print(f"发生错误: {e}") return None finally: client_socket.close() def main(): """主函数.""" host = "127.0.0.1" port = 12345 # 发送 PING 命令 send_mcp_message(host, port, "PING") # 发送 ECHO 命令 message = "Hello, MCP Server!" send_mcp_message(host, port, "ECHO", message) # 发送未知命令 send_mcp_message(host, port, "UNKNOWN") if __name__ == "__main__": main() ``` **如何运行:** 1. **保存:** 将代码分别保存为 `server.py` 和 `client.py`。 2. **运行服务器:** 在终端中运行 `python server.py`。 3. **运行客户端:** 在另一个终端中运行 `python client.py`。 **代码解释:** * **MCP 协议:** 这个例子定义了一个简单的 MCP 协议，包含： * **魔数 (Magic Number):** `0x4D4350` (ASCII 码 "MCP")，用于验证消息是否为 MCP 消息。 * **版本 (Version):** `1`，用于支持协议的未来版本。 * **命令 (Command):** 一个 3 字节的字符串，用于指定服务器要执行的操作。 * **状态 (Status):** 一个字节，表示操作是否成功 (`MCP_STATUS_OK` 或 `MCP_STATUS_ERROR`)。 * **数据长度 (Data Length):** 一个 4 字节的整数，表示数据的长度。 * **数据 (Data):** 实际的数据负载。 * **`struct` 模块:** 用于将数据打包成二进制格式，以便通过网络发送。 `!I` 表示网络字节序 (大端) 的无符号整数，`!B` 表示网络字节序的无符号字节。 * **服务器 (`server.py`):** * 监听指定的端口。 * 接受客户端连接。 * 读取 MCP 头部 (魔数, 版本, 命令)。 * 根据命令执行相应的操作 (PING, ECHO)。 * 构建并发送响应。 * 处理错误情况。 * **客户端 (`client.py`):** * 连接到服务器。 * 构建 MCP 头部和数据 (如果需要)。 * 发送消息。 * 接收响应。 * 解析响应。 * 打印结果。 **关键点:** * **字节序 (Endianness):** 使用 `!` 指定网络字节序 (大端) 以确保不同架构的机器可以正确地解释数据。 * **错误处理:** 包含基本的错误处理，例如检查魔数、版本和数据长度。 * **命令处理:** 服务器根据接收到的命令执行不同的操作。 * **数据打包和解包:** 使用 `struct` 模块将数据打包成二进制格式，以便通过网络发送，并在接收时解包。 * **编码:** 使用 UTF-8 编码来处理字符串数据。 **改进方向:** * **更健壮的错误处理:** 添加更详细的错误处理和日志记录。 * **线程/异步处理:** 使用线程或异步编程来处理多个客户端连接。 * **更复杂的命令:** 添加更多的命令和功能。 * **加密:** 使用 SSL/TLS 加密连接以提高安全性。 * **数据验证:** 对接收到的数据进行验证，以防止恶意攻击。 这个例子提供了一个使用 MCP 协议进行通信的基本框架。 你可以根据自己的需要进行修改和扩展。 **总结:** 这段代码演示了一个简单的客户端-服务器架构，使用自定义的 MCP 协议进行通信。 它展示了如何使用 `socket` 模块进行网络编程，以及如何使用 `struct` 模块打包和解包二进制数据。 这个例子可以作为学习网络编程和协议设计的起点。