Stream Utilities

Bidirectional message streams for ACP connections

Overview

The Stream interface and utilities enable bidirectional communication between ACP clients and agents. Streams handle the serialization and deserialization of JSON-RPC messages over various transport layers.

Stream Interface

type Stream = {
  writable: WritableStream<AnyMessage>;
  readable: ReadableStream<AnyMessage>;
};

The Stream type represents a bidirectional communication channel that:

• Receives JSON-RPC messages through the readable stream
• Sends JSON-RPC messages through the writable stream

Properties

writable

writable: WritableStream<AnyMessage>

A writable stream for sending JSON-RPC messages. Messages written to this stream are serialized and sent to the remote peer.

readable

readable: ReadableStream<AnyMessage>

A readable stream for receiving JSON-RPC messages. Messages arriving from the remote peer are deserialized and made available through this stream.

ndJsonStream

function ndJsonStream(
  output: WritableStream<Uint8Array>,
  input: ReadableStream<Uint8Array>
): Stream

Creates an ACP Stream from a pair of newline-delimited JSON streams. This is the typical way to handle ACP connections over stdio or other byte-oriented transports.

Parameters:

• output (WritableStream<Uint8Array>) - The writable stream to send encoded messages to
• input (ReadableStream<Uint8Array>) - The readable stream to receive encoded messages from

Returns: A Stream object for bidirectional ACP communication

How It Works

The ndJsonStream function:

1. Encoding (Writable Stream):

   • Takes AnyMessage objects
   • Serializes them to JSON
   • Appends a newline character
   • Encodes as UTF-8 bytes
   • Writes to the output stream

2. Decoding (Readable Stream):

   • Reads UTF-8 bytes from the input stream
   • Buffers incomplete lines
   • Splits on newline characters
   • Parses each line as JSON
   • Emits AnyMessage objects

Error Handling

The function handles errors gracefully:

• Parse errors: Logged to console but don’t stop the stream
• Empty lines: Silently skipped
• Incomplete messages: Buffered until complete

Usage Examples

Creating a Stream from stdin/stdout (Node.js)

import { ndJsonStream, AgentSideConnection } from "@agentclientprotocol/sdk";
import { ReadableStream, WritableStream } from "node:stream/web";

// Convert Node.js streams to Web Streams API
const input = ReadableStream.from(
  (async function* () {
    for await (const chunk of process.stdin) {
      yield chunk;
    }
  })()
);

const output = new WritableStream({
  write(chunk) {
    process.stdout.write(chunk);
  },
});

// Create the ACP stream
const stream = ndJsonStream(output, input);

// Use with AgentSideConnection
const connection = new AgentSideConnection(
  (conn) => createAgent(conn),
  stream
);

Creating a Stream from stdin/stdout (Deno)

import { ndJsonStream, AgentSideConnection } from "@agentclientprotocol/sdk";

// Deno provides Web Streams API directly
const stream = ndJsonStream(
  Deno.stdout.writable,
  Deno.stdin.readable
);

const connection = new AgentSideConnection(
  (conn) => createAgent(conn),
  stream
);

Creating a Stream from WebSockets

import { ndJsonStream } from "@agentclientprotocol/sdk";
import type { AnyMessage } from "@agentclientprotocol/sdk";

function createWebSocketStream(ws: WebSocket): Stream {
  // Create output stream (writable)
  const output = new WritableStream<Uint8Array>({
    write(chunk) {
      ws.send(chunk);
    },
    close() {
      ws.close();
    },
  });

  // Create input stream (readable)
  const input = new ReadableStream<Uint8Array>({
    start(controller) {
      ws.onmessage = (event) => {
        if (event.data instanceof ArrayBuffer) {
          controller.enqueue(new Uint8Array(event.data));
        } else if (typeof event.data === "string") {
          controller.enqueue(new TextEncoder().encode(event.data));
        }
      };

      ws.onclose = () => {
        controller.close();
      };

      ws.onerror = (error) => {
        controller.error(error);
      };
    },
  });

  return ndJsonStream(output, input);
}

