帮你快速理解、总结文档立即下载

使用 AgentBucket 挂载持久化存储

最近更新时间:2026-09-08 18:39:30
我的收藏
AgentBucket 是 Tencent Cloud Agent Runtime 基于 SMH(智能媒资托管)提供的沙箱持久化存储能力。本文通过“创建工具 → 启动沙箱 → 挂载 AgentBucket → 读写文件 → 验证持久化”的完整流程,展示如何在沙箱中使用 AgentBucket。

目标与结果

完成本快速入门后,您可以:
创建带 AgentBucket 存储挂载的沙箱工具。
启动沙箱实例,并将 AgentBucket 空间挂载到 /mnt/data/agentbucket
使用 E2B 文件 API 在挂载目录中写入和读回文件。
验证沙箱实例销毁后文件仍保留,并可被新实例读取。

前提条件

账号与资源配置

已开通腾讯云账号并取得 SecretIdSecretKey
已获取 E2B API Key 与 E2B 域名。
已创建 CAM Role,并取得完整 RoleArn
已申请 SMH library 和 spaceID

安装 SDK

python3 -m venv .venv
source .venv/bin/activate
python -m pip install -U pip tencentcloud-sdk-python
python -m pip install 'e2b_code_interpreter==2.2.1' 'e2b==2.23.1'

配置 RoleArn

使用 AgentBucket 时,RoleArn 必填。格式如下:
qcs::cam::uin/YOUR_UIN:roleName/YOUR_ROLE_NAME
建议按镜像来源关联以下 CAM 预设策略:
QcloudSMHFullAccess:AgentBucket 依赖的 SMH 访问权限。
QcloudCCRFullAccess:自定义镜像来自 CCR 时关联。
QcloudTCRFullAccess:自定义镜像来自 TCR 时关联。

配置 SMH

创建 AgentBucket 工具时传入 LibraryId,并在启动实例时通过 metadata.x-mounts[].subPath 传入 spaceID
字段
说明
LibraryId
用于创建 AgentBucket 工具级挂载,格式为 smh 后接 13 位小写字母或数字。
spaceID
用于启动实例时作为 metadata.x-mounts[].subPath 传入。

Custom 镜像配置

使用 custom 工具时,需要配置镜像、启动命令和 Probe。示例:
{
"Command": ["/bin/bash"],
"Args": ["-l", "-c", "/usr/bin/envd -port 49983 > /tmp/envd.log 2>&1 & sleep infinity"],
"Ports": [{"Name": "envd", "Port": 49983, "Protocol": "TCP"}],
"Probe": {
"HttpGet": {"Path": "/health", "Port": 49983, "Scheme": "HTTP"},
"ReadyTimeoutMs": 30000,
"ProbeTimeoutMs": 1000,
"ProbePeriodMs": 3000,
"SuccessThreshold": 1,
"FailureThreshold": 100
}
}
说明:
使用 ToolTypecode-interpreter 时无需 CustomConfiguration

挂载路径约束

推荐挂载到应用目录:
/mnt/data/agentbucket
/mnt/workspace
/mnt/input
/mnt/output
/mnt/logs
/opt/app/data
/home/USER/data
避免挂载到系统路径,例如 //bin/sbin/usr/lib/lib64/etc/proc/sys/var/dev/run

步骤概览

1. 配置环境变量。
2. 创建带 AgentBucket 的沙箱工具。
3. 通过 E2B 启动沙箱实例并读写挂载目录。
4. 验证持久化。

执行步骤

步骤 1:配置环境变量

export TENCENTCLOUD_SECRET_ID='YOUR_SECRET_ID'
export TENCENTCLOUD_SECRET_KEY='YOUR_SECRET_KEY'
export TENCENTCLOUD_REGION='ap-beijing'
export AGS_ENDPOINT='ags.ap-beijing.tencentcloudapi.com'

export TOOL_NAME='agentbucket-demo'
export ROLE_ARN='qcs::cam::uin/YOUR_UIN:roleName/YOUR_ROLE_NAME'
export IMAGE_ADDRESS='YOUR_REGISTRY/YOUR_NAMESPACE/YOUR_IMAGE:YOUR_TAG'
export IMAGE_REGISTRY_TYPE='personal'

