Skip to main content

Vercel AI SDK

The Alien platform is fully compatible with the Vercel AI SDK (ai ≥ 6) via @ai-sdk/openai's OpenAI Responses provider. This page shows how to wire up the provider, stream responses with platform extensions, handle multi-turn conversations, and build a complete Next.js App Router chat route — based on the reference implementation in lds-chatbot.

Dependencies​

npm install ai @ai-sdk/openai
# or
yarn add ai @ai-sdk/openai

The reference app uses:

{
  "ai": "^6.0.168",
  "@ai-sdk/openai": "^3.0.53"
}

Creating the provider​

The platform exposes the Responses API at /agent/:workflowId/responses. Point createOpenAI at that base URL and pass your access token in both auth headers — the platform accepts either Authorization: Bearer (API tokens starting with oat_) or x-oauth-access-token (OAuth JWT tokens), so sending both ensures compatibility regardless of token type.

import { createOpenAI } from "@ai-sdk/openai"

function platformProvider(workflowId: number, accessToken: string) {
    return createOpenAI({
        baseURL: `https://api.alien.club/agent/${workflowId}`,
        apiKey: "unused",          // field is required by the SDK but ignored by the platform
        headers: {
            "authorization":        `Bearer ${accessToken}`,
            "x-oauth-access-token": accessToken,
        },
    })
}
note

Pass apiKey: "unused" (or any non-empty string) — the SDK requires the field but the platform authenticates via the custom headers above, not via the OpenAI API key mechanism.

Basic single-turn stream​

Use provider.responses("agent") to target the Responses API endpoint. The "agent" string is the model name forwarded in the request body — the platform ignores it and uses the model configured on the workflow.

import { streamText } from "ai"

const provider = platformProvider(workflowId, accessToken)

const result = await streamText({
    model: provider.responses("agent"),
    prompt: "What is the SYNTEC collective agreement?",
})

for await (const chunk of result.textStream) {
    process.stdout.write(chunk)
}

Multi-turn with previousResponseId​

The Responses API maintains session state server-side. Pass the id from the previous response as previousResponseId to chain turns — no need to resend conversation history.

const result = await streamText({
    model: provider.responses("agent"),
    prompt: "What is the trial period for an executive?",
    providerOptions: {
        openai: { previousResponseId: "resp_KMen5Xw2..." },
    },
})

// Read the new response ID for the next turn
for await (const _ of result.textStream) { /* consume stream */ }

const finalResponse = await result.experimental_providerMetadata
const nextResponseId = (finalResponse?.openai as { responseId?: string })?.responseId

For a cleaner pattern, use includeRawChunks: true to intercept the response.completed event directly (see Platform extension chunks below).

Next.js App Router — complete chat route​

This is the full pattern used in lds-chatbot. It streams the response to the client while persisting the message and capturing the responseId in the background.

// app/api/chat/route.ts
import { createOpenAI }           from "@ai-sdk/openai"
import { streamText }             from "ai"
import { createUIMessageStream, createUIMessageStreamResponse } from "ai"

export async function POST(request: Request) {
    const { workflowId, message, previousResponseId, accessToken } = await request.json()

    const provider = createOpenAI({
        baseURL: `https://api.alien.club/agent/${workflowId}`,
        apiKey:  "unused",
        headers: {
            "authorization":        `Bearer ${accessToken}`,
            "x-oauth-access-token": accessToken,
        },
    })

    // Holds the response ID captured during the stream — read after the stream drains
    let capturedResponseId: string | undefined

    const stream = createUIMessageStream({
        execute: async (writer) => {
            const result = await streamText({
                model: provider.responses("agent"),
                prompt: message,
                providerOptions: previousResponseId
                    ? { openai: { previousResponseId } }
                    : undefined,
                includeRawChunks: true,
                // Optional: smooth word-by-word delivery
                // experimental_transform: smoothStream({ delayInMs: 20, chunking: "word" }),
            })

            // Emit a stable conversationId chunk so the client can track the session
            // before any text arrives
            writer.write({ type: "data-conversationId", value: { conversationId: previousResponseId ?? "pending" } })

            for await (const part of result.fullStream) {
                switch (part.type) {
                    case "text-delta":
                        writer.write({ type: "text-delta", textDelta: part.textDelta })
                        break

                    case "raw": {
                        // Extract responseId and other platform metadata from raw Responses API events
                        const raw = part.rawValue as Record<string, unknown>
                        const eventType = raw?.type as string | undefined

                        if (eventType === "response.completed") {
                            const response = raw?.response as Record<string, unknown> | undefined
                            capturedResponseId = response?.id as string | undefined
                        }
                        break
                    }
                }
            }
        },
    })

    // Tee: one branch goes to the client, the other to background persistence
    const [forClient, forPersist] = stream.tee()

    // Background: drain forPersist and save to DB
    void persistResponse(forPersist, capturedResponseId, workflowId)

    return createUIMessageStreamResponse({ stream: forClient })
}

