Give your code
eyes and hands.

pip install requests

{
  "request_id": "req_abc123",
  "actions": [
    {
      "action_type": "click",
      "params": { "x": 512, "y": 340, "button": "left", "clicks": 1 }
    },
    {
      "action_type": "type_text",
      "params": { "text": "hello world" }
    }
  ],
  "reasoning": "I see a search bar at (512, 340)...",
  "status": "continue",
  "usage": {
    "input_tokens": 1523,
    "output_tokens": 245,
    "credits_charged": 5,
    "cost_cents": 5
  }
}

# Automate YOUR screen — no VM needed. pip install requests mss pyautogui pillow
import base64, contextlib, email.utils, io, json, os, random, time, uuid, requests, mss, pyautogui
from datetime import timezone
from PIL import Image

API = "https://coasty.ai/v1"
KEY = os.environ["COASTY_API_KEY"]  # use a live key for real managed inference
HDRS = {"X-API-Key": KEY, "Content-Type": "application/json"}
pyautogui.FAILSAFE = True          # slam the mouse into a corner to abort instantly

REAL_W, REAL_H = pyautogui.size()  # your actual desktop resolution
SEND_W, SEND_H = 1280, 720         # what we tell the model (SD = 1 credit cheaper)
SX, SY = REAL_W / SEND_W, REAL_H / SEND_H   # scale model coords -> real pixels

def screenshot_b64():
    with mss.mss() as sct:
        shot = sct.grab(sct.monitors[1])                      # primary monitor
        img = Image.frombytes("RGB", shot.size, shot.bgra, "raw", "BGRX")
        img = img.resize((SEND_W, SEND_H))                    # MUST match screen_width/height
        buf = io.BytesIO(); img.save(buf, format="PNG")
        return base64.b64encode(buf.getvalue()).decode()

WINDOW_CLOSE_ACTIONS = {
    "close_window", "window_close", "browser_close", "browser_close_tab",
    "close_tab", "terminal_close",
}
KEY_ALIASES = {
    "esc": "escape", "return": "enter", "control": "ctrl", "ctl": "ctrl",
    "command": "cmd", "super": "cmd", "win": "cmd", "windows": "cmd", "meta": "cmd",
    "option": "alt",
    "ctrl_l": "ctrl", "ctrl_r": "ctrl", "control_l": "ctrl", "control_r": "ctrl",
    "cmd_l": "cmd", "cmd_r": "cmd", "command_l": "cmd", "command_r": "cmd",
    "meta_l": "cmd", "meta_r": "cmd", "super_l": "cmd", "super_r": "cmd",
    "win_l": "cmd", "win_r": "cmd", "windows_l": "cmd", "windows_r": "cmd",
    "alt_l": "alt", "alt_r": "alt", "option_l": "alt", "option_r": "alt",
}
PYAUTOGUI_KEY_ALIASES = {
    "cmd": "command", "meta": "command", "return": "enter",
    "esc": "escape", "option": "alt",
    "ctrl_l": "ctrl", "ctrl_r": "ctrl", "control_l": "ctrl", "control_r": "ctrl",
    "cmd_l": "command", "cmd_r": "command", "command_l": "command", "command_r": "command",
    "meta_l": "command", "meta_r": "command", "super_l": "command", "super_r": "command",
    "win_l": "command", "win_r": "command", "windows_l": "command", "windows_r": "command",
    "alt_l": "alt", "alt_r": "alt", "option_l": "alt", "option_r": "alt",
}

def pyautogui_key(value):
    key = str(value).strip().lower()
    return PYAUTOGUI_KEY_ALIASES.get(key, key)

def type_text_literal(value):
    # PyAutoGUI 0.9.x cannot faithfully type arbitrary Unicode. Never silently
    # drop patient names/notes: use an OS-native Unicode implementation in a
    # production desktop driver, or fail visibly as this portable sample does.
    if not value.isascii():
        raise RuntimeError("type_text contains non-ASCII text; configure an OS-native Unicode typer")
    pyautogui.write(value, interval=0.02)

@contextlib.contextmanager
def held_keys(values):
    keys = [pyautogui_key(value) for value in (values or [])]
    for key in keys:
        pyautogui.keyDown(key)
    try:
        yield
    finally:
        for key in reversed(keys):
            pyautogui.keyUp(key)

