0
# Webhooks
1

2
Verify and handle webhook events from OpenAI for asynchronous notifications about fine-tuning jobs, batch completions, and other events.
3

4
## Capabilities
5

6
### Verify Webhook Signature
7

8
Verify that a webhook request came from OpenAI.
9

10
```python { .api }
11
def verify_signature(
12
    payload: str | bytes,
13
    headers: dict[str, str] | list[tuple[str, str]],
14
    *,
15
    secret: str | None = None,
16
    tolerance: int = 300
17
) -> None:
18
    """
19
    Verify webhook signature to ensure authenticity.
20

21
    Args:
22
        payload: Raw request body (bytes or string).
23
        headers: Request headers containing webhook-signature, webhook-timestamp,
24
            and webhook-id. Can be dict or list of tuples.
25
        secret: Webhook secret from OpenAI dashboard. If None, uses client's
26
            webhook_secret or OPENAI_WEBHOOK_SECRET environment variable.
27
        tolerance: Maximum age of webhook in seconds (default: 300).
28
            Used to prevent replay attacks.
29

30
    Returns:
31
        None: If signature is valid.
32

33
    Raises:
34
        InvalidWebhookSignatureError: If signature is invalid or timestamp is too old/new.
35
        ValueError: If secret is not provided and not set on client.
36
    """
37
```
38

39
Usage example:
40

41
```python
42
from openai import OpenAI, InvalidWebhookSignatureError
43

44
client = OpenAI(webhook_secret="your-webhook-secret")
45

46
# In your webhook endpoint
47
def webhook_handler(request):
48
    payload = request.body  # Raw bytes
49
    headers = request.headers  # Headers dict or list
50

51
    try:
52
        # Verify signature
53
        client.webhooks.verify_signature(
54
            payload=payload,
55
            headers=headers,
56
            secret="your-webhook-secret"  # Optional if set on client
57
        )
58

59
        # Signature valid, process webhook
60
        event = json.loads(payload)
61
        handle_webhook_event(event)
62

63
    except InvalidWebhookSignatureError:
64
        # Invalid signature, reject request
65
        return {"error": "Invalid signature"}, 401
66
```
67

68
### Unwrap Webhook Event
69

70
Verify signature and parse webhook event in one call.
71

72
```python { .api }
73
def unwrap(
74
    payload: str | bytes,
75
    headers: dict[str, str] | list[tuple[str, str]],
76
    *,
77
    secret: str | None = None
78
) -> UnwrapWebhookEvent:
79
    """
80
    Verify signature and parse webhook event.
81

82
    Args:
83
        payload: Raw request body (bytes or string).
84
        headers: Request headers containing webhook-signature, webhook-timestamp,
85
            and webhook-id. Can be dict or list of tuples.
86
        secret: Webhook secret from OpenAI dashboard. If None, uses client's
87
            webhook_secret or OPENAI_WEBHOOK_SECRET environment variable.
88

89
    Returns:
90
        UnwrapWebhookEvent: Parsed and verified webhook event.
91

92
    Raises:
93
        InvalidWebhookSignatureError: If signature is invalid.
94
        ValueError: If secret is not provided and not set on client.
95
    """
96
```
97

98
Usage example:
99

100
```python
101
from openai import OpenAI
102

103
client = OpenAI(webhook_secret="your-webhook-secret")
104

105
# In webhook endpoint
106
def webhook_handler(request):
107
    try:
108
        # Verify and parse in one call
109
        event = client.webhooks.unwrap(
110
            payload=request.body,
111
            headers=request.headers,
112
            secret="your-webhook-secret"  # Optional if set on client
113
        )
114

115
        # Handle different event types
116
        if event.type == "fine_tuning.job.succeeded":
117
            handle_fine_tuning_success(event.data)
118

119
        elif event.type == "batch.completed":
120
            handle_batch_completion(event.data)
121

122
        return {"status": "ok"}, 200
123

124
    except Exception as e:
125
        return {"error": str(e)}, 400
126
```
127

128
## Webhook Event Types
129

130
### Fine-tuning Events
131

132
```python
133
# Job succeeded
134
{
135
    "type": "fine_tuning.job.succeeded",
136
    "data": {
137
        "id": "ftjob-abc123",
138
        "fine_tuned_model": "ft:gpt-3.5-turbo:org:model:abc",
139
        "status": "succeeded"
140
    }
141
}
142

143
# Job failed
144
{
145
    "type": "fine_tuning.job.failed",
146
    "data": {
147
        "id": "ftjob-abc123",
148
        "status": "failed",
149
        "error": {...}
150
    }
151
}
152

153
# Job cancelled
154
{
155
    "type": "fine_tuning.job.cancelled",
156
    "data": {
157
        "id": "ftjob-abc123",
158
        "status": "cancelled"
159
    }
160
}
161
```
162

163
### Batch Events
164

