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.7.0 ably openai \
  @temporalio/client @temporalio/worker @temporalio/workflow @temporalio/activity

The @ably/ai-transport/temporal subpath declares @temporalio/activity as an optional peer dependency, so install it alongside the SDK. 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 interact with the Ably AI Transport session directly. Instead, all interaction with the AI Transport session happens through Temporal activities.

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

Implement the agent

The agent side is made up of: three Temporal activities, the workflow that schedules them, and the route that starts a workflow per user run.

1. Creating a new run in its own activity

Creating a new run is its own activity, separate from the first model call. This allows the run's identifiers to be persisted in the Temporal workflow log, and later activities can re-use the same run across the workflow.

This activity calls createRun and run.start(), publishes the run-start event, and returns the run's ids without calling the model:

// 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';

export async function openRun({ invocation, invocationId }) {
  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.createRun(Invocation.fromJSON(invocation), {
      invocationId,
      // The route passes this id as both the workflowId and the invocationId,
      // so it is stable across retries. A continuation ignores it and reads
      // the run id off the channel.
      runId: invocationId,
    }, {
      signal: Context.current().cancellationSignal,
    });

    // Load the conversation history.
    while (run.view.hasOlder()) await run.view.loadOlder();
    await run.start();

    // detach (not end): the run is deliberately left active so the workflow's
    // first runInferenceStep can adopt it. session.end() would publish
    // `ai-run-end` and mark the run terminal.
    await session.detach();

    return {
      runId: run.runId,
      invocationId: run.invocationId,
    };
  } finally {
    ably.close();
  }
}

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. Use the invocation id as the Temporal workflow id and the run id.

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. Its outer catch handles the case where Temporal has given up on an activity:

import { proxyActivities } from '@temporalio/workflow';

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

// One attempt and a tight timeout, so cleanup cannot cascade its own failure.
const { cleanupRun } = proxyActivities({
  startToCloseTimeout: '30 seconds',
  retry: { maximumAttempts: 1 },
});

export async function chatWorkflow({ invocation, invocationId }) {
  let ids;
  try {
    ids = await openRun({ invocation, invocationId });

    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 });
    }
  } catch (error) {
    if (ids) {
      await cleanupRun({
        ids,
        invocation,
        errorMessage: error instanceof Error ? error.message : 'workflow failed',
      }).catch(() => {});
    }
    throw error;
  }
}

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.

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, invocationId }],
  });

  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

All runs should be ended with the workflow they belong to ends. So a workflow should implement error handling that starts a cleanup activity if the workflow catches an error.

// Agent-side activity, in the same module as the three above.
export async function cleanupRun({ ids, invocation, errorMessage }) {
  const ably = new Ably.Realtime({ key: process.env.ABLY_API_KEY });
  const session = createAgentSession({ client: ably, channelName: invocation.sessionName });
  try {
    await session.connect();
    const run = session.adoptRun(Invocation.fromJSON(invocation), ids);
    try {
      await run.load();
    } catch {
      // Already ended, so there is no open run to close.
      await session.detach();
      return;
    }
    await run.end({
      reason: 'error',
      error: new Ably.ErrorInfo(errorMessage, 104000, 500),
    });
    await session.detach();
  } finally {
    ably.close();
  }
}

run.load() promise checks the run status, rejecting if the run has already ended, so scheduling cleanup on every failure path is safe. Give it one attempt and a short timeout so its own failure cannot cascade.

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.
• stepIdFor API reference: the 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.