Arcade.dev Python 教程：用户授权、连接归属与敏感操作审批

python3 -m venv .venv
source .venv/bin/activate
python -m pip install arcadepy==1.10.0

from arcadepy import Arcade

client = Arcade(max_retries=0, timeout=20.0)
definition = client.tools.get("Gmail.SendEmail")
version = definition.toolkit.version
if not version:
    raise ValueError("无法确定工具版本，停止接入")
#在受控开发环境审核 definition.input，不把用户数据或凭据打印进公共日志。

"""Single-host example. Caller must supply authenticated server-side principals."""
from dataclasses import dataclass
import hashlib
import json
import sqlite3
import time
import uuid

TOOL = 'Gmail.SendEmail'


@dataclass(frozen=True)
class Principal:
    tenant: str
    user: str
    arcade_user: str
    can_approve: bool = False


def db(path):
    connection = sqlite3.connect(path, timeout=10)
    connection.row_factory = sqlite3.Row
    return connection


def init(path):
    with db(path) as connection:
        connection.execute('''CREATE TABLE IF NOT EXISTS actions(
          id TEXT PRIMARY KEY,tenant TEXT,user TEXT,request_id TEXT,
          snapshot TEXT,digest TEXT,state TEXT,expires REAL,receipt TEXT,
          UNIQUE(tenant,user,request_id))''')


def begin_auth(client, principal, version):
    return client.tools.authorize(tool_name=TOOL, tool_version=version,
                                  user_id=principal.arcade_user)


def confirm_identity(client, principal, flow_id):
    if not flow_id or len(flow_id) > 512:
        raise ValueError('invalid flow')
    result = client.auth.confirm_user(flow_id=flow_id, user_id=principal.arcade_user)
    status = client.auth.status(id=result.auth_id)
    if status.user_id != principal.arcade_user:
        raise ValueError('authorization belongs to another identity')
    return status.status


def stage(path, principal, request_id, version, arguments, allowed_recipients):
    if (not request_id or not version
        or set(arguments) != {'recipient','subject','body'}
        or not all(isinstance(v,str) and v.strip() for v in arguments.values())
        or arguments['recipient'] not in allowed_recipients):
        raise ValueError('input rejected')
    snapshot = json.dumps({
        'tenant':principal.tenant, 'user':principal.user,
        'arcade_user':principal.arcade_user, 'tool':TOOL, 'version':version,
        'input':arguments,
    }, ensure_ascii=False,sort_keys=True,separators=(',',':'))
    digest = hashlib.sha256(snapshot.encode()).hexdigest()
    action_id = uuid.uuid4().hex
    with db(path) as connection:
        connection.execute('INSERT INTO actions VALUES (?,?,?,?,?,?,?,NULL,NULL)',
                           (action_id,principal.tenant,principal.user,request_id,
                            snapshot,digest,'pending'))
    return action_id,digest


def approve(path, principal, action_id, displayed_digest):
    if not principal.can_approve:
        raise ValueError('review permission required')
    with db(path) as connection:
        changed = connection.execute('''UPDATE actions SET state='approved',expires=?
          WHERE id=? AND tenant=? AND user=? AND digest=? AND state='pending' ''',
          (time.time()+900,action_id,principal.tenant,principal.user,displayed_digest)).rowcount
        if changed != 1:
            raise ValueError('stale approval or wrong owner')


def execute(path, principal, action_id, allowed_recipients, client):
    with db(path) as connection:
        connection.execute('BEGIN IMMEDIATE')
        row = connection.execute('''SELECT * FROM actions
          WHERE id=? AND tenant=? AND user=?''',
          (action_id,principal.tenant,principal.user)).fetchone()
        if not row or row['state'] != 'approved' or row['expires'] < time.time():
            raise ValueError('not approved or expired')
        if hashlib.sha256(row['snapshot'].encode()).hexdigest() != row['digest']:
            raise ValueError('snapshot changed')
        snapshot = json.loads(row['snapshot'])
        if (snapshot['arcade_user'] != principal.arcade_user
            or snapshot['tool'] != TOOL
            or snapshot['input']['recipient'] not in allowed_recipients):
            raise ValueError('identity or policy changed')
        # Claim locally before any external side effect.
        connection.execute("UPDATE actions SET state='executing' WHERE id=?",(action_id,))
    try:
        auth = begin_auth(client,principal,snapshot['version'])
        if auth.status != 'completed' or auth.user_id != principal.arcade_user:
            raise ValueError('account authorization not completed')
        result = client.tools.execute(
            tool_name=snapshot['tool'],tool_version=snapshot['version'],
            user_id=principal.arcade_user,input=snapshot['input'])
        if result.success is not True or not result.output or result.output.error:
            raise ValueError('execution failed or ambiguous')
        receipt = json.dumps({'execution_id':result.execution_id,'id':result.id})
        with db(path) as connection:
            connection.execute("UPDATE actions SET state='completed',receipt=? WHERE id=?",
                               (receipt,action_id))
    except Exception:
        with db(path) as connection:
            connection.execute("UPDATE actions SET state='unknown' WHERE id=?",(action_id,))
        raise
    return receipt

from approval import init, begin_auth, stage, approve

init("actions.sqlite3")
auth = begin_auth(client, authenticated_principal, reviewed_version)
if auth.status != "completed":
    #仅向当前用户展示 auth.url，等待其完成授权。
    raise RuntimeError("账户授权尚未完成")

action_id, digest = stage(
    "actions.sqlite3", authenticated_principal, business_request_id,
    reviewed_version, reviewed_arguments, server_recipient_allowlist,
)
#在已鉴权 UI 展示完整身份、工具、版本和参数，用户明确确认后才调用。
approve("actions.sqlite3", authenticated_principal, action_id, digest)

from approval import execute

#真实接入会调用外部 Gmail，执行前必须确认此 action_id 已按业务流程批准。
receipt = execute(
    "actions.sqlite3", authenticated_principal, action_id,
    server_recipient_allowlist, client,
)