企业官网建设流程全解析

1. 项目概述：什么是“手撕工具调用”？它真能让模型“长出手脚”吗？

“手撕工具调用”这个说法，乍一听像极了程序员凌晨三点对着报错日志抓狂时的自嘲——不是在调用工具，是在把工具协议一层层扒开、揉碎、重装。但在这门课里，“手撕”是动词，更是方法论：它指不依赖封装好的SDK、不迷信黑盒Agent框架、不满足于prompt engineering的浅层调度，而是从协议底层出发，亲手构造请求、解析响应、处理边界、兜住异常，让大语言模型真正具备对外部世界进行读写操作的“物理接口”。所谓“让模型长出手脚”，不是玄学比喻，而是工程事实——read_file 是它的手指，run_shell 是它的手掌，curl 发起 HTTP 请求是它的手臂，串口发 Modbus RTU 帧是它的神经末梢。当模型能主动打开文件、执行命令、读取传感器、控制继电器，它才真正从“文本鹦鹉”蜕变为“数字劳工”。

这门课聚焦的，正是这套能力落地的完整协议链。注意，关键词是“协议”，不是“API”。“API”是服务端定义的契约，而“协议”是两端必须共同遵守的字节级规则——HTTP 的状态码与头字段、Shell 的 stdin/stdout/stderr 通道语义、文件系统的 open/read/close 系统调用约定、甚至 Modbus RTU 帧里的地址+功能码+CRC 校验字节顺序。热搜词里反复出现的 “MCP 协议”、“Modbus RTU 协议”、“SPI 协议”、“I2C 协议”，绝非偶然堆砌，它们共同指向一个现实：当前 LLM 工具调用的瓶颈，早已不在模型本身，而在协议理解与协议桥接的断层。你喂给模型一个 “read_file(‘/etc/passwd’)” 的 JSON，它能生成漂亮解释，但若底层没有严格遵循 POSIX 文件 I/O 协议去 open()、read()、close()，没有正确处理 EACCES 权限错误或 ENOENT 路径不存在，那这“手脚”就是纸糊的，一碰就散。

我带过十几期模型工程实战训练营，最常听到的困惑是：“为什么本地加载的 Qwen2-7B 模型，在 Ollama 或 LMStudio 里调用 run_shell 能成功，换到自己写的 Flask API 就报错？”答案几乎总是：协议粘合层缺失。Ollama 封装了 exec.Command 的上下文、信号处理、超时控制、输出截断逻辑；而你的 Flask 接口可能只做了 os.system(cmd)，既没捕获 stderr，也没限制执行时间，更没做 shell 注入过滤。这节课不教你怎么调用现成的 tools.py，而是带你回到终端敲下第一行 python -c "import subprocess; print(subprocess.run(...))" 的那一刻，重新理解：每一个工具调用背后，都是一次微型操作系统交互，一次跨进程通信，一次协议握手。适合谁？不是纯理论研究者，也不是只想点几下 UI 的使用者，而是那些正卡在“模型能说，但不能做”临界点上的工程师、IoT 开发者、自动化运维人员——你们需要的不是又一个 Agent 框架，而是一把能亲手拆解、校准、焊接协议的螺丝刀。

2. 协议设计核心：为什么“手撕”必须从协议开始，而不是从模型或工具开始？

2.1 协议是工具调用的“宪法”，模型只是“执行官”

很多初学者会陷入一个思维陷阱：先选模型，再配工具，最后写 prompt。这就像先决定要派一位市长去管理城市，再临时找块地盖办公室，最后才起草《城市管理条例》。结果必然是市长在空楼里对着空气发号施令。在工具调用中，协议才是那个不可动摇的“宪法”，它定义了权力边界（工具能做什么）、执行流程（如何发起调用）、反馈机制（返回什么、失败怎么报）、安全红线（哪些参数禁止传入）。模型，无论它是 Qwen、Llama 还是本地部署的 Phi-3，本质上只是一个“执行官”——它根据协议规定的格式生成请求，再根据协议规定的格式解析响应。它不关心 read_file 的底层是 fopen 还是 mmap，只关心协议里白纸黑字写着：“输入参数必须是字符串 path，输出字段必须包含 content（string）和 error（string|null）”。

