企业官网建设流程全解析

• 人工智能
• AI Agent
• Agent 框架
• 大模型
• 工具调用
• RAG
• 提示工程
• 强化学习

本文聚焦 openJiuwen agent-core 中沙箱操作层的核心入口SandboxGatewayClient，系统讲解其构造参数、invoke/invoke_stream全链路调用、端点解析与静态释放等全部 API，并结合SandboxGateway、SandboxRegistry、Provider 体系与真实测试用例，给出可直接运行的实战示例。读完本文，你将掌握如何通过沙箱网关客户端统一操作文件系统、Shell 与代码执行三类沙箱能力，并理解其底层“端点解析 → Provider 分发 → 方法调用”的完整链路。

SandboxGatewayClient是用户操作沙箱（Sandbox）的主接口，位于 gateway_client.py，同时支持端点解析（endpoint resolution）与全链路调用（full-chain invocation）两种模式。它是fs、shell、code三类沙箱操作的统一门面，也是所有沙箱模式业务（如 YuanRong、JiuwenBox、AIO 等 Provider）的请求入口。

类定义与构造参数

class SandboxGatewayClient( config: SandboxGatewayConfig, isolation_key: Optional[str], gateway: Optional[SandboxGateway] = None )

源码实现（gateway_client.py）中，gateway参数缺省时通过gateway or SandboxGateway.get_instance()回退到全局单例，因此业务侧通常只需要关心config与isolation_key两个参数。构造完成后，invoke/invoke_stream/get_endpoint/release四类 API 即可直接使用。

配置项：SandboxGatewayConfig 的核心字段

SandboxGatewayClient的行为高度依赖SandboxGatewayConfig（定义于 config.py），关键字段如下：

其中isolation（SandboxIsolationConfig）进一步包含：

• custom_id：核心身份覆盖，设置后取代自动生成的session_id/context_id；
• container_scope：容器粒度模板，取值为SYSTEM/SESSION/CUSTOM（ContainerScope枚举），默认SESSION；
• prefix：命名空间前缀，用于在同一作用域内隔离多个角色/任务。

launcher_config若使用PreDeployLauncherConfig（config.py），需要提供base_url（沙箱服务地址，http://或ws://）与sandbox_type（如aio、yuanrong），适用于沙箱进程由外部管理（已在服务器运行或由 sidecar 启动）的场景——launcher 直接返回给定base_url，不负责拉起任何进程。

完整配置定义可查阅 sandbox_config.md。

全链路调用：async invoke

async invoke(op_type: str, method: str, **params) -> Any

通过网关全链路路由发送调用请求。op_type为操作类型，取值"fs"（文件系统）、"shell"（Shell 执行）、"code"（代码执行）；method为具体方法名，如read_file、execute_cmd、execute_code。

实现上，invoke会先构造GatewayInvokeRequest(op_type=..., method=..., params=..., isolation_key=...)，再交给SandboxGateway.handle_request(config, request)处理，成功后将GatewayResponse.data返回（gateway_client.py）。

GatewayInvokeRequest（config.py）的定义：

class GatewayInvokeRequest(BaseModel): """Request model for Gateway full-chain routing.""" op_type: str = Field(description="Operation type: fs / shell / code") method: str = Field(description="Method name, e.g. read_file, execute_cmd") params: Dict[str, Any] = Field(default_factory=dict, description="Method parameters") isolation_key: Optional[str] = Field(default=None, description="Sandbox isolation key")

调用链拆解

在 gateway.py 中，handle_request遵循固定的三段式流程：

1. 解析端点：调用_get_or_create_provider(config, isolation_key, op_type)，内部先解析沙箱端点（_get_endpoint），再用SandboxRegistry.create_provider(sandbox_type=..., operation_type=op_type, endpoint=..., config=...)创建 Provider；
2. 缓存 Provider：以f"{isolation_key}:{op_type}"为缓存键，命中缓存则直接复用，避免重复创建；
3. 方法分发：通过getattr(provider, method)获取处理器并await handler(**params)执行，返回GatewayResponse(code=0, message=..., data=result)；方法不存在时返回Method '{method}' not found on provider错误。

调用示例

import asyncio from openjiuwen.core.sys_operation.config import ( SandboxGatewayConfig, PreDeployLauncherConfig, ) from openjiuwen.core.sys_operation.sandbox.gateway.gateway_client import SandboxGatewayClient async def main(): config = SandboxGatewayConfig( launcher_config=PreDeployLauncherConfig( base_url="http://127.0.0.1:8080", sandbox_type="aio", ), timeout_seconds=30, ) client = SandboxGatewayClient( config=config, isolation_key="my-sandbox-key", ) # fs 类型：读取文件 result = await client.invoke("fs", "read_file", path="/tmp/test.txt") print(result) # shell 类型：执行命令 result = await client.invoke("shell", "execute_cmd", command="echo hello") print(result.data.stdout) # code 类型：执行 Python 代码 result = await client.invoke("code", "execute_code", code="print(1+1)", language="python") print(result) # 释放沙箱资源 await SandboxGatewayClient.release("my-sandbox-key", on_stop="delete") asyncio.run(main())

