从Socket层手写MCP Server：零依赖实现工具通信协议-育师

1. 为什么现在必须亲手写一个MCP Server——不是用现成SDK，而是从Socket层开始

“MCP Server”这个词最近三个月在开发者社区的搜索量翻了4倍，但绝大多数人点开教程后第一眼看到的是“安装yakit插件”“配置Figma Token”“下载蓝湖客户端”，然后就卡在了“token在哪填”“怎么启动服务”“报错500 internal server error”上。我上周帮一位做工业设计协同的同事排查问题，他用的所谓“MCP Server”其实只是个封装了3层的Python脚本，底层连HTTP状态码都没正确返回，结果前端反复重试导致CAD插件直接崩溃。这让我意识到：当前所有公开资料里缺失的，不是“怎么调用MCP”，而是“MCP Server到底长什么样”。

MCP（Model Control Protocol）本质不是某种新协议，而是一套轻量级工具间通信的契约规范。它不规定传输层用HTTP还是WebSocket，不强制要求JSON Schema校验，甚至不定义认证方式——它只约定三件事：

工具A想调用工具B的某个能力时，必须发一个带tool_id和parameters字段的请求；
工具B收到后必须返回status、output和可选的error字段；
双方通过/tools端点交换能力清单，格式是固定结构的JSON数组。

这个契约简单到可以用10行bash脚本模拟，但正因如此，所有现成的“MCP Server”都成了黑盒：你不知道它什么时候会把{"status":"success"}错写成{"status":"ok"}，不知道它对超长参数是截断还是报错，更不知道当网络抖动时它会不会把部分响应体丢给前端。去年我们团队在对接KiCad和Blender的MCP桥接时，就因为某SDK默认开启gzip压缩，而Blender的Python HTTP库不支持自动解压，导致整个PCB布线流程卡死在“等待模型生成”状态——查了三天才发现是压缩头没处理。

所以这篇要做的，不是教你“如何接入MCP”，而是带你用Python原生socket+标准库，从零写出第一个能被Figma、Yakit、甚至自研工具识别的MCP Server。它不依赖任何第三方框架，代码全部展开，每个字节流走向都清晰可见。你会看到：

当Chrome浏览器发送GET /tools请求时，TCP包里实际包含多少个\r\n；
为什么Content-Type: application/json必须紧跟在HTTP/1.1 200 OK之后；
如何用select()系统调用同时监听多个连接而不阻塞主线程；
甚至当用户误输POST /tool/xxx（少了个s）时，怎样返回符合MCP规范的404错误体。

这不是理论课，这是给你一把螺丝刀，让你看清MCP Server的每一颗铆钉。接下来所有代码，你复制粘贴就能跑通，且能立刻用curl或Postman验证——因为真正的MCP Server，从来就不该需要“安装教程”。

2. 从TCP三次握手开始：手写Server的核心骨架与协议解析逻辑

很多教程一上来就写from flask import Flask，这等于直接跳过了MCP Server最本质的部分：它首先是个网络服务，其次才是业务逻辑容器。Flask这类框架帮你屏蔽了TCP连接管理、HTTP报文解析、并发处理等细节，但当你需要调试“为什么Figma发来的POST请求收不到body”时，这些被屏蔽的细节恰恰是故障根源。所以我们从最底层开始：用Python标准库socket模块构建一个能稳定处理HTTP请求的Server骨架。

2.1 基础Socket服务器：监听端口并接收原始字节流

