跳转到主要内容

​
代码执行器

使用安全的沙箱隔离执行或高性能本地执行方式运行Python代码片段。

​
执行器类型

​
内置代码执行器（沙箱隔离）

在隔离的E2B容器中安全执行，支持完整功能集。 
from ag_kit_py.tools.code_executers import BuiltInCodeExecutor

executor = BuiltInCodeExecutor(
    sandbox=sandbox,  # 可选：提供现有沙箱
    api_key="your_api_key",  # 可选：E2B应用程序接口密钥
    domain="your_domain",  # 可选：E2B域名
    timeout_ms=60000
)

​
非安全本地代码执行器

在本地系统直接执行，功能集完全相同。 
from ag_kit_py.tools.code_executers import UnsafeLocalCodeExecutor

executor = UnsafeLocalCodeExecutor(
    timeout_ms=30000,
    on_stdout=lambda data: print(f"OUT: {data['line']}"),
    on_stderr=lambda data: print(f"ERR: {data['line']}")
)

​
通用功能

两种执行器支持相同功能集：
功能内置（沙箱）非安全本地
安全性✅ 隔离环境⚠️ 完全系统访问
性能存在网络开销✅ 最高速度
上下文持久化✅ 支持✅ 支持
实时输出✅ on_stdout/on_stderr✅ on_stdout/on_stderr
环境变量✅ envs✅ envs
超时控制✅ timeout_ms✅ timeout_ms
多语言支持✅ Python/JS/TS/Java/R/Bash✅ Python/JS/TS/Bash

​
基础用法

from ag_kit_py.tools.code_executers import UnsafeLocalCodeExecutor

executor = UnsafeLocalCodeExecutor()

result = await executor.invoke({
    "code": 'x = "Hello World!"\nprint(x)',
    "language": "python"
})

if result.success:
    print(result.data.results[0].text)  # "Hello World!"

​
高级功能

​
持久化上下文

两种执行器均支持跨执行保留变量： 
# 首次执行
await executor.invoke({
    "code": "x = 10",
    "language": "python"
})

# 二次执行 - x仍可访问
result = await executor.invoke({
    "code": "print(x)",  # 输出：10
    "language": "python"
})

if result.success:
    print(result.data.logs.stdout)  # ['10']

​
实时输出流

executor = UnsafeLocalCodeExecutor(
    on_stdout=lambda data: print(f"OUT: {data['line']}", end=""),
    on_stderr=lambda data: print(f"ERR: {data['line']}", end="")
)

result = await executor.invoke({
    "code": """
import time
for i in range(3):
    print(f"Step {i}")
    time.sleep(1)
""",
    "language": "python"
})
# 实时输出：OUT: Step 0, OUT: Step 1, OUT: Step 2

​
环境变量

import os

# 执行前设置环境变量
os.environ["API_KEY"] = "secret-key"
os.environ["DEBUG"] = "true"

# 内置执行器
sandbox_executor = BuiltInCodeExecutor()

# 本地执行器  
local_executor = UnsafeLocalCodeExecutor()

result = await executor.invoke({
    "code": "import os; print(os.environ.get('API_KEY'))",
    "language": "python"
})

if result.success:
    print(result.data.logs.stdout)  # ['secret-key']

​
超时配置

executor = UnsafeLocalCodeExecutor(
    timeout_ms=60000  # 60秒
)

# 内置执行器配置
sandbox_executor = BuiltInCodeExecutor(
    timeout_ms=60000
)

​
安装指南

​
内置执行器配置

export AG_KIT_SANDBOX_API_KEY=your_sandbox_api_key
export AG_KIT_SANDBOX_DOMAIN=your_sandbox_domain  # 可选

import os
from ag_kit_py.tools.code_executers import BuiltInCodeExecutor
from ag_kit_py.tools.fs.env_utils import setup_e2b_env
from e2b_code_interpreter import Sandbox

# 配置E2B环境（自动加载.env文件）
api_key = setup_e2b_env()
if not api_key:
    raise ValueError("未找到E2B应用程序接口密钥。请在.env文件或环境变量中设置E2B_API_KEY")

