Build a robo-advisor with Hermes agents and Robinhood Chain

---
name: portfolio-wallet
description: "The user's portfolio wallet: the Privy wallet in the <your-app-name> app that holds their onchain stock positions. Use whenever the user asks to connect their wallet, check holdings, or buy or sell positions."
version: 1.0.0
platforms: [macos]
---

# Portfolio wallet: agent authorization and trading

Gives the agent authorized access to the user's wallet in the <your-app-name> Privy app
using the OAuth 2.0 Device Authorization Grant, then signs and sends trades on Robinhood
Chain via Privy's wallet RPC. No app secret lives on this machine.

**App config:**

- App ID: `<your-app-id>`
- Verification page: `https://your-app.com/authorize` (returned by the API)
- API base: `https://auth.privy.io`
- Trading chain: Robinhood Chain (`chain_id` 4663, CAIP-2 `eip155:4663`)

All logic lives in `scripts/privy_agent.py`. Tokens are stored in the macOS Keychain
(service `privy-agent`), never in plaintext files.

## Safety rules (non-negotiable)

1. **Always get explicit user confirmation before any trade** (`eth_sendTransaction` or
   anything that moves funds). Show the user the asset, share count, destination, value,
   and chain, then wait for a clear "yes".
2. Never print, log, or echo tokens or decrypted authorization keys.
3. If the user asks to revoke access, run `logout` locally and point them to their
   account settings in the app.

## Commands

All commands run via: `python3 <skill_dir>/scripts/privy_agent.py <command>`

### `login`: run the device authorization flow

    python3 -u scripts/privy_agent.py login > /tmp/privy-login.log 2>&1

Requests a device code, prints the verification URL and user code, then polls until the
user approves. Send the verification link to the user on Telegram, formatted as a consent
prompt:

    🔐 Authorization Request

    Hermes is requesting access to your portfolio wallet.

    This will allow the agent to:
      • View your wallet address and holdings
      • Sign messages on your behalf
      • Execute trades (with your confirmation)

    👉 Approve here: <verification_uri_complete>
    Verification code: XXXXX-XXXXX
    ⏱ Expires in 10 minutes.

On approval the script stores tokens in the Keychain and prints the wallet list.

### `status`: check auth state and list wallets

    python3 scripts/privy_agent.py status

Prints whether tokens exist, refreshes if needed, and lists wallets (id, address,
chain_type). Use the Privy wallet `id` (e.g. `wallet_abc123`), not the 0x address, in RPC
calls.

### `rpc`: sign or trade with a wallet

    # Buy an asset on Robinhood Chain (CONFIRM WITH USER FIRST!)
    python3 scripts/privy_agent.py rpc <wallet_id> '{"method":"eth_sendTransaction","params":{"transaction":{"to":"0xAssetContract","value":"0x0","data":"0x...","chain_id":4663}}}'

    # Sign a message
    python3 scripts/privy_agent.py rpc <wallet_id> '{"method":"personal_sign","params":{"message":"Hello from Hermes"}}'

### `logout`: drop stored tokens

    python3 scripts/privy_agent.py logout

#!/usr/bin/env python3
"""Portfolio agent wallet access: device-flow auth + wallet RPC.

Implements https://docs.privy.io/recipes/agent-integrations/agent-authorization
- OAuth 2.0 Device Authorization Grant against https://auth.privy.io
- Tokens stored in the macOS Keychain (service: privy-agent)
- Wallet RPC with an HPKE-decrypted ephemeral authorization key and
  RFC 8785 canonical JSON + ECDSA P-256 request signatures.

Commands: login | status | rpc <wallet_id> <json_body> | logout
"""
import base64
import functools
import json
import os
import subprocess
import sys
import time
import urllib.error
import urllib.request

print = functools.partial(print, flush=True)  # unbuffered for background use

APP_ID = os.environ["PRIVY_APP_ID"]
BASE = "https://auth.privy.io"
KEYCHAIN_SERVICE = "privy-agent"
KEYCHAIN_ACCOUNT = APP_ID


# ---------------------------------------------------------------- keychain --
def kc_get():
    try:
        out = subprocess.run(
            ["security", "find-generic-password", "-s", KEYCHAIN_SERVICE,
             "-a", KEYCHAIN_ACCOUNT, "-w"],
            capture_output=True, text=True, check=True)
        return json.loads(out.stdout.strip())
    except (subprocess.CalledProcessError, json.JSONDecodeError):
        return None


def kc_set(tokens: dict):
    subprocess.run(
        ["security", "add-generic-password", "-U", "-s", KEYCHAIN_SERVICE,
         "-a", KEYCHAIN_ACCOUNT, "-w", json.dumps(tokens)],
        capture_output=True, check=True)


def kc_delete():
    subprocess.run(
        ["security", "delete-generic-password", "-s", KEYCHAIN_SERVICE,
         "-a", KEYCHAIN_ACCOUNT],
        capture_output=True)


