feat(sdk): add streamInput / stream_input for commands and pty

.changeset/stream-input-sdk.md

@@ -0,0 +1,7 @@
+---
+"e2b": minor
+"@e2b/python-sdk": minor
+---
+
+- JS: `sandbox.commands.streamInput(pid)` / `sandbox.pty.streamInput(pid)` return a `CommandInputHandle` that sends ordered stdin/PTY input over one `StreamInput` RPC.
+- Python async: `stream_input(pid)` returns an `AsyncCommandInputHandle`; Python sync: `stream_input(pid, data)` blocks while it sends an iterable of input.

packages/js-sdk/src/connectionConfig.ts

@@ -8,6 +8,7 @@ import { InvalidArgumentError } from './errors'
 const supportedDomains = ['e2b.app', 'e2b.dev', 'e2b.pro', 'e2b-staging.dev']
 
 export const REQUEST_TIMEOUT_MS = 60_000 // 60 seconds
+export const STREAM_INPUT_TIMEOUT_MS = 60_000 // 60 seconds
 export const DEFAULT_RETRIES = 3
 export const KEEPALIVE_PING_INTERVAL_SEC = 50 // 50 seconds
 

packages/js-sdk/src/envd/streamingTransport.ts

@@ -0,0 +1,103 @@
+import type { Interceptor } from '@connectrpc/connect'
+import { createTransport } from '@connectrpc/connect/protocol-connect'
+import type {
+  UniversalClientFn,
+  UniversalClientRequest,
+  UniversalClientResponse,
+} from '@connectrpc/connect/protocol'
+
+function asyncIterableToReadableStream(
+  iterable: AsyncIterable<Uint8Array>
+): ReadableStream<Uint8Array> {
+  const iterator = iterable[Symbol.asyncIterator]()
+
+  return new ReadableStream({
+    async pull(controller) {
+      try {
+        const { done, value } = await iterator.next()
+        if (done) {
+          controller.close()
+        } else {
+          controller.enqueue(value)
+        }
+      } catch (error) {
+        controller.error(error)
+      }
+    },
+    async cancel(reason) {
+      await iterator.return?.(reason)
+    },
+  })
+}
+
+function readableStreamToAsyncIterable(
+  stream: ReadableStream<Uint8Array> | null
+): AsyncIterable<Uint8Array> {
+  return {
+    async *[Symbol.asyncIterator]() {
+      if (!stream) {
+        return
+      }
+
+      const reader = stream.getReader()
+      try {
+        while (true) {
+          const { done, value } = await reader.read()
+          if (done) {
+            return
+          }
+          yield value
+        }
+      } finally {
+        reader.releaseLock()
+      }
+    },
+  }
+}
+
+function createStreamingHttpClient(
+  fetch: typeof globalThis.fetch
+): UniversalClientFn {
+  return async (
+    request: UniversalClientRequest
+  ): Promise<UniversalClientResponse> => {
+    const response = await fetch(request.url, {
+      method: request.method,
+      headers: request.header,
+      body: request.body
+        ? asyncIterableToReadableStream(request.body)
+        : undefined,
+      signal: request.signal,
+      duplex: 'half',
+    } as RequestInit & { duplex: 'half' })
+
+    return {
+      status: response.status,
+      header: response.headers,
+      body: readableStreamToAsyncIterable(response.body),
+      trailer: new Headers(),
+    }
+  }
+}
+
+export function createClientStreamingTransport({
+  baseUrl,
+  fetch,
+  interceptors = [],
+}: {
+  baseUrl: string
+  fetch: typeof globalThis.fetch
+  interceptors?: Interceptor[]
+}) {
+  return createTransport({
+    baseUrl,
+    httpClient: createStreamingHttpClient(fetch),
+    useBinaryFormat: false,
+    interceptors,
+    acceptCompression: [],
+    sendCompression: null,
+    compressMinBytes: 1024,
+    readMaxBytes: 0xffffffff,
+    writeMaxBytes: 0xffffffff,
+  })
+}

packages/js-sdk/src/index.ts

@@ -51,6 +51,8 @@ export type {
 } from './sandbox/filesystem/watchHandle'
 
 export { CommandExitError } from './sandbox/commands/commandHandle'
+export { CommandInputHandle } from './sandbox/commands/inputHandle'
+export type { CommandStreamInputOpts } from './sandbox/commands/inputHandle'
 export type {
   CommandResult,
   Stdout,

packages/js-sdk/src/sandbox/commands/index.ts

@@ -28,7 +28,10 @@ import {
 import { ENVD_COMMANDS_STDIN, ENVD_ENVD_CLOSE } from '../../envd/versions'
 import { SandboxError } from '../../errors'
 import { CommandHandle, CommandResult } from './commandHandle'
+import { CommandInputHandle, CommandStreamInputOpts } from './inputHandle'
 export { Pty } from './pty'
+export { CommandInputHandle }
+export type { CommandStreamInputOpts }
 
 /**
  * Options for sending a command request.
@@ -131,6 +134,7 @@ export interface ProcessInfo {
  */
 export class Commands {
   protected readonly rpc: Client<typeof ProcessService>
+  private readonly streamingRpc: Client<typeof ProcessService>
 
   private readonly defaultProcessConnectionTimeout = 60_000 // 60 seconds
   private readonly envdVersion: string
@@ -139,9 +143,11 @@ export class Commands {
   constructor(
     transport: Transport,
     private readonly envdApi: EnvdApiClient,
-    private readonly connectionConfig: ConnectionConfig
+    private readonly connectionConfig: ConnectionConfig,
+    streamingTransport: Transport
   ) {
     this.rpc = createClient(ProcessService, transport)
+    this.streamingRpc = createClient(ProcessService, streamingTransport)
     this.envdVersion = envdApi.version
     this.checkHealth = () => checkSandboxHealth(this.envdApi)
   }
@@ -229,6 +235,42 @@ export class Commands {
     }
   }
 
+  /**
+   * Open a persistent input stream to a command.
+   *
+   * Input sent through this stream is ordered relative to other `send` calls.
+   * Use this instead of {@link Commands.sendStdin} when ordering matters; each
+   * `sendStdin` call is a separate request. The stream sends keepalive messages
+   * while idle.
+   *
+   * In browsers, this requires fetch request-body streaming support (Chromium
+   * over HTTP/2).
+   *
+   * @param pid Process ID of the command.
+   * @param opts Optional stream timeout and abort signal. `timeoutMs` defaults to `STREAM_INPUT_TIMEOUT_MS` (60_000 milliseconds).
+   * @returns A handle whose `send()` method queues input in order and whose `close()` method ends the stream and waits for envd.
+   * @throws {NotFoundError} If the command does not exist, surfaced from `send()` or `close()`.
+   * @throws {TimeoutError} If the stream timeout expires, surfaced from `send()` or `close()`.
+   * @example
+   * ```ts
+   * const input = sandbox.commands.streamInput(pid)
+   * await input.send('hello\n')
+   * await input.close()
+   * ```
+   */
+  streamInput(
+    pid: number,
+    opts: CommandStreamInputOpts = {}
+  ): CommandInputHandle {
+    return new CommandInputHandle(
+      this.streamingRpc,
+      pid,
+      'stdin',
+      opts,
+      this.checkHealth
+    )
+  }
+
   /**
    * Close command stdin.
    *