# 自定义沙箱配置
sandbox = Sandbox.create(api_key=api_key)

executor = BuiltInCodeExecutor(sandbox=sandbox)
环境配置：在项目根目录创建.env文件： 
E2B_API_KEY=your_e2b_api_key_here
从E2B控制台获取应用程序接口密钥。

​
本地执行器配置

确保已安装Python： 
python3 --version  # Python 3.8+
node --version     # Node.js
bash --version     # Bash

​
应用程序接口参考

​
通用接口

from typing import Optional, Callable, Dict, Literal, Any
from pydantic import BaseModel

class OutputMessage(BaseModel):
    """代码执行的输出消息"""
    line: str
    timestamp: Optional[float] = None

class CodeExecutorInput(BaseModel):
    """代码执行输入参数"""
    code: str
    language: Optional[Literal["python", "js", "ts", "java", "r", "bash"]] = "python"

class CodeExecutorOptions(BaseModel):
    """代码执行器配置选项"""
    timeout_ms: Optional[int] = 30000
    envs: Optional[Dict[str, str]] = None
    on_stdout: Optional[Callable[[OutputMessage], None]] = None
    on_stderr: Optional[Callable[[OutputMessage], None]] = None
    on_result: Optional[Callable[[Any], None]] = None
    on_error: Optional[Callable[[Exception], None]] = None

​
构造函数签名

# 内置执行器
BuiltInCodeExecutor(
    sandbox: Optional[Sandbox] = None,
    api_key: Optional[str] = None,
    domain: Optional[str] = None,
    timeout_ms: int = 3600 * 1000
)

# 本地执行器
UnsafeLocalCodeExecutor(
    timeout_ms: int = 30000,
    on_stdout: Optional[Callable[[Dict], None]] = None,
    on_stderr: Optional[Callable[[Dict], None]] = None
)

​
执行结果

from typing import Optional
from e2b_code_interpreter import Execution

class ToolResult:
    """执行结果包装器"""
    success: bool  # 是否执行成功
    data: Execution  # E2B执行对象
    error: Optional[str] = None  # 失败时的错误信息

# E2B执行对象结构：
class Execution:
    results: List[Result]  # 执行输出
    logs: Logs  # 捕获的日志
    error: Optional[ExecutionError] = None  # 执行失败时的错误
    execution_count: int  # 执行计数器

class Result:
    text: str  # 文本表示
    json_data: Optional[Any]  # 可选的解析后JSON数据
    # ... 其他字段

class Logs:
    stdout: List[str]  # 标准输出行
    stderr: List[str]  # 标准错误行

class ExecutionError:
    name: str  # 错误类型
    value: str  # 错误信息
    traceback: str  # 完整堆栈跟踪

​
语言支持

语言内置支持本地支持备注
Python✅✅完全支持
JavaScript✅✅完全支持
TypeScript✅✅完全支持
Bash✅✅完全支持
Java✅❌仅沙箱环境
R✅❌仅沙箱环境

​
示例

​
基础计算

from ag_kit_py.tools.code_executers import UnsafeLocalCodeExecutor

executor = UnsafeLocalCodeExecutor()

# 简单计算
result = await executor.invoke({
    "code": "print(2 + 2)",
    "language": "python"
})

if result.success:
    print(result.data.logs.stdout[0])  # "4"

​
数据处理

# 处理数据
result = await executor.invoke({
    "code": """
import json

data = [1, 2, 3, 4, 5]
result = {
    'sum': sum(data),
    'avg': sum(data) / len(data),
    'max': max(data)
}
print(json.dumps(result, indent=2))
""",
    "language": "python"
})

if result.success:
    print(result.data.logs.stdout)  # JSON格式输出字符串列表

​
函数定义与使用

# 定义并使用函数
result = await executor.invoke({
    "code": """
def process_data(items):
    return [item.upper() for item in items if len(item) > 3]

data = ['hello', 'hi', 'world', 'foo']
result = process_data(data)
print(result)
""",
    "language": "python"
})

if result.success:
    print(result.data.logs.stdout)  # ["['HELLO', 'WORLD']"]

​
基于环境的选择

import os
from ag_kit_p