0
# Deployments & Webhooks
1

2
Management of model deployments and webhook configuration for production use cases and real-time notifications.
3

4
## Capabilities
5

6
### Deployment Management
7

8
Manage dedicated model deployments for production workloads.
9

10
```python { .api }
11
class Deployments:
12
    def get(self, owner: str, name: str) -> Deployment:
13
        """
14
        Get a deployment by owner and name.
15

16
        Parameters:
17
        - owner: Deployment owner username
18
        - name: Deployment name
19

20
        Returns:
21
        Deployment object with configuration and status
22
        """
23

24
    def list(self, **params) -> Page[Deployment]:
25
        """
26
        List deployments.
27

28
        Returns:
29
        Paginated list of Deployment objects
30
        """
31

32
    def create(
33
        self,
34
        name: str,
35
        model: str,
36
        version: str,
37
        *,
38
        hardware: Optional[str] = None,
39
        min_instances: Optional[int] = None,
40
        max_instances: Optional[int] = None,
41
        **params
42
    ) -> Deployment:
43
        """
44
        Create a new deployment.
45

46
        Parameters:
47
        - name: Name for the deployment
48
        - model: Model name in format "owner/name"
49
        - version: Model version ID
50
        - hardware: Hardware tier for deployment
51
        - min_instances: Minimum number of instances
52
        - max_instances: Maximum number of instances
53

54
        Returns:
55
        Deployment object
56
        """
57

58
    def update(
59
        self,
60
        owner: str,
61
        name: str,
62
        *,
63
        version: Optional[str] = None,
64
        hardware: Optional[str] = None,
65
        min_instances: Optional[int] = None,
66
        max_instances: Optional[int] = None,
67
        **params
68
    ) -> Deployment:
69
        """
70
        Update an existing deployment.
71

72
        Parameters:
73
        - owner: Deployment owner username
74
        - name: Deployment name
75
        - version: New model version ID
76
        - hardware: New hardware tier
77
        - min_instances: New minimum instances
78
        - max_instances: New maximum instances
79

80
        Returns:
81
        Updated Deployment object
82
        """
83

84
    def delete(self, owner: str, name: str) -> None:
85
        """
86
        Delete a deployment.
87

88
        Parameters:
89
        - owner: Deployment owner username
90
        - name: Deployment name
91
        """
92
```
93

94
### Deployment Objects
95

96
Deployments represent dedicated model instances with scaling and performance configuration.
97

98
```python { .api }
99
class Deployment:
100
    owner: str
101
    """The owner of the deployment."""
102

103
    name: str
104
    """The name of the deployment."""
105

106
    model: str
107
    """The model being deployed."""
108

109
    version: str
110
    """The version of the model being deployed."""
111

112
    status: str
113
    """The current status of the deployment."""
114

115
    hardware: Optional[str]
116
    """The hardware tier for the deployment."""
117

118
    min_instances: int
119
    """The minimum number of instances."""
120

121
    max_instances: int
122
    """The maximum number of instances."""
123

124
    current_instances: Optional[int]
125
    """The current number of running instances."""
126

127
    created_at: str
128
    """When the deployment was created."""
129

130
    updated_at: str
131
    """When the deployment was last updated."""
132
```
133

134
### Webhook Management
135

136
Configure and validate webhooks for receiving real-time notifications.
137

138
```python { .api }
139
class Webhooks:
140
    @property
141
    def default(self) -> "Webhooks.Default":
142
        """Namespace for operations related to the default webhook."""
143

144
    class Default:
145
        def secret(self) -> WebhookSigningSecret:
146
            """
147
            Get the default webhook signing secret.
148

149
            Returns:
150
            WebhookSigningSecret object for signature validation
151
            """
152

153
        async def async_secret(self) -> WebhookSigningSecret:
154
            """
155
            Get the default webhook signing secret asynchronously.
156

157
            Returns:
158
            WebhookSigningSecret object for signature validation
159
            """
160

161
    @staticmethod
162
    def validate(
163
        request: Optional[httpx.Request] = None,
164
        headers: Optional[Dict[str, str]] = None,
165
        body: Optional[str] = None,
166
        secret: Optional[WebhookSigningSecret] = None,
167
        tolerance: Optional[int] = None
168
    ) -> None:
169
        """
170
        Validate the signature from an incoming webhook request.
171

172
        Parameters:
173
        - request: The HTTPX request object (alternative to headers/body)
174
        - headers: The request headers dict (alternative to request)  
175
        - body: The request body string (alternative to request)
176
        - secret: The webhook signing secret
177
        - tolerance: Maximum allowed time difference in seconds
178

179
        Raises:
180
        WebhookValidationError: If validation fails
181
        """
182
```
183

