企业官网建设流程全解析

多用户安全工具调用实战：用 Arcade.dev 与 LangGraph 构建生产级 Agent（Gmail / Slack / Notion 集成 + 人类审批）

【免费下载链接】agents-towards-productionEnd-to-end, code-first tutorials for building production-grade GenAI agents. From prototype to enterprise deployment.项目地址: https://gitcode.com/GitHub_Trending/ag/agents-towards-production

本指南基于 Agents Towards Production 仓库中的 Arcade 安全工具调用教程（见 multiuser-agent-arcade.ipynb），讲解如何用 LangGraph 与 Arcade.dev 构建真正面向多用户的生产级 Agent：从最简单的对话 Agent 起步，逐步接入 Gmail、Slack、Notion 等真实外部服务，并实现 OAuth2 多用户授权与 Human-in-the-Loop（人类审批）安全控制。读完本文你将掌握一套完整的"本地 Demo → 多用户生产系统"的进阶路线，理解工具级认证为何是生产化的关键瓶颈，以及如何用统一平台解决它。

为什么"本地好用"的 Agent 难以直接服务多用户

当一个 Agent 在自己电脑上运行良好时，它是一位出色的个人助理；但把它扩展给大量用户使用时，问题随之而来——本地部署的安全假设在规模化场景下完全不成立：

• Personal Access Token（个人访问令牌）无法支撑多用户：每个用户都需要独立的身份、独立的授权与独立的数据边界，共享一个 Token 意味着所有用户共享同一份权限，这在安全上是不可接受的。
• 远程 MCP 服务器也绕不开工具级认证：即使把所有功能封装进一个远程 MCP 服务器，工具层面的认证依然需要你为 Agent 依赖的每一个服务商（Gmail、Slack、Notion……）分别实现一套 OAuth 授权流程，工作量随服务数量线性膨胀。

Arcade 的解决思路是提供一个统一的 Agent 工具执行平台：由它代你处理认证流程，为 Agent 提供安全的多用户解决方案。教程的核心目标就是结合 Arcade 与 LangGraph，实现三类能力：

1. 构建 Agent；
2. 为 Agent 提供可安全交互的工具——Gmail、Slack、Notion；
3. 在调用特定工具时实现安全护栏（Human-in-the-Loop 人工审批）。

整个教程按难度递进为三个层次：基础对话 Agent → 工具增强 Agent（Gmail）→ 生产级 Agent（多服务协调 + 安全控制）。技术栈为LangGraph（Agent 编排与状态管理）、Arcade.dev（认证与安全 API 访问）、OAuth2（安全用户授权）。

环境准备：依赖安装

开始写代码之前，先搭建开发环境。教程使用的核心依赖包括：

• LangGraph：Agent 编排与状态管理；
• LangChain-Arcade：Arcade 工具与 LangChain/LangGraph 的集成层；
• LangChain（含 OpenAI 支持）：基础框架与模型调用。

在 Jupyter 环境中直接通过 pip 安装：

!pip install langgraph langchain-arcade "langchain[openai]"

API Key 与用户身份配置

运行本教程需要两个 API Key：

• OpenAI API Key：为 Agent 提供大模型推理能力；
• Arcade API Key：用于调用 Arcade 平台，管理工具执行与认证流程。

两个服务都提供简单的注册流程。为方便在 Notebook 中安全地设置环境变量，教程定义了一个_set_env辅助函数：若变量已存在于环境中则保留，否则若提供了默认值则写入，都没有则通过getpass交互式输入（避免明文出现在 Notebook 中）：

import getpass import os def _set_env(key: str, default: str | None): if key not in os.environ: if default: os.environ[key] = default else: os.environ[key] = getpass.getpass(f"{key}:") _set_env("OPENAI_API_KEY") _set_env("ARCADE_API_KEY")

用户身份（ARCADE_USER_ID）的作用

这是多用户安全模型的关键一环。Arcade 平台需要通过用户标识来管理工具授权、并在不同用户之间维持安全边界。该标识必须与注册 Arcade 账号时使用的邮箱一致，确保工具权限与 OAuth Token 能正确关联到对应的用户账号：