import socket import select import sys # 创建TCP socket server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server_socket.bind(('localhost', 8000)) server_socket.listen(5) print("MCP Server started on http://localhost:8000") print("Press Ctrl+C to stop") # 使用select实现非阻塞I/O，避免单连接阻塞整个服务 inputs = [server_socket] outputs = [] message_queues = {} try: while inputs: # 监听socket读事件（新连接/数据到达） readable, writable, exceptional = select.select(inputs, outputs, inputs, 1.0) for s in readable: if s is server_socket: # 新连接请求 client_socket, client_address = server_socket.accept() print(f"New connection from {client_address}") client_socket.setblocking(0) inputs.append(client_socket) message_queues[client_socket] = b"" else: # 已有连接收到数据 try: data = s.recv(4096) if data: message_queues[s] += data # 检查是否收到完整HTTP请求（以\r\n\r\n结尾） if b"\r\n\r\n" in message_queues[s]: # 解析HTTP请求 request_line, headers_and_body = message_queues[s].split(b"\r\n\r\n", 1) method, path, _ = request_line.split(b" ", 2) # 打印原始请求行（调试用） print(f"Raw request: {method.decode()} {path.decode()}") # 这里将触发业务逻辑处理 response = handle_http_request(method, path, headers_and_body) # 发送响应 s.sendall(response) # 关闭连接（简化版，实际应支持keep-alive） inputs.remove(s) if s in outputs: outputs.remove(s) s.close() else: # 客户端关闭连接 if s in inputs: inputs.remove(s) if s in outputs: outputs.remove(s) s.close() except ConnectionResetError: # 客户端异常断开 if s in inputs: inputs.remove(s) if s in outputs: outputs.remove(s) s.close() except KeyboardInterrupt: print("\nShutting down server...") finally: server_socket.close()

这段代码的关键不在语法，而在它暴露了HTTP协议的真实形态：

recv(4096)每次只取最多4096字节，意味着一个大JSON body可能被分多次接收；
b"\r\n\r\n"是HTTP头部与body的分界符，没有它你就无法确定header解析是否完成；
select.select()的1.0秒超时参数，决定了服务在空闲时每秒最多轮询1次，这是性能与资源消耗的平衡点。

提示：很多初学者以为HTTP请求是“原子性”的，实际上TCP层会根据网络状况自动分包。你用curl发的POST /tools请求，在Wireshark里可能看到3个TCP包：第一个含POST /tools HTTP/1.1\r\n，第二个含Content-Type: application/json\r\n，第三个含{"tool_id":"figma-export"...}。如果代码没处理分包逻辑，就会在第一次recv后就尝试解析，结果因缺少\r\n\r\n而卡死。

2.2 HTTP请求解析器：从原始字节流提取method/path/body

MCP Server不需要支持全部HTTP特性，但必须精准识别三类核心请求：

GET /tools：返回工具能力清单；
POST /tool/{id}：执行具体工具操作；
GET /health：健康检查（非MCP规范但工程必需）。

我们写一个极简解析器，不依赖http.server模块：

def parse_http_request(raw_data: bytes) -> dict: """ 解析原始HTTP请求字节流，返回结构化字典 返回格式：{'method': 'POST', 'path': '/tool/figma-export', 'headers': {...}, 'body': b'...'} """ if b"\r\n\r\n" not in raw_data: return {"error": "incomplete_request"} header_part, body_part = raw_data.split(b"\r\n\r\n", 1) lines = header_part.split(b"\r\n") if not lines: return {"error": "empty_request"} # 第一行：METHOD PATH VERSION try: method, path, version = lines[0].split(b" ", 2) except ValueError: return {"error": "invalid_request_line"} # 解析Headers headers = {} for line in lines[1:]: if b":" in line: key, value = line.split(b":", 1) headers[key.strip().decode()] = value.strip().decode() return { "method": method.decode(), "path": path.decode(), "version": version.decode(), "headers": headers, "body": body_part } def handle_http_request(method: bytes, path: bytes, raw_body: bytes) -> bytes: """根据HTTP方法和路径分发处理逻辑""" # 解析完整请求 req = parse_http_request(method + b" " + path + b" HTTP/1.1\r\n\r\n" + raw_body) if "error" in req: return build_http_response(400, {"error": req["error"]}) # 路由分发 if req["method"] == "GET": if req["path"] == "/tools": return build_tools_response() elif req["path"] == "/health": return build_health_response() else: return build_http_response(404, {"error": "not_found"}) elif req["method"] == "POST": if req["path"].startswith("/tool/"): tool_id = req["path"][6:] # 去掉"/tool/"前缀 try: # 尝试解析JSON body import json params = json.loads(req["body"].decode()) return execute_tool(tool_id, params) except json.JSONDecodeError: return build_http_response(400, {"error": "invalid_json_body"}) else: return build_http_response(404, {"error": "not_found"}) else: return build_http_response(405, {"error": "method_not_allowed"})

