vastai-webhooks-events

name	vastai-webhooks-events
description	Build event-driven workflows around Vast.ai instance lifecycle events. Use when monitoring instance status changes, implementing auto-recovery, or building event-driven GPU orchestration. Trigger with phrases like "vastai events", "vastai instance monitoring", "vastai status changes", "vastai lifecycle events".
allowed-tools	Read, Write, Edit, Bash(vastai:*), Bash(curl:*)
version	1.11.0
license	MIT
author	Jeremy Longshore <[e-mail]>
tags	["saas","vast-ai","webhooks"]
compatibility	Designed for Claude Code

Vast.ai Webhooks & Events

Overview

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", , ],
            capture_output=, text=)
        instances = json.loads(result.stdout)

         inst  instances:
            inst_id = inst[]
            status = inst.get(, )
            prev = .previous_states.get(inst_id)

             prev  prev != status:
                event = 
                 handler  .handlers.get(event, []):
                    handler(inst)
                 handler  .handlers.get(, []):
                    handler(inst, prev, status)

            .previous_states[inst_id] = status

     ():
        ()
         :
            .poll_once()
            time.sleep(.poll_interval)