// Usage
const ws = new WebSocket("ws://localhost:8080");
const stream = createWebSocketStream(ws);
const connection = new ClientSideConnection(
  (agent) => createClient(agent),
  stream
);

Creating a Custom Stream Implementation

For custom transport layers, you can create Stream objects directly without using ndJsonStream:

import type { Stream, AnyMessage } from "@agentclientprotocol/sdk";

function createCustomStream(): Stream {
  // Message queue for incoming messages
  const messageQueue: AnyMessage[] = [];
  let readController: ReadableStreamDefaultController<AnyMessage> | null = null;

  const readable = new ReadableStream<AnyMessage>({
    start(controller) {
      readController = controller;
      // Enqueue any buffered messages
      while (messageQueue.length > 0) {
        controller.enqueue(messageQueue.shift()!);
      }
    },
  });

  const writable = new WritableStream<AnyMessage>({
    async write(message) {
      // Send message over your custom transport
      await customTransport.send(message);
    },
  });

  // Handle incoming messages from your transport
  customTransport.onMessage((message: AnyMessage) => {
    if (readController) {
      readController.enqueue(message);
    } else {
      messageQueue.push(message);
    }
  });

  return { readable, writable };
}

Testing with In-Memory Streams

import type { Stream, AnyMessage } from "@agentclientprotocol/sdk";

function createTestStreamPair(): [Stream, Stream] {
  // Create two transformer streams
  const { readable: readable1, writable: writable1 } = 
    new TransformStream<AnyMessage>();
  const { readable: readable2, writable: writable2 } = 
    new TransformStream<AnyMessage>();

  // Cross-connect them
  const stream1: Stream = {
    readable: readable1,
    writable: writable2,
  };

  const stream2: Stream = {
    readable: readable2,
    writable: writable1,
  };

  return [stream1, stream2];
}

// Usage in tests
const [clientStream, agentStream] = createTestStreamPair();

const client = new ClientSideConnection(
  (agent) => createClient(agent),
  clientStream
);

const agent = new AgentSideConnection(
  (conn) => createAgent(conn),
  agentStream
);

// Now client and agent can communicate
await client.initialize({ protocolVersion: "0.1.0" });

Stream Lifecycle

Connection Establishment

1. Create a Stream with your transport layer
2. Pass the Stream to AgentSideConnection or ClientSideConnection
3. The connection automatically starts reading from the stream

Message Flow

// Outgoing messages
await connection.sendRequest("method", params);
// → Written to stream.writable
// → Serialized by transport
// → Sent to remote peer

// Incoming messages
// ← Received from remote peer
// ← Deserialized by transport
// ← Read from stream.readable
// → Processed by connection

Connection Closure

The connection closes when:

• The readable stream ends (remote peer disconnected)
• An unrecoverable error occurs
• You explicitly close the underlying transport

// Wait for connection to close
await connection.closed;
console.log("Connection closed");

// Or listen for closure
connection.signal.addEventListener("abort", () => {
  console.log("Connection closed");
});

Best Practices

1. Use ndJsonStream for stdio

For most use cases with stdin/stdout, ndJsonStream is the recommended approach:

const stream = ndJsonStream(stdout, stdin);

2. Handle Transport Errors

Ensure your transport layer handles errors gracefully:

const output = new WritableStream<Uint8Array>({
  write(chunk) {
    try {
      socket.write(chunk);
    } catch (error) {
      console.error("Write error:", error);
      throw error; // Propagate to stream
    }
  },
});

3. Clean Up Resources

Close streams and connections when done:

// Wait for connection to complete
await connection.closed;

// Clean up any resources
await cleanup();

4. Buffer Management

Be mindful of buffering when implementing custom streams:

// Avoid unbounded buffers
const messageQueue: AnyMessage[] = [];
const MAX_QUEUE_SIZE = 100;

if (messageQueue.length >= MAX_QUEUE_SIZE) {
  throw new Error("Message queue full");
}

See Also

• AgentSideConnection - Agent connection class
• ClientSideConnection - Client connection class
• JSON-RPC Types - Message type definitions
• Web Streams API - MDN documentation