Arcade.dev Python 教程：2026 年用户授权、连接归属、敏感操作审批与工具执行实操指南

给 Agent 接上 Gmail 等第三方账户时，先验证登录身份与账户归属，再决定当前工具能否执行。本文以 Arcade.dev 为例，把用户 OAuth、生产身份确认和敏感操作审批串起来，用 Python 与 SQLite 保存一个明确的批准版本，并处理重复请求和结果不明的外部调用。

核对日期：2026 年 10 月 16 日。 示例固定 arcadepy 1.10.0。已用官方 SDK 和 httpx.MockTransport 做 12 组离线检查；没有创建真实 Arcade 账户、连接 Gmail、运行真实 OAuth 或发送邮件。示例是单机应用核心，不包含登录系统、公开回调路由或完整审批 UI。

1. 三种身份与两次决定

应用用户：从已验证的登录会话取得本地 user 和 tenant。
Arcade 用户：由服务端数据库保存稳定映射，不能让浏览器或模型自行传 user_id。不同租户下的同名本地用户必须映射到不同身份。
第三方账户：用户在提供商页面选择并授权的账户。应用还应让用户确认它确实是业务所需的工作账户。

OAuth 决定应用可以访问哪些账户权限；业务审批决定这次操作是否按这些参数执行。前一次允许 Gmail.SendEmail，不代表后续任意收件人、正文和附件都已获批准。授权工具调用

2. 固定 SDK 并核对工具版本

bash

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

通过服务端秘密管理设置 ARCADE_API_KEY。生产客户端限制自动重试：写操作请求超时可能已生效，自动重试不是业务幂等保证。

python

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，不把用户数据或凭据打印进公共日志。

这里的 get 是真实 SDK 读取入口，只有在你准备好的测试账户中才调用。本文离线测试中的 fixture-version 仅是模拟值，不能复制到生产。应保存审核过的工具定义和版本，升级后重新检查 schema、授权范围与行为。

官方当前 Quickstart 使用 recipient、subject、body 调用 Gmail.SendEmail。下文只开放这三个字段，拒绝额外参数，不开放附件、抄送、任意工具名或调度发送。实际接入仍应对照你所固定的版本定义。官方示例

3. 授权入口与生产用户验证

应用后端取得可信会话后，加载 principal 映射，再请求授权。授权链接只在该用户的受控界面展示，不放到共享日志或当作团队通用连接。用户必须亲自完成提供商同意页面。

生产环境按官方指南配置自定义 user verifier。回调中的 flow_id 是不可信输入；服务端会话决定 principal，不能采用 query string 中的 user_id。confirm_user 再向 Arcade 验证该会话身份与原流程是否匹配。生产授权指南

公开路由还要处理登录、回调与待办流程绑定、CSRF / 重放、短时效、失败限流和固定跳转目的地。不要把“客户端返回成功”直接写成账户授权完成。下面的 confirm_identity 是已鉴权服务端路由调用的核心，既不是可直接暴露的 HTTP 接口，也不是自建 OAuth 协议实现。

常见问题

user_id 可以直接从模型或请求参数取得吗？

不应如此。后端先验证登录会话和租户，再从自己的数据库读取稳定 Arcade 身份映射。

账户授权 completed 就能执行任意操作吗？

不能。账户权限和本次业务批准不同，敏感操作仍要检查工具、版本、完整参数和有效期。

工具版本为什么也要进入审批快照？

同名工具升级可能改变 schema 或行为。把版本和参数一起审核，升级后重新验证，避免批准内容和执行内容不同。

重复点击会发送两次吗？

示例用业务 request_id 唯一约束和事务占位减少重复领取，但不承诺跨外部 API 的恰好一次效果。

超时后应该怎么恢复？

保留 executing / unknown 状态，先核对 Arcade 运行记录与上游实际结果。不要盲目重试或生成新业务 ID 绕过台账。

十二组测试包含真实 Gmail 发信吗？

没有。使用官方 arcadepy 1.10.0 和 MockTransport 验证请求与状态，真实 OAuth、账户撤销和外发需在自己的测试环境另行验收。

"""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

情况	示例处理	生产补齐
同一个业务 request_id 重复提交	唯一约束拒绝第二条	调用方查询原任务，不创建新 request_id 绕过
未审批、审批过期或摘要错误	执行前拒绝	UI 清楚显示待审版本与有效期
用户映射或收件策略改变	执行前拒绝	重新审核账户与内容
两个工作器同时读取批准任务	事务中只允许一个占位	多机改用共享数据库与事务
请求发出后超时	unknown，拒绝再次执行	核对 Arcade 运行 ID 与上游实际结果
进程退出后停在 executing	不自动重置	告警、核对、记录人工决定

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

1. 三种身份与两次决定

2. 固定 SDK 并核对工具版本

3. 授权入口与生产用户验证

常见问题

4. 保存候选参数与批准版本

5. 从你的内部应用调用

6. 去重与超时的实际含义

7. 验收、日志与数据处理