fix(sdk-py): percent-encode thread_id in v3 stream transport default paths (#7954)

## Problem

Fixes #7953.

The v3 SSE and WebSocket stream transports build their default paths by
interpolating `thread_id` directly into the URL:

```python
self._commands_url = commands_path or f"/threads/{thread_id}/commands"
self._stream_url   = stream_path   or f"/threads/{thread_id}/stream/events"
```

This skips the `_quote_path_param` escaping that the rest of the SDK
adopted in #7893. A `thread_id` containing reserved characters or
dot-segments is then normalized by the HTTP/WebSocket stack before
transmission. For example, `thread_id = "../assistants/abc"`:

| | before |
|---|---|
| constructed | `/threads/../assistants/abc/commands` |
| wire path | `/assistants/abc/commands` |

So the value stops being one opaque identifier under
`/threads/{thread_id}/...` and silently hits a different resource.

## Fix

Reuse the existing `_quote_path_param` helper for the **default** paths
in all four transports (`http`, `sync_http`, `ws`, `sync_ws`). Explicit
`commands_path` / `stream_path` overrides are left untouched, so callers
that pass their own paths opt out of encoding as before.

`_quote_path_param("../assistants/abc")` → `..%2Fassistants%2Fabc`,
which the HTTP/WS stack no longer collapses.

## Tests

Adds `tests/streaming/test_transport_path_encoding.py` covering all four
transports:
- default `_commands_url` / `_stream_url` / `_stream_path` are
percent-encoded,
- the actual SSE wire path (async + sync) stays under `/threads/`,
- the built WebSocket URL (async + sync) stays under `/threads/`,
- explicit path overrides are left untouched.

`make format`, `make lint`, and `make test` all pass in `libs/sdk-py`
(491 passed).

libs/sdk-py/langgraph_sdk/stream/transport/http.py

@@ -20,6 +20,7 @@ import httpx
 import orjson
 from langchain_protocol import Event
 
+from langgraph_sdk._shared.utilities import _quote_path_param
 from langgraph_sdk.sse import BytesLineDecoder, SSEDecoder
 from langgraph_sdk.stream.transport.base import (
     EventStreamHandle,
@@ -49,8 +50,12 @@ class ProtocolSseTransport:
     ) -> None:
         self._client = client
         self.thread_id = thread_id
-        self._commands_url = commands_path or f"/threads/{thread_id}/commands"
-        self._stream_url = stream_path or f"/threads/{thread_id}/stream/events"
+        self._commands_url = (
+            commands_path or f"/threads/{_quote_path_param(thread_id)}/commands"
+        )
+        self._stream_url = (
+            stream_path or f"/threads/{_quote_path_param(thread_id)}/stream/events"
+        )
         self._default_headers: dict[str, str] = dict(headers or {})
         self._max_queue_size = max_queue_size
         self._closed = False

libs/sdk-py/langgraph_sdk/stream/transport/sync_http.py

@@ -10,6 +10,7 @@ import httpx
 import orjson
 from langchain_protocol import Event
 
+from langgraph_sdk._shared.utilities import _quote_path_param
 from langgraph_sdk.sse import BytesLineDecoder, SSEDecoder
 from langgraph_sdk.stream.transport.base import (
     SyncEventStreamHandle,
@@ -31,8 +32,12 @@ class SyncProtocolSseTransport:
     ) -> None:
         self._client = client
         self.thread_id = thread_id
-        self._commands_url = commands_path or f"/threads/{thread_id}/commands"
-        self._stream_url = stream_path or f"/threads/{thread_id}/stream/events"
+        self._commands_url = (
+            commands_path or f"/threads/{_quote_path_param(thread_id)}/commands"
+        )
+        self._stream_url = (
+            stream_path or f"/threads/{_quote_path_param(thread_id)}/stream/events"
+        )
         self._default_headers: dict[str, str] = dict(headers or {})
         self._closed = False
         self._open_responses: list[httpx.Response] = []

libs/sdk-py/langgraph_sdk/stream/transport/sync_ws.py

@@ -11,6 +11,7 @@ import orjson
 from langchain_protocol import Event
 from websockets.sync.client import connect as websocket_connect
 
+from langgraph_sdk._shared.utilities import _quote_path_param
 from langgraph_sdk.stream.transport.base import (
     SyncEventStreamHandle,
     build_event_stream_body,
@@ -36,8 +37,12 @@ class SyncProtocolWebSocketTransport:
     ) -> None:
         self._client = client
         self.thread_id = thread_id
-        self._commands_url = commands_path or f"/threads/{thread_id}/commands"
-        self._stream_path = stream_path or f"/threads/{thread_id}/stream/events"
+        self._commands_url = (
+            commands_path or f"/threads/{_quote_path_param(thread_id)}/commands"
+        )
+        self._stream_path = (
+            stream_path or f"/threads/{_quote_path_param(thread_id)}/stream/events"
+        )
         self._default_headers: dict[str, str] = dict(headers or {})
         self._connect = connect
         self._ping_interval = ping_interval

libs/sdk-py/langgraph_sdk/stream/transport/ws.py

@@ -13,6 +13,7 @@ from langchain_protocol import Event
 from websockets.asyncio.client import connect as websocket_connect
 from websockets.exceptions import ConnectionClosedError, ConnectionClosedOK
 
+from langgraph_sdk._shared.utilities import _quote_path_param
 from langgraph_sdk.stream.transport.base import (
     EventStreamHandle,
     build_event_stream_body,
@@ -39,8 +40,12 @@ class ProtocolWebSocketTransport:
     ) -> None:
         self._client = client
         self.thread_id = thread_id
-        self._commands_url = commands_path or f"/threads/{thread_id}/commands"
-        self._stream_path = stream_path or f"/threads/{thread_id}/stream/events"
+        self._commands_url = (
+            commands_path or f"/threads/{_quote_path_param(thread_id)}/commands"
+        )
+        self._stream_path = (
+            stream_path or f"/threads/{_quote_path_param(thread_id)}/stream/events"
+        )
         self._default_headers: dict[str, str] = dict(headers or {})
         self._connect = connect
         self._max_queue_size = max_queue_size

libs/sdk-py/tests/streaming/test_transport_path_encoding.py

@@ -0,0 +1,136 @@
+"""Regression tests for #7953: v3 stream transports must percent-encode
+`thread_id` in their default paths so a value containing reserved characters
+or dot-segments stays an opaque identifier under `/threads/{thread_id}/...`
+instead of being normalized into a different resource path by the HTTP stack.
+"""
+
+from __future__ import annotations
+
+import httpx
+import pytest
+
+from langgraph_sdk.stream.transport.base import build_websocket_url
+from langgraph_sdk.stream.transport.http import ProtocolSseTransport
+from langgraph_sdk.stream.transport.sync_http import SyncProtocolSseTransport
+from langgraph_sdk.stream.transport.sync_ws import SyncProtocolWebSocketTransport
+from langgraph_sdk.stream.transport.ws import ProtocolWebSocketTransport
+
+# A thread_id that escapes the /threads/ namespace if interpolated raw: an HTTP
+# client collapses `/threads/../assistants/abc/...` to `/assistants/abc/...`.
+TRAVERSAL_THREAD_ID = "../assistants/abc"
+ENCODED_COMMANDS_PATH = "/threads/..%2Fassistants%2Fabc/commands"
+ENCODED_STREAM_PATH = "/threads/..%2Fassistants%2Fabc/stream/events"
+
+
+@pytest.mark.anyio
+async def test_async_sse_default_paths_encode_thread_id():
+    transport = ProtocolSseTransport(
+        client=httpx.AsyncClient(), thread_id=TRAVERSAL_THREAD_ID
+    )
+    assert transport._commands_url == ENCODED_COMMANDS_PATH
+    assert transport._stream_url == ENCODED_STREAM_PATH
+
+
+def test_sync_sse_default_paths_encode_thread_id():
+    transport = SyncProtocolSseTransport(
+        client=httpx.Client(), thread_id=TRAVERSAL_THREAD_ID
+    )
+    assert transport._commands_url == ENCODED_COMMANDS_PATH
+    assert transport._stream_url == ENCODED_STREAM_PATH
+
+
+@pytest.mark.anyio
+async def test_async_ws_default_paths_encode_thread_id():
+    transport = ProtocolWebSocketTransport(
+        client=httpx.AsyncClient(), thread_id=TRAVERSAL_THREAD_ID
+    )
+    assert transport._commands_url == ENCODED_COMMANDS_PATH
+    assert transport._stream_path == ENCODED_STREAM_PATH
+
+
+def test_sync_ws_default_paths_encode_thread_id():
+    transport = SyncProtocolWebSocketTransport(
+        client=httpx.Client(), thread_id=TRAVERSAL_THREAD_ID
+    )
+    assert transport._commands_url == ENCODED_COMMANDS_PATH
+    assert transport._stream_path == ENCODED_STREAM_PATH
+
+
+async def test_async_sse_wire_path_stays_under_threads_namespace():
+    """The path that actually goes on the wire must not be normalized away."""
+    seen: list[str] = []
+
+    async def handler(request: httpx.Request) -> httpx.Response:
+        seen.append(request.url.raw_path.decode("ascii"))
+        return httpx.Response(202)
+
+    async with httpx.AsyncClient(
+        transport=httpx.MockTransport(handler),
+        base_url="https://example.com",
+        trust_env=False,
+    ) as client:
+        transport = ProtocolSseTransport(client=client, thread_id=TRAVERSAL_THREAD_ID)
+        await transport.send_command({"id": 1, "method": "noop", "params": {}})
+
+    assert seen[0] == ENCODED_COMMANDS_PATH
+
+
+def test_sync_sse_wire_path_stays_under_threads_namespace():
+    seen: list[str] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        seen.append(request.url.raw_path.decode("ascii"))
+        return httpx.Response(202)
+
+    with httpx.Client(
+        transport=httpx.MockTransport(handler),
+        base_url="https://example.com",
+        trust_env=False,
+    ) as client:
+        transport = SyncProtocolSseTransport(
+            client=client, thread_id=TRAVERSAL_THREAD_ID
+        )
+        transport.send_command({"id": 1, "method": "noop", "params": {}})
+
+    assert seen[0] == ENCODED_COMMANDS_PATH
+
+
+@pytest.mark.anyio
+async def test_async_ws_url_stays_under_threads_namespace():
+    transport = ProtocolWebSocketTransport(
+        client=httpx.AsyncClient(base_url="https://example.com/api"),
+        thread_id=TRAVERSAL_THREAD_ID,
+    )
+    url = build_websocket_url(transport._client.base_url, transport._stream_path)
+    assert url == "wss://example.com/api/threads/..%2Fassistants%2Fabc/stream/events"
+
+
+def test_sync_ws_url_stays_under_threads_namespace():
+    transport = SyncProtocolWebSocketTransport(
+        client=httpx.Client(base_url="https://example.com/api"),
+        thread_id=TRAVERSAL_THREAD_ID,
+    )
+    url = build_websocket_url(transport._client.base_url, transport._stream_path)
+    assert url == "wss://example.com/api/threads/..%2Fassistants%2Fabc/stream/events"
+
+
+@pytest.mark.anyio
+async def test_explicit_path_overrides_are_left_untouched():
+    """Callers passing explicit paths opt out of default encoding entirely."""
+    sse = ProtocolSseTransport(
+        client=httpx.AsyncClient(),
+        thread_id=TRAVERSAL_THREAD_ID,
+        commands_path="/custom/commands",
+        stream_path="/custom/events",
+    )
+    assert sse._commands_url == "/custom/commands"
+    assert sse._stream_url == "/custom/events"
+
+    ws = ProtocolWebSocketTransport(
+        client=httpx.AsyncClient(),
+        thread_id=TRAVERSAL_THREAD_ID,
+        commands_path="/custom/commands",
+        stream_path="/custom/events",
+    )
+    assert ws._commands_url == "/custom/commands"
+    assert ws._stream_path == "/custom/events"