def normalized_keys(a):
    p = a.get("params", {})
    values = []
    for field in ("key", "keys", "hold_keys", "modifiers"):
        raw = p.get(field)
        values.extend([raw] if isinstance(raw, str) else (raw or []))
    keys = set()
    for value in values:
        for part in str(value).strip().lower().replace("-", "+").split("+"):
            part = part.strip()
            if part:
                keys.add(KEY_ALIASES.get(part, part))
    return keys

def prohibited(a):
    keys = normalized_keys(a)
    return (
        "escape" in keys
        or a["action_type"] in WINDOW_CLOSE_ACTIONS
        or {"alt", "f4"} <= keys
        or (("ctrl" in keys or "cmd" in keys) and "w" in keys)
        or {"cmd", "q"} <= keys
    )

def execute(a):
    t, p = a["action_type"], a["params"]
    if t in ("done", "fail"):
        return                          # terminal signals have no OS effect
    if t == "click":
        with held_keys(p.get("hold_keys")):
            pyautogui.click(p["x"] * SX, p["y"] * SY,
                            clicks=p.get("clicks", 1), button=p.get("button", "left"))
    elif t == "move":       pyautogui.moveTo(p["x"] * SX, p["y"] * SY)
    elif t == "type_text":  type_text_literal(p["text"])
    elif t == "key_press":  pyautogui.press([pyautogui_key(k) for k in p["keys"]])
    elif t == "key_combo":  pyautogui.hotkey(*[pyautogui_key(k) for k in p["keys"]])
    elif t == "scroll":
        if "x" in p:        pyautogui.moveTo(p["x"] * SX, p["y"] * SY)
        if p.get("direction", "vertical") == "horizontal":
            pyautogui.hscroll(p["clicks"])                    # +right / -left
        else:
            pyautogui.scroll(p["clicks"])                     # +up / -down
    elif t == "drag":
        pyautogui.moveTo(p["x1"] * SX, p["y1"] * SY)
        with held_keys(p.get("hold_keys")):
            pyautogui.dragTo(p["x2"] * SX, p["y2"] * SY, duration=0.4,
                             button=p.get("button", "left"))
    elif t == "wait":       time.sleep(p["seconds"])
    else:                   raise RuntimeError(f"unsupported action: {t}")

def parse_retry_after_seconds(raw):
    if raw is None:
        return None
    try:
        return max(0.0, float(raw))     # delta-seconds
    except (TypeError, ValueError):
        try:                            # IMF-fixdate HTTP-date
            dt = email.utils.parsedate_to_datetime(str(raw))
            if dt.tzinfo is None:
                dt = dt.replace(tzinfo=timezone.utc)
            return max(0.0, dt.timestamp() - time.time())
        except (TypeError, ValueError, OverflowError):
            return None

def retry_after_seconds(res, error):
    # Header and canonical body are independent signals. Honor the largest
    # valid delay so a smaller or malformed intermediary header cannot shorten
    # the server's requested wait.
    candidates = (
        parse_retry_after_seconds(res.headers.get("Retry-After")),
        parse_retry_after_seconds(error.get("retry_after")),
    )
    valid = [value for value in candidates if value is not None]
    return max(valid) if valid else None