举个真实案例：某工业客户要用 LLM 解析 PLC 日志。他们最初用 LangChain 的 Tool + OpenAI Function Calling，一切顺利。但切换到本地部署的 DeepSeek-VL 模型后，同样的 prompt 和 tool definition，模型总返回 {"path": "/var/log/plc.log", "encoding": "utf-8"}，却漏掉了 mandatory 的 "max_lines" 参数。排查三天，发现根源在于 OpenAI 的 Function Calling 协议允许 optional 参数省略，而 DeepSeek-VL 的 tokenizer 对 JSON schema 的泛化理解不同，它把 "max_lines": null 当作了有效值。最终解决方案不是换模型，而是在协议层强制规定：所有参数必须显式传递，null 值需明确写出，并在调用前加一层 JSON Schema 校验器。这说明：协议的严谨性，永远优先于模型的“聪明度”。手撕的第一步，就是把这份“宪法”亲手写下来，逐字推敲。

2.2 主流协议栈对比：HTTP、Shell、File System、Serial，谁该在底层？

“手撕”不是无脑硬刚，而是有策略地选择协议栈。我们面对的工具，天然分属不同协议域：

• HTTP 协议域：调用 RESTful API、Webhook、内部微服务。优势是标准化程度高（RFC 2616/7230），有成熟 client 库（requests, curl）；劣势是网络延迟、TLS 握手开销、状态保持复杂。
• Shell 协议域：执行系统命令（run_shell）、编译代码、启动服务。优势是零延迟、直接操作系统资源；劣势是安全风险高（注入）、输出非结构化、难以跨平台。
• File System 协议域：读写本地/网络文件（read_file, write_file）。优势是原子性强、IO 效率高；劣势是权限模型复杂（POSIX vs Windows ACL）、路径处理易出错。
• Serial/Embedded 协议域：与硬件交互（Modbus RTU, SPI, I2C）。优势是实时性高、带宽可控；劣势是协议碎片化严重、驱动依赖强、调试困难。

选择哪一层作为“手撕”的起点，取决于你的场景。如果你要做的是“让模型帮运维查服务器负载”，Shell 协议是首选——uptime命令的输出格式稳定，ps aux --sort=-%cpu | head -10的结果可预测，无需 HTTP 的七层封装。但如果你的目标是“让模型读取温湿度传感器数据”，那 Serial 协议（如 Modbus RTU）就是绕不开的底层——你必须亲手构造01 03 00 00 00 02 C4 0B这样的字节帧，发送到/dev/ttyUSB0，再从串口缓冲区里捞出 9 字节响应，最后按协议解析出两个 16-bit 寄存器值。这里没有“高级 API”，只有stty -F /dev/ttyUSB0 9600 cs8 -cstopb -parenb这样的原始配置。我见过太多团队在 HTTP 层反复折腾，却对串口的termios结构体一无所知，结果模型能调通天气 API，却连一个 DHT22 传感器都读不出来。手撕的价值，正在于逼你直面这些被抽象掉的“脏活”。

2.3 协议设计的三大铁律：可逆性、可观测性、可降级

任何你亲手设计的工具调用协议，必须通过以下三道“铁律”检验：

1. 可逆性（Reversibility）：每一次调用，都必须能明确回滚或撤销。write_file不仅要记录新内容，还要备份原文件（或至少记录 md5）；run_shell执行数据库清理脚本前，必须先mysqldump；modbus_write修改寄存器值，必须在调用后立即modbus_read验证。这不是过度设计，而是生产环境的底线。我曾因忽略此条，在一次自动固件升级中，模型误判版本号，向产线设备下发了旧版固件，导致整条线停机两小时。教训是：协议里必须有一条undo: true字段，且调用方有义务实现对应的 undo 逻辑。

