pypi-langgraph-sdk

1
# Client Management
2

3
Core client creation and HTTP operations for connecting to LangGraph servers with automatic discovery, custom authentication, and connection management.
4

5
## Capabilities
6

7
### Client Factory Functions
8

9
Create async and sync clients for interacting with LangGraph servers, with automatic local server discovery and configurable connection settings.
10

11
```python { .api }
12
from collections.abc import Mapping
13
from typing import Union, Optional
14
import httpx
15

16
# Type aliases
17
TimeoutTypes = Union[
18
    None,
19
    float,
20
    tuple[Optional[float], Optional[float]],
21
    tuple[Optional[float], Optional[float], Optional[float], Optional[float]],
22
    httpx.Timeout,
23
]
24

25
def get_client(
26
    *,
27
    url: str | None = None,
28
    api_key: str | None = None,
29
    headers: Mapping[str, str] | None = None,
30
    timeout: TimeoutTypes | None = None,
31
) -> LangGraphClient:
32
    """
33
    Create an async LangGraph client.
34

35
    Args:
36
        url: Base URL of the LangGraph API.
37
            – If `None`, the client first attempts an in-process connection via ASGI transport.
38
              If that fails, it falls back to `http://localhost:8123`.
39
        api_key: API key for authentication. If omitted, the client reads from environment
40
            variables in the following order:
41
              1. Function argument
42
              2. `LANGGRAPH_API_KEY`
43
              3. `LANGSMITH_API_KEY`
44
              4. `LANGCHAIN_API_KEY`
45
        headers: Additional HTTP headers to include in requests. Merged with authentication headers.
46
        timeout: HTTP timeout configuration. May be:
47
              – `httpx.Timeout` instance
48
              – float (total seconds)
49
              – tuple of timeouts (connect, read, write, pool)
50

51
    Returns:
52
        LangGraphClient: The top-level asynchronous client for accessing AssistantsClient,
53
        ThreadsClient, RunsClient, and CronClient.
54
    """
55

56
def get_sync_client(
57
    *,
58
    url: str | None = None,
59
    api_key: str | None = None,
60
    headers: Mapping[str, str] | None = None,
61
    timeout: TimeoutTypes | None = None,
62
) -> SyncLangGraphClient:
63
    """
64
    Create a sync LangGraph client.
65

66
    Args:
67
        url: The URL of the LangGraph API.
68
        api_key: The API key. If not provided, it will be read from the environment.
69
            Precedence:
70
                1. explicit argument
71
                2. LANGGRAPH_API_KEY
72
                3. LANGSMITH_API_KEY
73
                4. LANGCHAIN_API_KEY
74
        headers: Optional custom headers
75
        timeout: Optional timeout configuration for the HTTP client.
76
            Accepts an httpx.Timeout instance, a float (seconds), or a tuple of timeouts.
77
            Tuple format is (connect, read, write, pool)
78

79
    Returns:
80
        SyncLangGraphClient: The top-level synchronous client for accessing AssistantsClient,
81
        ThreadsClient, RunsClient, and CronClient.
82
    """
83
```
84

85
### Main Client Classes
86

87
Primary client interfaces providing access to all LangGraph Platform resources through specialized resource clients.
88

89
```python { .api }
90
class LangGraphClient:
91
    """
92
    Async client for LangGraph Platform API.
93

94
    Attributes:
95
    - assistants: AssistantsClient for managing AI assistants
96
    - threads: ThreadsClient for managing conversation threads
97
    - runs: RunsClient for executing workflows
98
    - crons: CronClient for scheduled tasks
99
    - store: StoreClient for persistent storage
100
    """
101
    assistants: AssistantsClient
102
    threads: ThreadsClient
103
    runs: RunsClient
104
    crons: CronClient
105
    store: StoreClient
106

107
    async def aclose(self) -> None:
108
        """Close the HTTP client and cleanup resources."""
109

110
class SyncLangGraphClient:
111
    """
112
    Sync client for LangGraph Platform API.
113

114
    Attributes:
115
    - assistants: SyncAssistantsClient for managing AI assistants
116
    - threads: SyncThreadsClient for managing conversation threads
117
    - runs: SyncRunsClient for executing workflows
118
    - crons: SyncCronClient for scheduled tasks
119
    - store: SyncStoreClient for persistent storage