def reconcile_idempotency(request_key, max_wait_seconds=600, initial_delay_seconds=0):
    """Collect a lost reserve-and-replay result; never invent a fresh operation."""
    url = f"{API}/idempotency/{request_key}"
    deadline = time.monotonic() + max_wait_seconds
    if initial_delay_seconds > max_wait_seconds:
        raise RuntimeError(
            f"server Retry-After exceeds foreground reconciliation budget; "
            f"resume GET {url} after {initial_delay_seconds:.1f}s"
        )
    if initial_delay_seconds > 0:
        time.sleep(initial_delay_seconds)
    attempt = 0
    while True:
        delay = None
        try:
            res = requests.get(url, headers=HDRS, timeout=15)
        except requests.RequestException:
            res = None
        if res is not None:
            try:
                envelope = res.json()
                if not isinstance(envelope, dict):
                    envelope = {}
                error = envelope.get("error") or {}
                if not isinstance(error, dict):
                    error = {}
            except (ValueError, AttributeError, requests.RequestException):
                envelope, error = {}, {}
            print({
                "reconcile_status": res.status_code,
                "x_request_id": res.headers.get("X-Request-Id"),
                "x_coasty_request_id": res.headers.get("X-Coasty-Request-Id"),
                "body_request_id": error.get("request_id") or envelope.get("request_id"),
                "cf_ray": res.headers.get("CF-Ray"),
                "idempotency_key": request_key,
            })
            if res.ok and envelope.get("status") == "completed":
                original_status = envelope.get("original_status")
                result = envelope.get("result")
                if isinstance(original_status, int) and 200 <= original_status < 300 and isinstance(result, dict):
                    return result
                raise RuntimeError(
                    f"original operation completed with HTTP {original_status}; "
                    f"inspect reconciled result for key={request_key}: {str(result)[:500]}"
                )
            if res.ok and envelope.get("status") != "processing":
                raise RuntimeError(f"malformed idempotency lookup for key={request_key}")
            if res.status_code == 404:
                raise RuntimeError(
                    f"idempotency key={request_key} is unknown or expired; outcome remains "
                    "ambiguous. Do not start a replacement operation until resource state is checked."
                )
            if not res.ok and res.status_code not in (429, 502, 503, 504):
                res.raise_for_status()
            delay = retry_after_seconds(res, error)

        remaining = deadline - time.monotonic()
        jitter = random.uniform(0, min(30, 2 ** min(attempt, 5)))
        delay = max(delay or 0, jitter)
        if remaining <= 0 or delay > remaining:
            raise RuntimeError(
                f"idempotency result still pending for key={request_key}; "
                f"resume GET {url} before any new operation"
            )
        time.sleep(delay)
        attempt += 1

def post_with_retry(url, request_key, payload):
    # These documented POST routes are reserve-and-replay capable: keep the
    # same key and byte-equivalent body for transport/edge retries. Typed
    # failures must also explicitly permit same-key replay.
    body = json.dumps(payload, separators=(",", ":"), ensure_ascii=False)
    first_failure_at = None
    for attempt in range(4):           # initial send + at most 3 retries
        delay = None
        remaining = (
            120.0 if first_failure_at is None
            else 120.0 - (time.monotonic() - first_failure_at)
        )
        if remaining <= 0:
            return reconcile_idempotency(request_key)
        try:
            res = requests.post(
                url,
                headers={**HDRS, "Idempotency-Key": request_key},
                data=body,                 # byte-identical across same-key retries
                timeout=max(0.001, min(120.0, remaining)),
            )
        except requests.RequestException:
            print(f"transport failure attempt={attempt + 1} idempotency_key={request_key}")
        else:
            if res.ok:
                try:
                    result = res.json()
                    if isinstance(result, dict):
                        return result
                except (ValueError, requests.RequestException):
                    pass
                # A 2xx with a truncated/malformed body may have committed. The
                # keyed lookup recovers the stored response without a new POST.
                return reconcile_idempotency(request_key)
            try:
                envelope = res.json()
                if not isinstance(envelope, dict):
                    envelope = {}
                error = envelope.get("error", {})
                if not isinstance(error, dict):
                    error = {}
            except (ValueError, AttributeError, requests.RequestException):
                envelope = {}
                error = {}
            ids = {
                "x_request_id": res.headers.get("X-Request-Id"),
                "x_coasty_request_id": res.headers.get("X-Coasty-Request-Id"),
                "body_request_id": error.get("request_id") or envelope.get("request_id"),
                "cf_ray": res.headers.get("CF-Ray"),
                "idempotency_key": request_key,
            }
            print(f"retry context attempt={attempt + 1}: {ids}")
            if error:
                if error.get("retryable") is not True:
                    res.raise_for_status()
                if error.get("retry_with_same_idempotency_key") is not True:
                    raise RuntimeError(
                        "Retryable response did not permit same-key replay; inspect state "
                        "before starting a new logical operation."
                    )
                if error.get("code") == "IDEMPOTENCY_IN_FLIGHT":
                    return reconcile_idempotency(
                        request_key,
                        initial_delay_seconds=retry_after_seconds(res, error) or 0,
                    )
            elif res.status_code not in (429, 502, 503, 504):
                res.raise_for_status()
            delay = retry_after_seconds(res, error)

        first_failure_at = first_failure_at or time.monotonic()
        delay = delay if delay is not None else random.uniform(0, min(30, 2 ** attempt))
        elapsed = time.monotonic() - first_failure_at
        if attempt == 3 or elapsed + delay > 120:
            # Do not poll before a server-directed Retry-After. Hand the delay
            # to reconciliation; if it exceeds that budget, the helper tells
            # the caller exactly when and where to resume GET polling.
            return reconcile_idempotency(
                request_key,
                initial_delay_seconds=delay,
            )
        time.sleep(delay)

