Agent Code Executor

为了提供 Agent 高度的灵活性，有些时候会需要 Agent 生成并执行一些代码，tRPC-Agent 框架为这种场景支持了 CodeExecutor。

当启用了该功能之后，如果LLM返回文本中有代码片段时，框架会调用相应的 CodeExecutor 执行代码，并将执行结果返回给 LLM，LLM 可以根据执行结果继续生成响应。

代码执行器类型

目前提供下面三种代码执行器：

UnsafeLocalCodeExecutor

特点： - 在 Agent 所在的进程中执行LLM生成的代码 - 非沙箱环境，直接使用本地 Python/Bash 环境，目前仅支持 Python/Bash 语言 - 执行速度快，无需 Docker 环境 - 安全警告：LLM 生成的代码可能有风险，不适合用在生产环境

适用场景： - 开发和测试环境 - 受信任的代码执行场景 - 需要快速迭代和调试的场景

ContainerCodeExecutor

特点： - Agent派发代码片段到 Docker 容器中执行，目前仅支持 Python/Bash 语言 - 沙箱环境，提供更好的隔离和安全性 - 支持自定义 Docker 镜像或 Dockerfile - 需要 Docker 环境支持

适用场景： - 生产环境 - 需要执行不可信代码的场景 - 需要环境隔离的场景

CubeCodeExecutor

特点： - Agent 派发代码片段到远端 Cube/E2B 沙箱中执行，支持 Python/Bash 语言 - 强沙箱环境，运行在远端宿主上，适合大规模执行不可信代码 - 生命周期解耦：通过 sandbox_id 可以跨进程重新挂接到同一个沙箱（提供 create / attach / create_or_recreate 三种工厂方法） - 附带可选的 CubeWorkspaceRuntime：提供按执行隔离的工作目录、文件/目录上传下载（目录走 tar 协议）、结构化程序运行能力，可用于 Skill 子系统 - 需要安装可选 extra [cube]（pip install 'trpc-agent-py[cube]'，会带上 e2b-code-interpreter），并能访问 Cube/E2B 兼容网关

适用场景： - 生产环境，且 Agent 宿主上没有 Docker - 对不可信代码需要强远端隔离的场景 - 需要长期复用工作空间、跨多次 execute_code 共享文件的 Skill / 代码执行任务 - 多租户 Agent 平台，共用一组远端沙箱集群

使用示例

创建 LlmAgent 时，构建 CodeExecutor 并配置code_executor参数，即可启用代码执行功能。

构建 CodeExecutor

from trpc_agent_sdk.agents import LlmAgent
from trpc_agent_sdk.models import LLMModel
from trpc_agent_sdk.models import OpenAIModel
from trpc_agent_sdk.code_executors import BaseCodeExecutor
from trpc_agent_sdk.code_executors import UnsafeLocalCodeExecutor
from trpc_agent_sdk.code_executors import ContainerCodeExecutor
# Cube 是可选 extra（`pip install 'trpc-agent-py[cube]'`），按需引入。
from trpc_agent_sdk.code_executors.cube import CubeCodeExecutor
from trpc_agent_sdk.code_executors.cube import CubeCodeExecutorConfig
from trpc_agent_sdk.log import logger

