본문으로 건너뛰기

연결 어댑터

연결 어댑터는 서버의 청크를 ChatClient로 전달하고, 이를 통해 프레임워크의 useChat으로 전달하는 _방법_을 결정하는 구성 요소입니다. TanStack AI의 나머지 기능인 청크 처리, 메시지 재조립, 도구 호출, UI 업데이트는 전송 방식과 무관합니다. 네트워크에 접근하는 것은 어댑터뿐입니다.

이 페이지에서는 지원되는 모든 전송 방식과 각각을 선택할 상황, 사용자 지정 어댑터를 만드는 방법을 설명합니다.

전송 방식 선택​

상황사용
일반 HTTP 서버에서 기본 방식을 사용하려는 경우fetchServerSentEvents
SSE를 차단하는 환경(일부 엣지 런타임, 엄격한 프록시)fetchHttpStream
React Native 또는 Expo기본적으로 xhrHttpStream, SSE에는 xhrServerSentEvents, 스트리밍 fetch를 사용할 수 있을 때만 fetchHttpStream
AsyncIterable<StreamChunk>을 동기적으로 반환하는 코드(프로세스 내 chat(), RSC 스트림, 테스트)stream
Response 또는 AsyncIterable<StreamChunk>으로 해석되는 비동기 호출(TanStack Start 서버 함수 또는 Promise를 반환하는 함수)fetcher
Cap'n Web, gRPC-Web, tRPC와 같은 RPC 프레임워크rpcStream
여러 실행을 처리하는 단일 장기 실행·재개 가능한 WebSocketwebSocket
BroadcastChannel, postMessage, 공유 워커 또는 다른 영속 전송사용자 지정 subscribe / send 어댑터
사용자 지정 fetch 래핑(인증 갱신, 재시도)이 필요한 표준 SSEfetchClient를 사용하는 [fetchServerSentEvents]
그 밖의 방식(HTTP/3, 다른 프로토콜을 통한 Server-Sent Events 등)사용자 지정 connect 어댑터

모든 어댑터는 동일한 StreamChunk 이벤트(AG-UI 프로토콜)를 생성합니다. 선택은 전적으로 전송 방식에 관한 것입니다.

서버 보낸 이벤트 (SSE)​

기본 방식입니다. SSE는 브라우저 전반에서 잘 지원되고 대부분의 프록시를 투명하게 통과하며 디버깅하기 쉽습니다. 서버에서는 toServerSentEventsResponse()와 함께 사용합니다.

import { useChat, fetchServerSentEvents } from "@tanstack/ai-react";

const { messages, sendMessage } = useChat({
  connection: fetchServerSentEvents("/api/chat"),
});

동적 URL 및 헤더. 값이 요청별 상태(현재 사용자, 새 토큰)에 따라 달라지면 함수를 전달합니다.

import { useChat, fetchServerSentEvents } from "@tanstack/ai-react";
import { currentUserId, getToken } from "./auth";

const { messages } = useChat({
  connection: fetchServerSentEvents(
    () => `/api/chat?user=${currentUserId}`,
    () => ({
      headers: { Authorization: `Bearer ${getToken()}` },
    }),
  ),
});

정적 본문. options.body의 모든 값은 서버로 전송되는 AG-UI forwardedProps 페이로드에 병합됩니다. 메시지별 sendMessage body가 이 값을 우선합니다.

import { useChat, fetchServerSentEvents } from "@tanstack/ai-react";

const { messages } = useChat({
  connection: fetchServerSentEvents("/api/chat", {
    body: { provider: "openai", model: "gpt-5.5" },
  }),
});

팁: body와 forwardedProps는 동일한 와이어 필드를 채웁니다. 정적 기본값에는 어댑터 body를 사용합니다. 변경되는 값에는 forwardedProps 생성자 옵션 또는 sendMessage(content, { body })를 사용합니다. 런타임 값이 항상 우선합니다.

호출별 본문. sendMessage의 두 번째 인수로 한 번의 전송에 사용할 추가 JSON을 전달합니다. 이 값은 채팅 수준의 body와 함께 forwardedProps에 얕게 병합됩니다({ ...chatBody, ...sendOptions.body }). 중복 키에는 sendMessage 값이 사용됩니다. 이 병합은 해당 요청에만 적용됩니다.

import { useChat, fetchServerSentEvents } from "@tanstack/ai-react";

const { sendMessage } = useChat({
  connection: fetchServerSentEvents("/api/chat"),
  forwardedProps: { provider: "openai" },
});

