# Streaming

> How invoke, stream, and astream work; StreamChunk fields; ResponseGranularity; and how to consume SSE in TypeScript.

Source: https://10xgraph.com/docs/concepts/streaming
Last updated: 2026-07-21

10xGraph gives you three execution modes — **invoke** (sync, wait for finish), **stream** (sync generator), and **astream** (async generator). Streaming is essential for chat UIs where users expect to see words appear as the model produces them.

---

## How it works

```mermaid
sequenceDiagram
  participant Client
  participant Graph

  Note over Client,Graph: invoke — wait for the full response
  Client->>Graph: invoke(input, config)
  Graph->>Graph: run all nodes
  Graph-->>Client: final result dict

  Note over Client,Graph: stream / astream — incremental chunks
  Client->>Graph: stream(input, config)
  Graph-->>Client: StreamChunk (message)
  Graph-->>Client: StreamChunk (message)
  Graph->>Graph: run remaining nodes
  Graph-->>Client: StreamChunk (state)
```

Use **invoke** when you need the full result before proceeding. Use **stream** / **astream** when the client should see partial responses immediately.

---

## StreamChunk

Every streaming event is a `StreamChunk` Pydantic model:

```python
from tenxgraph.core.state.stream_chunks import StreamChunk, StreamEvent

class StreamChunk(BaseModel):
    event: StreamEvent          # "message" | "state" | "error" | "updates"
    message: Message | None     # populated for StreamEvent.MESSAGE
    state: AgentState | None    # populated for StreamEvent.STATE
    data: dict | None           # populated for StreamEvent.ERROR / UPDATES
    thread_id: str | None
    run_id: str | None
    metadata: dict | None
    timestamp: float            # UNIX timestamp
```

### StreamEvent values

| `StreamEvent` | Value | When sent | Populated field |
|---|---|---|---|
| `StreamEvent.MESSAGE` | `"message"` | Each model output message | `chunk.message` |
| `StreamEvent.STATE` | `"state"` | After each node completes | `chunk.state` |
| `StreamEvent.ERROR` | `"error"` | Execution error | `chunk.data` |
| `StreamEvent.UPDATES` | `"updates"` | Custom node-level updates | `chunk.data` |

---

## ResponseGranularity

Both `stream()` and `astream()` accept a `response_granularity` parameter to control what is included in each `StreamChunk`:

```python
from tenxgraph.utils import ResponseGranularity
```

| Value | Description |
|---|---|
| `ResponseGranularity.LOW` | Only the latest messages (default) |
| `ResponseGranularity.PARTIAL` | Context, context_summary, and latest messages |
| `ResponseGranularity.FULL` | Complete state and messages |

---

## Synchronous streaming

`app.stream()` is a synchronous generator. Use it from non-async code:

```python
import asyncio
from tenxgraph.core.state import Message
from tenxgraph.utils import ResponseGranularity
from tenxgraph.core.state.stream_chunks import StreamEvent

for chunk in app.stream(
    {"messages": [Message.text_message("Tell me a short story.")]},
    config={"thread_id": "stream-1", "recursion_limit": 10},
    response_granularity=ResponseGranularity.LOW,
):
    if chunk.event == StreamEvent.MESSAGE and chunk.message is not None:
        print(chunk.message.text(), end="", flush=True)

print()  # trailing newline
```

---

## Asynchronous streaming

`app.astream()` is an async generator. Use it inside async code (e.g., FastAPI, async tests):

```python
import asyncio
from tenxgraph.core.state import Message
from tenxgraph.utils import ResponseGranularity
from tenxgraph.core.state.stream_chunks import StreamEvent

async def main():
    inp = {"messages": [Message.text_message("Call get_weather for Tokyo.")]}
    config = {"thread_id": "astream-1", "recursion_limit": 10}

    async for chunk in app.astream(inp, config, ResponseGranularity.LOW):
        if chunk.event == StreamEvent.MESSAGE and chunk.message is not None:
            print(chunk.message.text(), end="", flush=True)
        elif chunk.event == StreamEvent.STATE and chunk.state is not None:
            print(f"\n[state received, step={chunk.state.execution_meta.step}]")

asyncio.run(main())
```

### Inspecting all chunk types

```python
async for chunk in app.astream(inp, config):
    match chunk.event:
        case StreamEvent.MESSAGE:
            # new message from the model or a tool
            print("message:", chunk.message.text())
        case StreamEvent.STATE:
            # node completed — full or partial state
            print("state step:", chunk.state.execution_meta.step)
        case StreamEvent.ERROR:
            print("error:", chunk.data)
        case StreamEvent.UPDATES:
            print("updates:", chunk.data)
```

---

## invoke and ainvoke

For non-streaming use, `app.invoke()` and `app.ainvoke()` return a plain dict:

```python
# Sync
result = app.invoke(
    {"messages": [Message.text_message("Hello!")]},
    config={"thread_id": "t1"},
    response_granularity=ResponseGranularity.LOW,
)
messages = result["messages"]   # list of Message

# Async
result = await app.ainvoke(
    {"messages": [Message.text_message("Hello!")]},
    config={"thread_id": "t1"},
)
```

The returned dict keys depend on `response_granularity`:

| Granularity | Keys present |
|---|---|
| `LOW` | `messages` |
| `PARTIAL` | `messages`, `context`, `context_summary` |
| `FULL` | `messages`, `state` |

---

## Stopping a running stream

Call `app.stop()` / `app.astop()` to request cancellation. The graph checks a stop flag after each node and exits cleanly:

```python
# stop from another coroutine / thread
await app.astop({"thread_id": "stream-1"})
```

Via the REST API:

```bash
POST /v1/graph/stop
Content-Type: application/json

{"thread_id": "stream-1"}
```

---

## Streaming via the REST API

`POST /v1/graph/stream` streams one JSON-encoded `StreamChunk` per line.

The response carries a `text/event-stream` content type, but the body is **not** SSE-framed:
there are no `data:` prefixes and no blank-line separators. It is newline-delimited JSON
(NDJSON), so parse it by splitting on newlines rather than with an `EventSource` or an SSE
client library.

```bash
curl -N -X POST http://127.0.0.1:8000/v1/graph/stream \
  -H "Content-Type: application/json" \
  -d '{
    "messages": [{"role": "user", "content": "Tell me a short story."}],
    "config": {"thread_id": "rest-stream-1"}
  }'
```

Response:

```
{"event": "message", "message": {"role": "assistant", "content": [{"type": "text", "text": "Once"}]}, ...}
{"event": "message", "message": {"role": "assistant", "content": [{"type": "text", "text": " upon a time"}]}, ...}
{"event": "state", "state": {...}, ...}
```

---

## Streaming in TypeScript

`AgentFlowClient.stream` returns an async iterator of `StreamChunk`:

```typescript
import { AgentFlowClient, Message, StreamEventType } from "@10xscale/agentflow-client";

const client = new AgentFlowClient({ baseUrl: "http://127.0.0.1:8000" });

for await (const chunk of client.stream(
  [Message.text_message("Tell me a short story.")],
  { config: { thread_id: "ts-stream-1" } },
)) {
  if (chunk.event === StreamEventType.MESSAGE && chunk.message) {
    process.stdout.write(chunk.message.text());
  }
}
console.log();
```

---

## Related concepts

- [StateGraph and nodes](/docs/concepts/state-graph)
- [State and messages](/docs/concepts/state-and-messages)
- [Production runtime](/docs/concepts/production-runtime)