165
```python
166
# Batch completed
167
{
168
    "type": "batch.completed",
169
    "data": {
170
        "id": "batch_abc123",
171
        "status": "completed",
172
        "output_file_id": "file-xyz789"
173
    }
174
}
175

176
# Batch failed
177
{
178
    "type": "batch.failed",
179
    "data": {
180
        "id": "batch_abc123",
181
        "status": "failed",
182
        "errors": {...}
183
    }
184
}
185

186
# Batch cancelled
187
{
188
    "type": "batch.cancelled",
189
    "data": {
190
        "id": "batch_abc123",
191
        "status": "cancelled"
192
    }
193
}
194

195
# Batch expired
196
{
197
    "type": "batch.expired",
198
    "data": {
199
        "id": "batch_abc123",
200
        "status": "expired"
201
    }
202
}
203
```
204

205
### Eval Events
206

207
```python
208
# Eval run succeeded
209
{
210
    "type": "eval.run.succeeded",
211
    "data": {
212
        "id": "eval_run_abc123",
213
        "status": "succeeded",
214
        "results": {...}
215
    }
216
}
217

218
# Eval run failed
219
{
220
    "type": "eval.run.failed",
221
    "data": {
222
        "id": "eval_run_abc123",
223
        "status": "failed",
224
        "error": {...}
225
    }
226
}
227
```
228

229
## Types
230

231
```python { .api }
232
from typing import Union, Literal
233
from typing_extensions import TypeAlias
234

235
# UnwrapWebhookEvent is a union of all possible webhook event types
236
UnwrapWebhookEvent: TypeAlias = Union[
237
    BatchCancelledWebhookEvent,
238
    BatchCompletedWebhookEvent,
239
    BatchExpiredWebhookEvent,
240
    BatchFailedWebhookEvent,
241
    EvalRunCanceledWebhookEvent,
242
    EvalRunFailedWebhookEvent,
243
    EvalRunSucceededWebhookEvent,
244
    FineTuningJobCancelledWebhookEvent,
245
    FineTuningJobFailedWebhookEvent,
246
    FineTuningJobSucceededWebhookEvent,
247
    RealtimeCallIncomingWebhookEvent,
248
    ResponseCancelledWebhookEvent,
249
    ResponseCompletedWebhookEvent,
250
    ResponseFailedWebhookEvent,
251
    ResponseIncompleteWebhookEvent,
252
]
253

254
# Fine-tuning event types
255
class FineTuningJobSucceededWebhookEvent:
256
    type: Literal["fine_tuning.job.succeeded"]
257
    data: dict
258

259
class FineTuningJobFailedWebhookEvent:
260
    type: Literal["fine_tuning.job.failed"]
261
    data: dict
262

263
class FineTuningJobCancelledWebhookEvent:
264
    type: Literal["fine_tuning.job.cancelled"]
265
    data: dict
266

267
# Batch event types
268
class BatchCompletedWebhookEvent:
269
    type: Literal["batch.completed"]
270
    data: dict
271

272
class BatchFailedWebhookEvent:
273
    type: Literal["batch.failed"]
274
    data: dict
275

276
class BatchCancelledWebhookEvent:
277
    type: Literal["batch.cancelled"]
278
    data: dict
279

280
class BatchExpiredWebhookEvent:
281
    type: Literal["batch.expired"]
282
    data: dict
283

284
# Eval event types
285
class EvalRunSucceededWebhookEvent:
286
    type: Literal["eval.run.succeeded"]
287
    data: dict
288

289
class EvalRunFailedWebhookEvent:
290
    type: Literal["eval.run.failed"]
291
    data: dict
292

293
class EvalRunCanceledWebhookEvent:
294
    type: Literal["eval.run.canceled"]
295
    data: dict
296

297
# Response event types
298
class ResponseCompletedWebhookEvent:
299
    type: Literal["response.completed"]
300
    data: dict
301

302
class ResponseFailedWebhookEvent:
303
    type: Literal["response.failed"]
304
    data: dict
305

306
class ResponseCancelledWebhookEvent:
307
    type: Literal["response.cancelled"]
308
    data: dict
309

310
class ResponseIncompleteWebhookEvent:
311
    type: Literal["response.incomplete"]
312
    data: dict
313

314
# Realtime event types
315
class RealtimeCallIncomingWebhookEvent:
316
    type: Literal["realtime.call.incoming"]
317
    data: dict
318
```
319

320
## Complete Webhook Handler
321