def predict_with_retry(request_key, payload):
    return post_with_retry(f"{API}/sessions/{sid}/predict", request_key, payload)

def delete_session_with_reconciliation(session_id):
    url = f"{API}/sessions/{session_id}"
    deadline = time.monotonic() + 120
    for attempt in range(3):
        server_delay = None
        try:
            deleted = requests.delete(url, headers=HDRS, timeout=15)
        except requests.RequestException:
            deleted = None
        if deleted is not None:
            if deleted.ok or deleted.status_code == 404:
                return
            if deleted.status_code not in (429, 502, 503, 504):
                deleted.raise_for_status()
            try:
                envelope = deleted.json()
                if not isinstance(envelope, dict):
                    envelope = {}
                error = envelope.get("error") or {}
                if not isinstance(error, dict):
                    error = {}
            except (ValueError, AttributeError, requests.RequestException):
                error = {}
            server_delay = retry_after_seconds(deleted, error)
            print({
                "cleanup_status": deleted.status_code,
                "x_request_id": deleted.headers.get("X-Request-Id"),
                "x_coasty_request_id": deleted.headers.get("X-Coasty-Request-Id"),
                "cf_ray": deleted.headers.get("CF-Ray"),
            })
        # A lost DELETE response is ambiguous. GET is authoritative: 404 means
        # cleanup completed; 200 means the slot still exists and DELETE may retry.
        try:
            state = requests.get(url, headers=HDRS, timeout=15)
            if state.status_code == 404:
                return
        except requests.RequestException:
            state = None
        if attempt < 2:
            delay = max(
                server_delay or 0,
                random.uniform(0, min(4, 2 ** attempt)),
            )
            if delay > deadline - time.monotonic():
                raise RuntimeError(
                    f"session cleanup retry exceeds budget; honor Retry-After and resume GET {url}"
                )
            time.sleep(delay)
    raise RuntimeError(f"session cleanup unconfirmed; reconcile GET {url}")

# A live Coasty key gives real managed inference. A test key without explicit
# BYOK provider headers returns deterministic sandbox output rather than a model.
session_payload = {
    "cua_version": "v5",
    "screen_width": SEND_W, "screen_height": SEND_H,
    "action_policy": {
        "allowed_actions": [
            "click", "move", "type_text", "key_press", "key_combo",
            "scroll", "drag", "wait",
        ],
        "blocked_keys": ["escape"],
        "block_window_close": True,
        "max_actions": 5,
    },
}
create_key = f"session-create-{uuid.uuid4().hex}"
sess = post_with_retry(f"{API}/sessions", create_key, session_payload)
sid = sess["session_id"]

