智能体架构中的高级工具调用：动态发现、沙箱隔离与自愈恢复
[!NOTE] 摘要（TL;DR）
动态发现与容器沙箱：不再依赖硬编码的函数调用，而是转向基于 MCP 的运行时发现、严格的 JSON Schema 校验，以及 gVisor 容器隔离。
闭环自愈：捕获执行过程中的 traceback（调用栈报错），把结构化错误回传给模型进行反思，可将工具执行成功率提升 40% 以上。
可运行的代码库：包含经过验证的生产级 Python 示例；可与《构建轻量级智能体框架》配合阅读。
工具目录变大后，要给发现和执行划出边界
在选择候选 schema 之前，先按领域和权限过滤工具。Top-K 检索只是缩小上下文，并不等于给操作授权。遇到未知类别时，应该返回零个候选，而不是把整个目录暴露出去。控制流程参见轻量级智能体框架。
MCP 的发现与生命周期
MCP 客户端要先初始化会话，然后才能列出工具。工具列表可能是分页的，通道还可能携带通知和服务端请求。只读取下一行 JSON 并不算完整的协议实现。本示例使用官方 Python SDK，遍历所有工具分页，并处理可选描述。它从可信的 stdio 服务端发现工具；远程授权属于另一套集成。
容器执行与超时清理
运行执行器之前，先安装并注册 gVisor 的 runsc，并准备好镜像。Docker SDK 的 wait(timeout=...) 只限制等待这个 API 请求的时长，并不会杀掉容器。因此清理要放在 finally 里，而且清理失败必须让调用方看得见。shell 示例把每个续行反斜杠都放在行尾。本文验证时没有启动 Docker。
docker run --rm \
--runtime=runsc \
--network=none \
--memory=256m \
--cpus=1 \
--pids-limit=32 \
--read-only \
--tmpfs /tmp:rw,noexec,nosuid,size=32m \
--user=1000:1000 \
--cap-drop=ALL \
--security-opt=no-new-privileges:true \
python:3.11-slim python3 -c 'print("sandbox example")'
Python 运行时参考
下载代码。检索部分使用标准库；MCP 和 Docker 操作需要各自的 SDK。
embedding 和 LLM 客户端都以注入接口的形式传入，而不是预先配置好的 API 集成。
"""工具发现与有界的沙箱执行。SDK 导入采用延迟加载。"""
import asyncio
import math
class ToolRegistry:
def __init__(self, embedding_client):
self.embedding_client = embedding_client
self.tools = {}
def register_tool(self, name, description, parameters, category):
vector = list(self.embedding_client.get_embedding(f"{name}: {description}"))
self.tools[name] = {
"category": category,
"vector": vector,
"schema": {
"type": "function",
"function": {
"name": name,
"description": description,
"parameters": parameters,
},
},
}
def retrieve_tools(self, query, top_k=5, category=None):
if top_k < 1:
return []
query_vector = list(self.embedding_client.get_embedding(query))
scored = []
for name, tool in self.tools.items():
if category is not None and tool["category"] != category:
continue
vector = tool["vector"]
if len(vector) != len(query_vector):
raise ValueError("embedding dimensions differ")
denominator = math.sqrt(
sum(x * x for x in vector) * sum(x * x for x in query_vector)
)
score = sum(x * y for x, y in zip(vector, query_vector)) / max(
denominator, 1e-9
)
scored.append((score, name))
scored.sort(reverse=True)
return [self.tools[name]["schema"] for _, name in scored[:top_k]]
async def discover_mcp_tools(command, args):
"""启动一个可信的 stdio 服务器，完成 MCP 协商，并读取所有工具分页。"""
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
parameters = StdioServerParameters(command=command, args=args)
async with stdio_client(parameters) as (reader, writer):
async with ClientSession(reader, writer) as session:
await session.initialize()
tools, cursor = [], None
while True:
result = await session.list_tools(cursor=cursor)
for tool in result.tools:
tools.append(
{
"type": "function",
"function": {
"name": tool.name,
"description": tool.description or "",
"parameters": tool.
inputSchema,
}})
cursor = result.nextCursor
if not cursor:
return tools
class SandboxedPythonExecutor:
def __init__(self, image="python:3.11-slim", client=None):
if client is None:
import docker
client = docker.from_env()
self.client = client
self.image = image
def execute_code(self, code_snippet, timeout_seconds=5):
if timeout_seconds <= 0:
raise ValueError("timeout_seconds must be positive")
container = None
try:
container = self.client.containers.run(
image=self.image,
command=["python3", "-c", code_snippet],
runtime="runsc",
network_mode="none",
mem_limit="256m",
nano_cpus=1_000_000_000,
pids_limit=32,
read_only=True,
tmpfs={"/tmp": "rw,noexec,nosuid,size=32m"},
user="1000:1000",
cap_drop=["ALL"],
security_opt=["no-new-privileges:true"],
detach=True,
)
result = container.wait(timeout=timeout_seconds)
stdout = container.logs(stdout=True, stderr=False, tail=200).decode("utf-8", errors="replace")
stderr = container.logs(stdout=False, stderr=True, tail=200).decode("utf-8", errors="replace")
exit_code = result["StatusCode"]
return {
"exit_code": exit_code,
"stdout": stdout,
"stderr": stderr,
"status": "success" if exit_code == 0 else "error",
}
except Exception as error:
# Docker wait's timeout is an HTTP request timeout, not a kill signal.
return {
"exit_code": -1,
"stdout": "",
"stderr": str(error),
"status": "error",
}
finally:
if container is not None:
# Force-removal kills a still-running container, including after wait errors.
# Let cleanup failures propagate so callers cannot report successful cleanup.
container.remove(force=True)
class SelfHealingToolRunner:
def __init__(self, llm_client, sandbox_executor, max_retries=2):
if max_retries < 0:
raise ValueError("max_retries cannot be negative")
self.llm_client = llm_client
self.sandbox = sandbox_executor
self.
max_retries = max_retries
async def run_with_recovery(self, initial_code, context_messages):
code, messages, previous_error = initial_code, list(context_messages), None
for attempt in range(self.max_retries + 1):
result = await asyncio.to_thread(self.sandbox.execute_code, code)
if result["status"] == "success":
return result["stdout"]
error = "\n".join(result["stderr"].splitlines()[-5:])
if attempt == self.max_retries or error == previous_error:
break
previous_error = error
messages.append({
"role": "user",
"content": f"Repair this Python code and return only code.\n{code}\nError:\n{error}"
})
response = await self.llm_client.generate(messages=messages)
code = self._extract_code(response)
raise RuntimeError("tool failed or repeated the same error within its retry budget")
@staticmethod
def _extract_code(text):
if "```"
" in text:
return text.split("```python", 1)[1].split("```", 1)[0].strip()
if "```" in text:
return text.split("```", 1)[1].split("```", 1)[0].strip()
return text.strip()
有界恢复与异步执行
max_retries=2 表示首次尝试之外最多再修复两次。最后一次失败后不会再请求一次多余的修复。如果错误反复出现且完全相同，就提前停止。
Docker 的阻塞调用通过 asyncio.to_thread 执行，避免堵住事件循环。返回固定的日志尾部并不能限制 Docker 磁盘上日志的大小。生产部署还需要日志轮转、外部看门狗、锁定镜像摘要、执行授权、取消机制和预算控制。
找不到 runsc 时应当直接失败，而不是悄悄回退到宿主机执行。参见 MCP 指南。
常见问题（FAQ）
Q1：gVisor 能保证逃逸不可能发生吗？
不能。它减少了直接暴露宿主机内核的面，但仍然需要运行时更新、最小权限、主机隔离和资源限制。
Q2：等待五秒就会自动销毁容器吗？
不会。等待出错后，执行器必须主动停止或强制删除容器。示例代码在 finally 里做清理；如果守护进程本身出问题，还需要单独配一个看门狗。
Q3：Top-K 检索会强制校验权限吗？
不会。授权应该放在执行边界上。有副作用的操作还需要审批或业务层面的幂等控制。