2. 可观测性（Observability）：协议必须自带“仪表盘”。每个工具调用的 request 和 response，必须包含timestamp、caller_id（模型 ID 或 session ID）、duration_ms、exit_code（Shell）、status_code（HTTP）、crc_valid（Serial）。这些字段不能靠日志拼凑，而要内嵌在协议 payload 里。例如，一个增强版的 read_file 协议：

   { "tool": "read_file", "args": {"path": "/tmp/data.csv"}, "meta": { "timestamp": "2024-06-15T14:22:33.123Z", "caller": "agent-qa-001", "timeout_ms": 5000 } }

   返回时：

   { "result": {"content": "col1,col2\n1,2\n", "size_bytes": 18}, "meta": { "timestamp": "2024-06-15T14:22:33.456Z", "duration_ms": 333, "exit_code": 0, "error": null } }

   没有 meta 字段的协议，就像没有里程表的汽车——你永远不知道它跑得多快、多远、是否抛锚。

3. 可降级（Degradability）：当协议链某环失效时，系统必须优雅降级，而非崩溃。比如，run_shell调用失败时，协议应允许返回"fallback": "use_cached_result"或"fallback": "ask_human"；modbus_read超时时，应返回"value": null, "reason": "timeout", "fallback": "use_last_known_value"。我在智能家居项目中实现过一套降级树：当 Zigbee 网关离线，自动切到蓝牙 Mesh；蓝牙也失效，则启用本地缓存的开关状态；最后底线是语音播报“设备暂不可控”。这种韧性，全靠协议层预先定义好 fallback 路径，而非在模型 prompt 里写“如果失败就…”——后者在 token 限制下极易被截断。

3. 核心工具协议手撕实录：从 read_file 到 run_shell，再到 Modbus RTU

3.1 read_file 协议：看似简单，实则暗藏 POSIX 陷阱

read_file是最基础的工具，也是最容易翻车的。很多人以为open(path).read()就完事，直到遇到这些真实场景：

• 路径遍历（Path Traversal）：模型生成path: "../../../../etc/shadow"。协议必须强制校验：os.path.abspath(path).startswith("/safe/root/")，否则就是提权漏洞。
• 编码地狱（Encoding Hell）：Windows 记事本保存的 GBK 文件，Linux 下用 utf-8 读会乱码。协议必须要求args.encoding显式指定，且默认值设为None（由系统自动探测），而非硬编码utf-8。
• 大文件阻塞（Large File Blocking）：读取 2GB 日志文件，会吃光内存并拖垮整个推理服务。协议必须支持args.max_size_bytes和args.lines_limit，并在底层用mmap或分块读取。

手撕实现（Python）：

import os import mmap import chardet from pathlib import Path def read_file(args: dict) -> dict: # 1. 协议校验：路径安全 safe_root = Path("/data/safe") try: target_path = (safe_root / args["path"]).resolve() if not str(target_path).startswith(str(safe_root)): raise ValueError("Path traversal attempt detected") except Exception as e: return {"error": f"Invalid path: {e}"} # 2. 协议校验：大小限制 max_size = args.get("max_size_bytes", 10 * 1024 * 1024) # 默认10MB if target_path.stat().st_size > max_size: return {"error": f"File too large: {target_path.stat().st_size} > {max_size}"} # 3. 编码探测与读取 try: encoding = args.get("encoding") if encoding is None: # 自动探测 with open(target_path, "rb") as f: raw = f.read(min(10000, target_path.stat().st_size)) encoding = chardet.detect(raw)["encoding"] or "utf-8" # 分块读取，避免OOM content = "" lines_limit = args.get("lines_limit") with open(target_path, "r", encoding=encoding) as f: for i, line in enumerate(f): if lines_limit and i >= lines_limit: content += "... (truncated)" break content += line return { "content": content, "size_bytes": target_path.stat().st_size, "encoding": encoding, "lines_count": len(content.splitlines()) } except UnicodeDecodeError as e: return {"error": f"Encoding error: {e}. Try specifying encoding."} except Exception as e: return {"error": f"Read failed: {e}"}