_set_env("ARCADE_USER_ID")

理解这一点很重要：Arcade 的授权模型是"每个用户独立授权"的，user_id正是把一次工具调用绑定到具体用户身份的钥匙，后续所有工具执行都会携带它。

第一阶段：基础对话 Agent（无工具）

先从最朴素的对话 Agent 开始，它演示了 LangGraph 的核心能力，没有任何外部工具依赖。

核心实现：React Agent + 会话记忆

教程使用 LangGraph 预构建的create_react_agent创建 React 风格（Reasoning + Acting）Agent，并通过MemorySaver检查点（checkpointer）赋予其短期会话记忆——Agent 能在同一个会话线程（thread）内记住之前的交互：

from langgraph.prebuilt.chat_agent_executor import create_react_agent from langgraph.checkpoint.memory import MemorySaver from langchain_core.messages import HumanMessage import uuid # create a checkpointer to persist the graph's state checkpointer = MemorySaver() agent_a = create_react_agent( model="openai:gpt-5", prompt="You are a helpful assistant that can help with everyday tasks." " If the user's request is confusing you must ask them to clarify" " their intent, and fulfill the instruction to the best of your" " ability. Be concise and friendly at all times.", tools=[], # no tools for now! checkpointer=checkpointer )

注意tools=[]——此时 Agent 只具备对话能力。Prompt 中明确要求：请求含糊时必须主动向用户澄清意图，始终保持简洁友好。

交互工具函数：统一的消息流

为在整个教程中一致地观察 Agent 行为，定义run_graph工具函数：以stream_mode="values"流式输出图的每次状态更新，并打印每个事件中的最新一条消息：

from langgraph.graph.state import CompiledStateGraph def run_graph(graph: CompiledStateGraph, config, input): for event in graph.stream(input, config=config, stream_mode="values"): if "messages" in event: event["messages"][-1].pretty_print()

交互式聊天界面

下面给出完整的交互式聊天界面。系统为每次会话生成唯一的thread_id（config中configurable.thread_id），LangGraph 依据它区分不同会话并持久化记忆——本 Agent 虽未用到中断能力，但不同会话的记忆隔离机制已经就位。thread_id每次运行随机生成，如需测试记忆保持，可手动固定该值：

# the configuration helps LangGraph keep track of conversations and interrups # While it's not needed for this agent. The agent will remember different # conversations based on the thread_id. This code generates a random id every # time you run the cell, but you can hardcode the thread_id if you want to # test the memory. config = { "configurable": { "thread_id": uuid.uuid4() } } while True: user_input = input("👤: ") # let's use "exit" as a safe way to break the infinite loop if user_input.lower() == "exit": break user_message = {"messages": [HumanMessage(content=user_input)]} run_graph(agent_a, config, user_message)

输入exit即可安全退出循环。

测试 Agent 的边界：它做不了什么

为了理解基础 Agent 的能力边界，教程用两个典型请求做"负向测试"。

测试一：实时信息缺失。大多数大模型没有实时数据访问能力，可能给出过时或不准的日期信息：

config = { "configurable": { "thread_id": uuid.uuid4() } } print(f'thread_id = {config["configurable"]["thread_id"]}') prompt = "what's today's date?" user_message = {"messages": [HumanMessage(content=prompt)]} run_graph(agent_a, config, user_message)

测试二：无法访问私有认证数据。让 Agent 总结最近的 3 封邮件，它会因缺乏认证机制与授权的外部服务访问能力而完全无法推进：

config = { "configurable": { "thread_id": uuid.uuid4() } } print(f'thread_id = {config["configurable"]["thread_id"]}') prompt = "summarize my latest 3 emails please" user_message = {"messages": [HumanMessage(content=prompt)]} run_graph(agent_a, config, user_message)

这两个失败场景精准地指向了生产化 Agent 的两个刚需：实时工具与安全的私有数据访问——这正是接下来要解决的问题。

第二阶段：工具集成与安全认证（以 Gmail 为例）

