Optionally use loop-safe asgi transport (#4668)

libs/sdk-py/langgraph_sdk/client.py

@@ -10,6 +10,7 @@ document Store.
 from __future__ import annotations
 
 import asyncio
+import functools
 import logging
 import os
 import sys
@@ -22,6 +23,7 @@ from typing import (
     Literal,
     Optional,
     Sequence,
+    Type,
     Union,
     overload,
 )
@@ -162,7 +164,7 @@ def get_client(
     transport: Optional[httpx.AsyncBaseTransport] = None
     if url is None:
         if os.environ.get("__LANGGRAPH_DEFER_LOOPBACK_TRANSPORT") == "true":
-            transport = httpx.ASGITransport(app=None, root_path="/noauth")
+            transport = get_asgi_transport()(app=None, root_path="/noauth")
             _registered_transports.append(transport)
             url = "http://api"
         else:
@@ -170,7 +172,8 @@ def get_client(
                 from langgraph_api.server import app  # type: ignore
 
                 url = "http://api"
-                transport = httpx.ASGITransport(app, root_path="/noauth")
+
+                transport = get_asgi_transport()(app, root_path="/noauth")
             except Exception:
                 url = "http://localhost:8123"
 
@@ -5382,6 +5385,17 @@ def configure_loopback_transports(app: Any) -> None:
         transport.app = app
 
 
+@functools.lru_cache(maxsize=1)
+def get_asgi_transport() -> Type[httpx.ASGITransport]:
+    try:
+        from langgraph_api import asgi_transport
+
+        return asgi_transport.ASGITransport
+    except ImportError:
+        # Older versions of the server
+        return httpx.ASGITransport
+
+
 TimeoutTypes = Union[
     Optional[float],
     tuple[Optional[float], Optional[float], Optional[float], Optional[float]],

libs/sdk-py/pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "langgraph-sdk"
-version = "0.1.68"
+version = "0.1.69"
 description = "SDK for interacting with LangGraph API"
 authors = []
 license = "MIT"