async function persistResponse(stream: ReadableStream, responseId: string | undefined, workflowId: number) {
    // Drain the stream and save the assistant message + responseId to your database
    // ...
}

Platform extension chunks​

When you pass includeRawChunks: true, the raw Responses API events are exposed as part.type === "raw" in fullStream. The platform's reference implementation (responses_stream.ts) emits additional data-* chunks from these raw events to carry multi-agent context to the client.

data-conversationId​

Emitted once at the start of the stream, before any text. Carries the session identifier for this turn.

writer.write({
    type: "data-conversationId",
    value: { conversationId: string },
})

On the client, read it via useChat's data field and store it — pass it as previousResponseId on the next turn to maintain the session.

data-subagent​

Emitted when a subagent becomes active in a multi-agent run. Resolved from response.metadata.x_alien_agent_registry and the per-item agent:<id>:: prefix in output item IDs.

writer.write({
    type: "data-subagent",
    value: {
        id:       "subagent-6",         // node id in the workflow
        name:     "Légifrance researcher",
        kind:     "subagent",           // "main" | "subagent" | "tool"
        parentId: "MAIN",
    },
})

data-subagent-end​

Emitted with no value when the root agent resumes after a subagent's turn finishes. Use it to close a subagent panel in your UI.

writer.write({ type: "data-subagent-end", value: undefined })

data-toolCall​

Emitted on each tool invocation.

writer.write({
    type: "data-toolCall",
    value: {
        id:   "call_abc",
        name: "search_legal_database",
        args: { query: "SYNTEC trial period" },
    },
})

data-streamProgress​

Emitted on every raw Responses API event. Carries the sequence_number for resume and a terminal flag. Mark it transient: true so the AI SDK discards it and doesn't accumulate it in memory.

writer.write({
    type: "data-streamProgress",
    value: {
        responseId:     "resp_...",
        sequenceNumber: 7,
        terminal:       false,
    },
    transient: true,
})

Store the latest sequenceNumber client-side. The platform-to-route hop resumes on its own with the transport below (see Resume), this cursor is what the browser-to-route hop resumes from.

Parsing x_alien_agent_registry​

The agent registry arrives as a JSON-encoded string inside response.metadata. Double-parse it to get the array:

if (eventType === "response.created") {
    const response = raw?.response as Record<string, unknown>
    const metadata = response?.metadata as Record<string, string> | undefined

    const registryRaw = metadata?.["x_alien_agent_registry"]
    if (registryRaw) {
        const registry: Array<{ id: string; kind: string; name: string; parent_id: string | null }> =
            JSON.parse(registryRaw)
        // build a name-lookup map for subagent display names
    }

    const rootAgentId = metadata?.["x_alien_root_agent_id"]  // typically "MAIN"
}

Reading extension data in the client​

Use useChat from @ai-sdk/react. Data chunks written via writer.write({ type: "data-*", value: ... }) arrive in the data array returned by the hook.

"use client"
import { useChat } from "@ai-sdk/react"

export function ChatPanel({ workflowId }: { workflowId: number }) {
    const { messages, data, input, handleInputChange, handleSubmit } = useChat({
        api: "/api/chat",
        body: { workflowId },
    })

    // Read the platform conversationId from the data stream
    const conversationId = data?.findLast(
        (d): d is { conversationId: string } =>
            typeof d === "object" && d !== null && "conversationId" in d
    )?.conversationId

    // Active subagent (if any)
    const activeSubagent = data?.findLast(
        (d): d is { id: string; name: string } =>
            typeof d === "object" && d !== null && "id" in d && "name" in d
    )

    return (
        <div>
            {activeSubagent && <p>Subagent active: {activeSubagent.name}</p>}

            {messages.map((msg) => (
                <div key={msg.id}>
                    <strong>{msg.role}:</strong> {msg.content}
                </div>
            ))}

            <form onSubmit={handleSubmit}>
                <input value={input} onChange={handleInputChange} />
                <button type="submit">Send</button>
            </form>
        </div>
    )
}

Resume after an interruption​

