pypi-langgraph-sdk

1
# Scheduled Tasks
2

3
Create and manage cron jobs for automated execution of assistants on threads or with dynamic thread creation. Supports timezone handling, webhook notifications, and flexible scheduling.
4

5
## Capabilities
6

7
### Cron Job Creation
8

9
Create scheduled tasks for automated assistant execution with flexible scheduling and configuration options.
10

11
```python { .api }
12
from collections.abc import Mapping
13
from typing import Any
14
from langgraph_sdk.schema import (
15
    Cron, Config, Context, All, QueryParamTypes
16
)
17

18
# Via client.crons
19
async def create(
20
    assistant_id: str,
21
    *,
22
    schedule: str,
23
    input: Mapping[str, Any] | None = None,
24
    metadata: Mapping[str, Any] | None = None,
25
    config: Config | None = None,
26
    context: Context | None = None,
27
    checkpoint_during: bool | None = None,
28
    interrupt_before: All | list[str] | None = None,
29
    interrupt_after: All | list[str] | None = None,
30
    webhook: str | None = None,
31
    webhook_mode: str | None = None,
32
    headers: Mapping[str, str] | None = None,
33
    params: QueryParamTypes | None = None,
34
) -> Cron:
35
    """
36
    Create a cron job that will create a new thread for each run.
37

38
    Args:
39
        assistant_id: The assistant ID or graph name to cron.
40
        schedule: The schedule to run the assistant on.
41
        input: The input to the assistant.
42
        metadata: The metadata to add to the runs.
43
        config: The config to use for the runs.
44
        context: The context to add to the runs.
45
        checkpoint_during: Whether to checkpoint during the run.
46
        interrupt_before: Nodes to interrupt immediately before they run.
47
        interrupt_after: Nodes to interrupt immediately after they run.
48
        webhook: Webhook to call after the run is done.
49
        webhook_mode: Mode to call the webhook. Options are "GET" and "POST".
50
        headers: Optional custom headers to include with the request.
51
        params: Optional query parameters to include with the request.
52

53
    Returns:
54
        Cron: The created cron job.
55
    """
56

57
async def create_for_thread(
58
    thread_id: str,
59
    assistant_id: str,
60
    *,
61
    schedule: str,
62
    input: Mapping[str, Any] | None = None,
63
    metadata: Mapping[str, Any] | None = None,
64
    config: Config | None = None,
65
    context: Context | None = None,
66
    checkpoint_during: bool | None = None,
67
    interrupt_before: All | list[str] | None = None,
68
    interrupt_after: All | list[str] | None = None,
69
    webhook: str | None = None,
70
    webhook_mode: str | None = None,
71
    headers: Mapping[str, str] | None = None,
72
    params: QueryParamTypes | None = None,
73
) -> Cron:
74
    """
75
    Create a cron job that will run on a specific thread.
76

77
    Args:
78
        thread_id: The thread ID to cron.
79
        assistant_id: The assistant ID or graph name to cron.
80
        schedule: The schedule to run the assistant on.
81
        input: The input to the assistant.
82
        metadata: The metadata to add to the runs.
83
        config: The config to use for the runs.
84
        context: The context to add to the runs.
85
        checkpoint_during: Whether to checkpoint during the run.
86
        interrupt_before: Nodes to interrupt immediately before they run.
87
        interrupt_after: Nodes to interrupt immediately after they run.
88
        webhook: Webhook to call after the run is done.
89
        webhook_mode: Mode to call the webhook. Options are "GET" and "POST".
90
        headers: Optional custom headers to include with the request.
91
        params: Optional query parameters to include with the request.
92

93
    Returns:
94
        Cron: The created cron job.
95
    """
96
```
97

98
### Cron Job Management
99

100
Search, list, count, and delete scheduled tasks with filtering capabilities.
101