micro_goal = "Goal: compute 42 * 17. Expected screen: calculator shows 714. [done] when 714 is visible."
verify_goal = "Verify only whether the calculator visibly shows 714. Return done if visible; otherwise describe the mismatch. Do not click or type."
instruction = micro_goal
completed = False
rejections = 0
try:
    for step in range(25):
        request_key = f"step-{sid}-{step}-{uuid.uuid4().hex[:8]}"
        payload = {"screenshot": screenshot_b64(), "instruction": instruction}
        r = predict_with_retry(request_key, payload)
        print(f"step {r['step']}: {r['reasoning'][:80]}")

        policy_failure = next((
            a for a in r["actions"]
            if a["action_type"] == "fail"
            and a.get("params", {}).get("code") == "ACTION_POLICY_VIOLATION"
        ), None)
        rejected = next((a for a in r["actions"] if prohibited(a)), None)
        if policy_failure or rejected:
            # Server policy rejects its whole batch. The client check is defense
            # in depth. Execute NOTHING, observe again, and explain the rejection
            # to this same session with a new key.
            rejections += 1
            if rejections >= 3:
                raise RuntimeError(
                    "three prohibited-action proposals; human review required"
                )
            instruction = (
                "CONTROL FEEDBACK: the previous prohibited action was rejected and "
                "NOT executed. The next screenshot is authoritative. Continue without "
                "Escape or closing any window."
            )
            time.sleep(0.5)
            continue
        rejections = 0
        if r["status"] == "fail":
            raise RuntimeError(f"agent failed: {r['reasoning']}")

        for action_index, a in enumerate(r["actions"]):
            # A production driver durably journals (request_key, action_index)
            # as PREPARED before input and COMPLETED after it. On restart, a
            # leftover PREPARED action is ambiguous: capture a fresh screenshot
            # and re-plan; never blindly execute that action again.
            execute(a)
        time.sleep(0.5)                                       # let the UI settle

        if r["status"] == "done":
            # done is a model claim. Verify from pixels captured AFTER the last
            # action; also assert application/DOM state here when available.
            verify_payload = {"screenshot": screenshot_b64(), "instruction": verify_goal}
            verified = predict_with_retry(f"verify-{sid}-{uuid.uuid4().hex}", verify_payload)
            mutating = [
                a for a in verified["actions"]
                if a["action_type"] not in ("done", "fail")
            ]
            if mutating or verified["status"] != "done":
                raise RuntimeError(f"completion mismatch: {verified['reasoning']}")
            completed = True
            print("finished and freshly verified")
            break
        instruction = micro_goal
    if not completed:
        raise RuntimeError("step cap reached before verified completion")
finally:
    delete_session_with_reconciliation(sid)                   # stop the session clock

Be precise. Before clicking, confirm the target element is actually visible in the CURRENT screenshot — never click from memory of a previous screen. Click the visual center of elements, not their edges. If the element you need is not visible, scroll toward where it should be instead of guessing coordinates. If two elements look similar, prefer the one whose text matches the task exactly. After typing into a field, verify focus landed in the right field before continuing.

requests.post(f"{API}/sessions", headers=HDRS, json={
    "cua_version": "v5",
    "screen_width": 1280, "screen_height": 720,
    "instructions": PRESET,   # the text above — applies to every step in the session
})

import requests

# Provision a fresh Linux desktop VM. Sandbox keys (sk-coasty-test-*)
# return a mock machine instantly with no billing.
r = requests.post(
    "https://coasty.ai/v1/machines",
    headers={
        "X-API-Key": "sk-coasty-live-...",
        "Idempotency-Key": "provision-bot-001",   # safe to retry
    },
    json={
        "display_name": "automation-bot",
        "os_type": "linux",
        "desktop_enabled": True,
    },
)
machine = r.json()["machine"]
print(machine["id"], machine["status"])

import requests

machine_id = "..."  # from provision response
r = requests.post(
    f"https://coasty.ai/v1/machines/{machine_id}/actions",
    headers={"X-API-Key": "sk-coasty-live-..."},
    json={
        "command": "click",
        "parameters": {"x": 512, "y": 340},
    },
)
result = r.json()
print(result["success"], result["duration_ms"], "ms")

import requests

# Run a shell command (PowerShell on Windows, bash on Linux).
# Output is truncated VM-side to 5000 chars.
r = requests.post(
    f"https://coasty.ai/v1/machines/{machine_id}/terminal",
    headers={"X-API-Key": "sk-coasty-live-..."},
    json={
        "command": "uname -a && uptime",
        "timeout_ms": 10_000,
    },
)
print(r.json()["result"]["output"])

import requests

API_KEY = "sk-coasty-live-..."

# Start a task run. Create/exact-replay responses include webhook_secret
# when webhook_url is present; GET/list do not. Idempotency-Key deduplicates this
# top-level run creation; it does not make the run's later OS actions exactly once.
r = requests.post(
    "https://coasty.ai/v1/runs",
    headers={
        "X-API-Key": API_KEY,
        "Idempotency-Key": "invoice-run-4821",
    },
    json={
        "machine_id": "550e8400-e29b-41d4-a716-446655440000",
        "task": "Open the invoice in the browser and read the total.",
        "cua_version": "v5",          # any of v1/v3/v4/v5, all tiers; omit to use the v5 default
        "max_steps": 40,
        "on_awaiting_human": "pause", # pause | fail | cancel
        "webhook_url": "https://your.app/hooks/coasty",
    },
)
run = r.json()
print(run["id"], run["status"])               # ... queued
webhook_secret = run.get("webhook_secret")    # create/replay response only — store it now

