Temporal

The Ably AI Transport integration with Temporal lets an AI agent running as a Temporal workflow publish its output directly to clients over a durable session, so every device in the conversation receives messages in realtime, and survives reconnects and workflow retries without polluting the conversation.

Temporal AI Agent activities can publish their output directly to clients over Ably Pub/Sub channels using the AI Transport SDK, so clients can receive messages in realtime without the need for those messages to be proxied back through either Temporal or another database and API server.

Ably AI Transport is built on top of Ably Pub/Sub channels, so a conversation inherits all the features of the Ably platform. Tokens arrive in the order, a client recovers its stream after a disruption without gaps, and messages travel at low latency.

How AI Transport works with Temporal

These are AI Transport's four concepts and their Temporal equivalents:

How AI Transport handles Workflows and Activities

When an AI Agent's loop is implemented as a Temporal workflow the entire execution of the workflow is a single Ably AI Transport run.

Each activity within the workflow maps to a single step within the run. Steps within a run are deduplicated based on their stepId, which means that if an activity fails and Temporal retries it, the retried activity's output automatically supersedes the failed attempt's output in the run. This keeps the conversation clean, and prevents messages from a failed activity polluting what the user will see.

Cancellation and steering messages

The Ably Pub/Sub channel that the AI agent's output it published to also allows bi-directional streaming and control between the AI agent and the clients. This allows the client to cancel a run mid-stream, or to send interrupt or steering messages to the AI agent to shape the agent's output while it is running. This is possible without having to build a custom API server to proxy those messages and signals into the Temporal workflow.

Getting started

Prerequisites

• Working knowledge of Temporal workflows and activities. If you are new to Temporal, start with Understanding Temporal and the Temporal 101 course.
• A local Temporal development environment, including the Temporal CLI.
• An Ably account with an API key, and an OpenAI API key.

Install the packages

Install the SDK, the Ably client, the OpenAI client, and the Temporal packages:

npm install @ably/ai-transport@^0.8.0 ably openai \
  @temporalio/client @temporalio/worker @temporalio/workflow @temporalio/activity

The Temporal subpaths declare their Temporal packages as optional peer dependencies, so install them alongside the SDK. @ably/ai-transport/temporal is worker-side and needs @temporalio/activity and @temporalio/worker. @ably/ai-transport/temporal/workflow runs inside the workflow sandbox and needs @temporalio/workflow. Get started with Temporal covers the app around the worker: the Next.js configuration, the token endpoint the browser authenticates against, and the namespace setup.

Decide the agent layout

Temporal workflows should be free of I/O and deterministic, so the workflow itself does not call the model or run tools. The workflow also does not touch the Ably AI Transport session directly. All interaction with the session happens through Temporal activities.

The session, a run, and a step all have ids. These ids thread through the workflow and its activities.

The SDK splits along the same line. Your workflow imports @ably/ai-transport/temporal/workflow, where every call schedules an activity. Your worker imports @ably/ai-transport/temporal, which holds the Ably client and the session.

Implement the agent

The agent side is made up of: the SDK's worker plugin, two Temporal activities you write, the workflow that schedules them, and the route that starts a workflow per user run.

1. Register the framing activities on the worker

Framing is everything that brackets a turn: opening the run, publishing its terminal, suspending it, and closing it on a failure path. The SDK ships those four activities and a worker plugin that registers them, so you write only the inference and tool activities.

Pass createAblyTransportPlugin to Worker.create. Your own activities still go in activities; the plugin adds to them.

// Agent-side worker entrypoint.
import Ably from 'ably';
import { Worker } from '@temporalio/worker';
import { createAblyTransportPlugin } from '@ably/ai-transport/temporal';
import { ResponsesCodec } from '@ably/ai-transport/openai';
import * as activities from './activities.js';

const worker = await Worker.create({
  taskQueue: 'ai-transport-demo',
  workflowsPath: require.resolve('./workflows'),
  activities,
  plugins: [
    createAblyTransportPlugin({
      codec: ResponsesCodec,
      createClient: () => new Ably.Realtime({ key: process.env.ABLY_API_KEY }),
    }),
  ],
});