102
```python { .api }
103
from langgraph_sdk.schema import (
104
    CronSelectField, CronSortBy, SortOrder, QueryParamTypes
105
)
106

107
async def search(
108
    *,
109
    assistant_id: str | None = None,
110
    thread_id: str | None = None,
111
    limit: int = 10,
112
    offset: int = 0,
113
    sort_by: CronSortBy | None = None,
114
    sort_order: SortOrder | None = None,
115
    select: list[CronSelectField] | None = None,
116
    headers: Mapping[str, str] | None = None,
117
    params: QueryParamTypes | None = None,
118
) -> list[Cron]:
119
    """
120
    List cron jobs.
121

122
    Args:
123
        assistant_id: Assistant ID to filter by.
124
        thread_id: Thread ID to filter by.
125
        limit: Limit the number of cron jobs to return.
126
        offset: Offset to start from.
127
        sort_by: Field to sort by.
128
        sort_order: Order to sort by.
129
        select: Fields to include in the response.
130
        headers: Optional custom headers to include with the request.
131
        params: Optional query parameters to include with the request.
132

133
    Returns:
134
        list[Cron]: List of cron jobs.
135
    """
136

137
async def count(
138
    *,
139
    assistant_id: str | None = None,
140
    thread_id: str | None = None,
141
    headers: Mapping[str, str] | None = None,
142
    params: QueryParamTypes | None = None,
143
) -> int:
144
    """
145
    Count cron jobs matching filters.
146

147
    Args:
148
        assistant_id: Assistant ID to filter by.
149
        thread_id: Thread ID to filter by.
150
        headers: Optional custom headers to include with the request.
151
        params: Optional query parameters to include with the request.
152

153
    Returns:
154
        int: Number of crons matching the criteria.
155
    """
156

157
async def delete(
158
    cron_id: str,
159
    *,
160
    headers: Mapping[str, str] | None = None,
161
    params: QueryParamTypes | None = None,
162
) -> None:
163
    """
164
    Delete a cron job.
165

166
    Args:
167
        cron_id: The cron ID to delete.
168
        headers: Optional custom headers to include with the request.
169
        params: Optional query parameters to include with the request.
170
    """
171
```
172

173
## Types
174

175
```python { .api }
176
class Cron(TypedDict):
177
    """Scheduled task definition."""
178
    cron_id: str
179
    thread_id: str
180
    assistant_id: str
181
    schedule: str
182
    timezone: str
183
    created_at: str
184
    updated_at: str
185
    metadata: dict
186
    config: Config
187
    input: dict
188
    next_run_time: str
189
    last_run_time: str
190
    enabled: bool
191

192
CronSelectField = Literal[
193
    "cron_id", "thread_id", "assistant_id", "schedule",
194
    "timezone", "created_at", "updated_at", "metadata",
195
    "config", "input", "next_run_time", "last_run_time", "enabled"
196
]
197

198
CronSortBy = Literal["created_at", "updated_at", "next_run_time", "last_run_time"]
199
```
200

201
## Usage Examples
202

203
### Creating Scheduled Tasks
204

205
```python
206
# Daily report generation
207
daily_report = await client.crons.create(
208
    assistant_id="report-assistant",
209
    schedule="0 9 * * *",  # 9 AM daily
210
    input={"report_type": "daily", "email_list": ["admin@company.com"]},
211
    config={"timeout": 600},
212
    metadata={"purpose": "daily_report", "owner": "operations"},
213
    timezone="America/New_York",
214
    webhook="https://myapp.com/webhooks/report-complete"
215
)
216

217
# Hourly data processing for specific thread
218
hourly_processor = await client.crons.create_for_thread(
219
    thread_id="data-thread-123",
220
    assistant_id="data-processor",
221
    schedule="0 * * * *",  # Every hour
222
    input={"source": "api", "format": "json"},
223
    config={"batch_size": 1000},
224
    metadata={"environment": "production"}
225
)
226

227
# Weekly cleanup task
228
cleanup_job = await client.crons.create(
229
    assistant_id="cleanup-assistant",
230
    schedule="0 2 * * 0",  # 2 AM on Sundays
231
    input={"retention_days": 30, "dry_run": False},
232
    timezone="UTC",
233
    on_completion="keep"  # Keep run records for audit
234
)
235
```
236

237
### Managing Scheduled Tasks
238

