diff --git a/libs/sdk-py/langgraph_sdk/stream/transport/http.py b/libs/sdk-py/langgraph_sdk/stream/transport/http.py
index dfebf1ba4..501bc4125 100644
--- a/libs/sdk-py/langgraph_sdk/stream/transport/http.py
+++ b/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
diff --git a/libs/sdk-py/langgraph_sdk/stream/transport/sync_http.py b/libs/sdk-py/langgraph_sdk/stream/transport/sync_http.py
index 3ec4cb456..c60abe8f5 100644
--- a/libs/sdk-py/langgraph_sdk/stream/transport/sync_http.py
+++ b/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] = []
diff --git a/libs/sdk-py/langgraph_sdk/stream/transport/sync_ws.py b/libs/sdk-py/langgraph_sdk/stream/transport/sync_ws.py
index ec1289420..1db8ccd1c 100644
--- a/libs/sdk-py/langgraph_sdk/stream/transport/sync_ws.py
+++ b/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
diff --git a/libs/sdk-py/langgraph_sdk/stream/transport/ws.py b/libs/sdk-py/langgraph_sdk/stream/transport/ws.py
index 71bcf9411..af7d30ace 100644
--- a/libs/sdk-py/langgraph_sdk/stream/transport/ws.py
+++ b/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
diff --git a/libs/sdk-py/tests/streaming/test_transport_path_encoding.py b/libs/sdk-py/tests/streaming/test_transport_path_encoding.py
new file mode 100644
index 000000000..f1d7d838a
--- /dev/null
+++ b/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"