async def _create_code_executor(code_executor_type: str = "unsafe_local") -> BaseCodeExecutor:
    """Create a code executor.

    Args:
        code_executor_type: Type of code executor to use. Options:
            - "unsafe_local": Use UnsafeLocalCodeExecutor (default, no Docker required)
            - "container": Use ContainerCodeExecutor (requires Docker)
            - "cube": Use CubeCodeExecutor (requires the [cube] extra and a Cube/E2B gateway)
            - None: Auto-detect from environment variable CODE_EXECUTOR_TYPE,
                    or default to "unsafe_local"

    Returns:
        BaseCodeExecutor instance.

    Raises:
        RuntimeError: If container type is requested but Docker is not available.
            The error message will include detailed instructions on how to fix the issue.
    """
    # Get executor type from environment variable if not specified
    if code_executor_type == "unsafe_local":
        return UnsafeLocalCodeExecutor(timeout=10)
    elif code_executor_type == "container":
        # ContainerCodeExecutor will raise a clear error if Docker is not available
        # The error message includes detailed instructions on how to fix the issue
        executor = ContainerCodeExecutor(image="python:3-slim", error_retry_attempts=1)
        logger.info("ContainerCodeExecutor initialized successfully")
        return executor
    elif code_executor_type == "cube":
        # CubeCodeExecutor 在 cfg 字段未设时，会从环境变量读取
        # E2B_API_URL / E2B_API_KEY / CUBE_TEMPLATE_ID。
        # `create()` 会开一个新的远端沙箱；如果想挂接到已有沙箱，
        # 在 cfg 里传 `sandbox_id=...` 即可。
        cfg = CubeCodeExecutorConfig(execute_timeout=30.0, idle_timeout=600)
        executor = await CubeCodeExecutor.create(cfg)
        logger.info("CubeCodeExecutor initialized: sandbox_id=%s", executor.sandbox_id)
        return executor
    else:
        raise ValueError(f"Invalid code executor type: {code_executor_type}. "
                         "Valid options are: 'unsafe_local', 'container', 'cube'")

使用 UnsafeLocalCodeExecutor

# ...
def create_agent() -> LlmAgent:
    """Create an agent with code execution capabilities.

    The agent can:
    - Execute Python code blocks generated by the LLM
    - Use tools like get_weather_report
    - Perform calculations and data processing through code execution

    Note: UnsafeLocalCodeExecutor executes code in the current process context.
    For production use, consider using ContainerCodeExecutor for better security.
    """
    # 选择 unsafe_local
    executor = _create_code_executor(code_executor_type="unsafe_local")
    agent = LlmAgent(
        name="code_assistant",
        description="代码执行助手",
        model=_create_model(),  # You can change this to your preferred model
        instruction=INSTRUCTION,
        code_executor=executor,  # Enables code execution functionality
    )
    return agent


root_agent = create_agent()

执行结果示例：

UnsafeLocalCodeExecutor执行结果 UnsafeLocalCodeExecutor执行结果1

使用ContainerCodeExecutor


# ...
def create_agent() -> LlmAgent:
    """Create an agent with code execution capabilities.

    The agent can:
    - Execute Python code blocks generated by the LLM
    - Use tools like get_weather_report
    - Perform calculations and data processing through code execution

    Note: UnsafeLocalCodeExecutor executes code in the current process context.
    For production use, consider using ContainerCodeExecutor for better security.
    """
    # 选择 container
    executor = _create_code_executor(code_executor_type="container")
    agent = LlmAgent(
        name="code_assistant",
        description="代码执行助手",
        model=_create_model(),  # You can change this to your preferred model
        instruction=INSTRUCTION,
        code_executor=executor,  # Enables code execution functionality
    )
    return agent

# 使用前确保Docker已安装并运行
# Linux: sudo systemctl start docker
# Windows/Mac: 启动Docker Desktop

执行结果示例：

ContainerCodeExecutor执行结果 ContainerCodeExecutor执行结果1

使用 CubeCodeExecutor

# ...
async def create_agent() -> LlmAgent:
    """Create an agent backed by a remote Cube/E2B sandbox.

    必备环境变量（由 CubeCodeExecutorConfig.resolve_* 读取）：
    - E2B_API_URL:      Cube/E2B 兼容网关 URL
    - E2B_API_KEY:      网关 API Key
    - CUBE_TEMPLATE_ID: Cube 模板 id（如 `std-XXXXXXXX`）

    说明：`_create_code_executor` 改成 async 是因为 `CubeCodeExecutor.create`
    需要走网络打开远端沙箱。executor 持有该沙箱，Agent 退出时
    需要 `await executor.destroy()` 显式释放远端资源；`executor.close()`
    只丢本地句柄，沙箱会按 idle_timeout 自然过期。
    """
    # 选择 cube
    executor = await _create_code_executor(code_executor_type="cube")
    agent = LlmAgent(
        name="code_assistant",
        description="代码执行助手",
        model=_create_model(),  # You can change this to your preferred model
        instruction=INSTRUCTION,
        code_executor=executor,  # Enables code execution functionality
    )
    return agent

