0
# Client Management
1

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

4
## Capabilities
5

6
### Client Factory Functions
7

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

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

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

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

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

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

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

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

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

84
### Main Client Classes
85

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

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

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

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

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

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

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

130
### HTTP Client Operations
131

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

273
## Usage Examples
274

275
### Basic Client Setup
276

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

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

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

286
# Sync client
287
sync_client = get_sync_client()
288

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

296
### Resource Access
297

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

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

308
### Direct HTTP Operations
309

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

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