stream() in the TypeScript client

client.stream() sends a list of messages to the agent graph and returns an AsyncGenerator that yields StreamChunk objects as the server produces them. Use it to show incremental output (token by token, update by update) instead of waiting for the full response. For a single final result, use invoke().

Endpoint: POST /v1/graph/stream
Source: src/endpoints/stream.ts

TypeScript

import { TenxGraphClient, Message, StreamEventType } from '10xgraph-client';

Signature

TypeScript

client.stream(
  messages: Message[],
  options?: StreamOptions
): AsyncGenerator<StreamChunk, void, unknown>

stream() is synchronous: it returns the generator immediately without making a network call. The HTTP request starts when you begin iterating with for await.

Parameters

`messages`

Same as invoke(). An array of Message objects.

`options` (optional)

Field	Type	Default	Description
`initial_state`	`Record<string, any>`	`undefined`	Initial state values to seed the graph.
`config`	`Record<string, any>`	`undefined`	Run config. Set `thread_id` at the top level of this object for persistent state.
`recursion_limit`	`number`	`25`	Maximum recursion depth for the graph.
`response_granularity`	`'full' \| 'partial' \| 'low'`	`'low'`	How much data the server returns: `'full'` is state plus latest messages, `'partial'` is context, summary and latest messages, `'low'` is latest messages only. Streaming defaults to `'low'`, unlike `invoke()` which defaults to `'full'`.

Return value: `AsyncGenerator<StreamChunk>`

The generator yields StreamChunk objects. Iteration ends when the server closes the stream.

`StreamChunk`

TypeScript

interface StreamChunk {
  event: StreamEventType | string;
  message?: Message | null;
  state?: AgentState | null;
  data?: any;
  thread_id?: string;
  run_id?: string;
  metadata?: StreamMetadata | Record<string, any>;
  timestamp?: number;
}

interface StreamMetadata {
  is_new_thread: boolean;
  thread_id: string;
  [key: string]: any;
}

`StreamEventType`

The event field on a StreamChunk is a string matching one of these values (exported as the StreamEventType enum):

Event	Enum value	Description
`'message'`	`StreamEventType.MESSAGE`	A new or updated `Message` from the agent. The `message` field contains the full `Message` object. On streaming models this fires multiple times with partial token content (`message.delta = true`), followed by a final chunk with `delta = false`.
`'updates'`	`StreamEventType.UPDATES`	A graph execution update, such as a node starting or finishing. Inspect `chunk.data`, `chunk.state` and `chunk.metadata` for details.
`'state'`	`StreamEventType.STATE`	A state snapshot in the `state` field. Use `'full'` granularity if you need state in your responses.
`'error'`	`StreamEventType.ERROR`	An error occurred during graph execution. The `data` field contains the error details.

Async iterator examples

Basic streaming

TypeScript

import { TenxGraphClient, Message, StreamEventType } from '10xgraph-client';

const client = new TenxGraphClient({ baseUrl: 'http://localhost:8000' });

const stream = client.stream([
  Message.text_message('Write me a short poem about the ocean.'),
]);

for await (const chunk of stream) {
  if (chunk.event === StreamEventType.MESSAGE && chunk.message) {
    const textBlock = chunk.message.content.find(b => b.type === 'text');
    if (textBlock && 'text' in textBlock) {
      process.stdout.write(textBlock.text);
    }
  }
}
console.log('\n--- Stream complete ---');

Streaming with state updates

TypeScript

const stream = client.stream(
  [Message.text_message('Analyse this dataset.')],
  {
    config: { thread_id: 'analysis-001' },
    response_granularity: 'full',
  }
);

for await (const chunk of stream) {
  switch (chunk.event) {
    case StreamEventType.MESSAGE:
      if (chunk.message?.delta) {
        // Partial token: append to UI
        appendTokenToUI(chunk.message);
      } else if (chunk.message) {
        // Final message: replace partial content
        setFinalMessage(chunk.message);
      }
      break;

    case StreamEventType.UPDATES:
      console.log('State updated:', chunk.state);
      break;

    case StreamEventType.STATE:
      console.log('Full state snapshot:', chunk.state);
      break;

    case StreamEventType.ERROR:
      console.error('Stream error:', chunk.data);
      break;
  }
}

Streaming in a React component

TSX

import { useState } from 'react';
import { TenxGraphClient, Message, StreamEventType } from '10xgraph-client';

const client = new TenxGraphClient({ baseUrl: 'http://localhost:8000' });

function ChatBox() {
  const [output, setOutput] = useState('');

  async function sendMessage(text: string) {
    setOutput('');

    const stream = client.stream([Message.text_message(text)]);

    for await (const chunk of stream) {
      if (chunk.event === StreamEventType.MESSAGE && chunk.message?.delta) {
        const textBlock = chunk.message.content.find(b => b.type === 'text');
        if (textBlock && 'text' in textBlock) {
          setOutput(prev => prev + textBlock.text);
        }
      }
    }
  }

  return (
    <div>
      <button onClick={() => sendMessage('Hello!')}>Send</button>
      <pre>{output}</pre>
    </div>
  );
}