A turn outlives the connection that streams it. A stream ends before its terminal event when the platform pod serving it shuts down (a deploy, a scale-in, a node drain), when the event store is briefly unreachable, when the ingress cuts a connection at its 600 s cap, or on any network drop. The run keeps going on its worker, and any platform pod serves the rest of it from GET /agent/:id/responses/<responseId>?starting_after=<sequenceNumber>.

A shutting-down pod first lets open streams finish for up to 90 s. A stream still open after that ends with an error event whose code is stream_interrupted and whose sequence_number is the last one delivered. Other interruptions end the connection with no frame at all. Treat both the same way: any stream that ends without response.completed, response.failed or response.incomplete is resumed, never reported as a failure.

Auto-resume transport​

Hand the provider a fetch that resumes on its own, so streamText sees one uninterrupted stream however many times the platform connection is cut. It:

  • tracks the highest sequence_number and the response.id from response.created;
  • drops a frame whose (type, sequence_number) pair it already forwarded, the pair and not the number alone since response.tool_call.output and response.cost_breakdown reuse the sequence_number of the event they follow;
  • swallows the stream_interrupted error event, and on it or on a connection ending without a terminal event, reconnects with starting_after set to the highest sequence_number. The platform sends response.created as soon as the stream opens, so a stream that delivered nothing has no id and is not resumed;
  • backs off exponentially between attempts, gives up after a bounded number of them, and resets the count whenever an attempt delivers a new event.
// lib/platform/resumable_fetch.ts
const TERMINAL_EVENTS = new Set(["response.completed", "response.failed", "response.incomplete"])

export type ResumeOptions = {
    maxAttempts: number   // consecutive reconnects that deliver nothing before giving up
    baseDelayMs: number   // first backoff, doubled on every further attempt
    maxDelayMs:  number   // backoff ceiling
}

const DEFAULT_RESUME: ResumeOptions = { maxAttempts: 6, baseDelayMs: 500, maxDelayMs: 10_000 }

type Frame = { raw: string; type: string; sequenceNumber: number; payload: Record<string, unknown> }

/**
 * Parses one SSE frame, or returns null for a keep-alive comment.
 */
function parseFrame(raw: string): Frame | null {
    const data = raw.split("\n").find(line => line.startsWith("data:"))

    if (data === undefined) return null

    const payload = JSON.parse(data.slice("data:".length).trim()) as Record<string, unknown>

    return { raw, type: String(payload.type), sequenceNumber: Number(payload.sequence_number), payload }
}

/**
 * Wraps `fetch` so a streaming `POST .../responses` resumes across interruptions.
 */
export function resumableFetch(options: ResumeOptions = DEFAULT_RESUME): typeof fetch {
    return async (input, init) => {
        const response = await fetch(input, init)
        const url = new URL(input instanceof Request ? input.url : String(input))
        const isStreamingCreate =
            init?.method === "POST" &&
            url.pathname.endsWith("/responses") &&
            response.ok &&
            response.body !== null &&
            (response.headers.get("content-type") ?? "").startsWith("text/event-stream")

        if (!isStreamingCreate) return response

        const body = resumableBody(response.body!, `${url.origin}${url.pathname}`, init!.headers, options)

        return new Response(body, { status: response.status, headers: response.headers })
    }
}

