通过agentscope在EKS部署远程沙盒和代理应用

参考资料

• https://runtime.agentscope.io/zh/tool.html

AgentScope Runtime是一个面向 AI Agent 的全栈运行时，解决高效部署与服务化以及安全的沙箱化执行。

测试环境为python3.13，安装agentscope-runtime

uv add "agentscope-runtime[ext]"

Spring AI Alibaba 和 AgentScope-Java 的比较？

• Spring AI Alibaba：追求稳定可控、业务嵌入，覆盖 90% 企业级应用场景
• AgentScope-Java：若追求目标驱动、自主决策、多智能体协作，构建复杂智能体应用

sandbox沙盒工具和服务

Sandbox提供了一个安全且隔离的环境，用于工具执行、浏览器自动化、文件系统操作、训练评测等功能。Runtime支持即用和沙盒工具两种，本文只涉及到沙盒工具。

可以通过 Sandbox SDK 创建、连接、释放沙箱，默认使用docker运行sandbox，例如：

Base Image的示例如下

Filesystem Image的示例类似，但是额外可以通过Web桌面访问地址访问文件系统，这里的url很有趣，是一个vnc的link，即

• 通过websocket协议来连接浏览器的novnc客户端与容器中的vnc服务端，实现图形界面传输
• 而对于浏览器的操作，则使用了Chrome DevTools Protocol (CDP) 协议

sandbox sdk除了会启动对应的docker容器外，还会自动挂载本地目录到容器的/workspace下

"Binds": ["/home/xxx/sessions_mount_dir/c48d9ix2r7zl2zy766xwh0dq5:/workspace:rw"
]

runtime-sandbox-server

runtime-sandbox-server 是 Runtime Manager Service（运行时管理服务），用于集中管理和协调多个沙箱容器，如下日志的记录表明调用过程如下

• 为客户端动态创建沙箱容器，从沙箱池中分配可用实例
• 按照请求内容调用工具
• 使用完毕后释放和清理容器

runtime-sandbox-server 可以通过配置文件扩展功能，配置参考

export RUNTIME_SANDBOX_REGISTRY="agentscope-registry.ap-southeast-1.cr.aliyuncs.com"
runtime-sandbox-server --config k8s.env

通过额外的配置可以实现以下

• 身份令牌验证
• 预热容器池，当用户请求新沙箱时，系统将首先尝试从这个预热池中分配，相比从零开始创建容器显著减少启动时间。
• 指定容器运行时，支持docker、k8s、agentrun, fc、gvisor，关于不同后端的比较也进行了详细罗列
• 指定持久存储目录/workspace，默认为sessions_mount_dir
• 启用redis为沙箱状态和状态管理提供缓存，优势是简单且延迟更低

我们尝试在k8s上启动sandbox，配置如下

POOL_SIZE=1
DEFAULT_SANDBOX_TYPE=base,browser
CONTAINER_DEPLOYMENT=k8s
K8S_NAMESPACE=default
KUBECONFIG_PATH=/home/ec2-user/.kube/config
CONTAINER_PREFIX_KEY=runtime-sandbox-container-

启动结果如下

使用sandbox sdk连接远程执行，server报错如下，但是pod正常启动了

ERROR:agentscope_runtime.sandbox.manager.server.app:Error in call_tool: Runtime service did not start within the specified timeout.

检查发现server创建 Pod 时会自动创建一个 NodePort 类型的 名为runtime-sandbox-container-0ban2c523t0md7sx14w2c5g6v-service的svc，server会通过远程集群返回 Node 的 ExternalIP 或 InternalIP访问，优先使用ExternalIP 。在eks中配置的node的external ip地址为公网ip地址，由于没有放行规则导致无法连接。

暂时修改源码让InternalIP优先返回，始终返回internal ip，之后访问成功

runtime-sandbox-mcp

runtime-sandbox-mcp 具体在做什么？允许通过 MCP 协议远程访问沙箱功能，实际上通过mcp.run() 运行服务器，默认 transport 是 'stdio'，将sandbox-server暴露为mcp服务，例如如下的配置示例，通过远程的 sandbox 服务器 http://127.0.0.1:8000 来安全地执行沙箱中的命令

{"mcpServers": {"sandbox": {"command": "uvx","args": ["--from","agentscope-runtime","runtime-sandbox-mcp","--type=base","--base_url=http://127.0.0.1:8000"]}}
}

可见将两者搭配，可以通过runtime-sandbox-server在远程机器启动sandbox服务，然后通过runtime-sandbox-mcp将sandbox封装为mcp服务供agent使用

from agentscope_runtime.sandbox import BaseSandboxwith BaseSandbox(base_url="http://127.0.0.1:8000") as box:print(box.run_ipython_cell(code="print('hi')"))

还可以通过添加mcp的方式扩展sandbox tools

with BaseSandbox() as sandbox:mcp_server_configs = {...} # mcp配置服务提供get_current_time工具sandbox.add_mcp_servers(server_configs=mcp_server_configs) # # 将MCP服务器添加到沙箱print(sandbox.list_tools()) # 列出所有可用工具（现在包括MCP工具）sandbox.call_tool("get_current_time",arguments={"timezone": "America/New_York"})

sandbox service

• https://runtime.agentscope.io/v1.0.5/zh/service/sandbox.html

除了基本的运行方式，还可以通过 sandbox service 沙箱服务实现多会话场景下复用与隔离资源，具体是通过 session_id 和 user_id 来管理不同用户会话的沙箱环境