Collecting the full response from a stream

This helper reads only the final (delta: false) messages, so partial tokens are not counted twice:

TypeScript

async function streamToString(messages: Message[]): Promise<string> {
  const stream = client.stream(messages);
  let fullText = '';

  for await (const chunk of stream) {
    if (chunk.event === StreamEventType.MESSAGE && chunk.message && !chunk.message.delta) {
      for (const block of chunk.message.content) {
        if (block.type === 'text' && 'text' in block) {
          fullText += block.text;
        }
      }
    }
  }

  return fullText;
}

Stopping a stream mid-way

Breaking out of the loop stops reading locally, but the server keeps running the graph until you call stopGraph() with the thread ID:

TypeScript

const threadId = 'story-001';
const stream = client.stream(
  [Message.text_message('Tell me a very long story')],
  { config: { thread_id: threadId } }
);
let wordCount = 0;

for await (const chunk of stream) {
  if (chunk.event === StreamEventType.MESSAGE && chunk.message) {
    const textBlock = chunk.message.content.find(b => b.type === 'text');
    if (textBlock && 'text' in textBlock) {
      wordCount += textBlock.text.split(' ').length;
    }
  }

  if (wordCount > 500) {
    // Request a server-side stop, then stop iterating locally
    await client.stopGraph(threadId);
    break;
  }
}

`StreamRequest` (advanced)

The shape of the request body sent to POST /v1/graph/stream:

TypeScript

interface StreamRequest {
  messages: any[];
  initial_state?: Record<string, any>;
  config?: Record<string, any>;
  recursion_limit?: number;       // default 25
  response_granularity?: 'full' | 'partial' | 'low';
}

Stream format

The server sends the stream as NDJSON (newline-delimited JSON) over HTTP. Each line is a JSON object matching the StreamChunk shape. The client’s parser also handles concatenated JSON objects (no newlines) for compatibility with some proxy configurations.

The server sets the response header Content-Type: text/event-stream, but the body is NDJSON, not Server-Sent Events framing.

Example raw stream output:

plaintext

{"event":"message","message":{"role":"assistant","content":[{"type":"text","text":"The"}],"delta":true},"thread_id":"abc123"}
{"event":"message","message":{"role":"assistant","content":[{"type":"text","text":" capital"}],"delta":true},"thread_id":"abc123"}
{"event":"message","message":{"role":"assistant","content":[{"type":"text","text":" of France is Paris."}],"delta":false},"thread_id":"abc123"}
{"event":"updates","data":{"node":"MAIN"},"thread_id":"abc123"}

Using curl

Terminal

curl --no-buffer -X POST http://localhost:8000/v1/graph/stream \
  -H 'Content-Type: application/json' \
  -H 'Authorization: Bearer YOUR_TOKEN' \
  -d '{
    "messages": [{ "role": "user", "content": [{"type": "text", "text": "Hello"}], "message_id": "0" }],
    "response_granularity": "low"
  }'

invoke() vs stream() comparison

Aspect	`invoke()`	`stream()`
Network requests	One per loop iteration	One streaming connection per loop iteration
When result arrives	After full graph execution	Token by token in real time
Default `response_granularity`	`'full'`	`'low'`
Remote tool handling	Automatic loop	Automatic loop, chunks are still yielded as they arrive
Return type	`Promise<InvokeResult>`	`AsyncGenerator<StreamChunk>`
Best for	Background tasks, batch	Chat UIs, real-time displays

Common errors

Error	Cause	Fix
`TenxGraphError` status `401`	Missing or invalid auth.	Set `auth` in the client config. See auth.
`Request timeout after Nms`	The connection exceeded the client `timeout`.	Raise `timeout` in the client config or retry.
Stream stops mid-response	Server timeout or network issue.	Wrap the `for await` loop in try/catch and retry.
`event: 'error'` chunks	Graph execution error.	Read `chunk.data` for the error message and check server logs.

What you learned

stream() returns an AsyncGenerator: iterate it with for await.
StreamEventType.MESSAGE with delta: true is a partial token; delta: false is the final message.
response_granularity defaults to 'low' for streaming; use 'full' when you need state.
Remote tool calls are executed automatically between stream iterations, up to recursion_limit.
Use stopGraph() to cancel a running stream on the server.

Next step

See reference/client/threads to learn how to manage persistent conversation threads.

Signature

Parameters

messages

options (optional)

Return value: AsyncGenerator<StreamChunk>

StreamChunk

StreamEventType

Async iterator examples

Basic streaming

Streaming with state updates

Streaming in a React component

Collecting the full response from a stream

Stopping a stream mid-way

StreamRequest (advanced)

Stream format

Using curl

invoke() vs stream() comparison

Common errors

What you learned

Next step

`messages`

`options` (optional)

Return value: `AsyncGenerator<StreamChunk>`

`StreamChunk`

`StreamEventType`

`StreamRequest` (advanced)