HiMCP

Bio-OS MCP Server

Bio-OS 实例平台的 MCP 服务器。

Spring MCP Bridge

Spring MCP Bridge 是一个工具，它可以自动将 Spring Boot 应用程序中的 REST 端点转换为 MCP（消息会话协议）服务器，从而允许像 Claude、Cursor 和其他 MCP 兼容工具这样的人工智能助手直接与您的 API 进行交互。

Comfy MCP Pipeline

这是一个用于 Open WebUI 的 comfy-mcp-server 的管道包装器。

cosense-mcp-server MCP Server

镜子 (jìng zi)

Simple MCP MySQL Server

一个轻量级的 MySQL MCP 服务器，使用 Python 构建。

mcp-server-yuque MCP Server

Wait MCP Server

测试 MCP 服务器以进行等待。 (Cèshì MCP fúwùqì yǐ jìnxíng děngdài.)

OpenAPI to MCP Server Converter

一个自动将 OpenAPI 规范转换为模型上下文协议 (MCP) 服务器的工具

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 的核心思想：通过定义明确的消息格式和协议，实现服务器和客户端之间的可靠通信。在实际应用中，你需要根据你的具体场景选择合适的技术和工具，并进行充分的测试，确保系统的稳定性和可靠性。

MCP Server Demo 项目文档

好的，这是 "一个基于Spring Boot 的MCP（Model Control Protocol）服务示例项目" 的中文翻译： **一个基于 Spring Boot 的 MCP (模型控制协议) 服务示例项目** This translation is quite literal and accurate. It maintains the technical terms in English as they are commonly used in Chinese technical contexts.