239
```python
240
# List all cron jobs for an assistant
241
assistant_jobs = await client.crons.search(
242
    assistant_id="report-assistant",
243
    limit=50
244
)
245

246
# Find jobs for a specific thread
247
thread_jobs = await client.crons.search(
248
    thread_id="data-thread-123"
249
)
250

251
# Get total job count
252
total_jobs = await client.crons.count()
253

254
# Delete a scheduled task
255
await client.crons.delete("cron-456")
256
```
257

258
### Schedule Patterns
259

260
```python
261
# Common cron schedule patterns
262

263
# Every minute
264
await client.crons.create(
265
    assistant_id="monitoring-assistant",
266
    schedule="* * * * *"
267
)
268

269
# Every 15 minutes
270
await client.crons.create(
271
    assistant_id="health-check-assistant",
272
    schedule="*/15 * * * *"
273
)
274

275
# Daily at 3:30 AM
276
await client.crons.create(
277
    assistant_id="backup-assistant",
278
    schedule="30 3 * * *"
279
)
280

281
# Weekly on Monday at 9 AM
282
await client.crons.create(
283
    assistant_id="weekly-report-assistant",
284
    schedule="0 9 * * 1"
285
)
286

287
# Monthly on the 1st at midnight
288
await client.crons.create(
289
    assistant_id="monthly-billing-assistant",
290
    schedule="0 0 1 * *"
291
)
292

293
# Weekdays at 6 PM
294
await client.crons.create(
295
    assistant_id="daily-summary-assistant",
296
    schedule="0 18 * * 1-5"
297
)
298
```
299

300
### Timezone Handling
301

302
```python
303
# Schedule with specific timezone
304
tokyo_job = await client.crons.create(
305
    assistant_id="tokyo-report-assistant",
306
    schedule="0 9 * * *",  # 9 AM Tokyo time
307
    timezone="Asia/Tokyo",
308
    input={"region": "APAC"}
309
)
310

311
# Multiple timezone jobs for global operation
312
timezones = ["UTC", "America/New_York", "Europe/London", "Asia/Tokyo"]
313
for tz in timezones:
314
    await client.crons.create(
315
        assistant_id="regional-assistant",
316
        schedule="0 8 * * *",  # 8 AM local time
317
        timezone=tz,
318
        input={"timezone": tz, "region": tz.split("/")[-1]}
319
    )
320
```
321

322
### Webhook Integration
323

324
```python
325
# Job with webhook notifications
326
webhook_job = await client.crons.create(
327
    assistant_id="critical-task-assistant",
328
    schedule="0 */6 * * *",  # Every 6 hours
329
    input={"task": "system_health_check"},
330
    webhook="https://monitoring.company.com/webhooks/cron-complete",
331
    metadata={"priority": "critical", "alert_on_failure": True}
332
)
333

334
# Webhook payload will include:
335
# {
336
#   "cron_id": "cron-123",
337
#   "run_id": "run-456",
338
#   "status": "success|error|timeout",
339
#   "metadata": {...},
340
#   "completed_at": "2023-12-01T12:00:00Z"
341
# }
342
```
343

344
### Error Handling and Monitoring
345

346
```python
347
# Create jobs with error handling configuration
348
robust_job = await client.crons.create(
349
    assistant_id="data-sync-assistant",
350
    schedule="0 2 * * *",
351
    input={"source": "external_api", "retries": 3},
352
    config={"timeout": 1800, "retry_policy": "exponential_backoff"},
353
    multitask_strategy="enqueue",  # Queue if previous run still active
354
    webhook="https://alerts.company.com/cron-status"
355
)
356

357
# Monitor scheduled tasks
358
all_jobs = await client.crons.search(limit=100)
359
for job in all_jobs:
360
    next_run = job.get("next_run_time")
361
    last_run = job.get("last_run_time")
362
    enabled = job.get("enabled", True)
363

364
    print(f"Job {job['cron_id']}: next={next_run}, last={last_run}, enabled={enabled}")
365

366
    if not enabled:
367
        print(f"Warning: Job {job['cron_id']} is disabled")
368
```