await worker.run();

Each framing activity runs in a fresh process, so each one builds its own Ably client and session, does one thing, and tears both down. createClient supplies that client, and the SDK closes it before the activity returns.

Opening the run is its own activity, separate from the first model call. That persists the run's identifiers in the Temporal workflow log, so later activities re-use the same run across the workflow, and an inference failure retries the inference alone rather than re-opening the run.

A run's runId must be stable across retries, so an activity that retries in a fresh process re-enters the same run rather than opening a second one. The openRun activity pins the run id to the run's invocation id, which withRun takes from the Temporal workflow id by default.

This only covers a run the workflow opens itself. When a client resumes a suspended run, a second workflow continues it under a new workflow id, and AI Transport takes the run id from the resuming event's headers instead. So a run can span several workflow executions, and only the first one supplies the run id.

2. Run each model call in its own activity

Each model call is executed through the same activity. The activity adopts the open run, rebuilds the conversation from the session history, opens a step under the activity's id, and pipes one model response run into it:

// Agent-side activity.
import { Context } from '@temporalio/activity';
import Ably from 'ably';
import { createAgentSession, Invocation } from '@ably/ai-transport';
import { ResponsesCodec, toResponsesInput } from '@ably/ai-transport/openai';
import { stepIdFor } from '@ably/ai-transport/temporal';
import { createResponseStream } from './model.js';

// One Responses turn. Yields every event to the step and collects the function
// calls the model emitted, so the workflow can decide what to schedule next.
async function* modelTurn(input, signal, calls) {
  for await (const event of await createResponseStream({ input, signal })) {
    yield event;
    if (event.type === 'response.output_item.done' && event.item.type === 'function_call') {
      calls.push(event.item);
    }
  }
}

export async function runInferenceStep({ ids, invocation }) {
  const ably = new Ably.Realtime({ key: process.env.ABLY_API_KEY });
  const session = createAgentSession({
    client: ably,
    channelName: invocation.sessionName,
    codec: ResponsesCodec,
  });
  try {
    await session.connect();
    const run = session.adoptRun(Invocation.fromJSON(invocation), ids, {
      signal: Context.current().cancellationSignal,
    });
    await run.load();
    while (run.view.hasOlder()) await run.view.loadOlder();

    const step = run.createStep({ stepId: stepIdFor(ids.invocationId) });
    await step.start();

    // toResponsesInput flattens the conversation into the /responses input array.
    const input = toResponsesInput(run.view.getMessages().map((m) => m.message));
    const calls = [];
    const result = await step.pipe(modelTurn(input, run.abortSignal, calls));
    await step.end();

    if (result.reason === 'complete' && calls.length > 0) {
      // The only non-terminal outcome: the workflow schedules a tool activity per call.
      await session.detach();
      return { kind: 'server-tools', calls };
    }

    await run.end({ reason: result.reason });
    await session.detach();
    return { kind: result.reason };
  } finally {
    ably.close();
  }
}

3. Publish each tool result in its own activity

The workflow schedules one tool activity per call the model emitted. Each tool execution gets its own step, so a tool that throws retries and its retry's output supersedes the failed attempt.

// Agent-side activity.
import { Context } from '@temporalio/activity';
import Ably from 'ably';
import { createAgentSession, Invocation } from '@ably/ai-transport';
import { ResponsesCodec } from '@ably/ai-transport/openai';
import { stepIdFor } from '@ably/ai-transport/temporal';
import { executeTool } from './tools.js';

