人工智能AI AgentAgent 框架大模型工具调用RAG提示工程强化学习【免费下载链接】agent-coreopenJiuwen agent-core可提供AI Agent开发、运行、调优与演进相关的全套SDK能力项目地址https://gitcode.com/openJiuwen/agent-core点击查看免费下载本文聚焦 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 )参数类型说明configSandboxGatewayConfig沙箱网关配置决定沙箱隔离策略、启动方式与超时等isolation_keystr, optional沙箱隔离键用于区分不同的沙箱实例默认NonegatewaySandboxGateway, optional沙箱网关实例默认None此时自动取单例SandboxGateway.get_instance()源码实现gateway_client.py中gateway参数缺省时通过gateway or SandboxGateway.get_instance()回退到全局单例因此业务侧通常只需要关心config与isolation_key两个参数。构造完成后invoke/invoke_stream/get_endpoint/release四类 API 即可直接使用。配置项SandboxGatewayConfig 的核心字段SandboxGatewayClient的行为高度依赖SandboxGatewayConfig定义于 config.py关键字段如下字段类型默认值说明isolationSandboxIsolationConfigSandboxIsolationConfig()容器隔离与命名粒度策略launcher_configPreDeployLauncherConfig或SandboxLauncherConfigNone沙箱运行时的获取/连接方式现阶段支持PreDeployLauncherConfigaiotimeout_secondsint30统一超时请求 就绪探测单位秒auth_headersDict[str, str]{}鉴权 HTTP 请求头auth_query_paramsDict[str, str]{}鉴权查询参数其中isolationSandboxIsolationConfig进一步包含custom_id核心身份覆盖设置后取代自动生成的session_id/context_idcontainer_scope容器粒度模板取值为SYSTEM/SESSION/CUSTOMContainerScope枚举默认SESSIONprefix命名空间前缀用于在同一作用域内隔离多个角色/任务。launcher_config若使用PreDeployLauncherConfigconfig.py需要提供base_url沙箱服务地址http://或ws://与sandbox_type如aio、yuanrong适用于沙箱进程由外部管理已在服务器运行或由 sidecar 启动的场景——launcher 直接返回给定base_url不负责拉起任何进程。完整配置定义可查阅 sandbox_config.md。全链路调用async invokeasync invoke(op_type: str, method: str, **params) - Any通过网关全链路路由发送调用请求。op_type为操作类型取值fs文件系统、shellShell 执行、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。GatewayInvokeRequestconfig.py的定义class GatewayInvokeRequest(BaseModel): Request model for Gateway full-chain routing. op_type: str Field(descriptionOperation type: fs / shell / code) method: str Field(descriptionMethod name, e.g. read_file, execute_cmd) params: Dict[str, Any] Field(default_factorydict, descriptionMethod parameters) isolation_key: Optional[str] Field(defaultNone, descriptionSandbox isolation key)调用链拆解在 gateway.py 中handle_request遵循固定的三段式流程解析端点调用_get_or_create_provider(config, isolation_key, op_type)内部先解析沙箱端点_get_endpoint再用SandboxRegistry.create_provider(sandbox_type..., operation_typeop_type, endpoint..., config...)创建 Provider缓存 Provider以f{isolation_key}:{op_type}为缓存键命中缓存则直接复用避免重复创建方法分发通过getattr(provider, method)获取处理器并await handler(**params)执行返回GatewayResponse(code0, message..., dataresult)方法不存在时返回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_configPreDeployLauncherConfig( base_urlhttp://127.0.0.1:8080, sandbox_typeaio, ), timeout_seconds30, ) client SandboxGatewayClient( configconfig, isolation_keymy-sandbox-key, ) # fs 类型读取文件 result await client.invoke(fs, read_file, path/tmp/test.txt) print(result) # shell 类型执行命令 result await client.invoke(shell, execute_cmd, commandecho hello) print(result.data.stdout) # code 类型执行 Python 代码 result await client.invoke(code, execute_code, codeprint(11), languagepython) print(result) # 释放沙箱资源 await SandboxGatewayClient.release(my-sandbox-key, on_stopdelete) asyncio.run(main())流式调用async invoke_streamasync 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, commandping -c 3 localhost): print(item)在SandboxGateway.handle_stream_requestgateway.py中同样走“解析端点 → 选择 Provider → 调用流式方法”的链路但方法缺失时直接抛AttributeError而非返回错误响应。端点解析async get_endpointasync get_endpoint() - SandboxEndpoint获取必要时创建沙箱端点返回 SandboxEndpointclass 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(fInvalid endpoint payload: ...)。静态释放资源staticmethod releasestaticmethod async release(isolation_key: str, on_stop: str delete) - None静态方法仅凭隔离键即可通知网关回收资源。on_stop指定沙箱停止策略取值含义delete默认删除沙箱pause暂停沙箱下次启动可恢复keep保持沙箱运行由外部管理底层实现gateway.py对应release_sandbox_evict_provider_cache(isolation_key)清除该隔离键下的全部 Provider 缓存_store.hdel(isolation_key)取出并移除沙箱记录无记录时返回错误Sandbox record not found按on_stop分支执行keep不做任何操作pause通过 launcher 的pause(sandbox_id)暂停delete通过 launcher 的delete(...)删除沙箱实例。测试佐证在 test_yuanrong_shell_operation.py 中可以看到典型的释放用法——从 Runner 资源管理器取出 SysOperation 后用其isolation_key_template调用SandboxGatewayClient.release(..., on_stopdelete)完成沙箱回收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_stopdelete) except Exception as exc: if not found not in str(exc).lower(): raise Runner.resource_mgr.remove_sys_operation(sys_operation_idsys_operation_id)类似的释放调用还出现在 test_jiuwenbox.py、test_yuanrong.py 与各 fs/shell/code 端到端测试中是沙箱资源生命周期收尾的标准做法。与 fs / shell / code 操作及 SysOperation 的集成SandboxGatewayClient并非孤立存在而是被沙箱操作体系以 Mixin 方式复用。SandboxGatewayClientMixinsandbox_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转发给客户端。基于该 Mixinfs_operation.pyop_typefs、shell_operation.pyop_typeshell、code_operation.pyop_typecode三类操作均在构造时调用_init_sandbox_context(run_config, op_type...)随后所有方法统一走self.invoke(read_file, ...)/self.invoke(execute_cmd, ...)等网关调用。由此形成了用户侧的两级使用方式# 方式一直接使用 GatewayClient本文主接口 client SandboxGatewayClient(configconfig, isolation_keykey) 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(commandecho hello world)两种方式最终都会汇入SandboxGatewayClient.invoke / invoke_stream再进入SandboxGateway全链路路由。底层支撑SandboxGateway 与 SandboxRegistry沙箱生命周期管理SandboxGatewaygateway.py是管理沙箱生命周期的单例内部持有 Provider 缓存_provider_cache与InMemorySandboxStore内存沙箱记录存储SandboxRecord记录sandbox_id、base_url、状态、launcher 类型、容器配置哈希container_config_hash、最后使用时间等。其_get_endpointgateway.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 统一由SandboxRegistrysandbox_registry.py管理create_launcher(launcher_type)按类型创建 launcher内置pre_deploy启动器在SandboxGateway构造时注册create_provider(sandbox_type, operation_type, endpoint, config)创建操作 Provider。当前仓库内置三类 Provider 实现aio.pyAIOFSProvider/AIOShellProvider/AIOCodeProviderjiuwenbox.pyJiuwenBoxFSProvider/JiuwenBoxShellProvider/JiuwenBoxCodeProvideryuanrong.pyYuanrongFSProvider/YuanrongShellProvider/YuanrongCodeProvider。因此SandboxGatewayClient.invoke(op_type..., method...)中的method实际对应的是上述 Provider 暴露的方法例如AIOFSProvider.read_fileaio.py#L215、AIOShellProvider.execute_cmdaio.py#L849、AIOCodeProvider.execute_codeaio.py#L1007以及各自的_stream流式变体。错误处理与异常约定invoke/get_endpoint/release在失败时会通过_raise_if_failedgateway_client.py抛出异常兼容新旧两种成功判定优先读取response.successlegacy 字段不存在时以code 0判定成功失败时构造build_error(statusStatusCode.SYS_OPERATION_SANDBOX_GATEWAY_ERROR, operationfgateway_{op_type}, error_msg...)抛出其中operation形如gateway_fs、gateway_shell便于定位是哪一类操作出错错误消息取自response.error或response.message兜底为unknown error。invoke_stream本身不包装异常底层方法缺失时由SandboxGateway.handle_stream_request直接抛出AttributeError调用方需自行捕获处理。实战注意事项隔离键的重要性isolation_key是沙箱复用的关键。使用带{session_id}占位符的模板如sandbox-{session_id}可在多会话场景下自动生成互不干扰的隔离键避免沙箱串用on_stop 策略选择频繁启停的开发场景建议delete避免资源残留需要快速恢复且沙箱支持暂停/恢复如pause/resume能力时可选pausekeep仅适用于沙箱由外部独立管理、网关不负责生命周期的部署形态流式接口必须异步消费invoke_stream返回AsyncIterator只能用async for消费不可当作普通列表网关单例与缓存SandboxGateway以单例运行Provider 按isolation_key:op_type缓存release会清理对应缓存因此释放后再次调用会重新解析端点并新建沙箱配置一致性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 AgentAgent 框架大模型工具调用RAG提示工程强化学习【免费下载链接】agent-coreopenJiuwen agent-core可提供AI Agent开发、运行、调优与演进相关的全套SDK能力项目地址https://gitcode.com/openJiuwen/agent-core点击查看免费下载相关推荐openJiuwen agent-core 沙箱网关 SandboxGateway 深入解析全链路路由、生命周期管理与配置实战openJiuwen agent core 沙箱网关 SandboxGateway 深入解析全链路路由、生命周期管理与配置实战 导读 SandboxGatew人工智能AI AgentAgent 框架大模型工具调用RAG提示工程强化学习openJiuwen agent-core 沙箱网关调用基座SandboxGatewayClientMixin 与 BaseSandboxMixin 深入解析openJiuwen agent core 沙箱网关调用基座SandboxGatewayClientMixin 与 BaseSandboxMixin 深入解析人工智能AI AgentAgent 框架大模型工具调用RAG提示工程强化学习BentoML 客户端调用 API 端点SyncHTTPClient 与 AsyncHTTPClient 实战指南BentoML 客户端调用 API 端点SyncHTTPClient 与 AsyncHTTPClient 实战指南 BentoML 为 bentoml.Ser模型推理服务人工智能后端大模型MLOpsLLMOps上一篇避免磁盘空间耗尽websocketd日志轮转全攻略下一篇Vue.Draggable与GitHub Actions制品上传npm包发布创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