await sendMessage("Summarize the attached files", {
  body: { attachmentIds: ["att_1", "att_2"] },
});

서버는 chatParamsFromRequest에서 병합된 객체를 읽습니다. 모델이 해당 키를 보면 안 된다면 messages에 복사하지 않습니다.

import {
  chat,
  chatParamsFromRequest,
  toServerSentEventsResponse,
} from "@tanstack/ai";
import { openaiText } from "@tanstack/ai-openai";

export async function POST(request: Request) {
  const { messages, forwardedProps } = await chatParamsFromRequest(request);
  const stream = chat({
    adapter: openaiText("gpt-5.5"),
    messages,
  });
  if (
    forwardedProps &&
    typeof forwardedProps === "object" &&
    "attachmentIds" in forwardedProps
  ) {
    const { attachmentIds } = forwardedProps
    if (Array.isArray(attachmentIds) && attachmentIds.length > 0) {
      // Look up the uploads. Do not add them to `messages`.
    }
  }
  return toServerSentEventsResponse(stream);
}

ChatClient에서 직접 호출할 때도 위치 기반 두 번째 인수는 동일한 추가 JSON입니다. 이 값은 sendOptions.body와 얕게 병합되며, 키가 충돌하면 sendOptions.body가 우선합니다.

reload()은 새 요청을 시작합니다. 채팅 수준의 forwardedProps / body만 사용하며, 이전 전송의 호출별 body를 재생하지 않습니다.

재개 가능한 SSE​

fetchServerSentEvents는 SSE id: 값을 감시합니다. id를 받은 후 연결이 끊기면 Last-Event-ID를 사용해 재연결하고 재생된 접두부의 중복을 제거합니다. joinRun(runId)는 offset=-1과 실행 id를 사용해 읽기 전용 GET을 수행하고, 진행 중이거나 완료된 실행을 처음부터 재생합니다.

id는 서버가 toServerSentEventsResponse에 내구성 어댑터를 전달할 때만 나타납니다. id는 해당 어댑터가 소유하는 불투명 토큰이며, 채팅 클라이언트가 생성하거나 파싱하거나 영속화하지 않습니다. id가 없으면 동작은 일반적인 단일 fetch와 동일합니다. 재개 가능한 스트림을 참조하세요.

joinRun이 작동하려면 라우트에 POST와 함께 GET 핸들러가 필요합니다(두 번째 탭 또는 새로 고침). POST는 새 실행과 자동 재연결을 처리하고(Last-Event-ID와 함께 동일한 본문을 다시 전송), GET은 알려진 실행을 처음부터 재생합니다.

import {
  chat,
  chatParamsFromRequest,
  memoryStream,
  resumeServerSentEventsResponse,
  toServerSentEventsResponse,
} from "@tanstack/ai";
import { openaiText } from "@tanstack/ai-openai";

export async function POST(request: Request) {
  const { messages, threadId, runId } = await chatParamsFromRequest(request);
  const stream = chat({ adapter: openaiText("gpt-5.5"), messages, threadId, runId });
  return toServerSentEventsResponse(stream, {
    durability: { adapter: memoryStream(request) },
  });
}

// joinRun hits GET ?offset=-1&runId=... (replay only, no messages sent).
export async function GET(request: Request) {
  return resumeServerSentEventsResponse({ adapter: memoryStream(request) });
}

GET 핸들러는 provider를 호출하지 않습니다. 재생 시 내구성 어댑터의 resumeFrom()이 (?offset에서 가져온) null이 아니므로 로그를 대신 재생합니다. 요청에 재개 오프셋이 없으면 resumeServerSentEventsResponse는 400을 반환합니다. NDJSON 어댑터에는 resumeHttpResponse를 사용합니다.

fetchHttpStream과 xhrHttpStream은 NDJSON에서도 같은 방식으로 재개합니다. 이때 오프셋은 SSE id: 줄 대신 { id, chunk } 봉투에 포함됩니다(아래 참조). toHttpResponse에 내구성 어댑터를 전달해 활성화합니다. xhrServerSentEvents는 fetchServerSentEvents와 정확히 같은 방식으로 SSE에서 재개합니다 (toServerSentEventsResponse와 해당 id: 줄을 함께 사용).

HTTP 스트리밍 (NDJSON)​