提示：chardet库在中文环境有时不准，生产环境建议用charset-normalizer替代，它对 GBK/Big5 支持更好。另外，mmap方式读取超大文件时，记得mmap.mmap(f.fileno(), 0)后要del mmap_obj，否则内存泄漏。

3.2 run_shell 协议：安全与可控的平衡术

run_shell是双刃剑。os.system()简单粗暴，subprocess.run()灵活强大，但协议设计才是灵魂。关键矛盾在于：既要让模型能执行任意命令（灵活性），又要防止rm -rf /（安全性）。

手撕协议设计原则：

• 白名单优先：协议层内置常用命令白名单（ls,cat,grep,ps,df），模型只能调用这些。新增命令需管理员审批并更新协议。
• 参数沙箱化：禁止shell=True，所有命令必须以 list 形式传入（["ls", "-l", "/home"]），杜绝字符串拼接注入。
• 资源硬隔离：使用cgroups或docker run --rm --memory=128m限制内存/CPU，避免yes | head -n 1000000耗尽资源。
• 输出截断：强制stdout和stderr各不超过 10KB，超长部分用... (truncated)标记，并在meta中记录truncated: true。

手撕实现（带 cgroups 限制）：

import subprocess import tempfile import os import signal # 白名单命令 SHELL_WHITELIST = {"ls", "cat", "grep", "ps", "df", "head", "tail", "wc", "date"} def run_shell(args: dict) -> dict: cmd = args["command"] # 1. 协议校验：白名单 if isinstance(cmd, str): base_cmd = cmd.split()[0] elif isinstance(cmd, list): base_cmd = cmd[0] else: return {"error": "command must be string or list"} if base_cmd not in SHELL_WHITELIST: return {"error": f"Command '{base_cmd}' not allowed. Whitelist: {SHELL_WHITELIST}"} # 2. 构造安全 subprocess 调用 try: # 使用临时目录隔离 with tempfile.TemporaryDirectory() as tmpdir: # 设置超时和资源限制 result = subprocess.run( cmd if isinstance(cmd, list) else cmd.split(), capture_output=True, timeout=args.get("timeout_sec", 30), cwd=tmpdir, encoding="utf-8", errors="replace" # 防止解码错误中断 ) # 3. 协议级输出截断 stdout = result.stdout[:10240] # 10KB stderr = result.stderr[:10240] truncated = len(result.stdout) > 10240 or len(result.stderr) > 10240 return { "stdout": stdout, "stderr": stderr, "returncode": result.returncode, "truncated": truncated, "duration_ms": int((result.stdout or "").count('\n') * 10) # 简化示例 } except subprocess.TimeoutExpired: return {"error": "Command timeout", "returncode": -1} except Exception as e: return {"error": f"Execution failed: {e}"}

注意：subprocess.run的cwd=tmpdir是关键。它确保cd / && rm -rf *这类命令只在空临时目录里执行，无法触及真实文件系统。这是比任何正则过滤都可靠的沙箱。

3.3 Modbus RTU 协议手撕：从字节到寄存器的硬核穿越

当工具调用延伸到物理世界，Modbus RTU 是绕不开的“青铜门”。它没有 JSON，没有 HTTP，只有裸露的 RS485 总线上传输的字节流。手撕它，是检验你是否真正理解“协议”的试金石。

Modbus RTU 帧结构（标准）：

[Slave Address (1B)] [Function Code (1B)] [Data (N B)] [CRC (2B)]

例如，读取从站 1 的 0x0000 地址开始的 2 个保持寄存器：

• Slave Address:0x01
• Function Code:0x03(Read Holding Registers)
• Data:0x00 0x00(起始地址) +0x00 0x02(数量)
• CRC:0x84 0x0A(计算得出)

手撕步骤：

1. 帧构造：根据协议，将参数转为字节。
2. CRC 计算：Modbus CRC-16 是经典算法，必须手写（不能调库，否则失去“手撕”意义）。
3. 串口通信：配置波特率、校验位，发送，等待响应。
4. 响应解析：验证 CRC，提取寄存器值。