export AGENTBUCKET_LIBRARY_ID='YOUR_LIBRARY_ID'
export AGENTBUCKET_SPACE_ID='YOUR_SPACE_ID'

export E2B_DOMAIN='ap-beijing.tencentags.com'
export E2B_API_KEY='YOUR_ARK_API_KEY'

步骤 2:创建带 AgentBucket 的沙箱工具

创建 create_agentbucket_tool.py
import json
import os
import time

from tencentcloud.common import credential
from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException
from tencentcloud.common.profile.client_profile import ClientProfile
from tencentcloud.common.profile.http_profile import HttpProfile
from tencentcloud.ags.v20250920 import ags_client

def require_env(name: str) -> str:
value = os.getenv(name)
if not value:
raise RuntimeError(f'Missing required environment variable: {name}')
return value

def build_ags_client() -> ags_client.AgsClient:
cred = credential.Credential(
require_env('TENCENTCLOUD_SECRET_ID'),
require_env('TENCENTCLOUD_SECRET_KEY'),
)
http_profile = HttpProfile()
http_profile.endpoint = os.getenv(
'AGS_ENDPOINT', 'ags.ap-beijing.tencentcloudapi.com'
)

client_profile = ClientProfile()
client_profile.httpProfile = http_profile

return ags_client.AgsClient(
cred,
os.getenv('TENCENTCLOUD_REGION', 'ap-beijing'),
client_profile,
)

def get_tool_id(resp: dict) -> str:
response = resp['Response']
if 'Tool' in response and 'ToolId' in response['Tool']:
return response['Tool']['ToolId']
if 'ToolId' in response:
return response['ToolId']
raise RuntimeError(f'ToolId not found in response: {json.dumps(resp, ensure_ascii=False)}')

def wait_tool_active(tool_id: str) -> dict:
client = build_ags_client()
deadline = time.time() + 300
while time.time() < deadline:
resp = client.call_json('DescribeSandboxToolList', {'ToolIds': [tool_id]})
tool_set = resp['Response'].get('SandboxToolSet', [])
if not tool_set:
time.sleep(3)
continue
tool = tool_set[0]
status = tool.get('Status')
if status == 'ACTIVE':
return tool
if status == 'FAILED':
raise RuntimeError(f'tool create failed: {json.dumps(tool, ensure_ascii=False)}')
time.sleep(3)
raise TimeoutError(f'tool {tool_id} did not become ACTIVE in time')

def main():
client = build_ags_client()

payload = {
'ToolName': require_env('TOOL_NAME'),
'ToolType': 'custom',
'RoleArn': require_env('ROLE_ARN'),
'NetworkConfiguration': {'NetworkMode': 'PUBLIC'},
'CustomConfiguration': {
'Image': require_env('IMAGE_ADDRESS'),
'ImageRegistryType': os.getenv('IMAGE_REGISTRY_TYPE', 'personal'),
'Command': ['/bin/bash'],
'Args': [
'-l',
'-c',
'/usr/bin/envd -port 49983 > /tmp/envd.log 2>&1 & sleep infinity',
],
'Ports': [
{'Name': 'envd', 'Port': 49983, 'Protocol': 'TCP'}
],
'Resources': {'CPU': '1000m', 'Memory': '1Gi'},
'Probe': {
'HttpGet': {'Path': '/health', 'Port': 49983, 'Scheme': 'HTTP'},
'ReadyTimeoutMs': 30000,
'ProbeTimeoutMs': 1000,
'ProbePeriodMs': 3000,
'SuccessThreshold': 1,
'FailureThreshold': 100,
},
},
'StorageMounts': [
{
'Name': 'data',
'MountPath': '/mnt/data/agentbucket',
'ReadOnly': False,
'StorageSource': {
'AgentBucket': {
'LibraryId': require_env('AGENTBUCKET_LIBRARY_ID'),
}
},
}
],
}

create_resp = client.call_json('CreateSandboxTool', payload)
tool_id = get_tool_id(create_resp)
tool = wait_tool_active(tool_id)
print(json.dumps({'ToolId': tool_id, 'Status': tool['Status']}, ensure_ascii=False, indent=2))
print(json.dumps(tool.get('StorageMounts', []), ensure_ascii=False, indent=2))

