手把手带你使用无影 AgentBay + AgentScope 完成一站式智能体开发部署

# 安装 AgentScope Runtime
pip install agentscope-runtime
# 安装 AgentBay SDK
pip install wuying-agentbay-sdk
# 安装 AgentScope 框架（如果还没有安装）
pip install agentscope

# 设置 AgentBay API Key
export AGENTBAY_API_KEY=your_agentbay_api_key
# 设置 百炼 API Key（用于模型调用）
export DASHSCOPE_API_KEY=your_dashscope_api_key

# .env 文件
AGENTBAY_API_KEY=your_agentbay_api_key
DASHSCOPE_API_KEY=your_dashscope_api_key

import os
from agentscope_runtime.sandbox.box.agentbay.agentbay_sandbox import (
    AgentbaySandbox,
)
# 创建 AgentBay Sandbox
sandbox = AgentbaySandbox(
    api_key=os.getenv("AGENTBAY_API_KEY"),
    image_id="linux_latest",  # 或 "browser_latest", "windows_latest" 等
)
# 执行 Shell 命令
result = sandbox.call_tool(
    "run_shell_command",
    {"command": "echo 'Hello from AgentBay!'"},
)
print(result)
# 文件操作
sandbox.call_tool(
    "write_file",
    {"path": "/tmp/test.txt", "content": "Hello World"},
)
# 读取文件
result = sandbox.call_tool("read_file", {"path": "/tmp/test.txt"})
print(result)
# 清理资源
sandbox._cleanup()

import asyncio
import os
import logging
from agentscope_runtime.sandbox.enums import SandboxType
from agentscope_runtime.engine.services.sandbox import SandboxService
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
async def main():
        # Create environment manager context
        async with SandboxService(bearer_token=os.getenv("AGENTBAY_API_KEY")) as service:
            # Connect AgentBay sandbox
            sandboxes = service.connect(
                session_id="demo_service_session",
                user_id="demo_user",
                sandbox_types=[SandboxType.AGENTBAY],
            )
            if not sandboxes:
                logger.error("No sandboxes returned by SandboxService")
                return False
            sandbox = sandboxes[0]
            logger.info(
                f"Connected AgentBay sandbox via service: "
                f"{sandbox.sandbox_id}",
            )
            # Basic shell command
            result = sandbox.call_tool(
                "run_shell_command",
                {"command": "echo 'Service path OK'"},
            )
            logger.info(f"Service command result: {result}")
            # File write & read
            write_res = sandbox.call_tool(
                "write_file",
                {"path": "/tmp/svc_test.txt", "content": "hello"},
            )
            logger.info(f"Service write result: {write_res}")
            read_res = sandbox.call_tool(
                "read_file",
                {"path": "/tmp/svc_test.txt"},
            )
            logger.info(f"Service read result: {read_res}")
            # Session info
            info = sandbox.get_session_info()
            logger.info(f"Service session info: {info}")
asyncio.run(main())

async def setup_agent(self):
    """Setup AgentScope agent."""
    try:
        logger.info("Setting up AgentScope agent...")
        # Initialize AgentScope
        init()
        # Create DashScope model
        model = DashScopeChatModel(
            model_name="qwen-max",
            api_key=self.dashscope_api_key,
        )
        # Create DashScope formatter
        formatter = DashScopeChatFormatter()
        # Create agent
        self.agent = agent.ReActAgent(
            name="CloudAssistant",
            sys_prompt="""You are an AI assistant with access to a
            cloud-based sandbox environment. You can help users execute
            commands, manage files, and perform various tasks in the cloud.
            When users ask you to do something, you can use the available
            tools to:
            - Execute shell commands
            - Read and write files
            - List directory contents
            - Create directories
            - Run Python code
            - Take screenshots
            Always explain what you're doing and provide helpful
            responses.""",
            model=model,
            formatter=formatter,
            toolkit=self._get_toolkit(),
        )
        logger.info("AgentScope agent created successfully")
        return True
    except Exception as e:
        logger.error(f"Failed to setup agent: {e}")
        return False

# 在 setup_environment 中
sandboxes = self.sandbox_service.connect(
 session_id="demo_session",
 user_id="demo_user",
 sandbox_types=[SandboxType.AGENTBAY],
 # 注意：如果需要指定镜像类型，可能需要通过其他方式配置
 # AgentBay 会根据需要自动选择合适的镜像
)

def _get_toolkit(self) -> Toolkit:
        """Create and configure toolkit with AgentBay tools."""
        toolkit = Toolkit()
        # Register tool functions
        toolkit.register_tool_function(
            self.execute_command,
            func_description=(
                "Execute shell commands in the cloud environment"
            ),
        )
        toolkit.register_tool_function(
            self.write_file,
            func_description=(
                "Write content to a file in the cloud environment"
            ),
        )
        toolkit.register_tool_function(
            self.read_file,
            func_description=(
                "Read content from a file in the cloud environment"
            ),
        )
       #注册其他工具函数
       ......
        return toolkit