function resumableBody(
    first: ReadableStream<Uint8Array>,
    responsesUrl: string,
    headers: HeadersInit | undefined,
    options: ResumeOptions,
): ReadableStream<Uint8Array> {
    const encoder = new TextEncoder()
    const forwarded = new Set<string>()
    let responseId: string | null = null
    let highest = -1

    return new ReadableStream({
        async start(controller) {
            let body: ReadableStream<Uint8Array> | null = first
            let attempts = 0

            for (;;) {
                let isTerminal = false
                let isNewEvent = false

                if (body !== null) {
                    const decoder = new TextDecoder()
                    const reader = body.getReader()
                    let buffer = ""

                    try {
                        for (;;) {
                            const { value, done } = await reader.read()

                            if (done) break

                            buffer += decoder.decode(value, { stream: true })

                            let end = buffer.indexOf("\n\n")

                            while (end !== -1) {
                                const frame = parseFrame(buffer.slice(0, end))

                                buffer = buffer.slice(end + 2)
                                end = buffer.indexOf("\n\n")

                                if (frame === null) continue

                                // Not an event of the response: the cursor to resume from
                                if (frame.type === "error" && frame.payload.code === "stream_interrupted") continue

                                const key = `${frame.type}:${frame.sequenceNumber}`

                                if (forwarded.has(key)) continue

                                forwarded.add(key)
                                isNewEvent = true
                                highest = Math.max(highest, frame.sequenceNumber)

                                if (frame.type === "response.created") {
                                    responseId = String((frame.payload.response as { id: string }).id)
                                }

                                controller.enqueue(encoder.encode(`${frame.raw}\n\n`))

                                if (TERMINAL_EVENTS.has(frame.type)) isTerminal = true
                            }
                        }
                    } catch {
                        // A reset connection is one more interruption, resumed below
                    }
                }

                if (isTerminal) {
                    controller.close()

                    return
                }

                attempts = isNewEvent ? 1 : attempts + 1

                if (responseId === null || attempts > options.maxAttempts) {
                    controller.error(new Error(`response stream interrupted and not resumed after ${attempts - 1} attempts`))

                    return
                }

                const delay = Math.min(options.maxDelayMs, options.baseDelayMs * 2 ** (attempts - 1))

                // Jitter spreads the reconnects of every client a pod shutdown cut at once
                await new Promise(resolve => setTimeout(resolve, delay / 2 + Math.random() * (delay / 2)))

                // A known response id means response.created, sequence_number 0, was received
                const resumed = await fetch(`${responsesUrl}/${responseId}?starting_after=${highest}`, { headers }).catch(() => null)

                // A 503 with Retry-After means the event store is still unreachable, retry after the backoff
                body = resumed !== null && resumed.ok ? resumed.body : null
            }
        },
    })
}

Wire it into the provider, nothing else changes in the chat route:

const provider = createOpenAI({
    baseURL: `https://api.alien.club/agent/${workflowId}`,
    apiKey:  "unused",
    headers: {
        "authorization":        `Bearer ${accessToken}`,
        "x-oauth-access-token": accessToken,
    },
    fetch: resumableFetch(),
})

The same wrapper works where you consume the raw stream yourself: pass resumableFetch()(url, init)'s body to translateResponseStream.

note

A resume after a pod shutdown waits for the drain (up to 90 s) only for the streams still open at its end, every other stream finished on its own. A GET answering 410 Gone means the response is past its retention, re-issue the turn with POST. Chat Completions (/agent/:id/chat/completions) has no resume, the same interruption ends that stream with a final chunk carrying x_alien.error.code = "stream_interrupted", so prefer the Responses API for long turns.

Browser to route​

The transport above covers the platform-to-route hop. If the browser's own connection to your route drops, reconnect it through a resume route that starts from the sequenceNumber the client last received from data-streamProgress:

// app/api/chat/resume/route.ts
import { translateResponseStream } from "@/lib/platform/responses_stream"

export async function GET(request: Request) {
    const { searchParams } = new URL(request.url)
    const responseId    = searchParams.get("responseId")!
    const workflowId    = searchParams.get("workflowId")!
    const startingAfter = searchParams.get("startingAfter")
    const cursor        = startingAfter === null ? "" : `?starting_after=${startingAfter}`

    const platformResponse = await fetch(
        `https://api.alien.club/agent/${workflowId}/responses/${responseId}${cursor}`,
        { headers: { authorization: `Bearer ${accessToken}` } },
    )

    const stream = translateResponseStream(platformResponse.body!, {
        workflowId: Number(workflowId),
        accessToken,
        subagentNames: new Map(),
    })

    return createUIMessageStreamResponse({ stream })
}

SidecarState — full implementation reference​

The responses_stream.ts file in lds-chatbot contains a complete production implementation of all of the above patterns. It handles:

  • Registry parsing from response.metadata.x_alien_agent_registry
  • Per-item agent attribution from agent:<id>:: item ID prefixes
  • Subagent activation and deactivation signals
  • Tool call forwarding
  • TTFT timing measurement
  • Stream drain and result capture (responseId, usage, ok)

Copy and adapt lib/platform/responses_stream.ts as the starting point for your own integration rather than reimplementing from scratch.

Summary​

TaskHow
Create providercreateOpenAI({ baseURL: .../agent/:id, headers: { authorization, x-oauth-access-token } })
Target Responses APIprovider.responses("agent")
Multi-turnproviderOptions: { openai: { previousResponseId } }
Capture responseIdRead response.id from raw part where type === "response.completed"
Agent registryJSON.parse(metadata["x_alien_agent_registry"]) — double-parse required
Subagent contextdata-subagent / data-subagent-end chunks emitted by your raw handler
ResumecreateOpenAI({ fetch: resumableFetch() }), reconnecting with GET /agent/:id/responses/:respId?starting_after=<seq> on stream_interrupted or a stream ending without a terminal event

See also​