184
### Webhook Objects and Exceptions
185

186
```python { .api }
187
class WebhookSigningSecret:
188
    key: str
189
    """The webhook signing secret key."""
190

191
# Webhook validation exceptions
192
class WebhookValidationError(ValueError):
193
    """Base webhook validation error."""
194

195
class MissingWebhookHeaderError(WebhookValidationError):
196
    """Missing required webhook header."""
197

198
class InvalidSecretKeyError(WebhookValidationError):
199
    """Invalid webhook secret key."""
200

201
class MissingWebhookBodyError(WebhookValidationError):
202
    """Missing webhook request body."""
203

204
class InvalidTimestampError(WebhookValidationError):
205
    """Invalid or expired timestamp."""
206

207
class InvalidSignatureError(WebhookValidationError):
208
    """Invalid webhook signature."""
209
```
210

211
### Hardware Management
212

213
Access hardware configuration options for deployments.
214

215
```python { .api }
216
class Hardware:
217
    name: str
218
    """The hardware tier name."""
219

220
    sku: str
221
    """The hardware SKU identifier."""
222

223
class HardwareNamespace:
224
    def list(self, **params) -> List[Hardware]:
225
        """
226
        List available hardware options.
227

228
        Returns:
229
        List of Hardware objects with available tiers
230
        """
231
```
232

233
### Usage Examples
234

235
#### Create and Manage Deployments
236

237
```python
238
import replicate
239

240
# List available hardware options
241
hardware_options = replicate.hardware.list()
242
for hw in hardware_options:
243
    print(f"Hardware: {hw.name} (SKU: {hw.sku})")
244

245
# Create a deployment
246
deployment = replicate.deployments.create(
247
    name="my-production-model",
248
    model="stability-ai/stable-diffusion-3",
249
    version="version-id-here",
250
    hardware="gpu-t4",
251
    min_instances=1,
252
    max_instances=5
253
)
254

255
print(f"Deployment created: {deployment.owner}/{deployment.name}")
256
print(f"Status: {deployment.status}")
257
print(f"Hardware: {deployment.hardware}")
258
print(f"Instances: {deployment.min_instances}-{deployment.max_instances}")
259
```
260

261
#### Update Deployment Configuration
262

263
```python
264
import replicate
265

266
# Update deployment scaling
267
deployment = replicate.deployments.update(
268
    owner="myusername",
269
    name="my-production-model",
270
    min_instances=2,
271
    max_instances=10,
272
    hardware="gpu-a40-large"
273
)
274

275
print(f"Updated deployment: {deployment.status}")
276
print(f"Current instances: {deployment.current_instances}")
277
```
278

279
#### List and Monitor Deployments
280

281
```python
282
import replicate
283

284
# List all deployments
285
deployments = replicate.deployments.list()
286

287
for deployment in deployments.results:
288
    print(f"Deployment: {deployment.owner}/{deployment.name}")
289
    print(f"Model: {deployment.model} ({deployment.version})")
290
    print(f"Status: {deployment.status}")
291
    print(f"Instances: {deployment.current_instances}/{deployment.max_instances}")
292
    print(f"Created: {deployment.created_at}")
293
    print("---")
294

295
# Get specific deployment
296
deployment = replicate.deployments.get("myusername", "my-production-model")
297
print(f"Deployment status: {deployment.status}")
298
```
299

300
#### Run Predictions on Deployments
301

302
```python
303
import replicate
304

305
# Run prediction on deployment
306
deployment_url = "https://my-deployment.replicate.com"
307

308
output = replicate.run(
309
    "myusername/my-production-model",  # Deployment reference
310
    input={"prompt": "a beautiful landscape"},
311
    # Optionally specify deployment URL directly
312
    # model=deployment_url
313
)
314

315
with open("deployment_output.png", "wb") as f:
316
    f.write(output.read())
317
```
318

319
#### Webhook Configuration
320

321
```python
322
import replicate
323

324
# Get default webhook secret
325
secret = replicate.webhooks.default.secret()
326
print(f"Webhook secret: {secret.key}")
327

328
# Create prediction with webhook
329
prediction = replicate.predictions.create(
330
    model="stability-ai/stable-diffusion-3",
331
    input={"prompt": "webhook test"},
332
    webhook="https://myapp.com/webhook",
333
    webhook_events_filter=["start", "output", "logs", "completed"]
334
)
335

336
print(f"Prediction with webhook: {prediction.id}")
337
```
338

339
#### Webhook Validation
340

