代码沙箱

agentkit:
  tool_id: your-default-tool-id
  tool_id_script: your-script-tool-id
  tool_id_skills: your-skills-tool-id
  tool_id_opencode: your-opencode-tool-id

import asyncio

from veadk import Agent, Runner
from veadk.memory.short_term_memory import ShortTermMemory
from veadk.tools.builtin_tools.run_code import run_code
from veadk.tools.builtin_tools.web_search import web_search

agent = Agent(
    name="data_analysis_agent",
    model_name="doubao-seed-2-1-pro-260628",
    description="A data analysis agent for the stock market.",
    instruction=(
        "你是一名资深工程师，在沙箱中执行代码。可使用 web_search 搜索公司经营数据，"
        "可使用 akshare 等库下载股票数据；缺失依赖时用代码在沙箱中安装。"
    ),
    tools=[run_code, web_search],
)

runner = Runner(agent=agent, short_term_memory=ShortTermMemory())


async def main():
    response = await runner.run("分析阳光电源最近的股价走势")
    print(response)


if __name__ == "__main__":
    asyncio.run(main())

import asyncio

from google.adk.tools import ToolContext
from veadk import Agent, Runner
from veadk.memory.short_term_memory import ShortTermMemory
from veadk.tools.builtin_tools.run_code import run_code


def check_python(tool_context: ToolContext) -> str:
    """在远端沙箱中查看 Python 版本。"""
    return run_code(
        code="pwd && python --version",
        language="bash",
        tool_context=tool_context,
        timeout=60,
        exec_dir="/tmp",
        env={"APP_ENV": "test"},
        hard_timeout=120,
        max_output_length=5000,
    )


agent = Agent(
    name="shell_agent",
    instruction="调用工具查看远端沙箱的工作目录和 Python 版本。",
    tools=[check_python],
)
runner = Runner(agent=agent, short_term_memory=ShortTermMemory())


async def main():
    response = await runner.run("检查沙箱环境")
    print(response)


if __name__ == "__main__":
    asyncio.run(main())

import asyncio

from google.adk.tools import ToolContext
from veadk import Agent, Runner
from veadk.memory.short_term_memory import ShortTermMemory
from veadk.tools.builtin_tools.run_sandbox_agent import run_sandbox_agent


def run_custom_workflow(workflow_prompt: str, tool_context: ToolContext) -> str:
    """在指定的 AgentKit 沙箱中运行工作流。"""
    return run_sandbox_agent(
        workflow_prompt=workflow_prompt,
        tool_id="t-your-sandbox-tool-id",
        tool_context=tool_context,
        extra_env_vars={"DATA_SOURCE_URL": "https://example.com/data.csv"},
    )


agent = Agent(
    name="sandbox_agent",
    instruction="根据用户要求调用合适的沙箱工作流。",
    skills=["space:your-skill-space-id"],
    skills_mode="skills_sandbox",
    tools=[run_custom_workflow],
)
runner = Runner(agent=agent, short_term_memory=ShortTermMemory())


async def main():
    response = await runner.run("使用技能工作流读取知识库并生成摘要")
    print(response)


if __name__ == "__main__":
    asyncio.run(main())

import asyncio

from google.adk.tools import ToolContext
from veadk import Agent, Runner
from veadk.memory.short_term_memory import ShortTermMemory
from veadk.tools.builtin_tools.execute_skills import execute_skills


def run_skill_workflow(workflow_prompt: str, tool_context: ToolContext) -> str:
    """在技能沙箱中运行工作流，设置 600 秒超时。"""
    return execute_skills(
        workflow_prompt=workflow_prompt,
        tool_context=tool_context,
        timeout=600,
    )


agent = Agent(
    name="skill_agent",
    instruction="根据用户要求调用技能工作流。",
    skills=["space:your-skill-space-id"],
    skills_mode="skills_sandbox",
    tools=[run_skill_workflow],
)
runner = Runner(agent=agent, short_term_memory=ShortTermMemory())


async def main():
    response = await runner.run("使用技能工作流读取知识库并生成摘要")
    print(response)


if __name__ == "__main__":
    asyncio.run(main())