# -------------------------------------------------------------------- http --
def post(path, body, headers=None, bearer=None):
    # Send a custom User-Agent: Cloudflare blocks Python's default UA on
    # auth.privy.io with a 403 (error code 1010).
    h = {"Content-Type": "application/json", "privy-app-id": APP_ID,
         "User-Agent": "portfolio-agent/1.0"}
    if bearer:
        h["Authorization"] = f"Bearer {bearer}"
    if headers:
        h.update(headers)
    req = urllib.request.Request(BASE + path, data=json.dumps(body).encode(),
                                 headers=h, method="POST")
    try:
        with urllib.request.urlopen(req, timeout=30) as r:
            return r.status, json.loads(r.read().decode())
    except urllib.error.HTTPError as e:
        try:
            return e.code, json.loads(e.read().decode())
        except Exception:
            return e.code, {"error": "http_error", "detail": str(e)}


# ------------------------------------------------------------------ tokens --
def now():
    return int(time.time())


def refresh_tokens(tokens):
    status, data = post("/api/oauth/v2/token",
                        {"grant_type": "refresh_token",
                         "refresh_token": tokens["refresh_token"]})
    if status != 200:
        if data.get("error") == "access_denied":
            kc_delete()
            sys.exit("ERROR: refresh token expired or access revoked. "
                     "Run 'login' again; user must re-approve.")
        sys.exit(f"ERROR: token refresh failed ({status}): {data}")
    tokens = {
        "access_token": data["access_token"],
        "refresh_token": data["refresh_token"],  # rotates on every use; store the newest
        "expires_at": now() + int(data.get("expires_in", 900)),
    }
    kc_set(tokens)
    return tokens


def get_valid_tokens():
    tokens = kc_get()
    if not tokens:
        sys.exit("ERROR: not authorized. Run 'login' first.")
    if now() >= tokens.get("expires_at", 0) - 60:
        tokens = refresh_tokens(tokens)
    return tokens


# ------------------------------------------------------------------- login --
def cmd_login():
    status, data = post("/api/oauth/v2/device_authorization", {})
    if status == 403 and data.get("error") == "device_auth_not_enabled":
        sys.exit("ERROR: device auth not enabled in Privy Dashboard "
                 "(Authentication -> Advanced -> Enable for CLI and agent access)")
    if status != 200:
        sys.exit(f"ERROR: device_authorization failed ({status}): {data}")

    device_code = data["device_code"]
    interval = int(data.get("interval", 5))
    expires_in = int(data.get("expires_in", 600))

    print("=== AUTHORIZATION REQUIRED ===")
    print(f"URL:  {data['verification_uri_complete']}")
    print(f"Code: {data['user_code']}")
    print(f"(expires in {expires_in // 60} minutes)")
    print("Waiting for user approval...")

    deadline = now() + expires_in
    while now() < deadline:
        time.sleep(interval)
        # The poll grant_type MUST be the RFC 8628 URN, not the bare
        # "device_code" string.
        status, tok = post("/api/oauth/v2/token",
                           {"grant_type":
                            "urn:ietf:params:oauth:grant-type:device_code",
                            "device_code": device_code})
        if status == 200:
            kc_set({
                "access_token": tok["access_token"],
                "refresh_token": tok["refresh_token"],
                "expires_at": now() + int(tok.get("expires_in", 900)),
            })
            print("APPROVED. Tokens stored in Keychain.")
            cmd_status()
            return
        err = tok.get("error", "")
        if err == "authorization_pending":
            continue
        if err == "slow_down":
            interval += 5
            continue
        if err == "expired_token":
            sys.exit("ERROR: device code expired. Run 'login' again.")
        if err == "access_denied":
            sys.exit("ERROR: user denied the authorization request.")
        sys.exit(f"ERROR: unexpected polling response ({status}): {tok}")
    sys.exit("ERROR: device code expired (timeout). Run 'login' again.")


# ------------------------------------------------------------ hpke + p-256 --
def gen_p256_keypair():
    from cryptography.hazmat.primitives.asymmetric import ec
    from cryptography.hazmat.primitives import serialization
    priv = ec.generate_private_key(ec.SECP256R1())
    spki = priv.public_key().public_bytes(
        serialization.Encoding.DER,
        serialization.PublicFormat.SubjectPublicKeyInfo)
    return priv, base64.b64encode(spki).decode()


