SQLite Database Demo

好的，以下是一些在模型上下文协议 (Model Context Protocol, MCP) 中构建服务器、客户端测试的示例。由于 MCP 是一个比较宽泛的概念，我将提供一些通用的示例，并假设你指的是一种基于消息传递的协议，用于在服务器和客户端之间传递模型相关的信息。 **假设场景：** 我们假设有一个简单的场景，服务器提供一个模型推理服务，客户端发送数据给服务器，服务器使用模型进行推理，并将结果返回给客户端。 **1. 服务器端 (Server-side)** * **Python (使用 `socket` 模块):** ```python import socket import json # 假设的模型推理函数 def infer_model(data): # 模拟模型推理过程 # 这里可以替换成你的实际模型推理代码 result = {"prediction": data["input"] * 2} # 简单示例：将输入乘以2 return result def handle_client(conn, addr): print(f"Connected by {addr}") try: while True: data = conn.recv(1024) # 接收数据，最大1024字节 if not data: break # 客户端断开连接 try: # 尝试将接收到的数据解析为 JSON request = json.loads(data.decode('utf-8')) print(f"Received: {request}") # 调用模型推理函数 response = infer_model(request) # 将结果转换为 JSON 并发送回客户端 conn.sendall(json.dumps(response).encode('utf-8')) print(f"Sent: {response}") except json.JSONDecodeError: print("Invalid JSON received.") conn.sendall(b'{"error": "Invalid JSON"}') except Exception as e: print(f"Error processing request: {e}") conn.sendall(b'{"error": "Internal Server Error"}') except ConnectionResetError: print(f"Connection reset by {addr}") finally: conn.close() print(f"Connection closed with {addr}") def start_server(host='localhost', port=12345): with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.bind((host, port)) s.listen() print(f"Server listening on {host}:{port}") while True: conn, addr = s.accept() handle_client(conn, addr) if __name__ == "__main__": start_server() ``` * **代码解释:** * `infer_model(data)`: 模拟模型推理过程，你需要替换成你实际的模型推理代码。 * `handle_client(conn, addr)`: 处理客户端连接，接收数据，调用模型推理，并将结果返回给客户端。 * `start_server(host, port)`: 启动服务器，监听指定端口。 * 使用 `json` 模块进行数据的序列化和反序列化，方便传递复杂的数据结构。 * 异常处理，确保服务器的健壮性。 **2. 客户端 (Client-side)** * **Python (使用 `socket` 模块):** ```python import socket import json def send_request(host='localhost', port=12345, data=None): with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: try: s.connect((host, port)) print(f"Connected to {host}:{port}") # 将数据转换为 JSON 并发送 message = json.dumps(data).encode('utf-8') s.sendall(message) print(f"Sent: {data}") # 接收服务器的响应 response = s.recv(1024) if response: try: response_data = json.loads(response.decode('utf-8')) print(f"Received: {response_data}") return response_data except json.JSONDecodeError: print("Invalid JSON received from server.") return None else: print("No data received from server.") return None except ConnectionRefusedError: print(f"Connection refused by {host}:{port}") return None except Exception as e: print(f"An error occurred: {e}") return None if __name__ == "__main__": # 示例数据 request_data = {"input": 10} # 发送请求并接收响应 response = send_request(data=request_data) if response: print(f"Final Result: {response}") ``` * **代码解释:** * `send_request(host, port, data)`: 连接到服务器，发送数据，接收响应。 * 使用 `json` 模块进行数据的序列化和反序列化。 * 异常处理，处理连接错误和数据接收错误。 **3. 测试 (Testing)** * **手动测试:** 1. 先运行服务器端代码。 2. 再运行客户端代码，观察客户端的输出，以及服务器端的日志。 3. 可以修改客户端的 `request_data`，测试不同的输入。 * **自动化测试 (使用 `unittest` 模块):** ```python import unittest import socket import json import threading import time # 导入服务器端代码 (假设保存在 server.py 文件中) import server class ServerClientTest(unittest.TestCase): @classmethod def setUpClass(cls): # 启动服务器 (在后台线程中) cls.server_thread = threading.Thread(target=server.start_server, daemon=True) cls.server_thread.start() time.sleep(0.5) # 等待服务器启动 @classmethod def tearDownClass(cls): # 关闭服务器 (这里需要一种关闭服务器的机制，例如设置一个全局变量) # 注意：这个例子中没有实现优雅的关闭服务器，需要根据你的实际情况修改 pass def test_valid_request(self): # 测试发送有效请求 request_data = {"input": 5} response = self.send_request(data=request_data) self.assertIsNotNone(response) self.assertEqual(response["prediction"], 10) # 期望结果 def test_invalid_request(self): # 测试发送无效请求 (例如，非 JSON 数据) with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.connect(('localhost', 12345)) s.sendall(b"This is not JSON") response = s.recv(1024) self.assertIn(b"Invalid JSON", response) def send_request(self, host='localhost', port=12345, data=None): # 客户端代码 (与上面的客户端代码类似) with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: try: s.connect((host, port)) message = json.dumps(data).encode('utf-8') s.sendall(message) response = s.recv(1024) if response: return json.loads(response.decode('utf-8')) else: return None except Exception as e: print(f"An error occurred: {e}") return None if __name__ == '__main__': unittest.main() ``` * **代码解释:** * `setUpClass()`: 在所有测试用例运行之前启动服务器。 * `tearDownClass()`: 在所有测试用例运行之后关闭服务器。 **注意：** 这个例子中没有实现优雅的关闭服务器，需要根据你的实际情况修改。 通常，你需要一种机制来通知服务器停止监听连接。 * `test_valid_request()`: 测试发送有效的 JSON 请求，并验证服务器返回的结果是否符合预期。 * `test_invalid_request()`: 测试发送无效的 JSON 请求，并验证服务器是否返回错误信息。 * `send_request()`: 与上面的客户端代码相同，用于发送请求并接收响应。 **重要注意事项:** * **错误处理:** 在实际应用中，需要更完善的错误处理机制，例如重试机制、超时处理等。 * **数据格式:** 根据你的实际需求，选择合适的数据格式，例如 JSON, Protocol Buffers, Avro 等。 * **并发处理:** 如果服务器需要处理大量的并发请求，需要考虑使用多线程、多进程或异步编程。 * **安全性:** 如果服务器需要处理敏感数据，需要考虑安全性问题，例如身份验证、授权、加密等。 * **优雅关闭服务器:** 在测试和生产环境中，都需要一种优雅关闭服务器的机制，避免数据丢失或连接错误。 这通常涉及到向服务器发送一个信号，让它停止监听连接，并等待所有正在处理的请求完成。 **总结:** 这些示例提供了一个基本的框架，你可以根据你的实际需求进行修改和扩展。 关键在于理解 MCP 的核心思想：通过定义明确的消息格式和协议，实现服务器和客户端之间的可靠通信。 在实际应用中，你需要根据你的具体场景选择合适的技术和工具，并进行充分的测试，确保系统的稳定性和可靠性。 **中文翻译:** 好的，以下是一些在模型上下文协议 (Model Context Protocol, MCP) 中构建服务器、客户端测试的示例。由于 MCP 是一个比较宽泛的概念，我将提供一些通用的示例，并假设你指的是一种基于消息传递的协议，用于在服务器和客户端之间传递模型相关的信息。 **假设场景：** 我们假设有一个简单的场景，服务器提供一个模型推理服务，客户端发送数据给服务器，服务器使用模型进行推理，并将结果返回给客户端。 **1. 服务器端 (Server-side)** * **Python (使用 `socket` 模块):** ```python import socket import json # 假设的模型推理函数 def infer_model(data): # 模拟模型推理过程 # 这里可以替换成你的实际模型推理代码 result = {"prediction": data["input"] * 2} # 简单示例：将输入乘以2 return result def handle_client(conn, addr): print(f"Connected by {addr}") try: while True: data = conn.recv(1024) # 接收数据，最大1024字节 if not data: break # 客户端断开连接 try: # 尝试将接收到的数据解析为 JSON request = json.loads(data.decode('utf-8')) print(f"Received: {request}") # 调用模型推理函数 response = infer_model(request) # 将结果转换为 JSON 并发送回客户端 conn.sendall(json.dumps(response).encode('utf-8')) print(f"Sent: {response}") except json.JSONDecodeError: print("Invalid JSON received.") conn.sendall(b'{"error": "Invalid JSON"}') except Exception as e: print(f"Error processing request: {e}") conn.sendall(b'{"error": "Internal Server Error"}') except ConnectionResetError: print(f"Connection reset by {addr}") finally: conn.close() print(f"Connection closed with {addr}") def start_server(host='localhost', port=12345): with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.bind((host, port)) s.listen() print(f"Server listening on {host}:{port}") while True: conn, addr = s.accept() handle_client(conn, addr) if __name__ == "__main__": start_server() ``` * **代码解释:** * `infer_model(data)`: 模拟模型推理过程，你需要替换成你实际的模型推理代码。 * `handle_client(conn, addr)`: 处理客户端连接，接收数据，调用模型推理，并将结果返回给客户端。 * `start_server(host, port)`: 启动服务器，监听指定端口。 * 使用 `json` 模块进行数据的序列化和反序列化，方便传递复杂的数据结构。 * 异常处理，确保服务器的健壮性。 **2. 客户端 (Client-side)** * **Python (使用 `socket` 模块):** ```python import socket import json def send_request(host='localhost', port=12345, data=None): with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: try: s.connect((host, port)) print(f"Connected to {host}:{port}") # 将数据转换为 JSON 并发送 message = json.dumps(data).encode('utf-8') s.sendall(message) print(f"Sent: {data}") # 接收服务器的响应 response = s.recv(1024) if response: try: response_data = json.loads(response.decode('utf-8')) print(f"Received: {response_data}") return response_data except json.JSONDecodeError: print("Invalid JSON received from server.") return None else: print("No data received from server.") return None except ConnectionRefusedError: print(f"Connection refused by {host}:{port}") return None except Exception as e: print(f"An error occurred: {e}") return None if __name__ == "__main__": # 示例数据 request_data = {"input": 10} # 发送请求并接收响应 response = send_request(data=request_data) if response: print(f"Final Result: {response}") ``` * **代码解释:** * `send_request(host, port, data)`: 连接到服务器，发送数据，接收响应。 * 使用 `json` 模块进行数据的序列化和反序列化。 * 异常处理，处理连接错误和数据接收错误。 **3. 测试 (Testing)** * **手动测试:** 1. 先运行服务器端代码。 2. 再运行客户端代码，观察客户端的输出，以及服务器端的日志。 3. 可以修改客户端的 `request_data`，测试不同的输入。 * **自动化测试 (使用 `unittest` 模块):** ```python import unittest import socket import json import threading import time # 导入服务器端代码 (假设保存在 server.py 文件中) import server class ServerClientTest(unittest.TestCase): @classmethod def setUpClass(cls): # 启动服务器 (在后台线程中) cls.server_thread = threading.Thread(target=server.start_server, daemon=True) cls.server_thread.start() time.sleep(0.5) # 等待服务器启动 @classmethod def tearDownClass(cls): # 关闭服务器 (这里需要一种关闭服务器的机制，例如设置一个全局变量) # 注意：这个例子中没有实现优雅的关闭服务器，需要根据你的实际情况修改 pass def test_valid_request(self): # 测试发送有效请求 request_data = {"input": 5} response = self.send_request(data=request_data) self.assertIsNotNone(response) self.assertEqual(response["prediction"], 10) # 期望结果 def test_invalid_request(self): # 测试发送无效请求 (例如，非 JSON 数据) with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.connect(('localhost', 12345)) s.sendall(b"This is not JSON") response = s.recv(1024) self.assertIn(b"Invalid JSON", response) def send_request(self, host='localhost', port=12345, data=None): # 客户端代码 (与上面的客户端代码类似) with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: try: s.connect((host, port)) message = json.dumps(data).encode('utf-8') s.sendall(message) response = s.recv(1024) if response: return json.loads(response.decode('utf-8')) else: return None except Exception as e: print(f"An error occurred: {e}") return None if __name__ == '__main__': unittest.main() ``` * **代码解释:** * `setUpClass()`: 在所有测试用例运行之前启动服务器。 * `tearDownClass()`: 在所有测试用例运行之后关闭服务器。 **注意：** 这个例子中没有实现优雅的关闭服务器，需要根据你的实际情况修改。 通常，你需要一种机制来通知服务器停止监听连接。 * `test_valid_request()`: 测试发送有效的 JSON 请求，并验证服务器返回的结果是否符合预期。 * `test_invalid_request()`: 测试发送无效的 JSON 请求，并验证服务器是否返回错误信息。 * `send_request()`: 与上面的客户端代码相同，用于发送请求并接收响应。 **重要注意事项:** * **错误处理:** 在实际应用中，需要更完善的错误处理机制，例如重试机制、超时处理等。 * **数据格式:** 根据你的实际需求，选择合适的数据格式，例如 JSON, Protocol Buffers, Avro 等。 * **并发处理:** 如果服务器需要处理大量的并发请求，需要考虑使用多线程、多进程或异步编程。 * **安全性:** 如果服务器需要处理敏感数据，需要考虑安全性问题，例如身份验证、授权、加密等。 * **优雅关闭服务器:** 在测试和生产环境中，都需要一种优雅关闭服务器的机制，避免数据丢失或连接错误。 这通常涉及到向服务器发送一个信号，让它停止监听连接，并等待所有正在处理的请求完成。 **总结:** 这些示例提供了一个基本的框架，你可以根据你的实际需求进行修改和扩展。 关键在于理解 MCP 的核心思想：通过定义明确的消息格式和协议，实现服务器和客户端之间的可靠通信。 在实际应用中，你需要根据你的具体场景选择合适的技术和工具，并进行充分的测试，确保系统的稳定性和可靠性。