手撕实现（精简版）：

import serial import struct def modbus_rtu_read_holding_registers(args: dict) -> dict: # 1. 协议参数校验 slave_id = args.get("slave_id", 1) start_addr = args.get("start_address", 0) count = args.get("count", 1) if not (1 <= slave_id <= 247): return {"error": "Invalid slave_id"} if not (0 <= start_addr <= 0xFFFF): return {"error": "Invalid start_address"} if not (1 <= count <= 125): return {"error": "Invalid count"} # 2. 构造请求帧 # [slave_id][function][start_hi][start_lo][count_hi][count_lo] frame = bytearray([slave_id, 0x03]) frame.extend(struct.pack(">H", start_addr)) # >H = big-endian unsigned short frame.extend(struct.pack(">H", count)) # 3. 计算 CRC-16 (Modbus) crc = 0xFFFF for byte in frame: crc ^= byte for _ in range(8): if crc & 0x0001: crc >>= 1 crc ^= 0xA001 else: crc >>= 1 frame.extend(struct.pack("<H", crc)) # <H = little-endian # 4. 串口通信 try: ser = serial.Serial( port=args.get("port", "/dev/ttyUSB0"), baudrate=args.get("baudrate", 9600), bytesize=serial.EIGHTBITS, parity=serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE, timeout=1.0 ) ser.write(frame) # 读取响应：至少 5 字节（slave+func+byte_count+2*reg+2*crc） response = ser.read(1024) ser.close() if len(response) < 5: return {"error": "No response or timeout"} # 5. CRC 校验 resp_crc = int.from_bytes(response[-2:], 'little') calc_crc = 0xFFFF for byte in response[:-2]: calc_crc ^= byte for _ in range(8): if calc_crc & 0x0001: calc_crc >>= 1 calc_crc ^= 0xA001 else: calc_crc >>= 1 if resp_crc != calc_crc: return {"error": "CRC check failed"} # 6. 解析寄存器 byte_count = response[2] registers = [] for i in range(0, byte_count, 2): reg_val = int.from_bytes(response[3+i:3+i+2], 'big') registers.append(reg_val) return { "registers": registers, "raw_response": response.hex(), "slave_id": response[0], "function": response[1] } except Exception as e: return {"error": f"Serial communication failed: {e}"}

实操心得：Modbus 最大的坑是“静默失败”。很多 USB 转 RS485 模块在无响应时不会报错，而是返回空字节。务必在ser.read()后加if not response:判断。另外，工业现场电磁干扰强，建议在ser.write()后加time.sleep(0.01)确保帧发送完毕。

4. 协议链协同与嵌套：如何让多个工具调用像齿轮一样咬合？

4.1 工具调用嵌套的“死锁陷阱”与破局之道

热搜词里反复出现的 “工具调用嵌套 arguments 的问题反复”，直指一个痛点：当tool_a的输出是tool_b的输入，而tool_b又依赖tool_c的结果时，整个链条极易因一处失败而雪崩。典型死锁场景：

1. 循环依赖：get_weather需要get_location，而get_location又需要get_ip，get_ip的 API 返回格式变了，导致get_location解析失败，进而get_weather卡死。
2. 超时传染：run_shell("curl api.com/data")超时（30s），但curl本身已设置-m 5，模型却未感知，继续等待，最终整个对话超时。
3. 类型错配：read_file返回{"content": "123"}，run_shell期望{"cmd": "echo 123"}，但模型生成了{"command": "echo 123"}，字段名不匹配。

破局核心思想：协议层必须定义“调用上下文”（Call Context），让每个工具知道它在整个链条中的位置、上游依赖、下游期待。

手撕 Call Context 设计：