if __name__ == '__main__':
try:
main()
except TencentCloudSDKException as err:
print(f'CreateSandboxTool failed: {err}')
raise
运行并观察结果:
python create_agentbucket_tool.py
预期结果:
CreateSandboxTool 调用成功。
工具状态最终变为 ACTIVE
回查工具时,StorageMounts[].StorageSource.AgentBucket.LibraryId 与请求一致。
本示例创建工具时不传 SpaceId,目标 spaceID 在启动实例时通过 metadata.x-mounts[].subPath 传入。
配置了 CustomConfiguration 时,Probe 必须显式传入。

步骤 3:通过 E2B 启动实例并读写挂载目录

创建 e2b_agentbucket_files.py
import json
import os

from e2b_code_interpreter import Sandbox

def require_env(name: str) -> str:
value = os.getenv(name)
if not value:
raise RuntimeError(f'Missing required environment variable: {name}')
return value

os.environ['E2B_DOMAIN'] = require_env('E2B_DOMAIN')
os.environ['E2B_API_KEY'] = require_env('E2B_API_KEY')

mount_path = '/mnt/data/agentbucket'
sandbox = Sandbox.create(
template=require_env('TOOL_NAME'),
timeout=900,
metadata={
'x-mounts': json.dumps([
{
'name': 'data',
'mountPath': mount_path,
'readOnly': False,
'subPath': require_env('AGENTBUCKET_SPACE_ID'),
}
]),
},
)
try:
sandbox.files.write(f'{mount_path}/hello.txt', 'hello-agentbucket')
assert sandbox.files.exists(f'{mount_path}/hello.txt')
for item in sandbox.files.list(mount_path):
print(item.path, item.type)
print(sandbox.files.read(f'{mount_path}/hello.txt'))
finally:
sandbox.kill()
预期输出:
/mnt/data/agentbucket/hello.txt file
hello-agentbucket
如需执行 shell 任务,例如批量解压或运行脚本,可以使用 sandbox.commands.run(...)

步骤 4:验证持久化

第一次启动沙箱时写入文件,销毁实例后再使用相同 ToolName 和相同 metadata.x-mounts[].subPath 启动新实例并读取文件。
import json
import os

from e2b_code_interpreter import Sandbox

def require_env(name: str) -> str:
value = os.getenv(name)
if not value:
raise RuntimeError(f'Missing required environment variable: {name}')
return value

def create_sandbox():
mount_path = '/mnt/data/agentbucket'
sandbox = Sandbox.create(
template=require_env('TOOL_NAME'),
timeout=900,
metadata={
'x-mounts': json.dumps([
{
'name': 'data',
'mountPath': mount_path,
'readOnly': False,
'subPath': require_env('AGENTBUCKET_SPACE_ID'),
}
]),
},
)
return sandbox, mount_path

os.environ['E2B_DOMAIN'] = require_env('E2B_DOMAIN')
os.environ['E2B_API_KEY'] = require_env('E2B_API_KEY')

sb1, mount_path = create_sandbox()
try:
sb1.files.write(f'{mount_path}/persist.txt', 'persisted-by-agentbucket')
print('first sandbox id:', sb1.sandbox_id)
finally:
sb1.kill()

sb2, mount_path = create_sandbox()
try:
print('second sandbox id:', sb2.sandbox_id)
print(sb2.files.read(f'{mount_path}/persist.txt'))
finally:
sb2.kill()
预期结果:
两次的 sandbox_id 不同,说明确实启动了新实例。
第二个实例能读到 persisted-by-agentbucket,说明数据由 AgentBucket 持久化。
说明:
跨实例可见的前提是两次启动使用相同的 StorageMount.Name 与相同的 metadata.x-mounts[].subPath

结果验证

完成主流程后,可以观察到:
/mnt/data/agentbucket 可访问
/mnt/data/agentbucket/hello.txt 写入成功
cat /mnt/data/agentbucket/hello.txt 输出 hello-agentbucket
停止实例后使用相同 x-mounts[].subPath 重新启动,同一路径文件仍存在

使用示例

同一工具挂载不同 spaceID

同一个工具可以复用同一个 StorageMount.Name,不同实例启动时传入不同 subPath,实现数据空间隔离。
import json
import os