# 使用前先安装可选 extra：
#   pip install 'trpc-agent-py[cube]'
# 并导出网关凭据：
#   export E2B_API_URL=...
#   export E2B_API_KEY=...
#   export CUBE_TEMPLATE_ID=...

挂接到已存在的沙箱

CubeCodeExecutor 提供三个 async 工厂方法，让调用方显式选择生命周期策略。 三者都从 cfg.sandbox_id 读取目标沙箱 id（单一来源）：

# 1. 严格 create-or-attach：cfg.sandbox_id 已设则挂接并断言 RUNNING；
#    未设则新建一个。
executor = await CubeCodeExecutor.create(cfg)

# 2. 仅挂接：要求 cfg.sandbox_id 已设；不会新建。
executor = await CubeCodeExecutor.attach(cfg)

# 3. attach-or-recreate：沙箱已不存在时调用 `on_recreate`，再透明地新建。
#    适合需要在 recreate 时清理外部 locator 状态的长生命周期 Agent。
executor = await CubeCodeExecutor.create_or_recreate(
    cfg, on_recreate=lambda old_id: clear_locator(old_id),
)

close() 对远端沙箱是 no-op，只丢本地句柄；destroy() 才会显式杀掉 远端沙箱。

配置参数

UnsafeLocalCodeExecutor 参数

from trpc_agent_sdk.code_executors import UnsafeLocalCodeExecutor, CodeBlockDelimiter

code_executor = UnsafeLocalCodeExecutor(
    # 代码执行失败重试次数，默认是2次
    error_retry_attempts=2,

    # 代码块分隔符，用于识别LLM返回中的代码块
    # 默认支持: ```tool_code\n 和 ```python\n
    code_block_delimiters=[
        CodeBlockDelimiter(start="```python\n", end="\n```"),
        CodeBlockDelimiter(start="```tool_code\n", end="\n```"),
    ],

    # 工作目录，如果为空则使用临时目录
    work_dir="",

    # 代码执行超时时间（秒）
    timeout=10,

    # 是否在执行后清理临时文件，默认True
    clean_temp_files=True,
)

ContainerCodeExecutor 参数

from trpc_agent_sdk.code_executors import ContainerCodeExecutor, CodeBlockDelimiter

code_executor = ContainerCodeExecutor(
    # Docker镜像名称（必需，与docker_path二选一）
    image="python:3-slim",

    # Dockerfile路径（必需，与image二选一）
    # docker_path="/path/to/Dockerfile",

    # 远程Docker的base_url（可选）
    # base_url="tcp://remote-docker-host:2375",

    # 代码执行失败重试次数，默认是2次
    error_retry_attempts=2,

    # 代码块分隔符，默认使用 ```tool_code\n
    code_block_delimiters=[
        CodeBlockDelimiter(start="```tool_code\n", end="\n```"),
    ],
)

CubeCodeExecutor 参数

CubeCodeExecutor 按 ISP 拆成两个 dataclass：CubeCodeExecutorConfig 只承载沙箱生命周期 / 命令执行相关字段，CubeWorkspaceRuntimeConfig 只承载工作空间相关字段（见下一节）。

from trpc_agent_sdk.code_executors.cube import (
    CubeCodeExecutor,
    CubeCodeExecutorConfig,
)

