[使用教程] [体验WebUI] [English README] [示例]
核心能力:
工具沙箱化(Tool Sandboxing) —— 工具调用在加固的沙箱中运行
Agent 即服务(AaaS)API —— 将智能体以支持流式输出、可用于生产环境的 API 形式对外提供
可扩展部署 —— 支持本地、Kubernetes 或无服务器(Serverless)部署,实现弹性扩缩容
此外
全栈可观测性(日志 / 链路追踪)
框架兼容性 —— 兼容主流智能体框架
NOTE
推荐阅读顺序:
AgentApp 进行了核心架构重构。新版本采用直接继承 FastAPI 的设计,废弃了原有的工厂类模式,使开发者能够直接利用完整的 FastAPI 生态,显著提升了应用的可扩展性。此外,新版本引入了分布式任务中断管理服务,支持在 Agent 推理过程中进行实时干预,并允许灵活自定义中断前后的状态保存与恢复逻辑。完整更新内容与迁移说明请参考 CHANGELOG。BaseSandboxAsync、GuiSandboxAsync、BrowserSandboxAsync、FilesystemSandboxAsync、MobileSandboxAsync),支持在异步编程中进行非阻塞的并发工具执行。
同时优化了 run_ipython_cell 和 run_shell_command 方法的 并发与并行执行能力,提升沙箱运行效率。AgentApp方便部署并提供强大的自定义选项NOTE
关于框架无关:当前,AgentScope Runtime 支持 AgentScope 框架。未来我们计划扩展支持更多智能体开发框架。该表格展示了目前版本针对不同框架的适配器(adapter)支持情况,不同框架在各功能上的支持程度有所差异:
| 框架 / 功能项 | 消息 / 事件 | 工具 |
|---|---|---|
| AgentScope | ✅ | ✅ |
| LangGraph | ✅ | 🚧 |
| Microsoft Agent Framework | ✅ | ✅ |
| Agno | ✅ | ✅ |
| AutoGen | 🚧 | ✅ |
从PyPI安装:
# 安装核心依赖
pip install agentscope-runtime
# 安装拓展
pip install "agentscope-runtime[ext]"
# 安装预览版本
pip install --pre agentscope-runtime
(可选)从源码安装:
# 从 GitHub 拉取源码
git clone -b main https://github.com/agentscope-ai/agentscope-runtime.git
cd agentscope-runtime
# 安装核心依赖
pip install -e .
这个示例演示了如何使用 AgentScope 的 ReActAgent 和 AgentApp 创建一个代理 API 服务器。
要在 AgentScope Runtime 中运行一个最小化的 AgentScope Agent,通常需要实现以下内容:
定义生命周期 (lifespan) – 使用 contextlib.asynccontextmanager 管理服务启动时的资源初始化(如状态服务)和退出时的清理@agent_app.query(framework="agentscope") – 处理请求的核心逻辑,必须使用 stream_printing_messages 并 yield msg, last 来实现流式输出import os
from contextlib import asynccontextmanager
from fastapi import FastAPI
from agentscope.agent import ReActAgent
from agentscope.model import DashScopeChatModel
from agentscope.formatter import DashScopeChatFormatter
from agentscope.tool import Toolkit, execute_python_code
from agentscope.pipeline import stream_printing_messages
from agentscope.memory import InMemoryMemory
from agentscope.session import RedisSession
from agentscope_runtime.engine import AgentApp
from agentscope_runtime.engine.schemas.agent_schemas import AgentRequest
# 1. 定义生命周期管理器
@asynccontextmanager
async def lifespan(app: FastAPI):
"""管理服务启动和关闭时的资源"""
# 启动时:初始化 Session 管理器
import fakeredis
fake_redis = fakeredis.aioredis.FakeRedis(decode_responses=True)
# 注意:这个 FakeRedis 实例仅用于开发/测试。
# 在生产环境中,请替换为你自己的 Redis 客户端/连接
#(例如 aioredis.Redis)。
app.state.session = RedisSession(connection_pool=fake_redis.connection_pool)
yield # 服务运行中
# 关闭时:可以在此处添加清理逻辑(如关闭数据库连接)
print("AgentApp is shutting down...")
# 2. 创建 AgentApp 实例
agent_app = AgentApp(
app_name="Friday",
app_description="A helpful assistant",
lifespan=lifespan,
)
# 3. 定义请求处理逻辑
@agent_app.query(framework="agentscope")
async def query_func(
self,
msgs,
request: AgentRequest = None,
**kwargs,
):
session_id = request.session_id
user_id = request.user_id
toolkit = Toolkit()
toolkit.register_tool_function(execute_python_code)
agent = ReActAgent(
name="Friday",
model=DashScopeChatModel(
"qwen-turbo",
api_key=os.getenv("DASHSCOPE_API_KEY"),
stream=True,
),
sys_prompt="You're a helpful assistant named Friday.",
toolkit=toolkit,
memory=InMemoryMemory(),
formatter=DashScopeChatFormatter(),
)
agent.set_console_output_enabled(enabled=False)
# 加载状态
await agent_app.state.session.load_session_state(
session_id=session_id,
user_id=user_id,
agent=agent,
)
async for msg, last in stream_printing_messages(
agents=[agent],
coroutine_task=agent(msgs),
):
yield msg, last
# 保存状态
await agent_app.state.session.save_session_state(
session_id=session_id,
user_id=user_id,
agent=agent,
)
# 4. 启动应用
agent_app.run(host="127.0.0.1", port=8090)
运行后,服务器会启动并监听:http://localhost:8090/process。你可以使用 curl 向 API 发送 JSON 输入:
curl -N \
-X POST "http://localhost:8090/process" \
-H "Content-Type: application/json" \
-d '{
"input": [
{
"role": "user",
"content": [
{ "type": "text", "text": "What is the capital of France?" }
]
}
]
}'
你将会看到以 Server-Sent Events (SSE) 格式流式输出的响应:
data: {"sequence_number":0,"object":"response","status":"created", ... }
data: {"sequence_number":1,"object":"response","status":"in_progress", ... }
data: {"sequence_number":2,"object":"message","status":"in_progress", ... }
data: {"sequence_number":3,"object":"content","status":"in_progress","text":"The" }
data: {"sequence_number":4,"object":"content","status":"in_progress","text":" capital of France is Paris." }
data: {"sequence_number":5,"object":"message","status":"completed","text":"The capital of France is Paris." }
data: {"sequence_number":6,"object":"response","status":"completed", ... }
这些示例演示了如何创建沙箱环境并在其中执行工具,部分示例提供前端可交互页面(通过VNC,即Virtual Network Computing技术实现)
NOTE
如果你想在本地运行沙箱(sandbox),当前版本支持 Docker(可选配 gVisor) 或 BoxLite 作为后端,并且可以通过设置环境变量 CONTAINER_DEPLOYMENT 来切换(可选值包括 docker / gvisor / boxlite 等,默认 docker)。
对于大规模远程/生产环境部署,我们推荐使用 Kubernetes(K8s)、函数计算(Function Compute,FC),或 [阿里云容器服务 Kubernetes 版(ACK)](https://computenest.console.aliyun.com/service/instance/create/default?ServiceName=AgentScope Runtime 沙箱环境) 作为后端。更多细节请参考本教程。
TIP
AgentScope Runtime 为每种沙箱类型都提供了 同步版本 和 异步版本
| 同步类 | 异步类 |
|---|---|
BaseSandbox | BaseSandboxAsync |
GuiSandbox | GuiSandboxAsync |
FilesystemSandbox | FilesystemSandboxAsync |
BrowserSandbox | BrowserSandboxAsync |
MobileSandbox | MobileSandboxAsync |
TrainingSandbox | - (暂无异步版本) |
AgentbaySandbox | - (暂无异步版本) |
用于在隔离环境中运行 Python 代码 或 Shell 命令。
# --- 同步版本 ---
from agentscope_runtime.sandbox import BaseSandbox
with BaseSandbox() as box:
# 默认使用镜像 `agentscope/runtime-sandbox-base:latest` 从 DockerHub 拉取
print(box.list_tools()) # 列出所有可用工具
print(box.run_ipython_cell(code="print('你好')")) # 在沙箱中运行 Python 代码
print(box.run_shell_command(command="echo hello")) # 在沙箱中运行 Shell 命令
input("按 Enter 键继续...")
# --- 异步版本 ---
from agentscope_runtime.sandbox import BaseSandboxAsync
async with BaseSandboxAsync() as box:
# 默认使用镜像 `agentscope/runtime-sandbox-base:latest` 从 DockerHub 拉取
print(await box.list_tools_async()) # 列出所有可用工具
print(await box.run_ipython_cell(code="print('你好')")) # 在沙箱中运行 Python 代码
print(await box.run_shell_command(command="echo hello")) # 在沙箱中运行 Shell 命令
input("按 Enter 键继续...")
提供可视化桌面环境,可执行鼠标、键盘以及屏幕相关操作。
# --- 同步版本 ---
from agentscope_runtime.sandbox import GuiSandbox
with GuiSandbox() as box:
# 默认使用镜像 `agentscope/runtime-sandbox-gui:latest` 从 DockerHub 拉取
print(box.list_tools()) # 列出所有可用工具
print(box.desktop_url) # Web 桌面访问地址
print(box.computer_use(action="get_cursor_position")) # 获取鼠标位置坐标
print(box.computer_use(action="get_screenshot")) # 截取桌面截图
input("按 Enter 键继续...")
# --- 异步版本 ---
from agentscope_runtime.sandbox import GuiSandboxAsync
async with GuiSandboxAsync() as box:
# 默认使用镜像 `agentscope/runtime-sandbox-gui:latest` 从 DockerHub 拉取
print(await box.list_tools_async()) # 列出所有可用工具
print(box.desktop_url) # Web 桌面访问地址
print(await box.computer_use(action="get_cursor_position")) # 获取鼠标位置坐标
print(await box.computer_use(action="get_screenshot")) # 截取桌面截图
input("按 Enter 键继续...")
基于 GUI 的沙箱,可进行浏览器操作。
# --- 同步版本 ---
from agentscope_runtime.sandbox import BrowserSandbox
with BrowserSandbox() as box:
# 默认使用镜像 `agentscope/runtime-sandbox-browser:latest` 从 DockerHub 拉取
print(box.list_tools()) # 列出所有可用工具
print(box.desktop_url) # Web 桌面访问地址
box.browser_navigate("https://www.google.com/") # 打开网页
input("按 Enter 键继续...")
# --- 异步版本 ---
from agentscope_runtime.sandbox import BrowserSandboxAsync
async with BrowserSandboxAsync() as box:
# 默认使用镜像 `agentscope/runtime-sandbox-browser:latest` 从 DockerHub 拉取
print(await box.list_tools_async()) # 列出所有可用工具
print(box.desktop_url) # Web 桌面访问地址
await box.browser_navigate("https://www.google.com/") # 打开网页
input("按 Enter 键继续...")
基于 GUI 的隔离沙箱,可进行文件系统操作,如创建、读取和删除文件。
# --- 同步版本 ---
from agentscope_runtime.sandbox import FilesystemSandbox
with FilesystemSandbox() as box:
# 默认使用镜像 `agentscope/runtime-sandbox-filesystem:latest` 从 DockerHub 拉取
print(box.list_tools()) # 列出所有可用工具
print(box.desktop_url) # Web 桌面访问地址
box.create_directory("test") # 创建一个目录
input("按 Enter 键继续...")
# --- 异步版本 ---
from agentscope_runtime.sandbox import FilesystemSandboxAsync
async with FilesystemSandboxAsync() as box:
# 默认使用镜像 `agentscope/runtime-sandbox-filesystem:latest` 从 DockerHub 拉取
print(await box.list_tools_async()) # 列出所有可用工具
print(box.desktop_url) # Web 桌面访问地址
await box.create_directory("test") # 创建一个目录
input("按 Enter 键继续...")
提供一个沙箱化的 Android 模拟器环境,允许执行各种移动端操作,如点击、滑动、输入文本和截屏等。
Linux 主机:
该沙箱在 Linux 主机上运行时,需要内核加载 binder 和 ashmem 模块。如果缺失,请在主机上执行以下命令来安装和加载所需模块:
# 1. 安装额外的内核模块
sudo apt update && sudo apt install -y linux-modules-extra-`uname -r`
# 2. 加载模块并创建设备节点
sudo modprobe binder_linux devices="binder,hwbinder,vndbinder"
sudo modprobe ashmem_linux
架构兼容性: 在 ARM64/aarch64 架构(如 Apple M 系列芯片)上运行时,可能会遇到兼容性或性能问题,建议在 x86_64 架构的主机上运行。
# --- 同步版本 ---
from agentscope_runtime.sandbox import MobileSandbox
with MobileSandbox() as box:
# 默认使用镜像 'agentscope/runtime-sandbox-mobile:latest' 从 DockerHub 拉取
print(box.list_tools()) # 列出所有可用工具
print(box.mobile_get_screen_resolution()) # 获取屏幕分辨率
print(box.mobile_tap([500, 1000])) # 在坐标 (500, 1000) 点击
print(box.mobile_input_text("来自 AgentScope 的问候!")) # 输入文本
print(box.mobile_key_event(3)) # 发送 HOME 按键事件(KeyCode: 3)
screenshot_result = box.mobile_get_screenshot() # 截取屏幕
print(screenshot_result)
input("按 Enter 键继续...")
# --- 异步版本 ---
from agentscope_runtime.sandbox import MobileSandboxAsync
async with MobileSandboxAsync() as box:
# 默认使用镜像 'agentscope/runtime-sandbox-mobile:latest' 从 DockerHub 拉取
print(await box.list_tools_async()) # 列出所有可用工具
print(await box.mobile_get_screen_resolution()) # 获取屏幕分辨率
print(await box.mobile_tap([500, 1000])) # 在坐标 (500, 1000) 点击
print(await box.mobile_input_text("来自 AgentScope 的问候!")) # 输入文本
print(await box.mobile_key_event(3)) # 发送 HOME 按键事件(KeyCode: 3)
screenshot_result = await box.mobile_get_screenshot() # 截取屏幕
print(screenshot_result)
input("按 Enter 键继续...")
NOTE
要向 AgentScope 的 Toolkit 添加沙箱工具:
使用 sandbox_tool_adapter 包装沙箱工具,以便 AgentScope 中的 agent 可以调用它:
from agentscope_runtime.adapters.agentscope.tool import sandbox_tool_adapter
wrapped_tool = sandbox_tool_adapter(sandbox.browser_navigate)
使用 register_tool_function 注册工具:
toolkit = Toolkit() Toolkit.register_tool_function(wrapped_tool)
如果从 DockerHub 拉取镜像失败(例如由于网络限制),你可以将镜像源切换为阿里云容器镜像服务,以获得更快的访问速度:
export RUNTIME_SANDBOX_REGISTRY="agentscope-registry.ap-southeast-1.cr.aliyuncs.com"
命名空间用于区分不同的团队或项目镜像,你可以通过环境变量自定义 namespace:
export RUNTIME_SANDBOX_IMAGE_NAMESPACE="agentscope"
例如,这里会使用 agentscope 作为镜像路径的一部分。
镜像标签用于指定镜像版本,例如:
export RUNTIME_SANDBOX_IMAGE_TAG="preview"
其中:
latest,表示与PyPI发行版本适配的镜像版本preview 表示与 GitHub main 分支 同步构建的最新预览版本20250909,可以在DockerHub查看所有可用镜像版本沙箱 SDK 会根据上述环境变量拼接拉取镜像的完整路径:
<RUNTIME_SANDBOX_REGISTRY>/<RUNTIME_SANDBOX_IMAGE_NAMESPACE>/runtime-sandbox-base:<RUNTIME_SANDBOX_IMAGE_TAG>
示例:
agentscope-registry.ap-southeast-1.cr.aliyuncs.com/myteam/runtime-sandbox-base:preview
AgentScope Runtime 同样支持 serverless 部署,适用于在无服务器环境中运行沙箱,例如 阿里云函数计算(FC)。
首先,请参考文档配置 serverless 环境变量。将 CONTAINER_DEPLOYMENT 设置为 fc 以启用 serverless 部署。
然后,启动沙箱服务器,使用 --config 选项指定 serverless 环境配置:
# 此命令将加载 `fc.env` 文件中定义的设置
runtime-sandbox-server --config fc.env
服务器启动后,您可以通过URL http://localhost:8000 访问沙箱服务器,并调用上述描述的沙箱工具。
AgentApp 提供了一个 deploy 方法,该方法接收一个 DeployManager 实例并部署代理(agent)。
LocalDeployManager 时,通过参数 port 设置服务端口。endpoint_path 设置服务的端点路径为/process。部署后,可以通过 http://localhost:8090/process 访问该服务:
from agentscope_runtime.engine.deployers import LocalDeployManager
# 创建部署管理器
deployer = LocalDeployManager(
host="0.0.0.0",
port=8090,
)
# 部署应用
deploy_result = await app.deploy(
deployer=deployer,
endpoint_path="/process"
)
部署后用户也可以基于OpenAI SDK的Response API访问这个服务:
from openai import OpenAI
client = OpenAI(base_url="http://localhost:8090/compatible-mode/v1")
response = client.responses.create(
model="any_name",
input="杭州天气如何?"
)
print(response)
此外,DeployManager 也支持 Serverless 部署,例如将您的 agent 应用部署到
ModelStudio。
import os
from agentscope_runtime.engine.deployers.modelstudio_deployer import (
ModelstudioDeployManager,
OSSConfig,
ModelstudioConfig,
)
# 创建部署管理器
deployer = ModelstudioDeployManager(
oss_config=OSSConfig(
access_key_id=os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_ID"),
access_key_secret=os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_SECRET"),
),
modelstudio_config=ModelstudioConfig(
workspace_id=os.environ.get("MODELSTUDIO_WORKSPACE_ID"),
access_key_id=os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_ID"),
access_key_secret=os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_SECRET"),
dashscope_api_key=os.environ.get("DASHSCOPE_API_KEY"),
),
)
# 部署到 ModelStudio
result = await app.deploy(
deployer,
deploy_name="agent-app-example",
telemetry_enabled=True,
requirements=["agentscope", "fastapi", "uvicorn"],
environment={
"PYTHONPATH": "/app",
"DASHSCOPE_API_KEY": os.environ.get("DASHSCOPE_API_KEY"),
},
)
有关更高级的 serverless 部署指南,请参考文档。
欢迎加入我们的社区,获取最新的更新和支持!
| Discord | 钉钉群 |
|---|---|
![]() | ![]() |
我们欢迎来自社区的贡献!您可以提供以下帮助:
有关如何贡献的详细指南,请查看 如何贡献.
AgentScope Runtime 根据 Apache License 2.0 发布。
Copyright 2025 Tongyi Lab Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.
感谢这些优秀的贡献者们 (表情符号说明):
本项目遵循 all-contributors 规范。欢迎任何形式的贡献!