进程管理
AgentSphere沙箱实例支持通过PTY(伪终端)创建交互式终端会话,适用于需要实时交互的场景,如运行交互式程序、模拟终端操作、自动化Shell脚本测试等。与commands.run不同,PTY会话支持持续的输入输出流,可以模拟真实终端的行为,包括终端窗口大小调整、TTY属性检测等。
前提条件
以下示例均基于E2B SDK,请确保已完成配置本地E2B SDK执行环境。
交互式进程/PTY会话
创建PTY终端会话
您可以使用pty.create方法在沙箱实例中创建一个PTY伪终端会话,创建时需指定终端窗口大小和超时时间。返回的终端对象包含pid属性,用于后续操作。
from e2b import Sandbox, PtySize
sandbox = Sandbox.create(template="<template_id>")
# 创建PTY终端会话,指定窗口大小为24行80列
terminal = sandbox.pty.create(
PtySize(rows=24, cols=80),
timeout=30,
)
print(f"终端PID: {terminal.pid}") pty.create方法支持以下参数:
| 参数 | 是否必选 | 说明 |
|---|---|---|
| size | 是 | 终端窗口大小,通过PtySize(rows, cols)指定行数和列数。 |
| timeout | 否 | PTY终端会话的超时时间,单位为秒。超时后终端将被终止。 |
发送输入数据
您可以使用pty.send_stdin方法向PTY终端的标准输入发送数据,数据格式为bytes。支持多次发送,模拟用户逐行输入。
from e2b import Sandbox, PtySize
sandbox = Sandbox.create(template="<template_id>")
terminal = sandbox.pty.create(PtySize(rows=24, cols=80), timeout=30)
# 向终端发送命令
sandbox.pty.send_stdin(terminal.pid, b"echo 'hello from PTY'\n")
# 发送exit命令结束终端会话
sandbox.pty.send_stdin(terminal.pid, b"exit\n") pty.send_stdin方法支持以下参数:
| 参数 | 是否必选 | 说明 |
|---|---|---|
| pid | 是 | 目标PTY终端的进程ID。 |
| data | 是 | 发送到标准输入的数据,类型为bytes。 |
等待终端结束
您可以使用终端对象的wait方法等待PTY终端会话结束,并通过on_pty回调函数实时接收终端输出数据。wait方法会阻塞直到终端进程退出,返回包含退出码的结果对象。
from e2b import Sandbox, PtySize
sandbox = Sandbox.create(template="<template_id>")
terminal = sandbox.pty.create(PtySize(rows=24, cols=80), timeout=30)
sandbox.pty.send_stdin(terminal.pid, b"echo 'hello'\n")
sandbox.pty.send_stdin(terminal.pid, b"exit\n")
# 通过回调函数实时收集终端输出
output_parts = []
def collect(data):
output_parts.append(data.decode(errors="replace"))
# 等待终端结束,获取退出码
result = terminal.wait(on_pty=collect)
print(f"退出码: {result.exit_code}")
print(f"完整输出: {''.join(output_parts)}") terminal.wait方法支持以下参数:
| 参数 | 是否必选 | 说明 |
|---|---|---|
| on_pty | 否 | 终端输出回调函数,接收实时输出的bytes数据。 |
调整终端窗口大小
您可以使用pty.resize方法动态调整PTY终端的窗口大小,适用于模拟不同终端尺寸的场景。
from e2b import Sandbox, PtySize
sandbox = Sandbox.create(template="<template_id>")
terminal = sandbox.pty.create(PtySize(rows=24, cols=80), timeout=30)
# 将终端窗口调整为40行120列
sandbox.pty.resize(terminal.pid, PtySize(rows=40, cols=120))
print("终端窗口已调整为 40x120") pty.resize方法支持以下参数:
| 参数 | 是否必选 | 说明 |
|---|---|---|
| pid | 是 | 目标PTY终端的进程ID。 |
| size | 是 | 新的终端窗口大小,通过PtySize(rows, cols)指定。 |
终止PTY终端会话
您可以使用pty.kill方法终止指定的PTY终端会话。
终止PTY终端会话后,终端进程将被强制结束,未读取的输出数据将丢失。
from e2b import Sandbox, PtySize
sandbox = Sandbox.create(template="<template_id>")
terminal = sandbox.pty.create(PtySize(rows=24, cols=80), timeout=30)
# 终止PTY终端会话
sandbox.pty.kill(terminal.pid)
print(f"终端已终止: pid={terminal.pid}") pty.kill方法支持以下参数:
| 参数 | 是否必选 | 说明 |
|---|---|---|
| pid | 是 | 要终止的PTY终端的进程ID。 |
重新连接PTY终端
您可以使用pty.connect方法通过进程ID重新连接到已存在的PTY终端会话,适用于跨会话复用终端的场景。
from e2b import Sandbox, PtySize
sandbox = Sandbox.create(template="<template_id>")
terminal = sandbox.pty.create(PtySize(rows=24, cols=80), timeout=30)
pid = terminal.pid
# 通过PID重新连接到已存在的PTY终端
sandbox.pty.connect(pid)
print(f"已重新连接到终端: pid={pid}")
sandbox.pty.kill(pid) pty.connect方法支持以下参数:
| 参数 | 是否必选 | 说明 |
|---|---|---|
| pid | 是 | 要重新连接的PTY终端的进程ID。 |
完整代码示例
import os
import time
from e2b import Sandbox, PtySize
# 从环境变量中读取API Key
E2B_API_KEY = os.environ.get("E2B_API_KEY", "")
# 创建PTY终端会话
def test_pty_create(sandbox):
terminal = sandbox.pty.create(
PtySize(rows=24, cols=80),
timeout=30,
)
assert terminal.pid is not None
print(f"[PASS] pty create: pid={terminal.pid}")
sandbox.pty.kill(terminal.pid)
# 发送输入并等待终端结束
def test_pty_send_stdin_and_wait(sandbox):
terminal = sandbox.pty.create(
PtySize(rows=24, cols=80),
timeout=30,
)
sandbox.pty.send_stdin(
terminal.pid,
b"echo 'hello from PTY'\n",
)
sandbox.pty.send_stdin(terminal.pid, b"exit\n")
result = terminal.wait(on_pty=lambda data: None)
print(f"[PASS] pty send_stdin & wait: exit_code={result.exit_code}")
# 验证PTY是否为真实终端(isatty检查)
def test_pty_isatty_check(sandbox):
terminal = sandbox.pty.create(
PtySize(rows=24, cols=80),
timeout=30,
)
sandbox.pty.send_stdin(
terminal.pid,
b"python3 -c \"import sys; print(sys.stdout.isatty())\"\n",
)
sandbox.pty.send_stdin(terminal.pid, b"exit\n")
output_parts = []
def collect(data):
output_parts.append(data.decode(errors="replace"))
result = terminal.wait(on_pty=collect)
full_output = "".join(output_parts)
print(f"[PASS] pty isatty check: contains 'True' = {'True' in full_output}")
# 调整终端窗口大小
def test_pty_resize(sandbox):
terminal = sandbox.pty.create(
PtySize(rows=24, cols=80),
timeout=30,
)
sandbox.pty.resize(terminal.pid, PtySize(rows=40, cols=120))
print("[PASS] pty resize to 40x120")
sandbox.pty.kill(terminal.pid)
# 终止PTY终端会话
def test_pty_kill(sandbox):
terminal = sandbox.pty.create(
PtySize(rows=24, cols=80),
timeout=30,
)
sandbox.pty.kill(terminal.pid)
print(f"[PASS] pty kill: pid={terminal.pid}")
# 重新连接到已存在的PTY终端
def test_pty_connect(sandbox):
terminal = sandbox.pty.create(
PtySize(rows=24, cols=80),
timeout=30,
)
pid = terminal.pid
try:
sandbox.pty.connect(pid)
print(f"[PASS] pty connect: reconnected to pid={pid}")
except Exception as e:
print(f"[SKIP] pty connect: {e}")
sandbox.pty.kill(pid)
# 模拟Python交互式解释器会话
def test_pty_python_interactive(sandbox):
terminal = sandbox.pty.create(
PtySize(rows=24, cols=80),
timeout=30,
)
sandbox.pty.send_stdin(terminal.pid, b"python3\n")
time.sleep(1)
sandbox.pty.send_stdin(terminal.pid, b"x = 2 + 3\n")
time.sleep(1)
sandbox.pty.send_stdin(terminal.pid, b"print(x)\n")
time.sleep(1)
sandbox.pty.send_stdin(terminal.pid, b"exit()\n")
sandbox.pty.send_stdin(terminal.pid, b"exit\n")
output_parts = []
def collect(data):
output_parts.append(data.decode(errors="replace"))
result = terminal.wait(on_pty=collect)
full_output = "".join(output_parts)
print(f"[PASS] pty python interactive: output len={len(full_output)}")
def main():
sandbox = None
try:
sandbox = Sandbox.create(template="<template_id>", api_key=E2B_API_KEY, timeout=300)
print(f"Sandbox created: {sandbox.sandbox_id}")
time.sleep(1)
test_pty_create(sandbox)
test_pty_send_stdin_and_wait(sandbox)
test_pty_isatty_check(sandbox)
test_pty_resize(sandbox)
test_pty_kill(sandbox)
test_pty_connect(sandbox)
test_pty_python_interactive(sandbox)
finally:
if sandbox is not None:
sandbox.kill()
print("\n=== Sandbox 操作测试完成 ===")
if __name__ == "__main__":
main() 相关文档
- 创建沙箱模板:了解如何创建自定义沙箱模板,预装所需的运行环境和依赖包。
- 创建API Key:了解如何创建和管理API Key,获取访问凭据。
- 创建网关实例:了解Agent网关的配置和管理,获取数据面访问地址。
- E2B SDK官方文档:了解E2B SDK的完整接口说明和更多用法示例。
- 如果在使用SDK时遇到连接失败、认证错误等问题,请先检查环境变量配置是否正确,特别是E2B_API_URL、E2B_SANDBOX_URL和E2B_API_KEY是否与控制台中的信息一致。