fix(sdk-py): follow the SSE spec for multi-line data, empty ids and a leading BOM

- Repeated data fields are joined with a newline. They were concatenated, so
  a value split across lines could silently decode to a different one (a
  number 1 followed by a line 0 became 10).
- An empty id field resets the reconnect cursor: a reconnect after it sends
  no Last-Event-ID instead of the previous id.
- A UTF-8 BOM at the start of a stream is stripped, so the first event keeps
  its type.

Ideas from #7936, #7947, #9014 and #9236.

libs/sdk-py/langgraph_sdk/_async/http.py

@@ -217,7 +217,7 @@ class HttpClient:
             current_headers = dict(
                 request_headers if reconnect_path is None else reconnect_headers
             )
-            if last_event_id is not None:
+            if last_event_id:
                 current_headers["Last-Event-ID"] = last_event_id
 
             current_method = method if reconnect_path is None else "GET"

libs/sdk-py/langgraph_sdk/_sync/http.py

@@ -219,7 +219,7 @@ class SyncHttpClient:
             current_headers = dict(
                 request_headers if reconnect_path is None else reconnect_headers
             )
-            if last_event_id is not None:
+            if last_event_id:
                 current_headers["Last-Event-ID"] = last_event_id
 
             current_method = method if reconnect_path is None else "GET"

libs/sdk-py/langgraph_sdk/sse.py

@@ -79,18 +79,27 @@ class SSEDecoder:
     def __init__(self) -> None:
         self._event = ""
         self._data = bytearray()
-        self._last_event_id = ""
+        self._last_event_id: str | None = None
         self._retry: int | None = None
+        self._at_stream_start = True
 
     @property
     def last_event_id(self) -> str | None:
-        """Return the last event identifier that was seen."""
+        """Return the last event identifier that was seen.
 
-        return self._last_event_id or None
+        `None` until the stream sends an `id` field, and `""` after an empty
+        one, which resets the reconnect cursor.
+        """
+
+        return self._last_event_id
 
     def decode(self, line: bytes) -> StreamPart | None:
         # See: https://html.spec.whatwg.org/multipage/server-sent-events.html#event-stream-interpretation
 
+        if self._at_stream_start:
+            self._at_stream_start = False
+            line = line.removeprefix(b"\xef\xbb\xbf")
+
         if not line:
             if (
                 not self._event
@@ -103,7 +112,7 @@ class SSEDecoder:
             sse = StreamPart(
                 event=self._event,
                 data=orjson.loads(self._data) if self._data else None,  # ty: ignore[invalid-argument-type]
-                id=self.last_event_id,
+                id=self._last_event_id or None,
             )
 
             # NOTE: as per the SSE spec, do not reset last_event_id.
@@ -124,6 +133,8 @@ class SSEDecoder:
         if fieldname == b"event":
             self._event = value.decode()
         elif fieldname == b"data":
+            if self._data:
+                self._data.extend(b"\n")
             self._data.extend(value)
         elif fieldname == b"id":
             if b"\0" in value:

libs/sdk-py/tests/test_client_stream.py

@@ -5,6 +5,7 @@ from pathlib import Path
 from typing import Any
 
 import httpx
+import orjson
 import pytest
 from typing_extensions import assert_type
 
@@ -110,6 +111,25 @@ def test_stream_sse():
         assert len(parts) == 79
 
 
+def test_sse_decoder_joins_data_lines_with_a_newline():
+    decoder = SSEDecoder()
+    for line in (b"data: [1,", b"data: 2]"):
+        decoder.decode(line)
+    assert decoder.decode(b"") == StreamPart(event="", data=[1, 2], id=None)  # ty: ignore[invalid-argument-type]
+
+    for line in (b"data: [1", b"data: 0]"):
+        decoder.decode(line)
+    with pytest.raises(orjson.JSONDecodeError):
+        decoder.decode(b"")
+
+
+def test_sse_decoder_strips_a_leading_bom():
+    decoder = SSEDecoder()
+    for line in (b"\xef\xbb\xbfevent: values", b"data: 1"):
+        decoder.decode(line)
+    assert decoder.decode(b"") == StreamPart(event="values", data=1, id=None)  # ty: ignore[invalid-argument-type]
+
+
 # --- HTTP client streaming ---
 
 
@@ -292,6 +312,52 @@ async def test_http_client_stream_recovers_after_disconnect():
     ]
 
 
+def _drop_after_an_empty_id(stream_cls: type, reconnects: list[httpx.Request]):
+    def handler(request: httpx.Request) -> httpx.Response:
+        if request.method == "POST":
+            return httpx.Response(
+                200,
+                headers={"Content-Type": "text/event-stream", "Location": "/reconnect"},
+                stream=stream_cls(
+                    [
+                        b"id: 1\nevent: values\ndata: 1\n\n",
+                        b"id:\nevent: values\ndata: 2\n\n",
+                    ],
+                    httpx.RemoteProtocolError("incomplete chunked read"),
+                ),
+            )
+        reconnects.append(request)
+        return httpx.Response(
+            200,
+            headers={"Content-Type": "text/event-stream"},
+            stream=stream_cls([b"event: end\ndata: null\n\n"]),
+        )
+
+    return handler
+
+
+def test_sync_stream_reconnects_without_a_cursor_after_an_empty_id():
+    reconnects: list[httpx.Request] = []
+    transport = httpx.MockTransport(_drop_after_an_empty_id(ListByteStream, reconnects))
+    with httpx.Client(transport=transport, base_url="https://example.com") as client:
+        list(SyncHttpClient(client).stream("/stream", "POST", json={}))
+
+    assert "last-event-id" not in reconnects[0].headers
+
+
+async def test_async_stream_reconnects_without_a_cursor_after_an_empty_id():
+    reconnects: list[httpx.Request] = []
+    transport = httpx.MockTransport(
+        _drop_after_an_empty_id(AsyncListByteStream, reconnects)
+    )
+    async with httpx.AsyncClient(
+        transport=transport, base_url="https://example.com"
+    ) as client:
+        [part async for part in HttpClient(client).stream("/stream", "POST", json={})]
+
+    assert "last-event-id" not in reconnects[0].headers
+
+
 # --- _sse_to_v2_dict conversion ---