import requests

# Stream lifecycle as Server-Sent Events. Reconnect with Last-Event-ID to
# replay every successfully persisted frame after the last seq you saw.
# Intermediate appends are best-effort; GET the run for authoritative state.
with requests.get(
    f"https://coasty.ai/v1/runs/{run_id}/events",
    headers={"X-API-Key": API_KEY, "Last-Event-ID": "42"},
    stream=True,
) as r:
    for line in r.iter_lines(decode_unicode=True):
        if line.startswith("data:"):
            print(line[5:].strip())   # status / step / awaiting_human / terminal

import requests

# A workflow is a versioned JSON DSL composing many runs with branching,
# loops, parallelism, asserts, retries, and human approvals. {{var}} pulls
# from earlier steps' results (bound via save_as / step id).
r = requests.post(
    "https://coasty.ai/v1/workflows",
    headers={
        "X-API-Key": "sk-coasty-live-...",
        "Idempotency-Key": "ar-collections-v1",
    },
    json={
        "name": "AR collections",
        "definition": {
            "steps": [
                {"id": "invoice", "type": "task", "save_as": "invoice",
                 "machine_id": "550e8400-e29b-41d4-a716-446655440000",
                 "task": "Open the invoice and read its status + total."},
                {"id": "check", "type": "assert",
                 "condition": {"op": "truthy", "value": "{{invoice.passed}}"},
                 "message": "Agent failed to read the invoice"},
                {"id": "branch", "type": "if",
                 "condition": {"op": "contains",
                               "left": "{{invoice.result}}", "right": "PAID"},
                 "then": [{"id": "ok", "type": "succeed",
                           "output": {"state": "paid"}}],
                 "else": [{"id": "ask", "type": "human_approval",
                           "message": "Invoice unpaid — send reminder?"},
                          {"id": "fin", "type": "succeed",
                           "output": {"state": "reminded"}}]},
            ]
        },
    },
)
workflow = r.json()
print(workflow["id"], workflow["version"])

import requests

# Daily 9:00 AM ET email summary, fired by the Coasty scheduler.
# Published defaults per fire: >= 20 cr ($0.20) in your API wallet to dispatch (gate only),
# then 10 subscription credits/min. Read GET /v1/models pricing.schedules for effective values.
r = requests.post(
    "https://coasty.ai/v1/schedules",
    headers={
        "X-API-Key": "sk-coasty-live-...",
        "Idempotency-Key": "morning-briefing-001",
    },
    json={
        "name": "morning briefing",
        "machine_id": "550e8400-e29b-41d4-a716-446655440000",
        "task_prompt": "Summarize unread Gmail and post the top 5 to Slack.",
        "frequency": "daily",
        "time": "09:00",
        "timezone": "America/New_York",
    },
)
schedule = r.json()
print(schedule["id"], schedule["next_run_at"])

# Add a webhook trigger — returns the signing secret ONCE.
r = requests.post(
    f"https://coasty.ai/v1/schedules/{schedule_id}/triggers",
    headers={"X-API-Key": "sk-coasty-live-..."},
    json={"kind": "webhook", "rate_limit_per_minute": 60},
)
trigger = r.json()
webhook_url    = trigger["webhook_url"]      # https://coasty.ai/v1/triggers/webhook/whk_...
webhook_secret = trigger["webhook_secret"]   # whsec_<64 hex>  — STORE THIS

# Save webhook_secret in your secrets manager. Coasty encrypts it at rest.
# Exact bounded create replays return it; GET/list never do.

# Customer-side webhook signing — produces a Coasty-Signature header
# the public /v1/triggers/webhook/{id} endpoint accepts.
import hmac, hashlib, time

def sign_coasty_webhook(secret: str, body: bytes) -> dict:
    ts = int(time.time())
    signed_payload = f"{ts}.".encode("utf-8") + body
    sig = hmac.new(secret.encode("utf-8"), signed_payload, hashlib.sha256).hexdigest()
    return {"Coasty-Signature": f"t={ts},v1={sig}"}

# Fire the webhook from your own app:
import requests
body = b'{"event":"order.placed","order_id":"123"}'
headers = {**sign_coasty_webhook(webhook_secret, body), "Content-Type": "application/json"}
requests.post(webhook_url, data=body, headers=headers)