什么是沙箱（Sandbox）
智能体（Agent）需要生成代码、操作文件系统并运行 Shell 命令。由于我们无法预测智能体会执行什么操作，因此必须将其环境与主机系统隔离，防止其访问凭证、文件或网络。沙盒（Sandbox）通过创建智能体执行环境与主机系统之间的边界，提供了这种隔离能力。
在 Deep Agents 中，沙盒是一种特殊的后端（Backend）。与其他仅暴露文件操作的后端（State, Filesystem, Store）不同，沙盒后端还为智能体提供了一个 execute工具，用于在隔离环境中运行任意 Shell 命令。
沙盒就是给智能体准备的一个“封闭房间”：
智能体在里面随便折腾；
你的电脑、账号、文件、网络都被挡在外面。
沙箱的工具集
├── 文件系统操作工具
│ ├── ls - 列出目录内容
│ ├── read_file - 读取文件
│ ├── write_file - 写入文件
│ ├── edit_file - 编辑文件
│ ├── glob - 文件模式匹配
│ └── grep - 文本搜索
│
├── 执行工具
│ └── execute - 运行Shell命令
│ ├── 可以运行任意命令
│ ├── 返回stdout/stderr
│ └── 返回退出码
│
└── 安全边界
├── 无法访问主机文件
├── 无法读取主机环境变量
└── 无法干扰主机进程
为什么要用沙箱（Sandbox）
沙盒主要用于安全隔离。它们允许智能体执行任意代码、访问文件和联网，而不会泄露您的凭证、破坏本地文件或损害主机系统。当智能体自主运行时，这种隔离至关重要。
沙盒特别适用于以下场景：
编程智能体（Coding Agents）：需要运行 Shell、Git 操作、克隆仓库（许多供应商提供原生 Git API，如 Daytona 的 Git 操作），甚至运行 Docker-in-Docker 来构建和测试流水线。
数据分析智能体：在安全隔离的环境中加载文件、安装数据分析库（如 pandas, numpy）、运行统计计算，并生成如 PPT 等输出文件。
优势一：文件系统隔离
- ✅ 智能体只能访问沙盒内的文件
- ❌ 无法读取你的
/Documents、/Desktop等个人文件 - ❌ 无法修改你的项目源代码
优势二：大量输出造成上下文“爆炸”
- ✅ 智能体只需要知道临时文件的路径，以及临时文件内容的摘要信息
- ❌ 某个步骤处理完成后，输出：file1.py, file2.py, … 总大小：2MB
- ❌ 输出太大，直接返回会撑爆上下文窗口
优势三：环境隔离
- ✅ 智能体可以自由安装Python库，或者Java的lib，或者NodeJs的库，或者Go语音的第三方依赖。
- ❌ 不会污染你的全局项目代码的环境
- ❌ 不会与其他项目产生依赖冲突
优势四：进程隔离
- ✅ 智能体可以启动新进程
- ❌ 无法杀死你的其他进程
- ❌ 无法监控你的系统活动
优势五：网络控制
- ✅ 可以配置网络访问权限
- ❌ 防止数据外泄到未知服务器
主流沙盒提供商
提供商功能对比表
| 提供商 | 特点 | 适用场景 | 免费额度 | 网络控制 |
|---|---|---|---|---|
| Daytona | 简单易用，API直观 | 教学、快速原型 | 有 | 可配置 |
| Modal | 基于云函数，弹性强 | 生产环境、大规模 | 有 | 支持阻断 |
| Runloop | 开发箱模式，交互好 | 开发调试、团队协作 | 试用 | 可配置 |
| LangSmith | LangChain生态集成 | LangChain项目 | 内测中 | 待确认 |
| AgentCore | AWS Bedrock集成 | AWS用户、企业级 | 按使用量 | AWS VPC |
pip install langchain-daytona
# 2. 创建并启动沙箱（Daytona）
# 注意：此操作会向Daytona服务发起请求，创建一个远程隔离环境
print("步骤1: 创建远程沙箱环境...")
try:
# （假设在.env文件中配置了OPENAI_API_KEY和DAYTONA_API_KEY）
config = DaytonaConfig(
api_key=DAYTONA_API_KEY,
api_url=DAYTONA_BASE_URL, # "https://app.daytona.io/api"
target="us"
)
sandbox = Daytona(config).get('9058717f-98ce-442a-991a-69fece04c2e7') # 使用之前的docker容器作为远程沙箱
# sandbox = Daytona().create() # 第一次创建沙箱
print(f"沙箱创建成功！沙箱ID: {sandbox.id}")
except Exception as e:
print(f"创建沙箱失败，请检查网络和API密钥: {e}")
return
# 3. 将沙箱包装为Deep Agents可用的后端（Backend）
backend = DaytonaSandbox(sandbox=sandbox)
# 4. 创建Deep Agent，并授予其访问沙箱后端的能力
# 系统提示词定义了智能体的角色和能力
print("\n步骤2: 创建具有沙箱访问权限的智能体...")
agent = create_deep_agent(
model=llm, # 指定模型
backend=backend, # 关键：传入沙箱后端
system_prompt="""你是一个专业的Python编码助手，拥有在一个完全隔离的沙箱环境中执行命令和操作文件的能力。
你可以使用`execute`工具运行任何shell命令，使用`write_file`创建和编辑文件，使用`read_file`查看文件内容。
请用清晰、安全的代码回应用户的请求。""",
)
# 5. 定义任务：创建一个简单的Python程序并运行
user_request = """请完成以下任务：
1. 在沙箱的 /home/daytona/ 目录下，创建一个名为 ‘test_sandbox.py’ 的Python文件。
2. 文件内容应打印“Hello from the secure sandbox!”以及当前工作目录。
3. 运行这个Python脚本，并告诉我输出结果。"""
自定义企业服务器的SandBox
OpenSandbox相关介绍
OpenSandbox：是阿里巴巴开源的通用AI应用沙箱平台，提供多语言 SDK、统一的沙箱 API，以及 Docker/Kubernetes 运行时环境，专为解决AI时代代码执行的安全与效率问题而设计。它提供了一个安全、隔离、高效的执行环境，使AI模型能够安全地执行用户提交的代码，同时避免了对生产环境的潜在威胁。
它并非简单的Docker包装，而是一套完整的全栈平台，底层复用了阿里巴巴支撑大规模AI工作负载的内部基础设施，专门针对AIAgent的并行运行需求设计。其核心优势在于”全环境隔离”，这是它区别于Git Worktrees、普通容器化工具的核心亮点。
核心功能与技术特性
- 多语言 SDK：提供 Python、Java/Kotlin、JavaScript/TypeScript、C#/.NET、Go 的沙箱 SDK。
- 沙箱协议：定义了沙箱生命周期管理 API 和沙箱执行 API。你可以通过这些沙箱协议扩展自己的沙箱运行时。
- 沙箱运行时：沙箱全生命周期管理，支持 Docker 和自研高性能 Kubernetes 运行时，实现本地运行、企业级大规模分布式沙箱调度。
- 沙箱环境：内置 Command、Filesystem、Code Interpreter 实现。并提供 Coding Agent（Claude Code 等）、浏览器自动化（Chrome、Playwright）和桌面环境（VNC、VS Code）等示例。
- 网络策略：提供统一的 Ingress Gateway 实现，并支持多种路由策略；提供单实例级别的沙箱出口网络限制。
- 强隔离安全：支持 gVisor、Kata Containers 和 Firecracker 微虚拟机等安全容器运行时，为沙箱工作负载与宿主机之间提供增强的安全隔离。
安装并配置 Sandbox Server
环境要求：
- Docker（本地运行必需）
- Python 3.10+（示例和本地运行所需）
Linux中安装docker
基于docker部署Docker ≥ 24.0.0，下面在Linux中安装docker。这里默认用户已经安装好了Linux系统，
获取docker repo文件。
wget -O /etc/yum.repos.d/docker-ce.repo https://mirrors.aliyun.com/docker-ce/linux/centos/docker-ce.repo #如果下载docker对应版本有问题，可以选择使用镜像源2 wget -O /etc/yum.repos.d/docker-ce.repo https://download.docker.com/linux/centos/docker-ce.repo查看docker可以安装的版本：
yum list docker-ce.x86_64 --showduplicates | sort -r安装docker:这里指定docker版本为29.6.1版本
yum -y install docker-ce-29.6.1-1.el10设置docker 开机启动，并启动docker：
systemctl enable docker systemctl start docker修改cgroup的配置，尤其是镜像加速源，并重启docker。
# 如果文件不存在，需要创建。 vim /etc/docker/daemon.json添加如下内容
{ "features": { "containerd-snapshotter": true }, "exec-opts": ["native.cgroupdriver=systemd"], "registry-mirrors":[ "https://docker.m.daocloud.io", "https://docker.rainbond.cc", "https://docker.lmirror.top", "https://docker-0.unsee.tech", "https://docker.hlmirror.com" ] }重启docker
systemctl restart docker
Linux 安装OpenSandbox
安装python环境，可以先安装 Anaconda3
#创建 python环境 conda create --name sandbox python=3.11 #切换python环境 conda activate sandbox安装依赖-Code Interpreter SDK
pip install opensandbox-code-interpreter安装 Opensandbox-Server
pip install opensandbox-server opensandbox-server init-config ~/.sandbox.toml --example docker-zh 编辑： vi.sandbox.xml 1. 把ip改成linux节点ip 2. 配置 api_key
启动 Opensandbox-Server
opensandbox-server # 新开窗口Show help，查看帮助命令 opensandbox-server -hwindow本地测试sandbox是否正常可以连接
浏览器输入：http://192.168.11.2:8080/health ,可以看到{"status":"healthy"}就为正常。手动拉取镜像（在 opensandbox server 服务器端）
docker pull sandbox-registry.cn-zhangjiakou.cr.aliyuncs.com/opensandbox/code-interpreter:v1.0.2window本地创建代码测试创建沙箱
新建一个测试：test_opensanbox.py
import asyncio from datetime import timedelta from code_interpreter import CodeInterpreter, SupportedLanguage from opensandbox import Sandbox from opensandbox.config import ConnectionConfig from opensandbox.models import WriteEntry import os async def main() -> None: config = ConnectionConfig( domain="http://192.168.11.2:8080", use_server_proxy=True, api_key="1234567890abcdef", request_timeout=timedelta(seconds=60), # 请求超时时间，默认30秒 ) # 1. Create a sandbox sandbox = await Sandbox.create( "sandbox-registry.cn-zhangjiakou.cr.aliyuncs.com/opensandbox/code-interpreter:v1.0.2", entrypoint= ["/opt/opensandbox/code-interpreter.sh"], env={"PYTHON_VERSION": "3.13"}, timeout=timedelta(minutes=10), # 自动终止时间箱运行超时时间，默认10分钟，超过时间会自动终止沙箱 connection_config=config, ) async with sandbox: # 2. Execute a shell command execution = await sandbox.commands.run("echo 'Hello OpenSandbox!'") print(execution.logs.stdout[0].text) # 3. Write a file await sandbox.files.write_files([ WriteEntry(path="/tmp/hello.txt", data="Hello World", mode=644) ]) # 4. Read a file content = await sandbox.files.read_file("/tmp/hello.txt") print(f"Content: {content}") # Content: Hello World # 5. Create a code interpreter interpreter = await CodeInterpreter.create(sandbox) # 6. 执行 Python 代码（单次执行：直接传 language） result = await interpreter.codes.run( """ import sys print(sys.version) result = 2 + 2 result """, language=SupportedLanguage.PYTHON, ) print(result.result[0].text) # 4 print(result.logs.stdout[0].text) # 3.13.11 # 7. Cleanup the sandbox await sandbox.kill() if __name__ == "__main__": asyncio.run(main())配置沙箱的访问权限：
from opensandbox.models.sandboxes import NetworkPolicy, NetworkRule sandbox = await Sandbox.create( "python:3.11", ..... network_policy=NetworkPolicy( defaultAction="deny", egress=[ NetworkRule(action="allow", target="pypi.org"), NetworkRule(action="allow", target="*.github.com"), ] ) )
DeepAgents框架中自定义OpenSandboxBackend
在DeepAgents框架中有一个基类：BaseSandbox 所有沙盒功能都基于一个核心方法——execute()。这是一个巧妙的设计：
flowchart TD
%% 定义卡片内部文本左对齐，并赋予不同的背景颜色
%% 灰色代表抽象/基础方法，蓝色代表具体子类实现
classDef abstract fill:#f5f5f5,stroke:#9e9e9e,stroke-width:1px,text-align:left;
classDef concrete fill:#e3f2fd,stroke:#42a5f5,stroke-width:1px,text-align:left;
%% -------------------
%% 顶部：抽象基类沙盒
%% -------------------
subgraph Base [" BaseSandbox 基类 "]
direction TB
M1["<b>read_file()</b><br/>实现原理：<br/>execute("cat /path/to/file")"]:::abstract
M2["<b>write_file()</b><br/>实现原理：<br/>execute("echo 'content' > file")"]:::abstract
M3["<b>ls()</b><br/>实现原理：<br/>execute("ls -la")"]:::abstract
M4["<b>grep()</b><br/>实现原理：<br/>execute("grep pattern file")"]:::abstract
%% 使用隐形连接线强制内部这些节点垂直整齐排列
M1 ~~~ M2 ~~~ M3 ~~~ M4
end
%% -------------------
%% 底部：具体的厂商实现
%% -------------------
subgraph Provider [" 沙盒提供商的execute()实现 "]
direction TB
P1["<b>Daytona.execute("command")</b><br/>• 调用Daytona API<br/>• 在远程容器中执行命令"]:::concrete
P2["<b>Modal.execute("command")</b><br/>• 调用Modal云服务<br/>• 在隔离环境中执行"]:::concrete
P3["<b>Runloop.execute("command")</b><br/>• 调用Runloop API<br/>• 在开发箱中执行"]:::concrete
%% 使用隐形连接线强制内部这些节点垂直整齐排列
P1 ~~~ P2 ~~~ P3
end
%% -------------------
%% 连接两个模块主节点
%% -------------------
Base -->|"所有操作都转换为Shell命令"| Provider
两层文件访问平面
文件进出沙盒有两种截然不同的方式，理解何时使用哪种方式非常重要：
智能体文件系统工具 (Agent Filesystem Tools)
read_file、write_file、edit_file、ls、glob、grep和execute是 LLM 在执行过程中调用的工具。这些操作通过沙盒内部的execute()进行。智能体使用它们来读取代码、写入文件并作为任务的一部分运行命令。文件传输 API (File Transfer APIs)
uploadFiles()和downloadFiles()是您的应用程序代码调用的方法。它们使用提供商的本机文件传输 API（而非 Shell 命令），旨在在您的主机环境和沙盒之间移动文件。使用这些 API 来：- 初始化沙盒：在智能体运行前，向其填充源代码、配置或数据。
- 检索产物：在智能体完成后，获取生成的代码、构建输出、报告等。
- 预置依赖：预先准备好智能体需要的依赖项。
使用
upload_files()在智能体运行前填充沙盒。路径必须是绝对路径，内容必须是字节流（bytes）：使用
download_files()在智能体完成后从沙盒中检索文件或者下载文件到其他地方：
自定义OpenSandboxBackend类
from __future__ import annotations import logging from collections.abc import Callable from typing import cast from opensandbox import SandboxSync from deepagents.backends.protocol import ( ExecuteResponse, FileDownloadResponse, FileUploadResponse, ) from deepagents.backends.sandbox import BaseSandbox SyncPollingInterval = float | Callable[[float], float] PollingStrategy = Callable[[float], float] # 配置日志 logger = logging.getLogger(__name__) # logger.setLevel(logging.DEBUG) logger.setLevel(logging.ERROR) # 如果没有配置日志处理器，则添加一个 if not logger.handlers: handler = logging.StreamHandler() handler.setLevel(logging.DEBUG) formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') handler.setFormatter(formatter) logger.addHandler(handler) class OpenSandbox(BaseSandbox): """符合 SandboxBackendProtocol 协议的 OpenSandbox 实现。 该实现继承了 BaseSandbox 的所有文件操作方法， 并仅使用 OpenSandbox 的 API 实现了 execute()、download_files() 和 upload_files() 方法。 """ def __init__( self, *, sandbox: SandboxSync, timeout: int = 30 * 60, sync_polling_interval: SyncPollingInterval = 0.1, ) -> None: """创建一个包装已有 OpenSandbox 沙盒的后端实例。 Args： sandbox：要包装的现有 OpenSandbox 沙盒实例。 timeout：调用 `execute()` 且未显式指定 `timeout` 时使用的默认命令超时时间（秒）。 sync_polling_interval：在同步执行路径上，轮询 OpenSandbox 命令完成状态的间隔时间（秒）； 也可以是一个可调用对象，接收已执行的秒数并返回下一次轮询的延迟时间。 """ logger.info(f"正在初始化 OpenSandbox，沙盒 ID: {sandbox.id}") self._sandbox = sandbox # sandbox.kill() # 手动关闭沙箱 self._default_timeout = timeout # 处理轮询策略 if callable(sync_polling_interval): polling_strategy = cast("PollingStrategy", sync_polling_interval) else: def polling_strategy(_elapsed: float) -> float: return sync_polling_interval self._sync_polling_interval = polling_strategy logger.debug(f"OpenSandbox 初始化完成，默认超时时间={timeout}秒") @property def id(self) -> str: """返回 OpenSandbox 沙盒的 ID。""" sandbox_id = self._sandbox.id logger.debug(f"获取沙盒 ID: {sandbox_id}") return sandbox_id def execute( self, command: str, *, timeout: int | None = None, ) -> ExecuteResponse: """在沙盒内部执行一条 Shell 命令。 Args： command：要执行的 Shell 命令字符串。 timeout：等待命令完成的最大时间（秒）。 如果为 None，则使用后端默认的超时时间。 """ effective_timeout = timeout if timeout is not None else self._default_timeout logger.debug(f"准备执行命令：{command[:100]}...（超时时间={effective_timeout}秒）") return self._execute_command(command, timeout=effective_timeout) def _execute_command( self, command: str, *, timeout: int, ) -> ExecuteResponse: """使用 OpenSandbox 的 API 执行命令。""" try: logger.debug(f"通过 OpenSandbox API 执行命令：{command}") result = self._sandbox.commands.run(command) logger.debug(f"命令执行完成，退出码：{result.exit_code}") # 提取标准输出与标准错误 stdout = "" stderr = "" if result.logs.stdout: stdout = "\n".join([log.text for log in result.logs.stdout]) logger.debug(f"命令标准输出长度：{len(stdout)} 字符") if result.logs.stderr: stderr = "\n".join([log.text for log in result.logs.stderr]) logger.debug(f"命令标准错误长度：{len(stderr)} 字符") # 合并输出 output = stdout if stderr and stderr.strip(): output += f"\n<stderr>{stderr.strip()}</stderr>" logger.info(f"命令执行成功，退出码：{result.exit_code or 0}") return ExecuteResponse( output=output, exit_code=result.exit_code or 0, truncated=False, ) except Exception as e: error_msg = str(e) logger.error(f"执行命令时发生错误：{error_msg}", exc_info=True) if "timeout" in error_msg.lower(): logger.warning(f"命令在 {timeout} 秒后超时") return ExecuteResponse( output=f"命令在 {timeout} 秒后超时", exit_code=124, truncated=False, ) return ExecuteResponse( output=f"执行命令时出错：{error_msg}", exit_code=1, truncated=False, ) def download_files(self, paths: list[str]) -> list[FileDownloadResponse]: """从沙盒中下载文件。""" logger.info(f"开始下载 {len(paths)} 个文件：{paths}") responses: list[FileDownloadResponse] = [] for i, path in enumerate(paths): logger.debug(f"正在处理第 {i+1}/{len(paths)} 个文件：{path}") if not path.startswith("/"): logger.error(f"非法路径（必须是绝对路径）：{path}") responses.append( FileDownloadResponse(path=path, content=None, error="invalid_path") ) continue try: logger.debug(f"正在从沙盒读取文件：{path}") content = self._sandbox.files.read_file(path) content_bytes = content.encode("utf-8") if isinstance(content, str) else content logger.debug(f"文件读取成功，大小：{len(content_bytes)} 字节") responses.append( FileDownloadResponse( path=path, content=content_bytes, error=None, ) ) except Exception as e: logger.error(f"读取文件 {path} 时出错：{str(e)}", exc_info=True) # 尝试检查文件是否存在 try: logger.debug(f"检查文件是否存在：{path}") result = self._sandbox.commands.run(f"test -f '{path}' && echo 'exists'") if not result.logs.stdout or "exists" not in result.logs.stdout[0].text: logger.error(f"文件不存在：{path}") responses.append( FileDownloadResponse( path=path, content=None, error="file_not_found", ) ) else: logger.error(f"文件存在但读取失败：{path}，错误：{str(e)}") responses.append( FileDownloadResponse( path=path, content=None, error=f"read_error: {str(e)}", ) ) except Exception as check_error: logger.error(f"检查文件是否存在时出错：{check_error}", exc_info=True) responses.append( FileDownloadResponse( path=path, content=None, error=f"check_error: {str(check_error)}", ) ) success_count = sum(1 for r in responses if r.error is None) logger.info(f"文件下载完成，成功 {success_count}/{len(paths)} 个") return responses def upload_files(self, files: list[tuple[str, bytes]]) -> list[FileUploadResponse]: """将文件上传到沙盒中。""" from opensandbox.models import WriteEntry logger.info(f"准备上传 {len(files)} 个文件") responses: list[FileUploadResponse] = [] upload_entries = [] for i, (path, content) in enumerate(files): logger.debug(f"正在处理第 {i+1}/{len(files)} 个待上传文件：{path}，大小：{len(content)} 字节") if not path.startswith("/"): logger.error(f"非法路径（必须是绝对路径）：{path}") responses.append(FileUploadResponse(path=path, error="invalid_path")) continue try: # 将字节内容转换为字符串 if isinstance(content, bytes): try: content_str = content.decode("utf-8") logger.debug(f"已按 UTF-8 解码字节内容，长度：{len(content_str)} 字符") except UnicodeDecodeError as decode_error: logger.warning(f"UTF-8 解码失败，将以字符串形式存储：{decode_error}") content_str = str(content) else: content_str = str(content) upload_entries.append(WriteEntry(path=path, data=content_str, mode=0o644)) responses.append(FileUploadResponse(path=path, error=None)) logger.debug(f"文件已加入上传队列：{path}") except Exception as e: logger.error(f"准备上传文件 {path} 时出错：{str(e)}", exc_info=True) responses.append(FileUploadResponse(path=path, error=str(e))) # 如果有文件要上传 if upload_entries: logger.info(f"正在向沙盒写入 {len(upload_entries)} 个文件") try: self._sandbox.files.write_files(upload_entries) logger.info(f"成功上传 {len(upload_entries)} 个文件") except Exception as e: logger.error(f"上传文件时出错：{str(e)}", exc_info=True) # 如果有任何错误，更新所有响应 for i, resp in enumerate(responses): if resp.error is None: responses[i] = FileUploadResponse( path=resp.path, error=f"upload_failed: {str(e)}" ) else: logger.warning("没有有效的文件需要上传") # 统计上传结果 success_count = sum(1 for r in responses if r.error is None) error_count = len(responses) - success_count logger.info(f"文件上传完成，成功 {success_count} 个，失败 {error_count} 个") return responses