Minimal MCP Server

好的，以下是一个 Model Context Protocol (MCP) 服务器的最小实现，用 Python 编写，并使用 `asyncio` 库进行异步操作。 这个例子展示了如何监听连接，接收消息，并发送简单的响应。 ```python import asyncio import json async def handle_client(reader, writer): """处理单个客户端连接.""" addr = writer.get_extra_info('peername') print(f"连接来自 {addr}") try: while True: data = await reader.read(1024) # 读取最多 1024 字节 if not data: break message = data.decode() print(f"接收到消息: {message}") try: # 尝试解析 JSON request = json.loads(message) # 在这里添加你的 MCP 逻辑 # 例如，根据请求类型执行不同的操作 if "method" in request: method = request["method"] if method == "ping": response = {"result": "pong"} elif method == "get_model_info": response = {"result": {"model_name": "MyModel", "version": "1.0"}} # 示例模型信息 else: response = {"error": "Unknown method"} else: response = {"error": "Invalid request"} response_json = json.dumps(response) writer.write(response_json.encode()) await writer.drain() # 刷新缓冲区 print(f"发送响应: {response_json}") except json.JSONDecodeError: print("接收到无效的 JSON") writer.write(b'{"error": "Invalid JSON"}') await writer.drain() except ConnectionError as e: print(f"连接错误: {e}") finally: print(f"关闭连接 {addr}") writer.close() await writer.wait_closed() async def main(): """启动服务器.""" server = await asyncio.start_server( handle_client, '127.0.0.1', 8888) # 监听本地地址 127.0.0.1 端口 8888 addr = server.sockets[0].getsockname() print(f'服务于 {addr}') async with server: await server.serve_forever() if __name__ == "__main__": asyncio.run(main()) ``` **代码解释:** 1. **`handle_client(reader, writer)` 函数:** - 这是处理每个客户端连接的主要函数。 - `reader` 和 `writer` 是 `asyncio.StreamReader` 和 `asyncio.StreamWriter` 对象，用于读取和写入数据。 - `writer.get_extra_info('peername')` 获取客户端的地址。 - `reader.read(1024)` 从客户端读取最多 1024 字节的数据。 - `data.decode()` 将接收到的字节数据解码为字符串。 - `json.loads(message)` 尝试将接收到的消息解析为 JSON 对象。 - **MCP 逻辑:** 这是代码的核心部分，你需要根据 MCP 协议的要求实现你的逻辑。 在这个例子中，它检查请求中是否存在 "method" 字段，并根据方法名称返回不同的响应。 `ping` 方法返回 `pong`，`get_model_info` 返回一个示例模型信息。 - `json.dumps(response)` 将响应转换为 JSON 字符串。 - `writer.write(response_json.encode())` 将 JSON 字符串编码为字节数据并发送给客户端。 - `await writer.drain()` 刷新缓冲区，确保数据被发送。 - `writer.close()` 关闭连接。 - `await writer.wait_closed()` 等待连接完全关闭。 - 异常处理：包括 `json.JSONDecodeError` 处理无效的 JSON，以及 `ConnectionError` 处理连接错误。 2. **`main()` 函数:** - `asyncio.start_server(handle_client, '127.0.0.1', 8888)` 启动一个 TCP 服务器，监听本地地址 127.0.0.1 的 8888 端口。 `handle_client` 函数将处理每个新的客户端连接。 - `server.serve_forever()` 运行服务器，直到手动停止。 3. **`if __name__ == "__main__":` 块:** - 确保 `asyncio.run(main())` 只在脚本直接运行时执行，而不是作为模块导入时执行。 **如何运行:** 1. **保存代码:** 将代码保存为 `mcp_server.py`。 2. **运行脚本:** 在终端中运行 `python mcp_server.py`。 **如何测试:** 你可以使用 `telnet` 或 `netcat` 等工具来连接服务器并发送 MCP 请求。 例如： ```bash telnet 127.0.0.1 8888 ``` 然后，你可以发送一个 JSON 请求，例如： ```json {"method": "ping"} ``` 服务器应该返回： ```json {"result": "pong"} ``` 或者发送： ```json {"method": "get_model_info"} ``` 服务器应该返回： ```json {"result": {"model_name": "MyModel", "version": "1.0"}} ``` **重要注意事项:** * **错误处理:** 这个例子只包含基本的错误处理。 在实际应用中，你需要添加更完善的错误处理机制，例如记录错误日志，返回更详细的错误信息给客户端。 * **安全性:** 这个例子没有考虑安全性。 在生产环境中，你需要采取安全措施，例如身份验证、授权、加密等。 * **MCP 协议:** 这个例子只是一个框架。 你需要根据 MCP 协议的规范来实现你的 MCP 逻辑。 这包括定义消息格式、方法名称、参数、返回值等。 * **异步编程:** `asyncio` 是一个强大的异步编程库。 如果你不熟悉 `asyncio`，建议先学习一下 `async` 和 `await` 的用法。 * **依赖:** 这个例子只依赖于 Python 的标准库 `asyncio` 和 `json`，不需要安装额外的依赖。 * **扩展性:** 这个例子是一个最小实现。 你可以根据需要扩展它，例如添加多线程支持、使用更高效的序列化方法、支持更多的 MCP 方法等。 **下一步:** 1. **定义你的 MCP 协议:** 明确你的 MCP 协议的规范，包括消息格式、方法名称、参数、返回值等。 2. **实现你的 MCP 逻辑:** 根据你的 MCP 协议，实现 `handle_client` 函数中的 MCP 逻辑。 3. **添加错误处理:** 添加更完善的错误处理机制。 4. **考虑安全性:** 采取安全措施，例如身份验证、授权、加密等。 5. **测试和调试:** 充分测试和调试你的服务器。 这个最小实现为你提供了一个起点。 你可以根据你的具体需求来扩展和修改它。