本阶段解决核心难题：如何让 Agent 安全地访问外部服务。Arcade 将复杂的工具级 OAuth 集成封装为统一平台能力，可跨多用户、多服务平滑扩展。

初始化 Arcade 客户端与 ToolManager

首先建立与 Arcade 平台的连接：arcade_client负责底层认证基础设施，ToolManager是配置与授权工具的主要接口：

from langchain_arcade import ToolManager from arcadepy import Arcade arcade_client = Arcade(api_key=os.getenv("ARCADE_API_KEY")) manager = ToolManager(client=arcade_client)

初始化 Gmail 工具

第一个集成目标是 Gmail 的邮件列表能力——这正是基础 Agent 无法提供的功能。Gmail_ListEmails工具让 Agent 能检索并分析邮件数据，但在访问私有邮箱前，必须先完成用户授权：

gmail_tool = manager.init_tools(tools=["Gmail_ListEmails"])[0]

授权工具函数：OAuth 流程封装

要读取用户的邮件，需要以安全方式授予应用读取权限。Arcade 通过代管 OAuth2简化了这一过程。教程封装了可复用的authorize_tool函数：检查指定工具与用户组合的授权状态，必要时发起 OAuth 流程并输出授权 URL，然后阻塞等待用户完成授权：

def authorize_tool(tool_name, user_id, manager): # This line will check if this user is authorized to use the # tool, and return a response that we can use if the user # did not authorize the tool yet. auth_response = manager.authorize( tool_name=tool_name, user_id=user_id ) if auth_response.status != "completed": print(f"The app wants to use the {tool_name} tool.\n" f"Please click this url to authorize it {auth_response.url}") # wait until the user authorizes manager.wait_for_auth(auth_response.id)

执行 Gmail 授权

调用上述函数完成 Gmail 授权。若用户此前未授权，Arcade 会返回 OAuth URL 供用户点击完成授权；一旦授权成功，该权限会持久化保存，后续会话无需重复走授权流程：

authorize_tool(gmail_tool.name, os.getenv("ARCADE_USER_ID"), manager)

带 Gmail 能力的增强 Agent

授权完成后，创建增强版 Agent。与agent_a相比有三处关键差异：

1. Prompt 明确告知 Gmail 能力：指导 Agent 用 Gmail 工具处理邮件相关请求；
2. tools传入已授权的 Gmail 工具；
3. config中必须携带user_id：使用 Arcade 工具时，必须在 LangGraph 配置中提供user_id，Arcade 才能以该用户身份执行 Agent 调用的工具——这是多用户隔离的实现根基。

# define a new agent, this time with access to our tool! agent_b = create_react_agent( model="openai:gpt-5", prompt="You are a helpful assistant that can help with everyday tasks." " If the user's request is confusing you must ask them to clarify" " their intent, and fulfill the instruction to the best of your" " ability. Be concise and friendly at all times." # It's useful to let the agent know about the tools it has at its disposal. " Use the Gmail tools that you have to address requests about emails.", tools=[gmail_tool], # we pass the tool we previously authorized. checkpointer=checkpointer ) config = { "configurable": { "thread_id": uuid.uuid4(), "user_id": os.getenv("ARCADE_USER_ID") # When using Arcade tools, we must provide the user_id on the LangGraph config, so Arcade can execute the tool invoked by the agent. } } print(f'thread_id = {config["configurable"]["thread_id"]}') # we're using the same prompt we use before, but we're swapping the agent prompt = "summarize my latest 3 emails please" user_message = {"messages": [HumanMessage(content=prompt)]} run_graph(agent_b, config, user_message)

这一次，同一个请求得到了完全不同的结果：Agent 通过Gmail_ListEmails工具读取邮件并生成摘要。

第三阶段：多服务工具集成（Gmail + Slack + Notion）

单个服务集成成功后，下一步是让 Agent 同时协调多个外部服务。关键挑战变成：如何高效管理跨服务商的认证，同时保持安全与用户体验。

批量授权函数：按服务商合并 OAuth 作用域

逐个授权工具会随着能力扩张变得繁琐。教程的authorize_tools函数将所有工具的授权作用域（scopes）按服务商（provider）分组合并，从而把用户需要完成的 OAuth 流程数降到最低：

