AgentSession

The AgentSession is the server-side counterpart to ClientSession. It subscribes to the channel for cancel signals and creates AgentRun instances that publish lifecycle events, user messages, and streamed assistant output.

Construct one with createAgentSession from the core entry point. For Vercel UIMessage channels, use the pre-bound factory from @ably/ai-transport/vercel instead.

import * as Ably from 'ably';
import { createAgentSession, Invocation } from '@ably/ai-transport';
import { UIMessageCodec } from '@ably/ai-transport/vercel';

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

const invocation = Invocation.fromJSON(await req.json());

const session = createAgentSession({
  client: ably,
  channelName: invocation.sessionName,
  codec: UIMessageCodec,
});

await session.connect();
const run = session.createRun(invocation, { signal: req.signal });
await run.start();

Properties

Create an agent session

Construct an AgentSession bound to an Ably channel. The session does not attach until connect() resolves.

import * as Ably from 'ably';
import { createAgentSession } from '@ably/ai-transport';
import { UIMessageCodec } from '@ably/ai-transport/vercel';

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

const session = createAgentSession({
  client: ably,
  channelName: 'conversation-42',
  codec: UIMessageCodec,
});

Parameters

Subscribe to non-fatal session errors with on('error') rather than a constructor option.

Returns

AgentSession<TOutput, TProjection, TMessage>. The session instance. Call connect() before createRun.

Connect the session

Attaches and subscribes to the channel backing the session. Idempotent: subsequent calls return the same promise. All AgentRun methods (start, pipe, suspend, end) throw InvalidArgument until connect() has been called.

await session.connect();

Returns

Promise<void>. Resolves when the channel is attached and the session is ready to create runs.

Create a run

Create a new AgentRun for the input event named in the Invocation. Returns synchronously and publishes nothing to the channel until AgentRun.start is called. The run is registered for cancel routing immediately so early cancels fire the abortSignal.

const invocation = Invocation.fromJSON(await req.json());
const run = session.createRun(invocation, { signal: req.signal });

Parameters

Returns

An AgentRun<TOutput, TProjection, TMessage> handle for publishing lifecycle events, user messages, and streamed output. See AgentRun interface below.

AgentRun

The handle returned by createRun. It extends the shared BaseRun read-model (runId, status, error, messages) with the agent's lifecycle surface.

Properties

Start the run

Wait until the Run's triggering input has been observed on the channel (see located), then publish the opening lifecycle event (ai-run-start, or ai-run-resume for a continuation). Must be called before pipe, suspend, or end.

There is no built-in deadline: start() does not time out waiting for the trigger. It rejects only if the run is cancelled or the session is closed before the trigger is observed. Race it against your own timeout if you need one.

Pipe the response stream

Pipe a ReadableStream of outputs 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 pipe() returns.

Parameters

Returns

Promise<StreamResult>. Resolves when the stream ends. Pass result.reason to Run.end.

Suspend the run

Publish the ai-run-suspend event to the channel, pausing the Run pending external input (a tool approval, a human-in-the-loop response). The Run is not terminal: RunInfo.status becomes 'suspended', and a continuation Invocation resumes it via ai-run-resume.

Use suspend instead of end when you want the run to come back. Use end only for terminal outcomes.

End the run

Publish the ai-run-end event to the channel terminally and clean up. params is a RunEndParams object carrying the terminal reason and, when reason is 'error', an optional error. To pause a Run instead of ending it, use suspend.

Parameters

RunEndParams is discriminated on reason:

• { reason: 'complete' | 'cancelled' }: a non-error terminal reason that carries no error.
• { reason: 'error', error? }: the run ended in error. error is an optional Ably.ErrorInfo to surface to clients. Omit it to end in error without detail.

Subscribe to session errors

Subscribe to non-fatal session-level errors not scoped to any run: channel continuity loss (a re-attach with resumed: false, or FAILED / SUSPENDED / DETACHED), cancel-listener or attach failures, and any run-scoped error whose run supplied no onError. Returns an unsubscribe function. Once the session is closed this is a no-op.

const unsubscribe = session.on('error', (error) => {
  console.error('Session error:', error.code, error.message);
});

// later, when the listener is no longer needed
unsubscribe();

Parameters

Returns

() => void. An unsubscribe function. Call it to remove the listener.

Close the session

Cancel active runs, detach the channel the session attached, and clear handlers. Local-state-only for run lifecycle; it does not end in-progress Runs on the wire. End each Run explicitly before closing the session.

Invocation

A value object wrapping the JSON body a client sends to the agent's HTTP endpoint to start a Run.

Build from JSON

The entry point used by agent handlers: parse the request body and pass it to Invocation.fromJSON, then hand the result to createRun.

import { Invocation } from '@ably/ai-transport';

const data = await req.json();
const invocation = Invocation.fromJSON(data);

Parameters

Hydrate the conversation

Two accessors expose conversation content, at different scopes:

• run.messages is all of this Run's own messages: its triggering input plus its streamed output (across any suspend and resume). This is the unit to persist, not the value to feed the model in a multi-turn conversation.
• run.view is a read-only, leaf-pinned View of this Run's full branch, from its triggering input back to the conversation root. This is the value to feed the model.

run.view includes an ancestor turn only once its run has completed. An ancestor that is still active, suspended, cancelled, or errored is omitted, along with the input it replied to, so a dangling tool call from a concurrent or interrupted turn can't invalidate the prompt. The current run is always included, and an omitted ancestor reappears once it completes.

To rebuild the prior conversation for the model, drain run.view with loadOlder() for as much ancestor context as you want, then read getMessages():

// Rebuild the conversation from run.view before run.start(): draining pages in
// this run's triggering input (otherwise run.start() awaits it arriving live).
while (run.view.hasOlder()) {
  await run.view.loadOlder();
}
const conversation = run.view.getMessages().map(({ message }) => message);

await run.start();

For database-backed hydration, page run.view back only to the newest stored message with loadUntil instead of draining to the root.

RunEndReason

'complete' | 'cancelled' | 'error'. The terminal-reason discriminant: it is the reason field of the RunEndParams you pass to Run.end, the reason on StreamResult, and the value reflected on RunInfo.status once the Run terminates.

A Run that pauses for external input (tool approval, human-in-the-loop) uses Run.suspend instead of end, which publishes ai-run-suspend and leaves the Run alive at RunInfo.status === 'suspended'. A continuation Invocation resumes it via ai-run-resume.

Example

An HTTP handler that sets up the session, creates a Run, rebuilds the conversation from run.view, pipes the LLM stream, and ends the Run.

import * as Ably from 'ably';
import { createAgentSession, Invocation } from '@ably/ai-transport';
import { UIMessageCodec } from '@ably/ai-transport/vercel';

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

export async function POST(req: Request) {
  const invocation = Invocation.fromJSON(await req.json());

  const session = createAgentSession({
    client: ably,
    channelName: invocation.sessionName,
    codec: UIMessageCodec,
  });

  await session.connect();

  const run = session.createRun(invocation, { signal: req.signal });

  try {
    // Rebuild the conversation from run.view before run.start(): draining pages
    // in this run's triggering input (otherwise run.start() awaits it live).
    while (run.view.hasOlder()) {
      await run.view.loadOlder();
    }
    const conversation = run.view.getMessages().map(({ message }) => message);

    await run.start();

    const llmStream = await callMyLLM(conversation);
    const result = await run.pipe(llmStream);
    await run.end({ reason: result.reason });
  } catch (err) {
    await run.end({ reason: 'error' });
    throw err;
  } finally {
    await session.close();
  }

  return Response.json({ runId: run.runId, invocationId: run.invocationId });
}