FetchStreamTransport
FetchStreamTransport is the production transport for LangGraph Platform. It wraps the LangGraph SDK client, creates threads when needed, starts streaming runs, joins queued runs, cancels runs, loads history, and updates thread state for operations such as regenerate rollback.
When you interact with it directly
In most apps you never instantiate FetchStreamTransport directly. When no custom transport is configured, the runtime creates one from the resolved apiUrl, onThreadId, and clientOptions.
You reach for it explicitly when you want to share a transport instance, inspect behavior in tests, or implement a custom transport by matching the same AgentTransport interface.
// app.config.ts
import { signal, type ApplicationConfig } from '@angular/core';
import { provideAgent, FetchStreamTransport } from '@threadplane/langgraph';
const apiUrl = 'http://localhost:2024';
const threadId = signal<string | null>(localStorage.getItem('threadId'));
const transport = new FetchStreamTransport(apiUrl, (id) => {
threadId.set(id);
localStorage.setItem('threadId', id);
});
export const appConfig: ApplicationConfig = {
providers: [
provideAgent({
apiUrl,
assistantId: 'support-agent',
threadId,
transport,
}),
],
};The runtime only wraps onThreadId for the transport it constructs itself. A
transport you pass in owns its own thread creation, so every id it creates must
reach the runtime through the configured onThreadId callback or through the
threadId signal — as above, where the constructor callback writes both. Report
neither and the runtime keeps sending null, so every submit() starts a fresh
thread.
How it works
FetchStreamTransport creates a LangGraph SDK Client for the configured API URL. On stream(), it creates a thread when threadId is missing, calls client.runs.stream(...), normalizes SDK events into StreamEvent objects, and yields them to the stream manager.
The transport handles:
- Thread creation — calls
onThreadIdwhen the SDK creates a new thread. - Streaming modes — defaults to
values,messages-tuple,updates, andcustom. - Cancellation — forwards
AbortSignalto LangGraph run APIs. - Queued runs — supports create, cancel, and join flows when the backend supports them.
- History and state update — loads checkpoint history, reads a subagent's latest checkpoint under its child namespace (
getChildState()) when a thread is reloaded, and callsthreads.updateState()for regenerate rollback. - Error propagation — SDK and network errors surface through
injectAgent().error().
FetchStreamTransport implements AgentTransport. Custom transports must
implement stream(assistantId, threadId, payload, signal, options?); optional
methods such as joinStream, createQueuedRun, cancelRun, getHistory,
getChildState, and updateState enable advanced LangGraph features.