ServerTransport

The ServerTransport publishes to an Ably channel on behalf of the agent, manages the turn lifecycle, and routes cancel signals from clients to active turns. It is the server-side counterpart to ClientTransport.

Construct one with createServerTransport from the core entry point, or use the Vercel-pre-bound factory from @ably/ai-transport/vercel when the channel carries Vercel UIMessage content.

import { createServerTransport } from '@ably/ai-transport';
import { UIMessageCodec } from '@ably/ai-transport/vercel';

const transport = createServerTransport({
  channel,
  codec: UIMessageCodec,
});

const turn = transport.newTurn({ turnId, clientId });

Create a server transport

Construct a ServerTransport bound to an Ably channel. The transport attaches to the channel lazily on first use.

import Ably from 'ably';
import { createServerTransport } from '@ably/ai-transport';
import { UIMessageCodec } from '@ably/ai-transport/vercel';

const ably = new Ably.Realtime({ key: process.env.ABLY_API_KEY });
const channel = ably.channels.get('conversation-42');

const transport = createServerTransport({
  channel,
  codec: UIMessageCodec,
  onError(error) {
    console.error(`Transport error ${error.code}: ${error.message}`);
  },
});

Parameters

Returns

ServerTransport<TEvent, TMessage>

A new server transport. Lifecycle is managed by the caller; call close() to release resources.

Create a turn

Create a new turn. Synchronous; no channel activity until start() is called on the returned turn. The turn is registered for cancel routing immediately so that early cancels fire the abort signal.

const turn = transport.newTurn({
  turnId,
  clientId,
  parent,
  forkOf,
  onCancel: async (request) => {
    const owner = request.turnOwners.get(request.filter.turnId);
    return owner === request.message.clientId;
  },
});

Parameters

onError fires in two scenarios. First, stream failures in streamResponse (the underlying error is also returned on StreamResult.error, but this callback delivers it wrapped as an ErrorInfo with code StreamError for standardised observability). Second, failures in the onCancel handler. Publish failures in start, addMessages, addEvents, and end reject the returned promise instead and are not delivered to onError. Channel-wide events (continuity loss) are delivered via the transport-level onError on ServerTransportOptions.

Returns

Turn<TEvent, TMessage>

A new turn. Call its lifecycle methods (start, addMessages, streamResponse, addEvents, end) to publish to the channel.

Close the transport

Unsubscribe from cancel messages, abort all active turns, and clean up. Synchronous.

transport.close();

Turn

A Turn is the server-side handle for a single turn's lifecycle. Create one with transport.newTurn() and call its methods in order: start → optionally addMessages → streamResponse (and/or addEvents) → end.

Properties

Start the turn

Publish the turn-start event to the channel. Must be called before addMessages or streamResponse.

Returns

Promise<void>

Returns a promise. The promise is fulfilled when the turn-start event has been published, or rejected with an ErrorInfo object.

Add user messages

Publish user messages to the channel, scoped to this turn. Each node's msgId, parentId, and forkOf are used for message identity and branching. The node's headers override transport-generated defaults (used for optimistic reconciliation with the client's inserts).

Parameters

Returns

Promise<AddMessagesResult>

Returns a promise. The promise is fulfilled with an

Stream a response

Pipe a ReadableStream through the encoder to the channel. Returns when the stream completes, is cancelled, or errors. Does not call end(); the caller must call end() after streamResponse returns.

const result = streamText({
  model: anthropic('claude-sonnet-4-20250514'),
  messages: history,
  abortSignal: turn.abortSignal,
});

const { reason } = await turn.streamResponse(result.toUIMessageStream());
await turn.end(reason);

Parameters

Returns

Promise<StreamResult>

Returns a promise. The promise is fulfilled with a

Add events to existing messages

Publish events targeting existing messages in the tree. Each node specifies a target message (by msgId) and the events to apply. Events are encoded and published with the target's x-ably-msg-id, so receiving clients apply them to the existing node rather than creating a new one.

Used for cross-turn updates such as tool result delivery after approval or client-side tool execution.

Parameters

Returns

Promise<void>

Returns a promise. The promise is fulfilled when every targeted message has received its events, or rejected with an ErrorInfo object.

End the turn

Publish the turn-end event to the channel and clean up. reason is one of 'complete', 'cancelled', or 'error'.

await turn.end('complete');

Parameters

Returns

Promise<void>

Returns a promise. The promise is fulfilled when the turn-end event has been published, or rejected with an ErrorInfo object.

Example

import { after } from 'next/server';
import { streamText, convertToModelMessages } from 'ai';
import { anthropic } from '@ai-sdk/anthropic';
import Ably from 'ably';
import { createServerTransport } from '@ably/ai-transport/vercel';

const ably = new Ably.Realtime({ key: process.env.ABLY_API_KEY });

export async function POST(req) {
  const { messages, id, turnId, clientId, parent, forkOf } = await req.json();
  const channel = ably.channels.get(id);

  const transport = createServerTransport({
    channel,
    onError(error) {
      console.error('Transport error:', error);
    },
  });

  const turn = transport.newTurn({
    turnId,
    clientId,
    parent,
    forkOf,
    signal: req.signal,
    onCancel: async (request) => {
      const owner = request.turnOwners.get(request.filter.turnId);
      return owner === request.message.clientId;
    },
  });

  await turn.start();

  if (messages.length > 0) {
    await turn.addMessages(messages.map((m) => ({ kind: 'message', ...m })), { clientId });
  }

  const result = streamText({
    model: anthropic('claude-sonnet-4-20250514'),
    messages: await convertToModelMessages(messages),
    abortSignal: turn.abortSignal,
  });

  after(async () => {
    try {
      const { reason } = await turn.streamResponse(result.toUIMessageStream());
      await turn.end(reason);
    } catch {
      await turn.end('error');
    } finally {
      transport.close();
    }
  });

  return new Response(null, { status: 200 });
}