流式调用：async invoke_stream

async invoke_stream(op_type: str, method: str, **params) -> AsyncIterator

与invoke相同的参数语义，区别在于以异步迭代器方式消费结果，适用于日志、长输出、增量执行等流式场景。其实现（gateway_client.py）将SandboxGateway.handle_stream_request返回的AsyncIterator原样透出，调用方用async for逐条消费：

async for item in client.invoke_stream("shell", "execute_cmd_stream", command="ping -c 3 localhost"): print(item)

在SandboxGateway.handle_stream_request（gateway.py）中，同样走“解析端点 → 选择 Provider → 调用流式方法”的链路，但方法缺失时直接抛AttributeError，而非返回错误响应。

端点解析：async get_endpoint

async get_endpoint() -> SandboxEndpoint

获取（必要时创建）沙箱端点，返回 SandboxEndpoint：

class SandboxEndpoint(BaseModel): base_url: str sandbox_id: Optional[str] = None isolation_key: Optional[str] = None

该 API 属于兼容保留的端点-only 旧接口（源码注释明确标注 "Legacy endpoint-only API (kept for backward compatibility)"）。实现上（gateway_client.py）构造SandboxCreateRequest(isolation_key=..., config=...)后调用gateway.get_sandbox(request)；返回的data若已是SandboxEndpoint则直接返回，若是dict则通过SandboxEndpoint(**endpoint)重建，否则抛出TypeError(f"Invalid endpoint payload: ...")。

静态释放资源：staticmethod release

staticmethod async release(isolation_key: str, on_stop: str = "delete") -> None

静态方法，仅凭隔离键即可通知网关回收资源。on_stop指定沙箱停止策略：

底层实现（gateway.py）对应release_sandbox：

1. _evict_provider_cache(isolation_key)清除该隔离键下的全部 Provider 缓存；
2. _store.hdel(isolation_key)取出并移除沙箱记录，无记录时返回错误Sandbox record not found；
3. 按on_stop分支执行：keep不做任何操作；pause通过 launcher 的pause(sandbox_id)暂停；delete通过 launcher 的delete(...)删除沙箱实例。

测试佐证

在 test_yuanrong_shell_operation.py 中可以看到典型的释放用法——从 Runner 资源管理器取出 SysOperation 后，用其isolation_key_template调用SandboxGatewayClient.release(..., on_stop="delete")完成沙箱回收：

async def _remove_sys_operation_with_sandbox_release(sys_operation_id: str) -> None: sys_op = Runner.resource_mgr.get_sys_operation(sys_operation_id) if sys_op is not None and sys_op.isolation_key_template: try: await SandboxGatewayClient.release(sys_op.isolation_key_template, on_stop="delete") except Exception as exc: if "not found" not in str(exc).lower(): raise Runner.resource_mgr.remove_sys_operation(sys_operation_id=sys_operation_id)

类似的释放调用还出现在 test_jiuwenbox.py、test_yuanrong.py 与各 fs/shell/code 端到端测试中，是沙箱资源生命周期收尾的标准做法。

与 fs / shell / code 操作及 SysOperation 的集成

SandboxGatewayClient并非孤立存在，而是被沙箱操作体系以 Mixin 方式复用。SandboxGatewayClientMixin（sandbox_mixin.py）封装了客户端管理与统一调用：

• _init_client_context(run_config, op_type)：保存配置、隔离键模板与操作类型；
• _get_resolved_isolation_key()：解析隔离键模板，将{session_id}占位符替换为当前会话真实session_id（缺省回退default_session），见_resolve_isolation_key_template；
• _get_gateway_client()：惰性创建并缓存SandboxGatewayClient(config=..., isolation_key=解析后的键)；
• invoke(method, **params)/invoke_stream(method, **params)：自动带上op_type转发给客户端。

基于该 Mixin，fs_operation.py（op_type="fs"）、shell_operation.py（op_type="shell"）、code_operation.py（op_type="code"）三类操作均在构造时调用_init_sandbox_context(run_config, op_type=...)，随后所有方法统一走self.invoke("read_file", ...)/self.invoke("execute_cmd", ...)等网关调用。

由此形成了用户侧的两级使用方式：

# 方式一：直接使用 GatewayClient（本文主接口） client = SandboxGatewayClient(config=config, isolation_key="key") await client.invoke("fs", "read_file", path="/tmp/a.txt") # 方式二：通过 SysOperation 的沙箱操作对象（内部同样走 GatewayClient） sys_op = Runner.resource_mgr.get_sys_operation(card_id) await sys_op.shell().execute_cmd(command="echo hello world")