322
```python
323
from openai import OpenAI, InvalidWebhookSignatureError
324
from flask import Flask, request, jsonify
325

326
app = Flask(__name__)
327
client = OpenAI()
328

329
WEBHOOK_SECRET = "your-webhook-secret"
330

331
@app.route("/webhooks/openai", methods=["POST"])
332
def handle_webhook():
333
    # Get headers and payload
334
    payload = request.data
335
    headers = request.headers
336

337
    # Verify and parse
338
    try:
339
        event = client.webhooks.unwrap(
340
            payload=payload,
341
            headers=headers,
342
            secret=WEBHOOK_SECRET
343
        )
344
    except InvalidWebhookSignatureError:
345
        return jsonify({"error": "Invalid signature"}), 401
346

347
    # Handle event
348
    if event.type == "fine_tuning.job.succeeded":
349
        job_id = event.data["id"]
350
        model = event.data["fine_tuned_model"]
351
        print(f"Fine-tuning succeeded: {job_id} -> {model}")
352

353
        # Deploy model or notify user
354
        deploy_model(model)
355

356
    elif event.type == "batch.completed":
357
        batch_id = event.data["id"]
358
        output_file = event.data["output_file_id"]
359
        print(f"Batch completed: {batch_id}")
360

361
        # Process results
362
        process_batch_results(batch_id, output_file)
363

364
    elif event.type == "fine_tuning.job.failed":
365
        job_id = event.data["id"]
366
        error = event.data.get("error")
367
        print(f"Fine-tuning failed: {job_id}, Error: {error}")
368

369
        # Notify user of failure
370
        notify_failure(job_id, error)
371

372
    return jsonify({"status": "received"}), 200
373

374
if __name__ == "__main__":
375
    app.run(port=8080)
376
```
377

378
## Best Practices
379

380
```python
381
from openai import OpenAI, InvalidWebhookSignatureError
382
import hmac
383
import hashlib
384

385
client = OpenAI()
386

387
# 1. Always verify signatures
388
def is_valid_webhook(payload: bytes, headers: dict, secret: str) -> bool:
389
    try:
390
        client.webhooks.verify_signature(payload, headers, secret=secret)
391
        return True
392
    except InvalidWebhookSignatureError:
393
        return False
394

395
# 2. Handle replay attacks with timestamp
396
def is_recent_webhook(timestamp: str, max_age_seconds: int = 300) -> bool:
397
    import time
398

399
    event_time = int(timestamp)
400
    current_time = int(time.time())
401

402
    return (current_time - event_time) < max_age_seconds
403

404
# 3. Process events idempotently
405
processed_events = set()
406

407
def process_webhook_event(event_id: str, event_data: dict):
408
    if event_id in processed_events:
409
        print(f"Duplicate event: {event_id}")
410
        return
411

412
    # Process event
413
    handle_event(event_data)
414

415
    # Mark as processed
416
    processed_events.add(event_id)
417

418
# 4. Return 200 quickly, process async
419
from threading import Thread
420

421
def handle_webhook_async(event):
422
    # Process in background
423
    thread = Thread(target=process_event, args=(event,))
424
    thread.start()
425

426
    # Return immediately
427
    return {"status": "accepted"}, 200
428

429
# 5. Retry on failure
430
import time
431

432
def process_with_retry(event, max_retries=3):
433
    for attempt in range(max_retries):
434
        try:
435
            process_event(event)
436
            return
437
        except Exception as e:
438
            if attempt == max_retries - 1:
439
                log_failure(event, e)
440
                raise
441
            time.sleep(2 ** attempt)
442
```
443

444
## Testing Webhooks
445

446
```python
447
# Generate test signature for development
448
import hmac
449
import hashlib
450
import json
451
import time
452

453
def generate_test_signature(payload: dict, secret: str) -> tuple[str, str]:
454
    """Generate signature for testing."""
455
    timestamp = str(int(time.time()))
456
    payload_str = json.dumps(payload)
457

458
    # Create signature
459
    message = f"{timestamp}.{payload_str}"
460
    signature = hmac.new(
461
        secret.encode(),
462
        message.encode(),
463
        hashlib.sha256
464
    ).hexdigest()
465

466
    return signature, timestamp
467

468
# Test webhook handler
469
test_payload = {
470
    "type": "fine_tuning.job.succeeded",
471
    "data": {
472
        "id": "ftjob-test123",
473
        "fine_tuned_model": "ft:gpt-3.5-turbo:test"
474
    }
475
}
476

477
# Create test headers (actual implementation would vary)
478
test_headers = {
479
    "webhook-signature": "v1,test_signature",
480
    "webhook-timestamp": str(int(time.time())),
481
    "webhook-id": "test_webhook_id"
482
}
483

484
# Note: In production, signatures are generated by OpenAI
485
# This is a simplified example for testing
486
event = client.webhooks.unwrap(
487
    payload=json.dumps(test_payload),
488
    headers=test_headers,
489
    secret=WEBHOOK_SECRET
490
)
491

492
print(f"Test event: {event.type}")
493
```
494

495
## Security Considerations
496

497
1. **Always verify signatures** - Never trust webhook data without verification
498
2. **Use HTTPS** - Only accept webhooks over HTTPS
499
3. **Check timestamps** - Reject old events to prevent replay attacks
500
4. **Rate limit** - Limit webhook endpoint to prevent abuse
501
5. **Store secrets securely** - Use environment variables or secret management
502
6. **Log failures** - Track verification failures for security monitoring
503