341
```python
342
import replicate
343
from replicate.exceptions import WebhookValidationError
344

345
def handle_webhook(request):
346
    """Handle incoming webhook request."""
347
    try:
348
        # Extract headers and body
349
        signature = request.headers.get('Replicate-Signature')
350
        timestamp = request.headers.get('Replicate-Timestamp')
351
        body = request.body
352
        
353
        # Validate webhook
354
        replicate.webhooks.validate(
355
            headers={"webhook-signature": signature, "webhook-timestamp": timestamp, "webhook-id": "webhook-id"},
356
            body=body.decode('utf-8'),
357
            secret=secret,
358
            tolerance=300  # 5 minutes
359
        )
360
        
361
        # Process validated webhook
362
        import json
363
        payload = json.loads(body.decode('utf-8'))
364
        
365
        if payload['status'] == 'succeeded':
366
            print(f"Prediction {payload['id']} completed successfully")
367
            # Process output
368
            output_url = payload['output'][0] if payload['output'] else None
369
        elif payload['status'] == 'failed':
370
            print(f"Prediction {payload['id']} failed: {payload.get('error')}")
371
            
372
    except WebhookValidationError as e:
373
        print(f"Webhook validation failed: {e}")
374
        return {"error": "Invalid webhook"}, 400
375
    except Exception as e:
376
        print(f"Webhook processing error: {e}")
377
        return {"error": "Processing failed"}, 500
378
    
379
    return {"success": True}, 200
380
```
381

382
#### Advanced Webhook Handling
383

384
```python
385
import replicate
386
import hashlib
387
import hmac
388
import time
389
from replicate.exceptions import WebhookValidationError
390

391
class WebhookHandler:
392
    def __init__(self, secret_key: str):
393
        self.secret_key = secret_key
394
    
395
    def validate_signature(self, body: bytes, signature: str, timestamp: str):
396
        """Validate webhook signature manually."""
397
        try:
398
            # Parse signature
399
            if not signature.startswith('v1='):
400
                raise InvalidSignatureError("Invalid signature format")
401
            
402
            expected_signature = signature[3:]  # Remove 'v1=' prefix
403
            
404
            # Create signature payload
405
            payload = f"{timestamp}.{body.decode('utf-8')}"
406
            
407
            # Calculate expected signature
408
            expected = hmac.new(
409
                self.secret_key.encode('utf-8'),
410
                payload.encode('utf-8'),
411
                hashlib.sha256
412
            ).hexdigest()
413
            
414
            # Compare signatures
415
            if not hmac.compare_digest(expected, expected_signature):
416
                raise InvalidSignatureError("Signature mismatch")
417
            
418
            # Check timestamp freshness
419
            current_time = int(time.time())
420
            webhook_time = int(timestamp)
421
            if abs(current_time - webhook_time) > 300:  # 5 minutes
422
                raise InvalidTimestampError("Timestamp too old")
423
                
424
        except ValueError as e:
425
            raise WebhookValidationError(f"Validation error: {e}")
426
    
427
    def process_webhook(self, body: bytes, signature: str, timestamp: str):
428
        """Process validated webhook."""
429
        self.validate_signature(body, signature, timestamp)
430
        
431
        import json
432
        payload = json.loads(body.decode('utf-8'))
433
        
434
        event_type = payload.get('status')
435
        prediction_id = payload.get('id')
436
        
437
        if event_type == 'processing':
438
            print(f"Prediction {prediction_id} is processing...")
439
        elif event_type == 'succeeded':
440
            print(f"Prediction {prediction_id} completed!")
441
            self.handle_success(payload)
442
        elif event_type == 'failed':
443
            print(f"Prediction {prediction_id} failed: {payload.get('error')}")
444
            self.handle_failure(payload)
445
    
446
    def handle_success(self, payload):
447
        """Handle successful prediction."""
448
        output = payload.get('output')
449
        if output:
450
            # Download and process outputs
451
            for i, url in enumerate(output):
452
                print(f"Output {i}: {url}")
453
    
454
    def handle_failure(self, payload):
455
        """Handle failed prediction."""
456
        error = payload.get('error')
457
        logs = payload.get('logs')
458
        print(f"Error: {error}")
459
        if logs:
460
            print(f"Logs: {logs[-500:]}")  # Last 500 chars
461

462
# Usage
463
handler = WebhookHandler("your-webhook-secret")
464
# handler.process_webhook(request_body, signature, timestamp)
465
```
466

467
#### Delete Deployment
468

469
```python
470
import replicate
471

472
# Delete a deployment
473
replicate.deployments.delete("myusername", "my-production-model")
474
print("Deployment deleted")
475

476
# Verify deletion
477
try:
478
    deployment = replicate.deployments.get("myusername", "my-production-model")
479
except Exception as e:
480
    print(f"Deployment not found (expected): {e}")
481
```