这里有个关键细节：parse_http_request函数中，我们手动拼接了HTTP/1.1\r\n\r\n来构造完整请求头。这是因为recv()收到的数据可能只包含部分header（比如只有POST /tool/xxx HTTP/1.1\r\n），而b"\r\n\r\n"分割必须作用于完整header。这种“补全再解析”的思路，比盲目等待所有数据更可靠——毕竟HTTP协议本身允许客户端分多次发送header。

注意：真实生产环境需处理Transfer-Encoding: chunked和Content-Length两种body传输方式。本例为简化，假设所有请求都带Content-Length头（现代工具如Figma/Yakit默认如此）。若遇到无Content-Length的请求，需按chunked规则解析，这部分代码约200行，后续章节会展开。

2.3 构建HTTP响应：严格遵循RFC 7230的字节级规范

MCP Server的响应体必须让任意HTTP客户端（包括浏览器、curl、Figma插件）都能正确解析。这意味着不能只关注JSON内容，更要控制每一个\r\n和空格：

def build_http_response(status_code: int, body_dict: dict, content_type: str = "application/json") -> bytes: """ 构建符合RFC 7230的HTTP响应 确保：状态行正确、Header顺序合理、CRLF结尾、Body长度精确 """ import json # 构建状态行 status_text = { 200: "OK", 201: "Created", 400: "Bad Request", 404: "Not Found", 405: "Method Not Allowed", 500: "Internal Server Error" }.get(status_code, "Unknown") status_line = f"HTTP/1.1 {status_code} {status_text}\r\n" # 构建Headers（必须包含Date和Content-Length） import time date_str = time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime()) body_bytes = json.dumps(body_dict, ensure_ascii=False).encode('utf-8') content_length = len(body_bytes) headers = ( f"Date: {date_str}\r\n" f"Content-Type: {content_type}\r\n" f"Content-Length: {content_length}\r\n" "Connection: close\r\n" # 简化版，不支持keep-alive "\r\n" # Header与Body之间的空行 ) # 组合完整响应 return status_line.encode('utf-8') + headers.encode('utf-8') + body_bytes def build_tools_response() -> bytes: """返回MCP工具能力清单，严格按规范格式""" tools = [ { "tool_id": "figma-export", "name": "Figma Design Export", "description": "Export Figma design as PNG/SVG/PDF", "input_schema": { "type": "object", "properties": { "file_key": {"type": "string"}, "page_name": {"type": "string"}, "format": {"type": "string", "enum": ["png", "svg", "pdf"]} }, "required": ["file_key", "format"] } }, { "tool_id": "sql-query", "name": "SQL Query Executor", "description": "Run SQL queries against local database", "input_schema": { "type": "object", "properties": { "query": {"type": "string"}, "database": {"type": "string"} }, "required": ["query"] } } ] return build_http_response(200, tools)

这段代码的魔鬼细节在于：

Content-Length必须是body_bytes的真实字节长度，不是字符串长度。中文字符在UTF-8中占3字节，len("你好")是6而非2；
Date头必须用GMT时区，且格式严格为"Wed, 21 Oct 2015 07:28:00 GMT"，少一个空格或字母都会被严格模式客户端拒绝；
Connection: close确保客户端不会复用连接，避免后续请求被错误关联。

我曾在线上环境遇到过一个诡异问题：当build_http_response返回的Content-Length比实际body少1字节时，Postman显示“Response body is incomplete”，而Figma插件直接抛出SyntaxError: Unexpected end of JSON input——因为JSON解析器等不到最后一个}就超时了。这种字节级的精确性，正是手写Server不可替代的价值。