cfg = CubeCodeExecutorConfig(
    # 新建沙箱使用的 Cube 模板 id；为空时回退到环境变量 CUBE_TEMPLATE_ID。
    template=None,

    # E2B 兼容的 Cube API URL；为空时回退到环境变量 E2B_API_URL。
    api_url=None,

    # E2B API Key；为空时回退到环境变量 E2B_API_KEY。
    api_key=None,

    # 已存在的远端沙箱 id。设置后工厂方法会“挂接”而非新建。
    sandbox_id=None,

    # 单条命令默认超时（秒，float）。bare executor 与 workspace
    # runtime 共用该值。默认 60.0。
    execute_timeout=60.0,

    # 沙箱空闲生命周期（int 秒，>=1），每次命令都会续期。默认
    # 3600（1 小时）。底层 e2b API 接收 int 秒，所以这里禁止
    # 浮点值（在构造期就拒掉，避免 0.9 静默被截成 0）。
    idle_timeout=3600,
)

executor = await CubeCodeExecutor.create(cfg)

CubeCodeExecutor 也支持 code_block_delimiters，默认在标准的 python 和 tool_code 之外又加了一个 bash 分隔符，所以普通的 ```bash\n ... \n``` 围栏也能被识别。

CubeWorkspaceRuntime

对于 Skill 执行等需要按执行隔离工作目录（输入暂存、结构化程序运行、 输出收集）的场景，Cube 子包还提供了 CubeWorkspaceRuntime。它在同一个 CubeSandboxClient 之上组合了：

  • CubeWorkspaceManager：工作目录生命周期
  • CubeWorkspaceFS：文件 / 目录上传下载、按 glob 收集输出
  • CubeProgramRunner：结构化的 cmd + args 程序运行
from trpc_agent_sdk.code_executors._types import (
    WorkspaceOutputSpec,
    WorkspacePutFileInfo,
    WorkspaceRunProgramSpec,
)
from trpc_agent_sdk.code_executors.cube import (
    CubeCodeExecutor,
    CubeCodeExecutorConfig,
    CubeWorkspaceRuntimeConfig,
    create_cube_workspace_runtime,
)

executor = await CubeCodeExecutor.create(CubeCodeExecutorConfig())

# `workspace_cfg` 可选，省略时使用 DEFAULT_REMOTE_WORKSPACE
# = "/workspace/cube_agent" 作为根目录。
runtime = create_cube_workspace_runtime(
    executor,
    workspace_cfg=CubeWorkspaceRuntimeConfig(
        # Manager 在该路径下创建按执行隔离的
        # `ws_<exec_id>_<suffix>` 子目录。
        remote_workspace="/workspace/cube_agent",
    ),
)

manager = runtime.manager()
fs = runtime.fs()
runner = runtime.runner()

ws = await manager.create_workspace("demo-1")           # /workspace/cube_agent/ws_demo-1_<ts>

await fs.put_files(ws, [
    WorkspacePutFileInfo(path="work/script.py",
                         content=b"print('script ran')\n"),
])

run_result = await runner.run_program(
    ws,
    WorkspaceRunProgramSpec(cmd="python3", args=["work/script.py"], timeout=15.0),
)
print(run_result.exit_code, run_result.stdout)

outputs = await fs.collect_outputs(
    ws, WorkspaceOutputSpec(globs=["work/*.py"], inline=True),
)
for ref in outputs.files:
    print(ref.name, len(ref.content))

await manager.cleanup("demo-1")

该 runtime 可以直接接入 Skill 子系统 —— 在创建 skill repository 时 作为 workspace_runtime 传入即可（详见 skill.md）。

代码块格式

Agent会自动识别LLM返回中的代码块并执行。支持的代码块格式：

默认格式

```python
print("Hello, World!")
```

```tool_code
result = 15 + 27 * 3
print(result)
```

执行结果格式

代码执行后，结果会以以下格式返回给LLM：

```tool_output
96
```

支持的语言

UnsafeLocalCodeExecutor

  • Python (python, py, python3)
  • Bash (bash, sh)

ContainerCodeExecutor

  • Python (python, py, python3, 空字符串默认为Python)
  • Bash (bash, sh)

CubeCodeExecutor

  • Python (python, py, python3, 空字符串默认为Python)
  • Bash (bash, sh)