export async function runToolStep({ ids, invocation, call }) {
  const ably = new Ably.Realtime({ key: process.env.ABLY_API_KEY });
  const session = createAgentSession({
    client: ably,
    channelName: invocation.sessionName,
    codec: ResponsesCodec,
  });
  try {
    await session.connect();
    const run = session.adoptRun(Invocation.fromJSON(invocation), ids, {
      signal: Context.current().cancellationSignal,
    });
    await run.load();

    const step = run.createStep({ stepId: stepIdFor(ids.invocationId) });
    await step.start();
    const output = await executeTool(call.name, call.arguments);
    await step.send({
      type: 'function_call_output',
      item: {
        type: 'function_call_output',
        call_id: call.call_id,
        output: JSON.stringify(output),
      },
    });
    await step.end();

    await session.detach();
  } finally {
    ably.close();
  }
}

4. Drive the loop from the workflow

The workflow opens the run, runs the first model call, then handles tool activities and follow-up model calls until a model call returns a terminal outcome. withRun does the opening and the failure-path cleanup, so the loop is all that is left to write:

import { proxyActivities } from '@temporalio/workflow';
import { withRun } from '@ably/ai-transport/temporal/workflow';

const { runInferenceStep, runToolStep } = proxyActivities({
  startToCloseTimeout: '5 minutes',
  retry: { maximumAttempts: 3 },
});

export async function chatWorkflow({ invocation }) {
  await withRun(invocation, async (run) => {
    const ids = run.ids;

    let outcome = await runInferenceStep({ ids, invocation });
    while (outcome.kind === 'server-tools') {
      for (const call of outcome.calls) {
        await runToolStep({ ids, invocation, call });
      }
      outcome = await runInferenceStep({ ids, invocation });
    }
  });
}

server-tools is the only outcome the loop continues on. Every other kind means the activity that returned it has already published the run's terminal event, so the workflow returns without touching the session.

withRun takes the run's invocation id from the Temporal workflow id, and the route below starts every workflow with workflowId = invocationId. Override it with withRun(invocation, { invocationId }, body) when one workflow serves several turns, because that workflow keeps one workflow id across all of them and every turn would otherwise fold onto the first turn's run.

Override the framing activities' timeouts and retry policies through the same options object:

await withRun(invocation, {
  activityOptions: {
    default: { startToCloseTimeout: '5 minutes' },
    // openRun pages channel history, so a long conversation needs longer.
    openRun: { startToCloseTimeout: '10 minutes' },
  },
}, async (run) => {
  // ...
});

5. Trigger the workflow

The agent exposes an HTTP API endpoint to start a workflow. The endpoint returns immediately after the workflow is started, and the client receives the rest of the output by subscribing to the AI Transport session. Give the workflow the same id as the invocation, so every HTTP request gets its own workflow:

// Agent-side route handler.
import { Client, Connection } from '@temporalio/client';

export async function POST(req) {
  const invocation = await req.json();
  const invocationId = crypto.randomUUID();

  const connection = await Connection.connect({ address: 'localhost:7233' });
  const client = new Client({ connection, namespace: 'default' });
  await client.workflow.start('chatWorkflow', {
    workflowId: invocationId,
    taskQueue: 'ai-transport-demo',
    args: [{ invocation }],
  });

  return Response.json({ invocationId });
}

Connect the browser

Nothing on the client is workflow-aware. Create a client session naming the conversation, and authenticate the Ably client from a token endpoint rather than an API key:

// Client-side. Never put an API key here; the browser fetches a token instead.
import * as Ably from 'ably';
import { createClientSession } from '@ably/ai-transport';
import { ResponsesCodec } from '@ably/ai-transport/openai';

const ably = new Ably.Realtime({ authUrl: '/auth' });

const session = createClientSession({
  client: ably,
  channelName: conversationId,
  codec: ResponsesCodec,
});

await session.connect();

The session's view holds the conversation and publishes new messages. Sending is two steps, and that split keeps the client independent of Temporal. view.send publishes the user's message to the session and hands back a run, but the session only publishes to Ably and never makes an HTTP request. You wake the agent yourself, by posting the run's invocation pointer to your agent route. That pointer is the JSON the route's Invocation.fromJSON rebuilds before it starts the workflow:

// Client-side, on the session created above.
import { ResponsesCodec } from '@ably/ai-transport/openai';