3. MCP核心能力实现：/tools清单注册与/tool/{id}动态执行引擎

MCP Server的“智能”不在于算法多复杂，而在于如何让不同工具的能力像乐高一样即插即用。现成SDK往往把工具逻辑硬编码在if tool_id == "xxx"分支里，导致新增一个工具就要改一次Server源码。我们要实现的是：工具能力声明与执行逻辑完全解耦，新增工具只需提供一个Python文件，无需重启Server。

3.1 工具能力声明机制：基于文件系统的自动发现

MCP规范要求GET /tools返回一个JSON数组，每个元素描述一个工具。我们设计一个约定：所有工具定义放在./tools/目录下，每个工具一个.py文件，文件名即tool_id，内容必须包含TOOL_INFO字典和execute函数：

# ./tools/figma-export.py TOOL_INFO = { "tool_id": "figma-export", "name": "Figma Design Export", "description": "Export Figma design as PNG/SVG/PDF", "input_schema": { "type": "object", "properties": { "file_key": {"type": "string"}, "page_name": {"type": "string"}, "format": {"type": "string", "enum": ["png", "svg", "pdf"]} }, "required": ["file_key", "format"] } } def execute(parameters: dict) -> dict: """执行工具逻辑，返回标准MCP响应体""" import requests import os # 从环境变量读取Figma Token（安全实践） figma_token = os.getenv("FIGMA_TOKEN") if not figma_token: return {"status": "error", "error": "FIGMA_TOKEN not set"} # 调用Figma API（简化版） try: response = requests.get( f"https://api.figma.com/v1/files/{parameters['file_key']}", headers={"X-Figma-Token": figma_token} ) response.raise_for_status() file_data = response.json() # 模拟导出逻辑 export_url = f"https://figma.com/export/{parameters['file_key']}/{parameters['format']}" return { "status": "success", "output": { "export_url": export_url, "file_size": "2.4MB" } } except Exception as e: return {"status": "error", "error": str(e)}

Server启动时自动扫描此目录，动态加载所有工具：

import os import importlib.util from pathlib import Path TOOLS_DIR = Path("./tools") def load_all_tools() -> dict: """动态加载./tools/目录下所有工具模块""" tools = {} if not TOOLS_DIR.exists(): print(f"Warning: Tools directory {TOOLS_DIR} not found") return tools for py_file in TOOLS_DIR.glob("*.py"): if py_file.name.startswith("__"): continue tool_id = py_file.stem # 文件名去掉.py后缀 # 动态导入模块 spec = importlib.util.spec_from_file_location(tool_id, py_file) if spec is None: continue module = importlib.util.module_from_spec(spec) try: spec.loader.exec_module(module) # 验证模块包含必要属性 if not hasattr(module, "TOOL_INFO") or not hasattr(module, "execute"): print(f"Warning: {py_file} missing TOOL_INFO or execute function") continue tool_info = module.TOOL_INFO if tool_info.get("tool_id") != tool_id: print(f"Warning: {py_file} TOOL_INFO.tool_id mismatch: expected {tool_id}") continue tools[tool_id] = { "info": tool_info, "module": module } print(f"Loaded tool: {tool_id}") except Exception as e: print(f"Failed to load {py_file}: {e}") return tools # 全局工具注册表 ALL_TOOLS = load_all_tools() def build_tools_response() -> bytes: """返回所有已加载工具的清单""" tools_list = [tool["info"] for tool in ALL_TOOLS.values()] return build_http_response(200, tools_list)

这种设计带来三个关键优势：

热加载支持：修改./tools/figma-export.py后，下次GET /tools会自动返回新版本（实际生产中可加文件监控）；
权限隔离：每个工具在独立模块中运行，一个工具的异常（如无限循环）不会影响其他工具；
测试友好：直接import tools.figma_export即可单元测试execute函数，无需启动HTTP服务。

