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到底长什么样”。MCPModel 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 CtrlC 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(fNew 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(fRaw 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/bodyMCP 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 fHTTP/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_asciiFalse).encode(utf-8) content_length len(body_bytes) headers ( fDate: {date_str}\r\n fContent-Type: {content_type}\r\n fContent-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而非2Date头必须用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( fhttps://api.figma.com/v1/files/{parameters[file_key]}, headers{X-Figma-Token: figma_token} ) response.raise_for_status() file_data response.json() # 模拟导出逻辑 export_url fhttps://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(fWarning: 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(fWarning: {py_file} missing TOOL_INFO or execute function) continue tool_info module.TOOL_INFO if tool_info.get(tool_id) ! tool_id: print(fWarning: {py_file} TOOL_INFO.tool_id mismatch: expected {tool_id}) continue tools[tool_id] { info: tool_info, module: module } print(fLoaded tool: {tool_id}) except Exception as e: print(fFailed 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(fOperation 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: fTool {tool_id} not found}) tool ALL_TOOLS[tool_id] # 步骤1JSON Schema校验 try: schema tool[info][input_schema] jsonschema.validate(instanceparameters, schemaschema) 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: fschema_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: fMissing 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: fregistration_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, maxBytes10*1024*1024, # 10MB backupCount5 ) 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, maxBytes10*1024*1024, backupCount5 ) 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(fRequest ID: {request_id}, exc_infoexception) else: error_logger.error(Unidentified request, exc_infoexception)在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, timeout2) 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 ( fAccess-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