两种方式最终都会汇入SandboxGatewayClient.invoke / invoke_stream，再进入SandboxGateway全链路路由。

底层支撑：SandboxGateway 与 SandboxRegistry

沙箱生命周期管理

SandboxGateway（gateway.py）是管理沙箱生命周期的单例，内部持有 Provider 缓存_provider_cache与InMemorySandboxStore（内存沙箱记录存储，SandboxRecord记录sandbox_id、base_url、状态、launcher 类型、容器配置哈希container_config_hash、最后使用时间等）。其_get_endpoint（gateway.py）实现了完整的端点解析状态机：

• 记录存在且RUNNING：直接复用并更新last_used_ts；
• 记录不存在：调用 launcher 新建沙箱（_create_new_sandbox，期间先_evict_idle按idle_ttl_seconds清理空闲沙箱）；
• 记录存在但真实状态异常：通过 launchercheck_status探测，RUNNING则复用；PAUSED则resume恢复；已删除则清除记录并重建。

沙箱创建时，SandboxRecord会计算容器级配置哈希（image、env、volumes、resource_limits、network、service_port），便于识别配置变更。

Provider 注册与分发

Provider 与 Launcher 统一由SandboxRegistry（sandbox_registry.py）管理：create_launcher(launcher_type)按类型创建 launcher（内置pre_deploy启动器在SandboxGateway构造时注册），create_provider(sandbox_type, operation_type, endpoint, config)创建操作 Provider。当前仓库内置三类 Provider 实现：

• aio.py：AIOFSProvider/AIOShellProvider/AIOCodeProvider；
• jiuwenbox.py：JiuwenBoxFSProvider/JiuwenBoxShellProvider/JiuwenBoxCodeProvider；
• yuanrong.py：YuanrongFSProvider/YuanrongShellProvider/YuanrongCodeProvider。

因此SandboxGatewayClient.invoke(op_type=..., method=...)中的method实际对应的是上述 Provider 暴露的方法，例如AIOFSProvider.read_file（aio.py#L215）、AIOShellProvider.execute_cmd（aio.py#L849）、AIOCodeProvider.execute_code（aio.py#L1007），以及各自的_stream流式变体。

错误处理与异常约定

invoke/get_endpoint/release在失败时会通过_raise_if_failed（gateway_client.py）抛出异常：

• 兼容新旧两种成功判定：优先读取response.success（legacy 字段），不存在时以code == 0判定成功；
• 失败时构造build_error(status=StatusCode.SYS_OPERATION_SANDBOX_GATEWAY_ERROR, operation=f"gateway_{op_type}", error_msg=...)抛出，其中operation形如gateway_fs、gateway_shell，便于定位是哪一类操作出错；
• 错误消息取自response.error或response.message，兜底为unknown error。

invoke_stream本身不包装异常，底层方法缺失时由SandboxGateway.handle_stream_request直接抛出AttributeError，调用方需自行捕获处理。

实战注意事项

1. 隔离键的重要性：isolation_key是沙箱复用的关键。使用带{session_id}占位符的模板（如"sandbox-{session_id}"），可在多会话场景下自动生成互不干扰的隔离键，避免沙箱串用；
2. on_stop 策略选择：频繁启停的开发场景建议"delete"避免资源残留；需要快速恢复且沙箱支持暂停/恢复（如pause/resume能力）时可选"pause"；"keep"仅适用于沙箱由外部独立管理、网关不负责生命周期的部署形态；
3. 流式接口必须异步消费：invoke_stream返回AsyncIterator，只能用async for消费，不可当作普通列表；
4. 网关单例与缓存：SandboxGateway以单例运行，Provider 按isolation_key:op_type缓存；release会清理对应缓存，因此释放后再次调用会重新解析端点并新建沙箱；
5. 配置一致性：launcher_config.sandbox_type决定 Provider 类型，必须与base_url指向的服务能力匹配（如aio、yuanrong、jiuwenbox），否则SandboxRegistry.create_provider会抛出NotImplementedError。

总结

SandboxGatewayClient是 openJiuwen 沙箱体系的统一操作门面：invoke/invoke_stream承载fs、shell、code三类操作的全链路调用，get_endpoint提供端点解析能力，静态release完成按隔离键的资源回收。它上接SysOperation沙箱操作对象（Mixin 自动注入），下连SandboxGateway单例、SandboxRegistryProvider 工厂与各类 launcher/Provider 实现，形成了一条清晰、可扩展、可测试的沙箱调用链路。开发者既可以经由高级操作对象使用沙箱，也可以直接以SandboxGatewayClient为入口做细粒度的网关调用——两者殊途同归，最终都汇入本文所讲的网关全链路路由。

• 人工智能
• AI Agent
• Agent 框架
• 大模型
• 工具调用
• RAG
• 提示工程
• 强化学习