from e2b_code_interpreter import Sandbox

# 实例 A
sandbox_a = Sandbox.create(
template=os.environ['TOOL_NAME'],
timeout=900,
metadata={
'x-mounts': json.dumps([
{
'name': 'data',
'mountPath': '/mnt/data/agentbucket',
'readOnly': False,
'subPath': 'space-a',
}
]),
},
)

# 实例 B
sandbox_b = Sandbox.create(
template=os.environ['TOOL_NAME'],
timeout=900,
metadata={
'x-mounts': json.dumps([
{
'name': 'data',
'mountPath': '/mnt/data/agentbucket',
'readOnly': False,
'subPath': 'space-b',
}
]),
},
)

try:
sandbox_a.files.write('/mnt/data/agentbucket/only_in_a.txt', 'data-from-space-a')
sandbox_b.files.write('/mnt/data/agentbucket/only_in_b.txt', 'data-from-space-b')

assert not sandbox_a.files.exists('/mnt/data/agentbucket/only_in_b.txt')
assert not sandbox_b.files.exists('/mnt/data/agentbucket/only_in_a.txt')
assert sandbox_a.files.read('/mnt/data/agentbucket/only_in_a.txt') == 'data-from-space-a'
assert sandbox_b.files.read('/mnt/data/agentbucket/only_in_b.txt') == 'data-from-space-b'
print('隔离验证通过:两个实例文件互不可见')
finally:
sandbox_a.kill()
sandbox_b.kill()

只读挂载

实例级可以把可写挂载收紧为只读,但不能把工具级只读放宽为可写。
import json
import os

from e2b_code_interpreter import Sandbox

sandbox = Sandbox.create(
template=os.environ['TOOL_NAME'],
timeout=900,
metadata={
'x-mounts': json.dumps([
{
'name': 'data',
'mountPath': '/mnt/data/agentbucket',
'readOnly': True,
'subPath': os.environ['AGENTBUCKET_SPACE_ID'],
}
]),
},
)
预期结果:
files.list 能列出挂载目录文件。
files.write 写入失败或被拒绝。

常见问题

创建工具提示缺少 RoleArn

使用 AgentBucket 时,CreateSandboxTool 请求中的 RoleArn 必填。请检查是否填写了完整 ARN,而不是只填写 role name。

已配置 RoleArn 但实例启动或挂载失败

优先检查:
1. RoleArn 是否为完整 ARN。
2. 角色信任关系是否允许沙箱服务链路代入。
3. 角色权限是否覆盖目标 SMH library 或 space。
4. RoleArn 所属账号与 SMH 资源归属是否符合授权关系。

沙箱内看不到挂载目录

优先检查:
1. 实例是否启动成功并处于 RUNNING
2. metadata.x-mounts[].name 是否与工具级 StorageMount.Name 一致。
3. metadata.x-mounts[].mountPath 是否是您检查的路径。
4. metadata.x-mounts[].subPath 是否填写了目标 spaceID,且两次启动是否一致。
5. RoleArn 是否有访问目标 SMH 或 AgentBucket 的权限。
6. 当前地域是否同时开通 AgentBucket 服务和沙箱服务。
7. spaceID 是否真实存在。

写文件失败

优先检查:
1. 工具级 ReadOnly 是否为 true
2. 实例级 metadata.x-mounts[].readOnly 是否为 true
3. 写入路径是否位于挂载目录内。
4. 目标目录是否存在。

实例销毁后重新启动读不到文件

确认两次启动使用的是同一个实际目录:
metadata.x-mounts[].name 一致。
metadata.x-mounts[].subPath 一致,即传入同一个 spaceID
如果 mountPath 变化,请到新的容器路径下读取文件。

创建工具时 SpaceId 要不要填

在本文 E2B subPath = spaceID 模式下,创建工具时不传 StorageSource.AgentBucket.SpaceId,目标 spaceID 应在启动实例时通过 metadata.x-mounts[].subPath 传入。

metadata.x-mounts[].subPath 应该怎么填

在本文 AgentBucket E2B 模式下,subPath 填目标 spaceID。格式上不能以 / 开头,不能含连续斜杠,且不能包含 ... 路径段。