SSE를 지원하지 않는 환경(일부 엣지 런타임, 특정 모바일 WebView 또는 프록시가 text/event-stream을 제거하는 환경)에서는 줄바꿈으로 구분된 원시 JSON을 사용합니다. 와이어 형식은 줄마다 하나의 JSON StreamChunk입니다.

import { useChat, fetchHttpStream } from "@tanstack/ai-react";

const { messages } = useChat({
  connection: fetchHttpStream("/api/chat"),
});

서버에서는 각 청크를 JSON.stringify(chunk) + "\n" 형식으로 응답 본문에 쓰거나 toHttpResponse(stream)을 사용합니다. 옵션(url, headers, body, fetchClient, 동적 함수)은 fetchServerSentEvents와 정확히 같습니다.

fetchHttpStream도 재개할 수 있습니다. toHttpResponse에 내구성 어댑터를 전달하면 각 줄이 { id, chunk } 봉투가 됩니다. 연결이 끊기면 Last-Event-ID로 재연결하고 재생된 접두부의 중복을 제거하며, joinRun(runId)는 기존 실행에 연결합니다. NDJSON을 사용하지만 재개 가능한 SSE와 동일한 보장을 제공합니다.

React Native 및 Expo​

동일 출처 브라우저 라우트가 아닌 자체 백엔드를 호출해야 하는 네이티브 앱이 있다고 가정합니다. 명시적인 채팅 전송 방식과 절대 URL을 사용해 @tanstack/ai-react의 useChat을 사용합니다. 이 섹션을 마치면 React Native 또는 Expo에 맞게 클라이언트 어댑터와 서버 응답 헬퍼가 올바르게 연결됩니다.

const baseUrl =
  process.env.EXPO_PUBLIC_TANSTACK_AI_BASE_URL ??
  'http://127.0.0.1:8787'
const httpUrl = `${baseUrl}/chat/http`
const sseUrl = `${baseUrl}/chat/sse`

런타임에서 접근할 수 있는 URL을 사용합니다.

  • iOS 시뮬레이터: 대개 localhost 또는 127.0.0.1입니다.
  • Android 에뮬레이터: 호스트 머신에 접근할 때 일반적으로 10.0.2.2를 사용합니다.
  • 실제 기기: LAN 또는 터널링된 URL을 사용합니다.

Expo와 React Native에서는 xhrHttpStream()을 우선 사용합니다. toHttpResponse()와 함께 사용하며 증분 XHR 진행 이벤트를 통해 줄바꿈으로 구분된 JSON을 읽습니다.

import { useChat, xhrHttpStream } from "@tanstack/ai-react";

const baseUrl = process.env.EXPO_PUBLIC_TANSTACK_AI_BASE_URL ?? 'http://127.0.0.1:8787';
const httpUrl = `${baseUrl}/chat/http`;

const chat = useChat({
  connection: xhrHttpStream(httpUrl),
});

모바일 연결은 자주 끊기므로 이 경우 재개 기능의 효과가 가장 큽니다. 서버가 내구성 어댑터를 추가하면 두 XHR 어댑터 모두 재연결하고 joinRun을 수행합니다. 재개 가능한 스트림을 참조하세요.

서버가 toServerSentEventsResponse()를 통해 text/event-stream을 반환할 때는 xhrServerSentEvents()를 사용합니다.

import { useChat, xhrServerSentEvents } from "@tanstack/ai-react";

const baseUrl = process.env.EXPO_PUBLIC_TANSTACK_AI_BASE_URL ?? 'http://127.0.0.1:8787';
const sseUrl = `${baseUrl}/chat/sse`;

const chat = useChat({
  connection: xhrServerSentEvents(sseUrl),
});

정확히 사용하는 React Native 런타임이 스트리밍 fetch 응답, Response.body.getReader(), TextDecoder를 노출하는 경우에만 fetchHttpStream()을 사용합니다. 서버는 여전히 toHttpResponse()로 줄바꿈으로 구분된 JSON을 반환합니다.

import { useChat, fetchHttpStream } from "@tanstack/ai-react";

const baseUrl = process.env.EXPO_PUBLIC_TANSTACK_AI_BASE_URL ?? 'http://127.0.0.1:8787';
const httpUrl = `${baseUrl}/chat/http`;

const chat = useChat({
  connection: fetchHttpStream(httpUrl),
});

이러한 fetch 스트리밍 API 중 하나라도 없으면 fetchHttpStream()이 UnsupportedResponseStreamError를 발생시킵니다. 응답을 버퍼링하는 폴리필은 fetch 스트리밍을 호환되게 만들지 않습니다. 어댑터에는 증분 바이트가 필요합니다. 대신 xhrHttpStream() 또는 xhrServerSentEvents()로 전환합니다.