async def main():# sandbox_service = SandboxService() # 启动本地服务器# 指定远程服务器地址sandbox_service = SandboxService(base_url="http://your_IP_address:8000",  # 替换为实际的服务器IPbearer_token="your_token"  # 可选：如果需要身份验证)await sandbox_service.start()...sandboxes = sandbox_service.connect(session_id=session_id,user_id=user_id,sandbox_types=["base"],)base_sandbox = sandboxes[0]result = base_sandbox.run_ipython_cell("print('Hello, World!')")base_sandbox.run_ipython_cell("a=1")# 可以使用相同的 session_id 和 user_id 会复用同一个沙箱实例new_sandboxes = sandbox_service.connect(session_id=session_id,user_id=user_id,sandbox_types=["base"],)...

在EKS上部署agent应用

• https://runtime.agentscope.io/v1.0.5/zh/agent_app.html

LocalDeployManager部署

AgentScope Runtime提供多种不同的部署方式，LocalDeployManager部署模式示例如下，适合开发、测试和需要手动控制的持久服务的单用户场景。

import osfrom agentscope.agent import ReActAgent
from agentscope.model import OpenAIChatModel
from agentscope.formatter import OpenAIChatFormatter
from agentscope.tool import Toolkit, execute_python_code
from agentscope.pipeline import stream_printing_messages
from agentscope.memory import InMemoryMemoryfrom agentscope_runtime.engine import AgentApp
from agentscope_runtime.engine.schemas.agent_schemas import AgentRequest
from agentscope_runtime.engine.services.agent_state import (InMemoryStateService,
)
from agentscope_runtime.engine.deployers.local_deployer import LocalDeployManageragent_app = AgentApp(app_name="Friday",app_description="A helpful assistant",
)@agent_app.init
async def init_func(self):self.state_service = InMemoryStateService()  # 创建内存状态服务实例await self.state_service.start()  # 启动状态服务@agent_app.shutdown
async def shutdown_func(self):await self.state_service.stop()  # 停止并清理状态服务@agent_app.query(framework="agentscope")
async def query_func(self,msgs,request: AgentRequest = None,**kwargs,
):session_id = request.session_id  # 获取会话IDuser_id = request.user_id  # 获取用户ID# 尝试从状态服务加载之前的状态state = await self.state_service.export_state(session_id=session_id,user_id=user_id,)# 创建工具包并注册Python代码执行功能toolkit = Toolkit()toolkit.register_tool_function(execute_python_code)# 创建ReAct智能体实例agent = ReActAgent(name="Friday",model=OpenAIChatModel(model_name="qwen3-vl",api_key="sk-uzpq0u0n5FN14HorW45hUw",client_kwargs={"base_url": "http://localhost:4000"},stream=True,  # 启用流式响应),sys_prompt="You're a helpful assistant named Friday.",  # 系统提示词toolkit=toolkit,  # 绑定工具包memory=InMemoryMemory(),  # 使用内存存储记忆formatter=OpenAIChatFormatter(),  # 设置消息格式化器)agent.set_console_output_enabled(enabled=False)  # 禁用控制台输出# 如果存在之前的状态，则恢复智能体状态if state:agent.load_state_dict(state)# 执行智能体并流式返回消息async for msg, last in stream_printing_messages(agents=[agent],  # 指定要使用的智能体coroutine_task=agent(msgs),  # 执行智能体任务):yield msg, last  # 逐个返回消息# 获取智能体当前状态以供后续使用state = agent.state_dict()# 保存会话状态到状态服务await self.state_service.save_state(user_id=user_id,session_id=session_id,state=state,)async def main():# 部署agent_app到本地服务器，监听所有IP地址的8091端口deploy_manager = LocalDeployManager(host="0.0.0.0", port=8091)# 使用 standalone 模式，阻塞运行直到手动停止# 1. daemon_thread - 守护线程模式（默认）# 2. detached_process - 独立进程模式# 3. standalone - 独立模式await agent_app.deploy(deploy_manager, mode='standalone')if __name__ == "__main__":import asyncioasyncio.run(main())

之后使用openai客户端测试访问

from openai import OpenAI
client = OpenAI(base_url="http://127.0.0.1:8654/compatible-mode/v1",api_key="sk-placeholder"
)
response = client.responses.create(model="any_name",input="杭州天气如何？"
)
print(response)

KubernetesDeployManager部署

将local中的部署代码替换即可，agent app不变，示例如下

import asyncio
import os
from agentscope_runtime.engine.deployers.kubernetes_deployer import (KubernetesDeployManager,RegistryConfig,K8sConfig,
)# 使用agent_app
...async def deploy_to_k8s():# 配置镜像仓库和 K8s 连接deployer = KubernetesDeployManager(kube_config=K8sConfig(k8s_namespace="agentscope-runtime",kubeconfig_path=None,),registry_config=RegistryConfig(registry_url="000000000.dkr.ecr.cn-north-1.amazonaws.com.cn",namespace="agentscope-runtime",),use_deployment=True,)# 执行部署result = await agent_app.deploy(deployer,port="8080",replicas=1,image_name="agent_app",image_tag="v1.0",requirements=["agentscope", "fastapi", "uvicorn"],base_image="python:3.10-slim-bookworm",environment={"PYTHONPATH": "/app",},runtime_config={"resources": {"requests": {"cpu": "200m", "memory": "512Mi"},"limits": {"cpu": "1000m", "memory": "2Gi"},},},platform="linux/amd64",push_to_registry=True,)print(f"✅ 部署成功：{result['url']}")return result, deployerif __name__ == "__main__":asyncio.run(deploy_to_k8s())

实际上会使用本地docker打包镜像，并推送到指定的镜像仓库中（000000000000.dkr.ecr.cn-north-1.amazonaws.com.cn/agentscope-runtime/agent_app:v1.0），然后在eks中创建pod

结果如下，调用正常