taolib.symphony.agent.transport

传输层抽象。

定义智能体进程启动和钩子执行的传输接口， 并提供本地（LocalTransport）和 SSH（SSHTransport）两种实现。

Classes

AgentProcess	智能体子进程的 I/O 句柄。
AgentTransport	传输层抽象基类。
LocalTransport	本地传输实现。
SSHTransport	SSH 远程传输实现。

Module Contents

class taolib.symphony.agent.transport.AgentProcess

智能体子进程的 I/O 句柄。

stdin: asyncio.StreamWriter

stdout: asyncio.StreamReader

stderr: asyncio.StreamReader

pid: str | None = None

class taolib.symphony.agent.transport.AgentTransport

Bases: abc.ABC

传输层抽象基类。

定义启动智能体进程和执行钩子脚本的接口， 使编排器可与本地或远程执行环境解耦。

abstractmethod start_process(command: str, cwd: str) → AgentProcess

Async:

启动智能体进程。

参数:

• command -- 要执行的命令。

• cwd -- 工作目录。

返回:

进程 I/O 句柄。

abstractmethod run_hook(script: str, cwd: str, timeout_ms: int) → int

Async:

执行钩子脚本。

参数:

• script -- Shell 命令。

• cwd -- 工作目录。

• timeout_ms -- 超时毫秒数。

返回:

进程退出码。

class taolib.symphony.agent.transport.LocalTransport

Bases: AgentTransport

本地传输实现。

在本机以 asyncio 子进程方式启动智能体和执行钩子。

async start_process(command: str, cwd: str) → AgentProcess

在本机启动子进程。

async run_hook(script: str, cwd: str, timeout_ms: int) → int

在本机执行钩子脚本。

class taolib.symphony.agent.transport.SSHTransport(host: str, **ssh_opts: object)

Bases: AgentTransport

SSH 远程传输实现。

通过 asyncssh 连接远程主机，在远端启动智能体进程和执行钩子。 使用前必须先调用 connect() 建立连接。

_host

_ssh_opts

_conn: asyncssh.SSHClientConnection | None = None

async connect() → None

建立 SSH 连接。

async start_process(command: str, cwd: str) → AgentProcess

在远程主机启动子进程。

async run_hook(script: str, cwd: str, timeout_ms: int) → int

在远程主机执行钩子脚本。

async disconnect() → None

关闭 SSH 连接。