provider SDK와 서버 헬퍼는 백엔드에 둡니다. React Native 번들은 OpenAI/Anthropic/Gemini SDK, React DOM UI, devtools UI 또는 다른 프레임워크 패키지가 아니라 훅과 연결 어댑터를 가져와야 합니다. 전체 모바일 안내는 빠른 시작: React Native를 참조하세요.

서버 함수 및 직접 AsyncIterable​

클라이언트가 HTTP를 거치지 않고 서버를 호출할 수 있다면(RSC 스트림, 프로세스 내 테스트, 직접적인 프로세스 내 chat() 호출) 전송 방식을 완전히 건너뜁니다. stream()은 동기적으로 AsyncIterable<StreamChunk>을 반환하는 팩토리를 받아 클라이언트에 직접 연결합니다. (TanStack Start 서버 함수는 Promise를 반환하므로 stream()이 아니라 fetcher가 필요합니다. 다음 섹션을 참조하세요.)

import { useChat, stream } from "@tanstack/ai-react";
import { chatServerFn } from "./server/chat.server";

// `chatServerFn` is an in-process server-side function that synchronously
// returns an AsyncIterable<StreamChunk> — e.g. the result of
// `chat({ adapter, model, messages })` on the server.
const { messages } = useChat({
  connection: stream((messages, data) => chatServerFn({ messages, ...data })),
});

팩토리는 대화 메시지와 sendMessage에 전달한 요청별 data를 받습니다. StreamChunk 객체를 생성하는 비동기 이터러블이면 무엇이든 반환할 수 있습니다. 생성기, 서버의 chat() 출력, 변환된 스트림 등이 해당합니다.

팁: stream()은 요청 범위입니다. 팩토리는 sendMessage마다 한 번 호출되고, 이터러블이 완료될 때까지 실행된 후 연결이 닫힙니다. 여러 전송을 다중화하는 하나의 장기 실행 채널(예: WebSocket)이 필요하면 대신 subscribe / send를 사용합니다.

stream()은 영속성 핸들러로 구성된 선택적 두 번째 인수도 받아 어댑터에 펼칩니다. 따라서 HTTP 엔드포인트 없이도 서버 주도 영속성(persistence: true)이 작동합니다. 각 핸들러는 일반적으로 서버를 호출하는 한 줄입니다.

  • hydrate: 채팅 스레드를 복원합니다.
  • hydrateGeneration: 생성의 마지막 실행을 복원합니다.
  • joinRun: 아직 진행 중인 실행을 재생합니다.

생성 영속성에서 전체 서버 함수 연결 방법을 확인하세요.

fetcher를 통한 서버 함수​

비동기 함수로 서버를 호출할 때(TanStack Start 서버 함수처럼 항상 Promise를 반환하는 경우) 연결 어댑터 대신 최상위 fetcher 옵션을 사용합니다. fetcher는 connection과 나란한 옵션이며(정확히 하나만 제공), 일반 비동기 함수를 받습니다. 생성 훅의 fetcher 옵션과 같은 방식입니다. 가장 일반적인 형태는 toServerSentEventsResponse(...)로 끝나 Response로 해석되는 핸들러입니다.

// server/chat.server.ts
import { createServerFn } from "@tanstack/react-start";
import { chat, toServerSentEventsResponse } from "@tanstack/ai";
import { openaiText } from "@tanstack/ai-openai";
import type { UIMessage } from "@tanstack/ai";

export const chatFn = createServerFn({ method: "POST" })
  .inputValidator((data: { messages: Array<UIMessage> }) => data)
  .handler(({ data }) =>
    toServerSentEventsResponse(
      chat({ adapter: openaiText("gpt-5.5"), messages: data.messages }),
    ),
  );
import { useChat } from "@tanstack/ai-react";
import { chatFn } from "./server/chat.server";

const { messages, sendMessage } = useChat({
  fetcher: ({ messages }, { signal }) => chatFn({ data: { messages }, signal }),
});

fetcher는 { messages, data, threadId, runId }와 AbortSignal을 받습니다(stop() 호출 또는 전송이 다른 전송으로 대체될 때 트리거됨). 다음 중 하나를 반환합니다.

  • Response: 채팅 클라이언트가 SSE 본문을 대신 파싱합니다.
  • AsyncIterable<StreamChunk>: 직접 생성됩니다. 스트림 자체를 반환하고 Response로 감싸지 않는 서버 함수를 지원합니다.

