Implement Vast.ai webhook signature validation and event handling.
Use when setting up webhook endpoints, implementing signature verification,
or handling Vast.ai event notifications securely.
Trigger with phrases like "vastai webhook", "vastai events",
"vastai webhook signature", "handle vastai events", "vastai notifications".
Build event-driven workflows around Vast.ai GPU instance lifecycle. Vast.ai does not provide traditional webhooks, so event detection relies on polling the REST API at cloud.vast.ai/api/v0 and reacting to instance status transitions (loading, running, exited, error, offline).
Prerequisites
Vast.ai CLI authenticated
Understanding of instance lifecycle states
Python 3.8+ for event loop implementation
Instructions
Step 1: Instance Status Poller
import time, json, subprocess
from typing import Callable, Dict, List
class InstanceEventPoller:
"""Poll Vast.ai API and emit events on status transitions."""
def __init__(self, api_key: str, poll_interval: int = 30):
self.api_key = api_key
self.poll_interval = poll_interval
self.previous_states: Dict[int, str] = {}
self.handlers: Dict[str, List[Callable]] = {}
def on(self, event: str, handler: Callable):
self.handlers.setdefault(event, []).append(handler)
def poll_once(self):
result = subprocess.run(
["vastai", "show", "instances", "--raw"],
capture_output=True, text=True)
instances = json.loads(result.stdout)
for inst in instances:
inst_id = inst["id"]
status = inst.get("actual_status", "unknown")
prev = self.previous_states.get(inst_id)
if prev and prev != status:
event = f"{prev}_to_{status}"
for handler in self.handlers.get(event, []):
handler(inst)
for handler in self.handlers.get("any_change", []):
handler(inst, prev, status)
self.previous_states[inst_id] = status
def run(self):
print(f"Polling every {self.poll_interval}s...")
while True:
self.poll_once()
time.sleep(self.poll_interval)
Step 2: Event Handlers
def on_instance_running(instance):
print(f"Instance {instance['id']} is RUNNING")
print(f" SSH: ssh -p {instance['ssh_port']} root@{instance['ssh_host']}")
# Trigger: start training job, send notification, etc.
def on_instance_exited(instance):
print(f"Instance {instance['id']} EXITED")
# Trigger: collect results, check for errors, notify team
def on_spot_preemption(instance, old_status, new_status):
if old_status == "running" and new_status in ("exited", "offline"):
print(f"ALERT: Instance {instance['id']} may have been preempted")
# Trigger: auto-recovery, provision replacement
# Wire up handlers
poller = InstanceEventPoller(api_key)
poller.on("loading_to_running", on_instance_running)
poller.on("running_to_exited", on_instance_exited)
poller.on("any_change", on_spot_preemption)
poller.run()
For performance optimization, see vastai-performance-tuning.
Examples
Slack notifications: Wire on_instance_running to send a Slack message with SSH connection details. Wire on_spot_preemption to alert the team.
Training monitor: Track running_to_exited events. If exit was expected (job complete), collect results. If unexpected, trigger auto-recovery with checkpoint resume.