实操心得：我们最初把工具逻辑写在同一个tools.py里，结果某次SQL查询工具因数据库连接池耗尽，导致整个Server的/tools接口超时。拆分为独立文件后，问题被精准定位到sql-query.py，修复后其他工具毫秒级恢复。

3.2 工具执行引擎：参数校验、超时控制与错误标准化

POST /tool/{id}的执行不是简单调用函数，而是需要一套健壮的执行管道。我们构建一个execute_tool函数，它串联四个环节：

路由匹配：确认tool_id是否存在；
参数校验：用JSON Schema验证parameters是否符合input_schema；
沙箱执行：设置超时、内存限制、子进程隔离；
错误归一化：将各种异常（网络超时、JSON解析失败、业务逻辑异常）统一为MCP标准错误格式。

import jsonschema import signal import threading from contextlib import contextmanager @contextmanager def timeout(seconds): """上下文管理器：执行代码块，超时则抛出TimeoutError""" def timeout_handler(signum, frame): raise TimeoutError(f"Operation timed out after {seconds} seconds") signal.signal(signal.SIGALRM, timeout_handler) signal.alarm(seconds) try: yield finally: signal.alarm(0) def execute_tool(tool_id: str, parameters: dict) -> bytes: """执行指定工具，返回标准化MCP响应""" if tool_id not in ALL_TOOLS: return build_http_response(404, {"error": f"Tool '{tool_id}' not found"}) tool = ALL_TOOLS[tool_id] # 步骤1：JSON Schema校验 try: schema = tool["info"]["input_schema"] jsonschema.validate(instance=parameters, schema=schema) except jsonschema.ValidationError as e: return build_http_response(400, { "error": "validation_failed", "details": str(e) }) except Exception as e: return build_http_response(500, {"error": f"schema_validation_error: {e}"}) # 步骤2：沙箱执行（超时5秒） try: with timeout(5): result = tool["module"].execute(parameters) # 步骤3：结果标准化（确保包含status字段） if not isinstance(result, dict) or "status" not in result: return build_http_response(500, { "error": "invalid_tool_response", "details": "Tool must return dict with 'status' key" }) return build_http_response(200, result) except TimeoutError: return build_http_response(504, {"error": "execution_timeout", "timeout_seconds": 5}) except Exception as e: # 捕获所有未预期异常，避免暴露内部信息 error_msg = str(e) if "password" in error_msg.lower() or "token" in error_msg.lower(): error_msg = "internal_error_occurred" return build_http_response(500, { "error": "execution_failed", "details": error_msg }) # 示例：添加一个内存敏感型工具（如图像处理） # ./tools/image-resize.py # TOOL_INFO = {...} # def execute(parameters): # # 使用PIL处理图片，但限制最大尺寸防止OOM # max_pixels = 10000000 # 1000万像素 # if parameters["width"] * parameters["height"] > max_pixels: # raise ValueError("Image too large") # ...

这个执行引擎的关键设计点：

超时非装饰器而是上下文管理器：signal.alarm()在多线程环境下更可靠，避免Flask等框架的线程安全问题；
错误消息脱敏：当异常信息含password或token时，自动替换为泛化提示，防止敏感信息泄露；
Schema校验前置：在执行业务逻辑前就拦截非法参数，避免无效计算浪费资源。

我们曾在线上遇到一个案例：某用户传入{"query": "SELECT * FROM users"}（无WHERE条件），SQL工具执行耗时12秒。通过timeout(5)直接中断，返回504错误，既保护了数据库，又让前端能及时提示“查询超时，请添加筛选条件”。

3.3 动态能力扩展：如何让MCP Server支持新工具零代码修改

真正的“专属”Server，意味着你能随时接入任何新工具，而无需改动Server核心代码。我们设计一个register_tool端点，允许通过HTTP请求动态注册工具（适合CI/CD场景）：