동기 반환과 Promise로 래핑된 반환을 모두 허용합니다.

팁: 생성 훅(useGenerateImage 및 관련 훅)은 동일한 서버 함수 형태를 한 단계 확장합니다. fetcher와 함께 hydrateGeneration 및 joinRun 옵션을 허용하므로 HTTP 라우트 없이 서버 함수를 통해 persistence: true 상태를 복원하고 다시 연결합니다. 생성 영속성 — 서버 함수 / 직접 호출을 참조하세요.

팁: fetcher와 stream() 중 선택은 Response와 이터러블의 차이가 아니라 비동기와 동기의 차이입니다. 둘 다 AsyncIterable<StreamChunk>을 생성할 수 있습니다. stream()의 팩토리는 해당 이터러블을 동기적으로 반환해야 하므로 Promise를 반환하는 서버 함수 호출은 여기서 타입 검사를 통과하지 못합니다. 이 간극을 fetcher가 해결합니다(이슈 #509). 비동기 이터러블을 동기적으로 반환할 수 있으면 stream()을 사용하고(프로세스 내 chat(), RPC 클라이언트, 테스트), await해야 하는 항목에는 fetcher를 사용합니다. 둘 다 동일한 요청 범위 어댑터로 정규화되므로 stop()/abort, 오류 처리, 도구 호출은 동일하게 동작합니다.

RPC 스트림​

rpcStream()은 동작 면에서 stream()과 동일하지만 RPC 클라이언트에 넘기는 호출 위치에서 의미가 더 분명합니다. Cap'n Web, gRPC-Web, tRPC 구독 또는 이미 비동기 이터러블을 반환하는 RPC 프레임워크와 통합할 때 사용합니다.

import { useChat, rpcStream } from "@tanstack/ai-react";
import { api } from "./rpc-client";

// `api.chat.stream` is your RPC method; it must return an AsyncIterable<StreamChunk>.
const { messages } = useChat({
  connection: rpcStream((messages, data) =>
    api.chat.stream({ messages, ...data }),
  ),
});

stream()과 마찬가지로 rpcStream()은 영속성 핸들러({ hydrate, hydrateGeneration, joinRun })로 구성된 선택적 두 번째 인수를 받아 RPC에서도 서버 주도 영속성이 작동하도록 합니다. 각 핸들러는 대개 한 줄의 RPC 호출입니다.

WebSocket​

영속적이고 재개 가능한 WebSocket에는 SubscribeConnectionAdapter를 직접 작성하는 대신 기본 제공 webSocket() 어댑터를 사용합니다. 전체 대화에 하나의 소켓을 열고, 끊긴 내구성 실행에 자동으로 재연결하며, 서버의 toWebSocketStream / toWebSocketResponse와 함께 사용합니다.

import { useChat, webSocket } from "@tanstack/ai-react";

const connection = webSocket("/api/chat-ws");

const { messages, sendMessage } = useChat({ connection });

Cloudflare Workers 또는 Durable Objects에서는 해당 클라이언트를 toWebSocketResponse와 함께 사용합니다. 그 외 환경에서는 소켓을 직접 수락하고 toWebSocketStream에 전달합니다(WebSockets 참조).

import { chat, memoryStream, toWebSocketResponse } from "@tanstack/ai";
import { openaiText } from "@tanstack/ai-openai";

export default {
  fetch(request: Request): Response {
    return toWebSocketResponse(request, {
      durability: (ctx) => memoryStream(ctx.request),
      onRun: ({ messages, threadId, runId }) =>
        chat({
          adapter: openaiText("gpt-5.5"),
          messages,
          threadId,
          runId,
        }),
    });
  },
};

WebSocket에서 와이어 프로토콜, 재연결 세부 정보, Node와 Cloudflare에서의 호스팅을 확인하세요.

영속 전송(WebSocket 및 유사 방식)​

영속 전송(WebSocket, BroadcastChannel, iframe 간 postMessage, 공유 워커)은 요청/응답과 근본적으로 다릅니다. 채널을 한 번 연 다음 클라이언트의 수명 동안 이를 통해 송수신합니다. stream()/connect()는 요청마다 하나의 비동기 이터러블을 가정하므로 이를 깔끔하게 모델링할 수 없습니다.

위의 기본 제공 webSocket() 어댑터는 일반적인 재개 가능한 WebSocket 사례를 다룹니다. 그 밖의 영속 전송에는 SubscribeConnectionAdapter 인터페이스를 직접 구현합니다. 형태는 다음과 같습니다(전체 정의는 어댑터 인터페이스 참조).

import type { SubscribeConnectionAdapter } from "@tanstack/ai-react";

// subscribe(abortSignal?): AsyncIterable<StreamChunk>   — long-lived
// send(messages, data?, abortSignal?, runContext?): Promise<void> — one per user message
  • subscribe()는 ChatClient가 한 번 호출하며 채널이 생성하는 모든 청크의 장기 실행 비동기 이터러블을 반환합니다.
  • send()는 사용자 메시지마다 한 번 호출해 채널에 요청 프레임을 전달합니다. 프레임이 기록되면 반환되고, 청크는 subscribe()를 통해 별도로 도착합니다.

런타임은 이를 서로 연결합니다. send()와 다음 종료 이벤트(RUN_FINISHED / RUN_ERROR) 사이에 구독 큐에서 생성된 청크는 해당 실행에 귀속됩니다.

사용자 지정 WebSocket 예시​

기본 제공 webSocket() 어댑터 대신 자체 프로토콜을 구축하려는 경우(다른 와이어 형식, 재개 지원 불필요 또는 제어할 수 없는 서버) SubscribeConnectionAdapter를 직접 구현합니다.

import { useChat, type SubscribeConnectionAdapter } from "@tanstack/ai-react";
import type { StreamChunk } from "@tanstack/ai";

function websocketConnection(url: string): SubscribeConnectionAdapter {
  const ws = new WebSocket(url);
  const queue: Array<StreamChunk> = [];
  let pending: ((chunk: StreamChunk | null) => void) | null = null;
  let closed = false;

  const ready = new Promise<void>((resolve) => {
    ws.addEventListener("open", () => resolve(), { once: true });
  });

  function deliver(chunk: StreamChunk | null) {
    const resolve = pending;
    if (resolve) {
      pending = null;
      resolve(chunk);
    } else if (chunk !== null) {
      queue.push(chunk);
    }
  }

  ws.addEventListener("message", (event) => {
    const chunk: StreamChunk = JSON.parse(event.data);
    deliver(chunk);
  });
  ws.addEventListener("close", () => {
    closed = true;
    deliver(null);
  });

  return {
    async *subscribe(abortSignal) {
      // Register the abort listener once (not per-iteration) so it can't
      // accumulate on a long-lived socket.
      const onAbort = () => deliver(null);
      abortSignal?.addEventListener("abort", onAbort, { once: true });
      try {
        while (!abortSignal?.aborted) {
          // Drain buffered chunks BEFORE honoring `closed`: a burst of messages
          // followed by a close event (common within one macrotask) must still
          // deliver the queued chunks (including a trailing RUN_FINISHED),
          // otherwise the client would hang waiting for a terminal it dropped.
          const buffered = queue.shift();
          if (buffered !== undefined) {
            yield buffered;
            continue;
          }
          if (closed) return;
          const chunk = await new Promise<StreamChunk | null>((resolve) => {
            pending = resolve;
          });
          if (chunk === null) return;
          yield chunk;
        }
      } finally {
        abortSignal?.removeEventListener("abort", onAbort);
      }
    },

    async send(messages, data, _abortSignal, runContext) {
      await ready;
      ws.send(
        JSON.stringify({
          threadId: runContext?.threadId,
          runId: runContext?.runId,
          messages,
          data,
        }),
      );
    },
  };
}

const { messages } = useChat({
  connection: websocketConnection("wss://example.com/chat"),
});

팁: 각 실행이 끝날 때 RUN_FINISHED(또는 RUN_ERROR)를 생성하는 책임은 서버에 있습니다. 이 이벤트가 없으면 클라이언트는 어시스턴트 차례가 끝났는지 알 수 없어 무기한 대기합니다. 전체 이벤트 수명 주기는 스트리밍을 참조하세요.

요청 범위 방식 대신 영속 방식을 선택할 때​

다음 중 하나라도 해당하면 subscribe / send를 선택합니다.

  • 하나의 연결이 여러 실행을 다중화합니다(채팅 스레드가 메시지 사이에도 소켓을 열어 둠).
  • 서버가 요청 외부에서 청크를 푸시합니다(상태 업데이트, 서버가 시작한 도구 호출, 브로드캐스트 알림).
  • 여러 탭(BroadcastChannel) 또는 워커에서 하나의 연결을 공유하려고 합니다.

그 외에는 fetchServerSentEvents 또는 stream()을 우선 사용합니다. 더 간단하고 연결 수명 주기 관리가 필요하지 않습니다.

사용자 지정 Fetch 클라이언트​

SSE 또는 HTTP 스트리밍을 유지하면서 인증 갱신, 재시도, 로깅 또는 엣지 프록시 라우팅을 위해 fetch를 래핑해야 한다면 fetchClient를 전달합니다.

import { useChat, fetchServerSentEvents } from "@tanstack/ai-react";
import { refreshToken } from "./auth";

async function authedFetch(input: RequestInfo | URL, init?: RequestInit) {
  let response = await fetch(input, init);
  if (response.status === 401) {
    await refreshToken();
    response = await fetch(input, init);
  }
  return response;
}

const { messages } = useChat({
  connection: fetchServerSentEvents("/api/chat", {
    fetchClient: authedFetch,
  }),
});

fetchClient는 표준 fetch 시그니처를 충족해야 합니다. fetchHttpStream도 동일한 옵션을 허용합니다.

사용자 지정 요청 범위 어댑터​

기본 제공 어댑터가 맞지 않지만 전송이 여전히 요청 범위(사용자 메시지마다 하나의 요청)라면 ConnectConnectionAdapter를 직접 구현합니다. 이는 영속 방식으로 전환하는 것 다음으로 가장 낮은 수준의 확장 지점입니다.

import { useChat, type ConnectConnectionAdapter } from "@tanstack/ai-react";
import type { StreamChunk } from "@tanstack/ai";

const myAdapter: ConnectConnectionAdapter = {
  async *connect(messages, data, abortSignal, runContext) {
    const response = await fetch("/api/chat", {
      method: "POST",
      headers: {
        "Content-Type": "application/json",
        ...runContext?.headers,
      },
      body: JSON.stringify({
        threadId: runContext?.threadId,
        runId: runContext?.runId,
        messages,
        ...data,
      }),
      ...(abortSignal ? { signal: abortSignal } : {}),
    });

    if (!response.ok) throw new Error(`HTTP ${response.status}`);
    if (!response.body) throw new Error("Response has no body");

    // Example: newline-delimited JSON. Replace this loop with whatever
    // framing your wire format uses, yielding one `StreamChunk` per event.
    const reader = response.body.getReader();
    const decoder = new TextDecoder();
    let buffer = "";
    while (true) {
      const { done, value } = await reader.read();
      if (done) break;
      buffer += decoder.decode(value, { stream: true });
      const lines = buffer.split("\n");
      buffer = lines.pop() ?? "";
      for (const line of lines) {
        if (line.trim()) {
          const chunk: StreamChunk = JSON.parse(line);
          yield chunk;
        }
      }
    }
  },
};

const { messages } = useChat({ connection: myAdapter });

runContext에는 threadId, runId, clientTools, forwardedProps, headers가 포함됩니다. 서버가 AG-UI 호환 응답을 만들 수 있도록 앞의 네 항목을 JSON 페이로드에 포함합니다.

runContext.headers는 본문이 아니라 POST 요청 헤더에 복사합니다. 기본 제공 fetch 및 XHR 어댑터가 이렇게 처리합니다. stream()과 rpcStream()은 runContext를 볼 수 없으므로 이렇게 처리하지 않습니다. headers를 건너뛰는 사용자 지정 connect는 BYOK 키(x-byok-*)를 누락합니다. 자체 키 사용을 참조하세요.

어느 경우든 런타임이 종료 이벤트를 처리합니다.

  • connect 스트림이 RUN_FINISHED를 생성하지 않고 완료되면 런타임이 이를 대신 생성합니다.
  • connect 스트림이 예외를 발생시키면 RUN_ERROR가 생성됩니다.

어댑터 인터페이스​

ConnectionAdapter는 유니온입니다. connect를 제공하거나 subscribe와 send를 모두 제공해야 합니다. 두 모드를 함께 제공하면 안 됩니다.

import type { UIMessage } from "@tanstack/ai-client";
import type { ModelMessage, StreamChunk } from "@tanstack/ai";

export interface RunAgentInputContext {
  threadId: string;
  runId: string;
  parentRunId?: string;
  clientTools?: Array<{ name: string; description: string; parameters: unknown }>;
  forwardedProps?: Record<string, unknown>;
  headers?: Record<string, string>;
}

export interface ConnectConnectionAdapter {
  connect(
    messages: UIMessage[] | ModelMessage[],
    data?: Record<string, any>,
    abortSignal?: AbortSignal,
    runContext?: RunAgentInputContext,
  ): AsyncIterable<StreamChunk>;
}

export interface SubscribeConnectionAdapter {
  subscribe(abortSignal?: AbortSignal): AsyncIterable<StreamChunk>;
  send(
    messages: UIMessage[] | ModelMessage[],
    data?: Record<string, any>,
    abortSignal?: AbortSignal,
    runContext?: RunAgentInputContext,
  ): Promise<void>;
}

export type ConnectionAdapter =
  | ConnectConnectionAdapter
  | SubscribeConnectionAdapter;

내부적으로 ChatClient는 normalizeConnectionAdapter()를 통해 두 형태를 하나의 subscribe/send 쌍으로 정규화합니다.

  • connect를 제공하면 비동기 큐로 래핑됩니다.
  • subscribe + send를 기본 형태로 제공하면 그대로 사용됩니다.

인증​

정적 헤더는 options.headers에 넣습니다.

import { useChat, fetchServerSentEvents } from "@tanstack/ai-react";
import { token } from "./auth";

const { messages } = useChat({
  connection: fetchServerSentEvents("/api/chat", {
    headers: { Authorization: `Bearer ${token}` },
  }),
});

요청마다 변경되는 토큰(갱신 토큰, 단기 JWT)의 경우 함수를 전달합니다. 모든 전송마다 호출되므로 헤더에는 항상 최신 토큰이 반영됩니다.

import { useChat, fetchServerSentEvents } from "@tanstack/ai-react";
import { getToken } from "./auth";

const { messages } = useChat({
  connection: fetchServerSentEvents("/api/chat", () => ({
    headers: { Authorization: `Bearer ${getToken()}` },
  })),
});

credentials가 "same-origin"(기본값) 또는 "include"이면 쿠키가 자동으로 전송됩니다.

취소​

기본 제공 어댑터와 사용자 지정 어댑터를 포함한 모든 어댑터는 AbortSignal을 받습니다. 기본 제공 어댑터는 이를 fetch로 전달하고, 사용자 지정 어댑터는 직접 준수해야 합니다. useChat의 stop()은 신호를 트리거해 현재 실행을 중단합니다.

import { useChat, fetchServerSentEvents } from "@tanstack/ai-react";

const { stop } = useChat({ connection: fetchServerSentEvents("/api/chat") });
stop(); // aborts the active stream

SubscribeConnectionAdapter에서 subscribe()의 신호는 전체 구독을 종료하고(컴포넌트 마운트 해제), send()의 신호는 진행 중인 전송만 종료합니다.

오류 처리​

어댑터는 전송 오류(HTTP non-2xx, 파싱 실패, 끊긴 소켓)에서 예외를 발생시켜야 합니다. ChatClient는 예외를 포착하고 아직 생성된 청크가 없으면 RUN_ERROR 청크를 생성하며, onError / error 상태를 통해 오류를 노출합니다.

import { useChat, fetchServerSentEvents } from "@tanstack/ai-react";

const { error } = useChat({
  connection: fetchServerSentEvents("/api/chat"),
  onError: (err) => console.error("Chat failed:", err),
});

AbortError를 삼키지 말고 전파해 클라이언트가 중단에 성공했음을 알 수 있게 합니다.

모범 사례​

  • SSE를 기본으로 사용합니다. 호환성이 가장 높고 디버깅하기 쉽습니다. 무언가가 차단할 때만 전환합니다.
  • 가능하면 stream()을 사용합니다. 양쪽을 모두 제어하고 HTTP 의미 체계가 필요 없다면 사용자 지정 어댑터를 만드는 것보다 서버 함수를 더 빠르게 연결할 수 있습니다.
  • 영속성이 필요할 때만 subscribe/send를 선택합니다. WebSocket은 강력하지만 재연결, 실행 연결, 수명 주기를 직접 처리해야 합니다.
  • 항상 abortSignal을 준수합니다. 마운트 해제 및 stop() 시 클라이언트가 정리하는 방식입니다.
  • 서버에서 RUN_FINISHED를 생성합니다. 이 이벤트가 없으면 클라이언트는 차례가 끝났는지 알 수 없습니다.

다음 단계​