120
    """
121
    assistants: SyncAssistantsClient
122
    threads: SyncThreadsClient
123
    runs: SyncRunsClient
124
    crons: SyncCronClient
125
    store: SyncStoreClient
126

127
    def close(self) -> None:
128
        """Close the HTTP client and cleanup resources."""
129
```
130

131
### HTTP Client Operations
132

133
Low-level HTTP operations for direct API access when needed, supporting all standard HTTP methods with streaming capabilities.
134

135
```python { .api }
136
from collections.abc import AsyncIterator, Iterator, Mapping, Callable
137
from typing import Any
138
from langgraph_sdk.schema import QueryParamTypes, StreamPart
139
import httpx
140

141
class HttpClient:
142
    """Low-level HTTP client for direct API access."""
143

144
    async def get(
145
        self,
146
        path: str,
147
        *,
148
        params: QueryParamTypes | None = None,
149
        headers: Mapping[str, str] | None = None,
150
        on_response: Callable[[httpx.Response], None] | None = None,
151
    ) -> Any:
152
        """Send a GET request."""
153

154
    async def post(
155
        self,
156
        path: str,
157
        *,
158
        json: dict[str, Any] | list | None,
159
        params: QueryParamTypes | None = None,
160
        headers: Mapping[str, str] | None = None,
161
        on_response: Callable[[httpx.Response], None] | None = None,
162
    ) -> Any:
163
        """Send a POST request."""
164

165
    async def put(
166
        self,
167
        path: str,
168
        *,
169
        json: dict,
170
        params: QueryParamTypes | None = None,
171
        headers: Mapping[str, str] | None = None,
172
        on_response: Callable[[httpx.Response], None] | None = None,
173
    ) -> Any:
174
        """Send a PUT request."""
175

176
    async def patch(
177
        self,
178
        path: str,
179
        *,
180
        json: dict,
181
        params: QueryParamTypes | None = None,
182
        headers: Mapping[str, str] | None = None,
183
        on_response: Callable[[httpx.Response], None] | None = None,
184
    ) -> Any:
185
        """Send a PATCH request."""
186

187
    async def delete(
188
        self,
189
        path: str,
190
        *,
191
        json: Any | None = None,
192
        params: QueryParamTypes | None = None,
193
        headers: Mapping[str, str] | None = None,
194
        on_response: Callable[[httpx.Response], None] | None = None,
195
    ) -> None:
196
        """Send a DELETE request."""
197

198
    async def stream(
199
        self,
200
        path: str,
201
        method: str,
202
        *,
203
        json: dict[str, Any] | None = None,
204
        params: QueryParamTypes | None = None,
205
        headers: Mapping[str, str] | None = None,
206
        on_response: Callable[[httpx.Response], None] | None = None,
207
    ) -> AsyncIterator[StreamPart]:
208
        """Stream results using SSE."""
209

210
class SyncHttpClient:
211
    """Sync HTTP client with identical methods but without async/await."""
212

213
    def get(
214
        self,
215
        path: str,
216
        *,
217
        params: QueryParamTypes | None = None,
218
        headers: Mapping[str, str] | None = None,
219
        on_response: Callable[[httpx.Response], None] | None = None,
220
    ) -> Any: ...
221

222
    def post(
223
        self,
224
        path: str,
225
        *,
226
        json: dict[str, Any] | list | None,
227
        params: QueryParamTypes | None = None,
228
        headers: Mapping[str, str] | None = None,
229
        on_response: Callable[[httpx.Response], None] | None = None,
230
    ) -> Any: ...
231

232
    def put(
233
        self,
234
        path: str,
235
        *,
236
        json: dict,
237
        params: QueryParamTypes | None = None,
238
        headers: Mapping[str, str] | None = None,
239
        on_response: Callable[[httpx.Response], None] | None = None,
240
    ) -> Any: ...
241

242
    def patch(
243
        self,
244
        path: str,
245
        *,
246
        json: dict,
247
        params: QueryParamTypes | None = None,
248
        headers: Mapping[str, str] | None = None,
249
        on_response: Callable[[httpx.Response], None] | None = None,
250
    ) -> Any: ...
251

252
    def delete(
253
        self,
254
        path: str,
255
        *,
256
        json: Any | None = None,
257
        params: QueryParamTypes | None = None,
258
        headers: Mapping[str, str] | None = None,
259
        on_response: Callable[[httpx.Response], None] | None = None,
260
    ) -> None: ...
261

262
    def stream(
263
        self,
264
        path: str,
265
        method: str,
266
        *,
267
        json: dict[str, Any] | None = None,
268
        params: QueryParamTypes | None = None,
269
        headers: Mapping[str, str] | None = None,
270
        on_response: Callable[[httpx.Response], None] | None = None,
271
    ) -> Iterator[StreamPart]: ...
272
```
273

274
## Usage Examples
275

276
### Basic Client Setup
277

278
```python
279
from langgraph_sdk import get_client, get_sync_client
280

281
# Async client with auto-discovery
282
async_client = await get_client()
283

284
# Async client with specific server
285
async_client = await get_client(url="https://api.langgraph.com", api_key="your-api-key")
286

287
# Sync client
288
sync_client = get_sync_client()
289

290
# With custom headers
291
client = await get_client(
292
    headers={"User-Agent": "MyApp/1.0"},
293
    timeout=60.0
294
)
295
```
296

297
### Resource Access
298

299
```python
300
# Access different resource managers
301
assistants = await client.assistants.search()
302
threads = await client.threads.search()
303
runs = await client.runs.list(thread_id="thread-123")
304

305
# Cleanup when done
306
await client.aclose()
307
```
308

309
### Direct HTTP Operations
310

311
```python
312
# Low-level API access
313
response = await client.http.get("/assistants", params={"limit": 20})
314
new_thread = await client.http.post("/threads", json={"metadata": {"user": "john"}})
315

316
# Streaming
317
async for chunk in client.http.stream("/runs/stream", method="POST", json=run_data):
318
    print(chunk)
319
```