def handle_register_tool(request_body: bytes) -> bytes: """处理POST /register-tool请求，动态注册新工具""" try: import json payload = json.loads(request_body.decode()) # 验证payload结构 required_fields = ["tool_id", "name", "description", "input_schema", "code"] for field in required_fields: if field not in payload: return build_http_response(400, {"error": f"Missing field: {field}"}) tool_id = payload["tool_id"] code = payload["code"] # 写入tools目录（生产环境应存入数据库或Redis） tool_file = TOOLS_DIR / f"{tool_id}.py" tool_file.write_text(code) # 重新加载工具（实际中可触发reload） global ALL_TOOLS ALL_TOOLS = load_all_tools() return build_http_response(201, { "status": "registered", "tool_id": tool_id, "tools_count": len(ALL_TOOLS) }) except json.JSONDecodeError: return build_http_response(400, {"error": "invalid_json_payload"}) except Exception as e: return build_http_response(500, {"error": f"registration_failed: {e}"}) # 在handle_http_request中添加路由 if req["method"] == "POST" and req["path"] == "/register-tool": return handle_register_tool(req["body"])

使用示例（curl命令）：

curl -X POST http://localhost:8000/register-tool \ -H "Content-Type: application/json" \ -d '{ "tool_id": "git-commit-lint", "name": "Git Commit Message Linter", "description": "Validate commit messages against Conventional Commits", "input_schema": { "type": "object", "properties": { "message": {"type": "string"} }, "required": ["message"] }, "code": "TOOL_INFO = {...}\\ndef execute(parameters):\\n import re\\n if re.match(r\"^(feat|fix|docs|style|refactor|test|chore):\", parameters[\"message\"]):\\n return {\"status\": \"success\", \"output\": {\"valid\": True}}\\n return {\"status\": \"error\", \"error\": \"Invalid commit format\"}" }'

这个设计让MCP Server真正成为“工具连接中枢”：

DevOps流水线可在部署时自动注册CI检查工具；
设计师团队可上传Figma插件专用工具；
甚至用户可通过Web界面粘贴Python代码，即时创建个人工作流。

注意事项：生产环境启用/register-tool必须加鉴权（如API Key），否则等于开放远程代码执行。本例为演示省略，实际应集成JWT或Basic Auth。

4. 生产级加固：日志审计、健康检查与跨域支持实战

写完能跑通的Server只是第一步，要让它在真实环境中稳定服役，必须解决三个工程问题：

问题可追溯：当Figma插件报错“connection refused”时，你得知道是Server崩了，还是网络策略拦截了8000端口；
服务可观测：运维需要知道QPS、平均延迟、错误率，而不是靠ps aux | grep python；
前端能调用：浏览器同源策略会阻止https://figma.com向http://localhost:8000发请求。

我们不用引入Prometheus或ELK，仅用标准库和几行代码解决。

4.1 结构化日志系统：区分访问日志与错误日志

Python的logging模块足够强大，但多数教程只教logging.info()。我们要实现：

访问日志单独文件，格式为[2023-10-05 14:22:31] GET /tools 200 12ms；
错误日志包含完整traceback，且自动关联请求ID便于追踪；
日志滚动，避免单文件过大。

import logging from logging.handlers import RotatingFileHandler import uuid import time # 创建两个logger access_logger = logging.getLogger("access") access_logger.setLevel(logging.INFO) error_logger = logging.getLogger("error") error_logger.setLevel(logging.ERROR) # 访问日志处理器（按大小滚动） access_handler = RotatingFileHandler( "logs/access.log", maxBytes=10*1024*1024, # 10MB backupCount=5 ) access_handler.setFormatter(logging.Formatter( "%(asctime)s %(message)s", datefmt="[%Y-%m-%d %H:%M:%S]" )) access_logger.addHandler(access_handler) # 错误日志处理器（同样滚动） error_handler = RotatingFileHandler( "logs/error.log", maxBytes=10*1024*1024, backupCount=5 ) error_handler.setFormatter(logging.Formatter( "%(asctime)s %(levelname)s %(name)s %(message)s\n%(exc_text)s", datefmt="[%Y-%m-%d %H:%M:%S]" )) error_logger.addHandler(error_handler) def log_access(client_ip: str, method: str, path: str, status_code: int, duration_ms: float): """记录访问日志""" access_logger.info(f'{client_ip} {method} {path} {status_code} {duration_ms:.1f}ms') def log_error(exception: Exception, request_id: str = None): """记录错误日志，附带request_id""" if request_id: error_logger.error(f"Request ID: {request_id}", exc_info=exception) else: error_logger.error("Unidentified request", exc_info=exception)