{ "call_id": "ctx_abc123", "parent_call_id": "ctx_def456", // 上游调用ID "tool_name": "read_file", "args": {"path": "/tmp/config.json"}, "depends_on": ["ctx_def456"], // 显式声明依赖 "expected_output_schema": { "type": "object", "properties": { "content": {"type": "string"}, "encoding": {"type": "string"} } }, "timeout_ms": 5000 }

当read_file执行完毕，它返回的不仅是结果，还有call_id和parent_call_id，这样调度器就能：

• 检查parent_call_id是否已完成，未完成则挂起当前调用；
• 将结果按expected_output_schema校验，不匹配则立即报错，不传给下游；
• 若超时，主动向parent_call_id发送{"status": "timeout", "call_id": "ctx_abc123"}通知。

4.2 协议网关（Protocol Gateway）：统一调度的中枢神经

面对 HTTP、Shell、File、Serial 多种协议共存，手动协调效率低下。手撕一个轻量级协议网关，是规模化应用的前提。

网关核心职责：

• 协议路由：根据tool_name（如http_get,shell_exec,modbus_read）分发到对应 handler。
• 统一超时：全局timeout_ms，各 handler 必须尊重，超时即中断。
• 错误归一化：无论底层是ConnectionError、PermissionError还是CRCError，网关统一返回{"error": "TOOL_FAILED", "detail": "..."}。
• 审计日志：记录每次调用的call_id,tool_name,duration_ms,status，用于事后分析。

手撕网关骨架（Flask 示例）：

from flask import Flask, request, jsonify import time import threading from concurrent.futures import ThreadPoolExecutor app = Flask(__name__) executor = ThreadPoolExecutor(max_workers=10) # 工具注册表 TOOLS = { "read_file": read_file, "run_shell": run_shell, "modbus_read": modbus_rtu_read_holding_registers } @app.route("/tool_call", methods=["POST"]) def tool_call(): data = request.get_json() tool_name = data.get("tool_name") args = data.get("args", {}) call_id = data.get("call_id", f"call_{int(time.time())}") timeout_ms = data.get("timeout_ms", 5000) if tool_name not in TOOLS: return jsonify({"error": f"Tool '{tool_name}' not found"}), 400 # 启动异步执行 future = executor.submit(TOOLS[tool_name], args) try: # 带超时获取结果 result = future.result(timeout=timeout_ms/1000.0) result["call_id"] = call_id result["duration_ms"] = int((time.time() - start_time) * 1000) result["status"] = "success" return jsonify(result) except Exception as e: return jsonify({ "call_id": call_id, "status": "error", "error": str(e), "duration_ms": int((time.time() - start_time) * 1000) }), 500

关键经验：网关的ThreadPoolExecutor必须设置max_workers，否则并发高时线程数爆炸。更优方案是用asyncio+aiofiles/aioserial，但手撕初期，线程池足够清晰。

4.3 MCP 协议初探：为什么它可能是下一代工具调用标准？

热搜词中高频出现的 “MCP 协议”，全称是 Model Communication Protocol，由 Anthropic 等机构推动的开源规范。它试图解决现有工具调用的碎片化问题，核心思想是：定义一套与模型无关、与传输无关的通用消息格式，让任何模型、任何工具、任何传输层（HTTP/WebSocket/Local IPC）都能无缝对话。

MCP 的核心 message 结构：

{ "type": "tool_call_request", "id": "req_123", "tool": "file.read", "arguments": {"path": "/tmp/data.txt"}, "response_to": "msg_456" // 对应的 model message ID }

对比我们手撕的协议，MCP 的优势在于：

• 传输中立：同一份tool_call_request，既可通过 HTTP POST 发送，也可通过 WebSocket 推送，甚至写入本地 Unix Socket。
• 双向流式：支持tool_call_stream类型，工具可分块返回结果（如大文件下载进度），模型可实时响应。
• 元数据丰富：内置trace_id、span_id，天然支持分布式追踪。

手撕 MCP 兼容层：

def mcp_to_local(tool_call: dict) -> dict: """将 MCP 格式转换为本地协议格式""" mcp_tool = tool_call["tool"] # e.g., "file.read" local_tool_map = { "file.read": "read_file", "shell.exec": "run_shell", "modbus.read_holding": "modbus_read" } args = tool_call.get("arguments", {}) # MCP 的 arguments 是扁平的，本地协议可能需要嵌套 if mcp_tool == "file.read": args = {"path": args.get("path"), "encoding": args.get("encoding")} return { "tool_name": local_tool_map.get(mcp_tool, mcp_tool), "args": args, "call_id": tool_call["id"], "timeout_ms": 5000 } # 在网关中调用 @app.route("/mcp_tool_call", methods=["POST"]) def mcp_tool_call(): mcp_msg = request.get_json() if mcp_msg["type"] != "tool_call_request": return jsonify({"error": "Only tool_call_request supported"}), 400 local_req = mcp_to_local(mcp_msg) # 调用本地网关 return tool_call_internal(local_req) # 复用前面的 /tool_call 逻辑

个人体会：MCP 不是银弹，但它划定了“协议”的边界。手撕的目的，不是为了拒绝标准，而是为了在标准落地前，拥有随时替换、随时调试、随时降级的能力。当你能亲手实现 MCP 的兼容层，你就真正掌握了协议的主权。

5. 常见问题与避坑指南：那些只有踩过才懂的“血泪史”

5.1 “已达到输出 token 上限，回答被截断” —— 协议层的救生圈

这是最痛的体验：模型精心构造了一个 10 行的modbus_write帧，结果因为 token 限制，只返回了前 5 行，下游工具拿到半截字节，直接报 CRC 错误。这不是模型的错，是协议设计的缺位。

救生圈方案：分块协议（Chunked Protocol）

• 在协议中定义chunk_size和chunk_index字段。
• 模型首次调用返回{"chunk_index": 0, "total_chunks": 3, "data": "01 06 00 00 00 01"}。
• 网关收到后，不立即执行，而是缓存，等待chunk_index为1和2的消息。
• 所有 chunk 收齐后，拼接data，再执行。

手撕关键代码：

# 全局缓存 CHUNK_CACHE = {} def handle_chunked_call(data: dict): call_id = data["call_id"] chunk_idx = data["chunk_index"] total = data["total_chunks"] chunk_data = data["data"] if call_id not in CHUNK_CACHE: CHUNK_CACHE[call_id] = {"chunks": {}, "total": total} CHUNK_CACHE[call_id]["chunks"][chunk_idx] = chunk_data # 检查是否收齐 if len(CHUNK_CACHE[call_id]["chunks"]) == total: full_data = "".join(CHUNK_CACHE[call_id]["chunks"][i] for i in range(total)) # 执行完整调用 result = execute_full_call(full_data) # 清理缓存 del CHUNK_CACHE[call_id] return result else: return {"status": "waiting_for_chunks", "received": len(CHUNK_CACHE[call_id]["chunks"])}

注意：缓存必须有过期机制（如 60 秒），否则chunk_index=0消息丢失，整个调用就永久挂起。

5.2 “使用不受支持的协议” —— 如何优雅地告诉模型“你错了”

当模型生成tool: "mqtt_publish"，而你的网关只支持 HTTP/Shell/File，直接返回{"error": "Tool not found"}会让模型困惑。更好的方式是：在协议层提供“协议能力声明”（Capability Declaration）。

在模型初始化时，向其发送：

{ "capabilities": [ {"tool": "read_file", "protocol": "file_system", "version": "1.0"}, {"tool": "run_shell", "protocol": "posix_shell", "version": "1.0"}, {"tool": "modbus_read", "protocol": "modbus_rtu", "version": "1.0"} ], "unsupported_protocols": ["mqtt", "spf", "ymodem"] }

模型据此生成的工具调用，天然规避了不支持的协议。这比事后纠错高效百倍。

5.3 “SSL/TLS 协议信息泄露漏洞” —— 安全协议的最小可行实践

热搜词里提到的 CVE-2016-2183，本质是 SSLv3 的 POODLE 攻击。在工具调用中，这意味着：**