const { view } = session;

// Fires on every change: tokens appended, steps opening, runs ending.
view.on('update', () => render(view.getMessages()));

// Reveal a page of history before the first paint.
await view.loadOlder(30);

async function ask(text) {
  const run = await view.send(
    ResponsesCodec.createUserMessage({
      role: 'user',
      items: [{ type: 'message', role: 'user', content: [{ type: 'input_text', text }] }],
    }),
  );

  // The session is pure Ably transport, so the application owns the wake POST.
  await fetch('/api/chat', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify(run.toInvocation().toJSON()),
  });

  return run;
}

Cancel a run with run.cancel() on the handle ask returned. The cancel is keyed on the id of the message the client published, which the client knows as soon as it publishes, so a cancel sent before the agent has created the run id still works. That matters with Temporal, because the run id does not exist until the workflow has been scheduled and openRun has run, and a user who wants to stop the agent usually clicks during that gap.

End the run when retries are exhausted

Every run should end with the workflow it belongs to. An unclosed run leaves every subscribed client waiting on a stream that never ends.

withRun handles that. When the body throws, it schedules the SDK's cleanupRun activity, which adopts the run and ends it as error. The attempt runs in a non-cancellable scope, so a cancelled or terminated workflow still closes its run, and cleanup's own failure is swallowed so the body's error reaches Temporal unmasked.

The cleanup is best-effort, in three ways worth knowing about:

• It gets one attempt with a short timeout. Retrying would let a hanging cleanup hold up a terminate.
• It no-ops when the run has already finished, or is parked suspended.
• It fires only on a throw. A body that returns without publishing a terminal leaves the run open.

On the happy path withRun publishes nothing, and your application publishes the run's terminal itself. Doing that inside an activity that already has the run loaded costs no extra adopt, which is why the inference activity above calls run.end directly. The run.end on the handle is there for the case where the workflow is the thing that knows the turn is over, and it costs a fresh adopt and load.

Use openRun instead of withRun when the handle has to outlive a single lexical scope, and take on the cleanup yourself through run.cleanup(message).

FAQ

Do I need this to run an agent in Temporal?

Temporal is a workflow engine that can run the agent loop. Ably AI Transport is how you get AI agent responses from those activities to clients in realtime.

Why not send the output out of the workflow/activity directly?

Temporal workflows are started by a HTTP request, but the workflow's activities are scheduled and executed on a Temporal worker. This worker might not be the same process that received the HTTP request, so it won't be able to reach the client that made the request. AI Transport solves that problem by publishing the output to a channel that any client can subscribe to, regardless of which process started the workflow.

What if nobody is watching while the agent runs?

The run completes anyway. Each activity publishes its output as it goes, so a client that attaches later loads the finished answer rather than needing to have been connected throughout. How far back a client can load depends on how long the session retains history; for conversations longer than that, hydrate from your own database.

Does the browser talk to Temporal?

No. The browser attaches to the session and posts to your agent's HTTP endpoint, and the endpoint starts the workflow. Nothing on the client is workflow-aware, so you can move the agent into or out of Temporal without changing client code.

Demo app

Next, explore the Temporal demo app. It is a complete Next.js chat app with the Temporal worker built in, and a deliberately flaky tool that throws on roughly half its attempts so you can see Temporal retries in action. The demo app is built on the Vercel codec rather than the OpenAI one, but the workflow and activity structure is the same.

Read next

• Get started with Temporal: the same worker built out in full inside a Next.js chat app, with project setup and the auth endpoint.
• Durable execution: the same pattern, generalised to any workflow engine.
• Steps: the retry unit Temporal activities map to.
• Temporal workflow: openRun, withRun, and the RunHandle they return.
• Temporal worker: createAblyTransportPlugin, and the stepIdFor helper that produces a Temporal-safe step id.
• AgentSession API reference: adoptRun, createStep, detach, and end.
• ClientSession API reference: createClientSession, view.send, view.loadOlder, and run.cancel.