Custom Transports

Custom transports plug into the shared client facade by implementing AgentClientTransport.

type AgentClientTransport<TContract extends AgentContract = AgentContract> = {
  getManifest(): Promise<AgentManifest<TContract>>;
  listThreads(options?: ThreadListOptions): Promise<ThreadListResult>;
  renameThread(input: ThreadRenameInput): Promise<ThreadRenameResult>;
  deleteThread(input: { threadId: ThreadId }): Promise<void>;

  getThreadState(input: GetThreadStateInput): Promise<PaiThreadTransportState<TContract> | null>;
  send(input: ThreadSendInput<TContract>): Promise<ThreadSendResult<TContract>>;
  regenerate(input: ThreadRegenerateInput<TContract>): Promise<ThreadRegenerateResult<TContract>>;
  stopThread(input: StopThreadInput): Promise<ThreadMutationResult<TContract>>;
  stopRun(input: StopRunInput): Promise<ThreadMutationResult<TContract>>;
  respondToAction(input: RespondToActionInput<TContract>): Promise<RespondToActionResult<TContract>>;
  writeToolData(input: WriteToolDataInput<TContract>): Promise<WriteToolDataResult<TContract>>;
  recallQueuedItem(input: RecallQueuedItemInput): Promise<RecallQueuedItemResult<TContract>>;
  executeCommand<TName extends CommandName<TContract>>(
    input: ExecuteCommandInput<TContract, TName>,
  ): Promise<CommandOutput<TContract, TName>>;
  getRun(input: GetRunInput): Promise<RunStateDTO | null>;
  uploadFile(input: FileUploadInput): Promise<UploadedFile>;
  createFileUrl(input: ClientCreateFileUrlInput): Promise<FileUrlResult | null>;
  readFile(input: ClientReadFileInput): Promise<StoredFileBody>;

  /** Optional live watch channel. If omitted, the client falls back to snapshot polling. */
  watchThread?(input: ThreadWatchTransportInput): AsyncIterable<PaiThreadTransportEvent<TContract>>;
  watchThreadHeads?(input: ThreadHeadWatchInput): AsyncIterable<ThreadHeadWatchEvent>;

  /** Release transport resources. */
  close?(): Promise<void>;
};

AgentClientTransport is intentionally lower-level than the app API. It does not implement thread.send(), thread.watch(), or run.waitUntilIdle() itself. Those are client semantics built by @pai/client on top of transport methods.

If watchThread() is present, the client uses it for live state. If it is absent or fails and requires repair, the client can refresh the authoritative snapshot through getThreadState().

PaiThreadTransportState and PaiThreadTransportEvent are framework transport contracts from @pai/protocol/internal; they are not application state. A transport must carry them losslessly and let @pai/client perform the sole public PaiMessage projection.

When ThreadWatchTransportInput.includeInitial is not false, watchThread() must emit an authoritative raw snapshot before live events. For an unrealized thread, that snapshot contains the empty authorized transport state. When includeInitial is false, the stream may begin directly with live events.

Call input.onConnectionEstablished?.() once the transport has accepted the subscription and can receive events. Do not wait for the first event: a healthy thread or thread-head stream may remain quiet indefinitely. The client falls back to first-event reporting only for transports that omit this signal.

watchThread() emits an authoritative snapshot, ordered native generation.message.chunk updates for the active generation, typed control events such as queue add/remove, and state.changed invalidations for other durable state changes. The high-level client owns local state caching, keeps one AI SDK reducer session per active message, repairs gaps from snapshots, and projects the result into the same ThreadState that direct clients expose.

Simple transports can emit snapshots for initial load and repair. Streaming transports should use compact frames for high-frequency changes:

snapshot version=instance_1:3
generation.started version=instance_1:3 runId=run_1 seq=0 baseMessages=[...]
generation.message.chunk version=instance_1:3 runId=run_1 seq=1 messageId=msg_a1 chunk.type=start
generation.message.chunk version=instance_1:3 runId=run_1 seq=2 messageId=msg_a1 chunk.type=start-step
generation.message.chunk version=instance_1:3 runId=run_1 seq=3 messageId=msg_a1 chunk.type=text-start
generation.message.chunk version=instance_1:3 runId=run_1 seq=4 messageId=msg_a1 chunk.type=text-delta
generation.message.chunk version=instance_1:3 runId=run_1 seq=5 messageId=msg_a1 chunk.type=tool-input-available
queue-item-added version=instance_1:4 queued_msg_1
generation.message.chunk version=instance_1:3 runId=run_1 seq=6 messageId=msg_a1 chunk.type=tool-output-available
generation.message.chunk version=instance_1:3 runId=run_1 seq=7 messageId=msg_a1 chunk.type=finish-step
generation.message.chunk version=instance_1:3 runId=run_1 seq=8 messageId=msg_a2 chunk.type=start
generation.message.chunk version=instance_1:3 runId=run_1 seq=9 messageId=msg_a2 chunk.type=text-start
generation.message.chunk version=instance_1:3 runId=run_1 seq=10 messageId=msg_a2 chunk.type=text-delta
generation.completed version=instance_1:3 runId=run_1

If the transport cannot deliver generation events in sequence, emit state.changed or close the stream. The client will repair by reading the latest state through getThreadState(), then replaying the active generation stream when available. generation.started makes that replay authoritative for the run's model messages, so a transport must not translate native chunks into its own text/tool grammar or mix checkpointed parts into the replay baseline.

The paired transport and receiver own their wire contract. HTTP uses routes, request bodies, status codes, and SSE; a WebSocket or IPC implementation may define its own method discriminator or envelope. None of those encodings is a universal @pai/protocol request shape.

Preserve canonical ids such as threadId, messageId, and fileId when encoding the matching AgentClientTransport method, even when an HTTP route also contains an id as a path parameter.

Adapter authors should preserve:

• request validation;
• authentication and scope resolution on trusted backend boundaries;
• scope isolation;
• caller-chosen ids on operations that support retry deduplication;
• typed errors;
• watch recovery through snapshots.

Do not implement separate queueing, pending action, or tool semantics in a transport.

Test a custom transport through the client facade rather than by calling transport methods directly: construct createAgentClient(createMyTransport(...)) and assert the app-facing behaviour. PAI ships no shared conformance suite for this today.