fix(sdk-py): percent-encode thread_id and assistant_id in thread stream requests (#9213)

## Summary

Two REST helpers on the thread stream built their paths by interpolating
the raw identifier:

- `thread.agent.get_tree()` built `/assistants/{assistant_id}/graph`
- the state fetch behind `thread.output` built
`/threads/{thread_id}/state`

httpx collapses `..` segments and treats `#` as a fragment, so an
identifier containing those characters was rewritten into a different
path before the request was sent. Both now go through
`_quote_path_param` in the async and sync stream modules, matching the
other SDK endpoints and the stream transports.

## Tests

`tests/streaming/test_thread_stream_path_encoding.py` drives
`threads.stream(...)` followed by `thread.output` or
`thread.agent.get_tree()`, async and sync, with identifiers containing
`..`, `/`, and `#`. It asserts that the request path keeps each
identifier as a single encoded segment. All four tests fail without the
change.

`make format`, `make lint`, and `make test` pass in `libs/sdk-py`.

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

@@ -24,6 +24,7 @@ from langchain_core.language_models.chat_model_stream import AsyncChatModelStrea
 from langchain_protocol import Event, SubscribeParams
 
 from langgraph_sdk._async.http import HttpClient
+from langgraph_sdk._shared.utilities import _quote_path_param
 from langgraph_sdk.schema import LangSmithTracing, QueryParamTypes
 from langgraph_sdk.stream.controller import _SeenEventIds
 from langgraph_sdk.stream.decoders import (
@@ -151,7 +152,7 @@ class _AgentModule:
             query_params.update(dict(params))
         request_headers = {**self._owner._headers, **dict(headers or {})}
         return await self._owner._http.get(
-            f"/assistants/{self._owner.assistant_id}/graph",
+            f"/assistants/{_quote_path_param(self._owner.assistant_id)}/graph",
             params=query_params,
             headers=request_headers or None,
         )
@@ -1902,7 +1903,7 @@ class AsyncThreadStream:
     async def _fetch_state(self) -> dict[str, Any]:
         """Fetch the current thread state from the REST endpoint."""
         return await self._http.get(
-            f"/threads/{self.thread_id}/state",
+            f"/threads/{_quote_path_param(self.thread_id)}/state",
             headers=self._headers or None,
         )
 

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

@@ -22,6 +22,7 @@ from typing import Any, Literal, TypedDict, cast
 from langchain_core.language_models.chat_model_stream import ChatModelStream
 from langchain_protocol import Event, SubscribeParams
 
+from langgraph_sdk._shared.utilities import _quote_path_param
 from langgraph_sdk._sync.http import SyncHttpClient
 from langgraph_sdk.schema import LangSmithTracing, QueryParamTypes
 from langgraph_sdk.stream.decoders import (
@@ -194,7 +195,7 @@ class _SyncAgentModule:
             query_params.update(dict(params))
         request_headers = {**self._owner._headers, **dict(headers or {})}
         return self._owner._http.get(
-            f"/assistants/{self._owner.assistant_id}/graph",
+            f"/assistants/{_quote_path_param(self._owner.assistant_id)}/graph",
             params=query_params,
             headers=request_headers or None,
         )
@@ -1552,7 +1553,7 @@ class SyncThreadStream:
     def _fetch_state(self) -> dict[str, Any]:
         """Fetch the current thread state from the REST endpoint."""
         return self._http.get(
-            f"/threads/{self.thread_id}/state",
+            f"/threads/{_quote_path_param(self.thread_id)}/state",
             headers=self._headers or None,
         )
 

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

@@ -0,0 +1,92 @@
+"""Thread stream REST helpers must percent-encode `thread_id` and `assistant_id`
+so a value containing reserved characters or dot-segments stays an opaque path
+segment instead of being normalized into a different path by the HTTP stack.
+
+Each test drives a public entry point (`threads.stream(...)` followed by
+`thread.output` or `thread.agent.get_tree()`) with an identifier containing
+`..`, `/`, and `#`, and asserts the request path keeps it as one segment.
+"""
+
+from __future__ import annotations
+
+import httpx
+
+from langgraph_sdk._async.http import HttpClient
+from langgraph_sdk._async.threads import ThreadsClient
+from langgraph_sdk._sync.http import SyncHttpClient
+from langgraph_sdk._sync.threads import SyncThreadsClient
+
+# Interpolated raw, httpx would collapse the `..` segments and treat `#` as a
+# fragment, producing `/assistants/a/graph` and `/other/path`.
+THREAD_ID = "../assistants/a/graph#"
+ASSISTANT_ID = "../../other/path#"
+ENCODED_STATE_PATH = "/threads/..%2Fassistants%2Fa%2Fgraph%23/state"
+ENCODED_GRAPH_PATH = "/assistants/..%2F..%2Fother%2Fpath%23/graph"
+
+TERMINAL_STATE = {"values": {"ok": True}, "next": [], "tasks": []}
+
+
+def _recording_handler(gets: list[str]):
+    def handler(request: httpx.Request) -> httpx.Response:
+        if request.method == "GET":
+            path = request.url.raw_path.decode("ascii").split("?", 1)[0]
+            gets.append(path)
+            return httpx.Response(200, json=TERMINAL_STATE)
+        # Lifecycle watcher subscription: an empty event stream.
+        return httpx.Response(
+            200, headers={"content-type": "text/event-stream"}, content=b""
+        )
+
+    return handler
+
+
+async def test_async_output_encodes_thread_id():
+    gets: list[str] = []
+    async with httpx.AsyncClient(
+        transport=httpx.MockTransport(_recording_handler(gets)),
+        base_url="https://example.com",
+    ) as raw:
+        threads = ThreadsClient(HttpClient(raw))
+        async with threads.stream(thread_id=THREAD_ID, assistant_id="agent") as thread:
+            assert await thread.output == {"ok": True}
+
+    assert gets == [ENCODED_STATE_PATH]
+
+
+async def test_async_agent_get_tree_encodes_assistant_id():
+    gets: list[str] = []
+    async with httpx.AsyncClient(
+        transport=httpx.MockTransport(_recording_handler(gets)),
+        base_url="https://example.com",
+    ) as raw:
+        threads = ThreadsClient(HttpClient(raw))
+        async with threads.stream(thread_id="t-1", assistant_id=ASSISTANT_ID) as thread:
+            await thread.agent.get_tree()
+
+    assert gets == [ENCODED_GRAPH_PATH]
+
+
+def test_sync_output_encodes_thread_id():
+    gets: list[str] = []
+    with httpx.Client(
+        transport=httpx.MockTransport(_recording_handler(gets)),
+        base_url="https://example.com",
+    ) as raw:
+        threads = SyncThreadsClient(SyncHttpClient(raw))
+        with threads.stream(thread_id=THREAD_ID, assistant_id="agent") as thread:
+            assert thread.output == {"ok": True}
+
+    assert gets == [ENCODED_STATE_PATH]
+
+
+def test_sync_agent_get_tree_encodes_assistant_id():
+    gets: list[str] = []
+    with httpx.Client(
+        transport=httpx.MockTransport(_recording_handler(gets)),
+        base_url="https://example.com",
+    ) as raw:
+        threads = SyncThreadsClient(SyncHttpClient(raw))
+        with threads.stream(thread_id="t-1", assistant_id=ASSISTANT_ID) as thread:
+            thread.agent.get_tree()
+
+    assert gets == [ENCODED_GRAPH_PATH]