#!/usr/bin/env python3 """Quattrino sandbox bootstrap - the OWNER side of the quickstart, via the API. Creates (or reuses) in your Quattrino organization, TEST/SANDBOX only: 1. an agent (the thing that will call authorize) 2. a standing policy (its standing authority: $50 per action by default) 3. a sandbox wallet (test money; nothing real can move) 4. a scoped agent API key (the credential your code / MCP client uses) Everything here can also be done by hand in the web app (Getting started -> Connect agent -> Connect money -> Set limits). This script just makes the 15-minute quickstart reproducible. Usage: export QUATTRINO_BASE=https://quattrino.io # or your runtime export QUATTRINO_EMAIL=you@company.com export QUATTRINO_PASSWORD='...' python bootstrap_sandbox.py [--standing-cents 5000] [--agent-name "My agent"] [--signup] Writes ./.quattrino_sandbox.json with {base, agent_id, api_key, policy_id, wallet_id, owner_email}. The api_key is shown ONCE by the API; keep the file private. Only dependency: the Python standard library. """ import argparse import getpass import json import os import sys import time import urllib.error import urllib.request def _call(base, method, path, body=None, token=None, headers=None): data = json.dumps(body).encode() if body is not None else None req = urllib.request.Request(base + "/api/v1" + path, data=data, method=method) req.add_header("Content-Type", "application/json") if token: req.add_header("Authorization", "Bearer " + token) for k, v in (headers or {}).items(): req.add_header(k, v) try: with urllib.request.urlopen(req, timeout=30) as r: return r.status, json.loads(r.read() or b"{}") except urllib.error.HTTPError as e: try: return e.code, json.loads(e.read() or b"{}") except ValueError: return e.code, {"error": {"code": "http_" + str(e.code)}} def _err(payload): return ((payload or {}).get("error") or {}).get("code") or (payload or {}).get("detail") or payload def main(): ap = argparse.ArgumentParser() ap.add_argument("--base", default=os.environ.get("QUATTRINO_BASE", "https://quattrino.io")) ap.add_argument("--email", default=os.environ.get("QUATTRINO_EMAIL")) ap.add_argument("--password", default=os.environ.get("QUATTRINO_PASSWORD")) ap.add_argument("--signup", action="store_true", help="create the account if it does not exist") ap.add_argument("--org-name", default=None) ap.add_argument("--agent-name", default="Quickstart agent") ap.add_argument("--standing-cents", type=int, default=5000, help="standing per-action ceiling (minor units)") ap.add_argument("--sandbox-authority-cents", type=int, default=100000, help="sandbox test money on the wallet") ap.add_argument("--out", default=".quattrino_sandbox.json") a = ap.parse_args() base = a.base.rstrip("/") email = a.email or input("Quattrino email: ").strip() password = a.password or getpass.getpass("Quattrino password: ") t0 = time.time() # ---- 0. sign in (or sign up) ------------------------------------------------ st, j = _call(base, "POST", "/auth/login", {"email": email, "password": password}) if st == 200 and j.get("mfa_required"): sys.exit("This account has two-factor authentication enabled. Sign in through the web app and create the " "agent + key there (Agents -> New agent -> Issue key), then use quickstart.py with that key.") if st != 200 and a.signup: st, j = _call(base, "POST", "/auth/signup", { "email": email, "password": password, "name": email.split("@")[0], "organization_name": a.org_name or (email.split("@")[1].split(".")[0].title() + " Sandbox"), "accepted_documents": ["TERMS_OF_SERVICE", "PRIVACY_POLICY"]}) if st not in (200, 201): sys.exit("signup failed: " + json.dumps(_err(j))) print("account created for", email) if st != 200 or not j.get("access_token"): sys.exit("login failed (" + str(st) + "): " + json.dumps(_err(j)) + "\n" "Hint: add --signup to create the account, or sign in on the website first.") tok = j["access_token"] print("signed in as", email) # ---- 1. agent ---------------------------------------------------------------- st, j = _call(base, "GET", "/agents", token=tok) agent = next((x for x in (j.get("agents") or []) if x.get("name") == a.agent_name), None) if not agent: st, j = _call(base, "POST", "/agents", {"name": a.agent_name, "description": "Quattrino quickstart", "environment": "sandbox"}, token=tok) if st != 200: sys.exit("agent create failed: " + json.dumps(_err(j))) agent = j["agent"] print("agent created:", agent["agent_id"]) else: print("agent reused:", agent["agent_id"]) # ---- 2. standing policy (its standing authority) --------------------------- policy_id = agent.get("default_policy_id") if not policy_id: st, j = _call(base, "POST", "/policies", { "name": "Quickstart standing authority ($%.2f per action)" % (a.standing_cents / 100), "currency": "USD", "transaction_limit": a.standing_cents, # per-action ceiling = standing authority "transaction_auto_limit": a.standing_cents, # autonomous ceiling (above -> APPROVAL_REQUIRED) "daily_limit": a.standing_cents * 20, "monthly_limit": a.standing_cents * 200, "interpretation": "Quickstart: the agent may spend up to the standing ceiling per action; " "tasks may narrow it further, never widen it."}, token=tok) if st != 200: sys.exit("policy create failed: " + json.dumps(_err(j))) policy_id = j["policy"]["policy_id"] st, j = _call(base, "POST", "/policies/" + policy_id + "/activate", {"confirmed": True}, token=tok) if st != 200: sys.exit("policy activate failed: " + json.dumps(_err(j))) print("policy active:", policy_id, "standing ceiling $%.2f" % (a.standing_cents / 100)) else: print("policy reused:", policy_id) # ---- 3. sandbox wallet (test money) --------------------------------------- wallet_id = agent.get("wallet_id") if not wallet_id: st, j = _call(base, "POST", "/wallets", { "display_name": a.agent_name + " sandbox wallet", "wallet_type": "agent", "currency": "USD", "authority_minor": a.sandbox_authority_cents, "agent_id": agent["agent_id"], "default_policy_id": policy_id, "environment": "sandbox"}, token=tok) if st != 200: sys.exit("wallet create failed: " + json.dumps(_err(j))) wallet_id = j["wallet"]["wallet_id"] print("sandbox wallet:", wallet_id) if agent.get("default_policy_id") != policy_id: _call(base, "PATCH", "/agents/" + agent["agent_id"], {"default_policy_id": policy_id}, token=tok) # ---- 4. scoped agent credential (shown once) ------------------------------- st, j = _call(base, "POST", "/agents/" + agent["agent_id"] + "/credentials", {"name": "quickstart " + time.strftime("%Y-%m-%d %H:%M")}, token=tok) if st != 200: sys.exit("credential create failed: " + json.dumps(_err(j))) api_key = j["credential"]["api_key"] out = {"base": base, "owner_email": email, "agent_id": agent["agent_id"], "api_key": api_key, "policy_id": policy_id, "wallet_id": wallet_id, "standing_cents": a.standing_cents, "environment": "sandbox"} with open(a.out, "w") as f: json.dump(out, f, indent=2) print("\nready in %.1fs -> %s" % (time.time() - t0, a.out)) print("agent key (shown once):", api_key[:18] + "...") print("next: python quickstart.py (first authorization)") print(" python differentiation_journeys.py (journeys A-G)") if __name__ == "__main__": main()