def authorize_tools(tools, user_id, client): # This will map all the providers to the specific scopes they need provider_to_scopes = {} for tool in tools: provider = tool.requirements.authorization.provider_id if provider not in provider_to_scopes: provider_to_scopes[provider] = set() if tool.requirements.authorization.oauth2.scopes: provider_to_scopes[provider] |= set(tool.requirements.authorization.oauth2.scopes) # Each provider will handle its own scopes, we iterate and present the # auth URL for all providers that need it for provider, scopes in provider_to_scopes.items(): # start auth auth_response = client.auth.start( user_id=user_id, scopes=list(scopes), provider=provider ) # show the url to the user if needed if auth_response.status != "completed": print(f"🔗 Please click here to authorize: {auth_response.url}") print(f"⏳ Waiting for authorization completion...") # Wait for the authorization to complete with timeout client.auth.wait_for_completion(auth_response),

这里的核心逻辑是读取每个工具的requirements.authorization元数据（provider_id与oauth2.scopes），以 provider 为键做集合合并，再对每个 provider 调用一次client.auth.start发起授权。工具自身的元数据驱动着认证流程，这正是 Arcade 统一平台能力的体现。

配置完整工具套件：单个工具 + 整个工具包

接下来扩展能力：加入邮件发送（Gmail）、Slack 通信与 Notion 内容管理。ToolManager支持两种注册方式——add_tool添加单个工具，add_toolkit一次添加整组相关工具（工具包）：

# add a single tool manager.add_tool("Gmail.SendEmail") # add an entire toolkit (a collection of tools) manager.add_toolkit("Slack") manager.add_toolkit("NotionToolkit")

manager.definitions即当前已注册的全部工具定义，可直接用于批量授权：

authorize_tools( tools=manager.definitions, user_id=os.getenv("ARCADE_USER_ID"), client=arcade_client )

多服务 Agent：借助 ToolManager 无缝接入 LangGraph

授权完成后创建能力最强的 Agent。关键一行是tools=manager.to_langchain()——ToolManager 的 LangChain 转换功能把 Arcade 工具定义无缝桥接到 LangGraph 的执行框架。Prompt 中为每类服务明确分工：Gmail 处理邮件读写、Slack 处理用户与频道交互、Notion 处理页面内容管理，并鼓励 Agent 优先选择最相关的工具：

# define a new agent, this time with access to our tool! agent_c = create_react_agent( model="openai:gpt-5", prompt="You are a helpful assistant that can help with everyday tasks." " If the user's request is confusing you must ask them to clarify" " their intent, and fulfill the instruction to the best of your" " ability. Be concise and friendly at all times." # It's useful to let the agent know about the tools it has at its disposal. " Use the Gmail tools to address requests about reading or sending emails." " Use the Slack tools to address requests about interactions with users and channels in Slack." " Use the Notion tools to address requests about managing content in Notion Pages." " In general, when possible, use the most relevant tool for the job.", tools=manager.to_langchain(), checkpointer=checkpointer )

复杂跨服务任务演示

下面的请求要求 Agent 同时完成三项工作：分析邮件数据、检索 Slack 通信、探索 Notion 工作区结构——充分展示其跨服务工具选择与执行协调能力：