工作流程

  1. 用户查询 → Agent接收用户查询
  2. LLM生成响应 → LLM生成包含代码块的响应
  3. 代码提取 → 框架自动提取代码块（根据code_block_delimiters）
  4. 代码执行 → CodeExecutor执行代码
  5. 结果返回 → 执行结果返回给LLM
  6. 最终响应 → LLM根据执行结果生成最终响应

123 沙箱环境 CodeExecutor 使用

参考：Pcg123 沙箱环境使用示例（示例待补充）

常见问题

1. Docker连接失败

问题： 使用ContainerCodeExecutor时提示Docker连接失败

解决方案： - Linux: 确保Docker daemon正在运行 sudo systemctl start docker - Windows/Mac: 启动Docker Desktop应用 - 检查Docker权限: sudo chmod 666 /var/run/docker.sock 或添加用户到docker组 - 验证Docker运行: docker ps - 如果使用远程Docker，检查base_url配置

2. 代码执行超时

问题： 代码执行时间过长导致超时

解决方案：

# UnsafeLocalCodeExecutor设置超时
code_executor = UnsafeLocalCodeExecutor(timeout=30)  # 30秒超时

3. 代码执行失败但无错误信息

问题： 代码执行失败但没有看到错误信息

解决方案： - 检查error_retry_attempts设置，增加重试次数 - 查看日志输出，框架会记录详细的错误信息 - 对于ContainerCodeExecutor，检查容器日志

4. CubeCodeExecutor 无法连接 / 鉴权到错误租户

问题： CubeCodeExecutor.create 抛出形如 Cube sandbox requires \api_url` or E2B_API_URL env、... api_key ...、 或... template ... CUBE_TEMPLATE_ID ...` 的错误。

解决方案： - 安装可选 extra：pip install 'trpc-agent-py[cube]' - 导出三个必备环境变量（或在 CubeCodeExecutorConfig 上显式传入）： E2B_API_URL、E2B_API_KEY、CUBE_TEMPLATE_ID - 多租户部署建议显式给 cfg 字段赋值，避免不同 Agent 实例落到同一份 进程级环境变量上

5. CubeCodeExecutor 沙箱在两次调用之间消失

问题： 通过 cfg.sandbox_id 挂接的沙箱，在下一次命令时抛 SandboxNotFoundException（已销毁）或 SandboxException（PAUSED）。

解决方案： - 长生命周期 Agent 推荐使用 CubeCodeExecutor.create_or_recreate(cfg, on_recreate=...)，让 executor 透明地新建沙箱，并通过 callback 通知调用方清理外部 locator 状态 - 如果确实需要更长的空闲窗口，调大 idle_timeout（默认 3600 秒）， 每次命令都会续期 - 仅想丢掉某个工作目录而保留沙箱时，用 CubeWorkspaceManager.cleanup(exec_id) 而不是 executor.destroy()

完整示例

查看完整示例代码：examples/code_executors/agent/agent.py

Cube 端到端示例（executor + workspace runtime）： examples/code_executors/cube_demo.py

安全建议

  1. 生产环境：强烈建议使用ContainerCodeExecutor，提供沙箱隔离
  2. 代码审查：在生产环境部署前，审查LLM生成的代码
  3. 资源限制：为 ContainerCodeExecutor 设置适当的资源限制
  4. 权限控制：限制代码执行器的文件系统访问权限
  5. 网络隔离：根据需要限制容器的网络访问

性能考虑

  • UnsafeLocalCodeExecutor：执行速度快，适合快速迭代
  • ContainerCodeExecutor：首次启动需要拉取镜像，后续执行速度较快
  • CubeCodeExecutor：每条命令多一次远端网络往返，但因为沙箱可以跨多次调用复用（且通过 sandbox_id 跨进程复用），长会话场景摊销得很好；工作空间文件传输使用 tar 协议，目录上传/下载都是单次往返
  • 建议在生产环境使用 ContainerCodeExecutor 或 CubeCodeExecutor，开发环境可以使用 UnsafeLocalCodeExecutor