def hpke_decrypt(priv, encapsulated_key_b64, ciphertext_b64):
    """Decrypt the authorization key. Privy uses DHKEM-P256 + HKDF-SHA256;
    AEAD is ChaCha20Poly1305, with AES-GCM tried as a fallback."""
    from pyhpke import AEADId, CipherSuite, KDFId, KEMId, KEMKey
    from cryptography.hazmat.primitives import serialization
    enc = base64.b64decode(encapsulated_key_b64)
    ct = base64.b64decode(ciphertext_b64)
    pem = priv.private_bytes(
        serialization.Encoding.PEM,
        serialization.PrivateFormat.PKCS8,
        serialization.NoEncryption()).decode()
    last_err = None
    for aead in (AEADId.CHACHA20_POLY1305, AEADId.AES256_GCM, AEADId.AES128_GCM):
        try:
            suite = CipherSuite.new(KEMId.DHKEM_P256_HKDF_SHA256,
                                    KDFId.HKDF_SHA256, aead)
            recipient = suite.create_recipient_context(enc, KEMKey.from_pem(pem))
            return recipient.open(ct).decode()
        except Exception as e:
            last_err = e
    raise RuntimeError(f"HPKE decryption failed with all AEADs: {last_err}")


def canonicalize(obj):
    """RFC 8785 (JCS) canonical JSON: sorted keys + compact separators."""
    return json.dumps(obj, sort_keys=True, separators=(",", ":"),
                      ensure_ascii=False)


def sign_payload(auth_key_str, payload_obj):
    """ECDSA P-256 / SHA-256 over canonical JSON; DER signature, base64."""
    from cryptography.hazmat.primitives.asymmetric import ec
    from cryptography.hazmat.primitives import hashes, serialization
    # Authorization keys arrive prefixed with "wallet-auth:"; strip it.
    key_b64 = auth_key_str.replace("wallet-auth:", "").strip()
    priv = serialization.load_der_private_key(base64.b64decode(key_b64),
                                              password=None)
    sig = priv.sign(canonicalize(payload_obj).encode(),
                    ec.ECDSA(hashes.SHA256()))
    return base64.b64encode(sig).decode()


# ----------------------------------------------------------- authenticate --
def wallet_authenticate(tokens):
    priv, pub_spki_b64 = gen_p256_keypair()
    body = {"encryption_type": "HPKE", "recipient_public_key": pub_spki_b64}
    hdrs = {"privy-grant-type": "device_code"}
    status, data = post("/api/oauth/v2/wallets/authenticate", body,
                        headers=hdrs, bearer=tokens["access_token"])
    if status == 401:
        tokens = refresh_tokens(tokens)
        status, data = post("/api/oauth/v2/wallets/authenticate", body,
                            headers=hdrs, bearer=tokens["access_token"])
    if status != 200:
        sys.exit(f"ERROR: wallets/authenticate failed ({status}): {data}")
    ek = data["encrypted_authorization_key"]
    auth_key = hpke_decrypt(priv, ek["encapsulated_key"], ek["ciphertext"])
    return tokens, auth_key, data.get("wallets", [])


# ------------------------------------------------------------------ status --
def cmd_status():
    tokens = get_valid_tokens()
    _tokens, _auth_key, wallets = wallet_authenticate(tokens)
    print("Authorized: YES")
    print(f"Wallets ({len(wallets)}):")
    for w in wallets:
        print(f"  id={w.get('id')}  address={w.get('address')}  "
              f"chain={w.get('chain_type')}")


# --------------------------------------------------------------------- rpc --
def cmd_rpc(wallet_id, body_json):
    body = json.loads(body_json)
    tokens = get_valid_tokens()
    tokens, auth_key, _wallets = wallet_authenticate(tokens)

    url = f"{BASE}/api/oauth/v2/wallets/{wallet_id}/rpc"
    # The signature payload includes ONLY privy-prefixed headers.
    payload = {"version": 1, "method": "POST", "url": url, "body": body,
               "headers": {"privy-app-id": APP_ID}}
    signature = sign_payload(auth_key, payload)

    rpc_headers = {"privy-grant-type": "device_code",
                   "privy-authorization-signature": signature}
    status, data = post(f"/api/oauth/v2/wallets/{wallet_id}/rpc", body,
                        headers=rpc_headers, bearer=tokens["access_token"])
    if status == 401:
        tokens = refresh_tokens(tokens)
        tokens, auth_key, _ = wallet_authenticate(tokens)
        rpc_headers["privy-authorization-signature"] = sign_payload(auth_key, payload)
        status, data = post(f"/api/oauth/v2/wallets/{wallet_id}/rpc", body,
                            headers=rpc_headers, bearer=tokens["access_token"])
    print(json.dumps({"http_status": status, "response": data}, indent=2))
    if status != 200:
        sys.exit(1)


# -------------------------------------------------------------------- main --
def main():
    if len(sys.argv) < 2:
        sys.exit(__doc__)
    cmd = sys.argv[1]
    if cmd == "login":
        cmd_login()
    elif cmd == "status":
        cmd_status()
    elif cmd == "rpc":
        if len(sys.argv) < 4:
            sys.exit("usage: privy_agent.py rpc <wallet_id> '<json_body>'")
        cmd_rpc(sys.argv[2], sys.argv[3])
    elif cmd == "logout":
        kc_delete()
        print("Tokens removed from Keychain.")
    else:
        sys.exit(f"unknown command: {cmd}\n{__doc__}")


if __name__ == "__main__":
    main()