更新时间:2026-09-07 GMT+08:00
分享

进程管理

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是否与控制台中的信息一致。

相关文档