在handle_http_request中注入日志：

# 在处理请求前生成request_id request_id = str(uuid.uuid4())[:8] # 记录开始时间 start_time = time.time() try: response = ... # 原有逻辑 duration = (time.time() - start_time) * 1000 log_access(client_ip, req["method"], req["path"], status_code, duration) return response except Exception as e: log_error(e, request_id) return build_http_response(500, {"error": "internal_server_error"})

这样当问题发生时，你打开error.log就能看到：

[2023-10-05 14:22:31] ERROR error Request ID: a1b2c3d4 Traceback (most recent call last): File "server.py", line 123, in handle_http_request result = tool["module"].execute(parameters) File "./tools/sql-query.py", line 45, in execute cursor.execute(query) sqlite3.OperationalError: no such table: users

再查access.log同一时间点的记录，就能确认是哪个请求触发了此错误。

4.2 健康检查端点：让Kubernetes/Liveness Probe真正有用

GET /health不能只返回{"status":"ok"}，它必须反映服务真实状态。我们检查三项：

TCP端口是否可监听；
工具目录是否可读；
至少一个工具能成功执行（如调用/tools接口）。

def build_health_response() -> bytes: """深度健康检查，返回详细状态""" import os checks = { "tcp_port_listening": False, "tools_directory_readable": False, "tools_api_available": False } # 检查端口（本机连接测试） try: import socket s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.connect(("localhost", 8000)) s.close() checks["tcp_port_listening"] = True except ConnectionRefusedError: pass # 检查tools目录 if TOOLS_DIR.exists() and os.access(TOOLS_DIR, os.R_OK): checks["tools_directory_readable"] = True # 检查/tools接口 try: # 本地调用自身API（避免网络依赖） import http.client conn = http.client.HTTPConnection("localhost", 8000, timeout=2) conn.request("GET", "/tools") resp = conn.getresponse() if resp.status == 200: checks["tools_api_available"] = True conn.close() except Exception: pass # 计算整体状态 overall_status = "healthy" if all(checks.values()) else "degraded" return build_http_response(200, { "status": overall_status, "checks": checks, "timestamp": int(time.time()) })

这个/health端点可直接被Kubernetes的livenessProbe使用：

livenessProbe: httpGet: path: /health port: 8000 initialDelaySeconds: 30 periodSeconds: 10

当tools_api_available为False时，K8s会自动重启Pod，而不是让服务挂着“假活”。

4.3 跨域支持（CORS）：让浏览器前端能直接调用

MCP Server常被Figma、VSCode等桌面应用调用，它们不受同源策略限制。但如果你开发Web管理界面（如工具配置面板），就必须处理CORS。我们手动添加响应头，不依赖第三方库：

def build_http_response(status_code: int, body_dict: dict, content_type: str = "application/json", cors_origin: str = None) -> bytes: # ... 原有状态行和headers构建 ... # 添加CORS头（如果指定了origin） if cors_origin: headers += ( f"Access-Control-Allow-Origin: {cors_origin}\r\n" "Access-Control-Allow-Methods: GET, POST, OPTIONS\r\n" "Access-Control-Allow-Headers: Content-Type, Authorization\r\n" "Access-Control-Allow-Credentials: true\r\n" ) # ... 后续组合响应 ... return status_line.encode('utf-8') + headers.encode('utf-8') + body_bytes # 在handle_http_request中处理OPTIONS预检 if req["method"] == "OPTIONS": # 返回CORS预检响应 origin = req["headers"].get("Origin", "*") return