## Type-safe stream parts for v2 streaming
### Review recommendations
* Don't fear the diff! It's not so bad, I swear! The PR description
below gives a nice overview of changes
* Check out my explicit comments below, those should help to orient you
to the important changes :)
* On a first pass, ignore the test files! Just check out the new types,
overloads, and minor logical changes (diff behavior based on the flag)
## Summary
Adds a `stream_version="v2"` option that emits typed `{"type", "ns",
"data"}` dicts instead of raw tuples/SSE events. Each stream mode gets
its own `TypedDict` with a `Literal` type field, enabling full type
narrowing on `part["type"]`.
This is **opt in**, so it's **non-breaking**!!
```python
async for part in graph.astream(
inputs, stream_mode=["messages", "custom"], stream_version="v2"
):
if part["type"] == "messages":
msg, metadata = part["data"] # tuple[AnyMessage, dict] ✅
print(msg.content)
elif part["type"] == "custom":
part["data"] # Any ✅
```
Before v2 you'd get `tuple[str, Any]` with no way to narrow `data` based
on mode.
### What changed
**`langgraph` (core):** New `StreamPart` discriminated union + per-mode
TypedDicts in `types.py`. Stream-emit code in `pregel/` refactored to
use the new types. `RemoteGraph` gains a `stream_version` param.
**`sdk-py`:** Client-side v2 wrapper that converts SSE events into typed
dicts. No server API changes — v2 is purely a client-side rewrite of the
stream format.
### Stream part types
#### `langgraph` (core)
| `type` | `data` |
|---|---|
| `"values"` | `dict[str, Any]` — full state after each step |
| `"updates"` | `dict[str, Any]` — node name → output |
| `"messages"` | `tuple[AnyMessage, dict]` — message + metadata |
| `"custom"` | `Any` — whatever was passed to `StreamWriter` |
| `"tasks"` | `TaskPayload \| TaskResultPayload` |
| `"checkpoints"` | `CheckpointPayload` |
| `"debug"` | `DebugPayload` |
#### `sdk-py` (additional types from SSE events)
| `type` | `data` | Description |
|---|---|---|
| `"messages/partial"` | `list[dict]` | Partial message chunks |
| `"messages/complete"` | `list[dict]` | Complete messages |
| `"messages/metadata"` | `dict` | Message metadata |
| `"metadata"` | `RunMetadataPayload` | Run-level metadata (`run_id`,
etc.) |
All parts share the shape `{"type": Literal[...], "ns": list[str],
"data": ...}`.
## Release plan
* Release as a part of langgraph 1.1
* Before release, I'd like to do more experimentation with support for
pydantic + dataclasses and/or input/output runtime validation, as that
would help resolve the lack of typing for `values` mode.
### Notes
- Docs need a mass update to cover v2 streaming usage and the new types
- Changing the default `stream_version` to `"v2"` in a future release
would be breaking but backwards compatible (users can pin `"v1"` to keep
current behavior)
- I called out specifically relevant parts of the code in comments on
the PR :)
- **Description:** Provide the id of the event for routes that use SSE
streams. This will allow for more custom retry logic when streams
disconnect if needed.
- **Issue:** N/A
- **Dependencies:** N/A
- **Twitter handle:** N/A
---------
Co-authored-by: William FH <13333726+hinthornw@users.noreply.github.com>
## Summary
- add a public accessor for the last received SSE event id
- retry async and sync SSE streams using the Location reconnect path and
Last-Event-ID while skipping empty events
- add regression tests that simulate interrupted SSE streams for both
async and sync clients
## Testing
- make format
- make lint
- make test
------
https://chatgpt.com/codex/tasks/task_e_68ca8bfa26cc832d98bcb359884962ec
## Summary
- ensure both async and sync HTTP clients flush the SSE decoder after
streaming
- add regression tests covering trailing SSE events without a
terminating blank line
## Testing
- make format
- make lint
- make test
------
https://chatgpt.com/codex/tasks/task_e_68c9727ca9f8832d9f207323c5e02a72