config = { "configurable": { "thread_id": uuid.uuid4(), "user_id": os.getenv("ARCADE_USER_ID") # When using Arcade tools, we must provide the user_id on the LangGraph config, so Arcade can execute the tool invoked by the agent. } } print(f'thread_id = {config["configurable"]["thread_id"]}') # we're using the same prompt we use before, but we're swapping the agent prompt = "summarize my latest 3 emails, then show me the latest 3 messages in the #general Slack channel, and tell me about the structure of my Notion Workspace" user_message = {"messages": [HumanMessage(content=prompt)]} run_graph(agent_c, config, user_message)

第四阶段：Human-in-the-Loop 安全控制

多服务 Agent 能力强大，但生产系统必须防范非预期操作。本阶段为敏感操作引入人类审批机制：可能有害或不可逆的动作，必须在执行前获得用户明确批准。

识别敏感操作

先枚举当前注册的全部工具，基于影响面与不可逆性进行分类：

for tool_name, _ in manager: print(tool_name)

据此，教程将"创建、发送、修改数据"类的工具（而非只读检索）判定为敏感——它们可能产生外部影响或危及用户隐私/系统完整性：

tools_to_protect = [ "Gmail_SendEmail", "Slack_SendDmToUser", "Slack_SendMessage", "Slack_SendMessageToChannel", "NotionToolkit_AppendContentToEndOfPage", "NotionToolkit_CreatePage", ]

人类审批工具包装器：基于 LangGraph interrupt 机制

核心是add_human_in_the_loop包装函数：它把普通工具转换为"人工监督版"，拦截工具执行请求、向用户展示将要执行的动作、仅在获得明确同意后放行。实现依托LangGraph 的interrupt机制——暂停图执行并等待外部输入：

from typing import Callable, Any from langchain_core.tools import tool, BaseTool from langgraph.types import interrupt, Command from langchain_core.runnables import RunnableConfig import pprint def add_human_in_the_loop( target_tool: Callable | BaseTool, ) -> BaseTool: """Wrap a tool to support human-in-the-loop review.""" if not isinstance(target_tool, BaseTool): target_tool = tool(target_tool) @tool( target_tool.name, description=target_tool.description, args_schema=target_tool.args_schema ) def call_tool_with_interrupt(config: RunnableConfig, **tool_input): arguments = pprint.pformat(tool_input, indent=4) response = interrupt( f"Do you allow the call to {target_tool.name} with arguments:\n" f"{arguments}" ) # approve the tool call if response == "yes": tool_response = target_tool.invoke(tool_input, config) # deny tool call elif response == "no": tool_response = "The User did not allow the tool to run" else: raise ValueError( f"Unsupported interrupt response type: {response}" ) return tool_response return call_tool_with_interrupt

值得注意的细节：包装器通过@tool重新注册同名工具，并保留原工具的 description 与 args_schema，从而不破坏 Agent 对工具的调用协议。interrupt会抛出待审批内容（工具名 + 参数），只有收到"yes"才真正执行原工具，"no"则返回拒绝消息。

选择性应用保护：只包装敏感工具

为了保持安全操作的高效性，只对敏感列表内的工具应用包装，只读工具保持原样：

protected_tools = [ add_human_in_the_loop(t) if t.name in tools_to_protect else t for t in manager.to_langchain() ]

中断处理工具：审批交互与恢复执行

LangGraph 的中断需要专门处理才能恢复执行。yes_no_loop强制用户给出明确的 y/n 决定；handle_interrupts遍历图中挂起的中断，逐一向用户展示审批内容，并把用户的决定通过Command(resume=...)恢复图的执行：

def yes_no_loop(prompt: str) -> str: """ Force the user to say yes or no """ print(prompt) user_input = input("Your response [y/n]: ") while user_input.lower() not in ["y", "n"]: user_input = input("Your response (must be 'y' or 'n'): ") return "yes" if user_input.lower() == "y" else "no" def handle_interrupts(graph: CompiledStateGraph, config): for interr in graph.get_state(config).interrupts: approved = yes_no_loop(interr.value) run_graph(graph, config, Command(resume=approved))

受保护的生产级 Agent

最终 Agent 在保留全部多服务能力的同时叠加了安全控制——这是一个"功能与安全平衡"的生产就绪系统：常规任务自动化，敏感操作由用户掌控：

# define a new agent, this time with access to our tool! agent_hitl = create_react_agent( model="openai:gpt-5", prompt="You are a helpful assistant that can help with everyday tasks." " If the user's request is confusing you must ask them to clarify" " their intent, and fulfill the instruction to the best of your" " ability. Be concise and friendly at all times." # It's useful to let the agent know about the tools it has at its disposal. " Use the Gmail tools to address requests about reading or sending emails." " Use the Slack tools to address requests about interactions with users and channels in Slack." " Use the Notion tools to address requests about managing content in Notion Pages." " In general, when possible, use the most relevant tool for the job.", tools=protected_tools, checkpointer=checkpointer )

安全机制演示：拦截"机密邮件"

下面用发送潜在敏感邮件的场景验证安全系统。请求让 Agent 给指定地址发送一封含"机密数据"标题的邮件——Human-in-the-loop 机制会在此拦截动作、展示细节并等待用户明确批准：

config = { "configurable": { "thread_id": uuid.uuid4(), "user_id": os.getenv("ARCADE_USER_ID") # When using Arcade tools, we must provide the user_id on the LangGraph config, so Arcade can execute the tool invoked by the agent. } } print(f'thread_id = {config["configurable"]["thread_id"]}') # we're using the same prompt we use before, but we're swapping the agent prompt = 'send an email with subject "confidential data" and body "this is top secret information" to random-dude@example.com' user_message = {"messages": [HumanMessage(content=prompt)]} run_graph(agent_hitl, config, user_message)

查看中断状态

安全系统触发时，Agent 执行暂停并进入中断状态。通过get_state(config).interrupts可直接检查挂起的审批请求——其中包含了待审批动作的完整细节：

agent_hitl.get_state(config).interrupts

处理用户决定

接着处理挂起的中断：向用户展示动作细节、收集审批决定。这演示了用户如何审查潜在敏感操作并决定是否放行 Agent 提出的操作：

handle_interrupts(agent_hitl, config)

选择y则继续执行邮件发送，选择n则工具返回"用户不允许执行"。

完整交互系统

最后把全部能力组装为完整交互系统：自然对话 + 多服务访问 + 敏感操作人工审批 + 自动授权与工具执行，全部封装在无缝的用户体验中：

config = { "configurable": { "thread_id": uuid.uuid4() } } while True: user_input = input("👤: ") # let's use "exit" as a safe way to break the infinite loop if user_input.lower() == "exit": break user_message = {"messages": [HumanMessage(content=user_input)]} run_graph(agent_hitl, config, user_message) handle_interrupts(agent_hitl, config)

每一轮用户输入之后，先运行 Agent（可能触发中断），再统一处理所有挂起的中断，形成"对话 → 提议 → 审批 → 执行"的完整闭环。

贯穿全文的生产级安全设计要点

回顾整条进阶路线，可以提炼出多用户生产 Agent 的几条核心设计原则：

1. 用户身份贯穿始终：user_id从授权到执行的每一个环节都不可缺失（ARCADE_USER_ID），它是 OAuth Token 与工具调用绑定的锚点，也是用户间安全隔离的基础；
2. 认证一次、长期有效：Arcade 将 OAuth 授权结果持久化，用户无需重复授权；批量授权函数按服务商合并作用域，把用户体验摩擦降到最低；
3. 元数据驱动的认证：工具的requirements.authorization（provider 与 scopes）驱动认证流程，接入新服务无需手写认证逻辑；
4. 敏感操作分级治理：并非所有工具都需要审批——只读检索直接放行，发送/创建/修改类操作强制人类确认，兼顾效率与安全；
5. 中断与恢复的标准化：interrupt暂停执行、get_state().interrupts检查挂起请求、Command(resume=...)恢复执行，构成了可复用的审批工作流范式。

进一步学习

• 完整可运行的教程代码见 multiuser-agent-arcade.ipynb，教程的速览说明见 README.md；
• 仓库根目录 README.md 的教程列表中，本主题被定位为"Secure Tool Calling (Arcade)"，与 LangGraph Agent、安全护栏（如 agent-security-apex）、带 MCP 的 Agent 等教程共同构成从原型到企业部署的完整学习路径；
• 体系架构图 arcade-diagram.png 总结了"用户请求 → 是否需要工具 → 是否已授权（OAuth）→ 是否敏感（人工审批）→ 执行/阻止"的完整决策流。

【免费下载链接】agents-towards-productionEnd-to-end, code-first tutorials for building production-grade GenAI agents. From prototype to enterprise deployment.项目地址: https://gitcode.com/GitHub_Trending/ag/agents-towards-production