多用户安全工具调用实战用 Arcade.dev 与 LangGraph 构建生产级 AgentGmail / 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实现三类能力构建 Agent为 Agent 提供可安全交互的工具——Gmail、Slack、Notion在调用特定工具时实现安全护栏Human-in-the-Loop 人工审批。整个教程按难度递进为三个层次基础对话 Agent → 工具增强 AgentGmail→ 生产级 Agent多服务协调 安全控制。技术栈为LangGraphAgent 编排与状态管理、Arcade.dev认证与安全 API 访问、OAuth2安全用户授权。环境准备依赖安装开始写代码之前先搭建开发环境。教程使用的核心依赖包括LangGraphAgent 编排与状态管理LangChain-ArcadeArcade 工具与 LangChain/LangGraph 的集成层LangChain含 OpenAI 支持基础框架与模型调用。在 Jupyter 环境中直接通过 pip 安装!pip install langgraph langchain-arcade langchain[openai]API Key 与用户身份配置运行本教程需要两个 API KeyOpenAI 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 ActingAgent并通过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 graphs state checkpointer MemorySaver() agent_a create_react_agent( modelopenai:gpt-5, promptYou are a helpful assistant that can help with everyday tasks. If the users 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! checkpointercheckpointer )注意tools[]——此时 Agent 只具备对话能力。Prompt 中明确要求请求含糊时必须主动向用户澄清意图始终保持简洁友好。交互工具函数统一的消息流为在整个教程中一致地观察 Agent 行为定义run_graph工具函数以stream_modevalues流式输出图的每次状态更新并打印每个事件中的最新一条消息from langgraph.graph.state import CompiledStateGraph def run_graph(graph: CompiledStateGraph, config, input): for event in graph.stream(input, configconfig, stream_modevalues): if messages in event: event[messages][-1].pretty_print()交互式聊天界面下面给出完整的交互式聊天界面。系统为每次会话生成唯一的thread_idconfig中configurable.thread_idLangGraph 依据它区分不同会话并持久化记忆——本 Agent 虽未用到中断能力但不同会话的记忆隔离机制已经就位。thread_id每次运行随机生成如需测试记忆保持可手动固定该值# the configuration helps LangGraph keep track of conversations and interrups # While its 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(: ) # lets use exit as a safe way to break the infinite loop if user_input.lower() exit: break user_message {messages: [HumanMessage(contentuser_input)]} run_graph(agent_a, config, user_message)输入exit即可安全退出循环。测试 Agent 的边界它做不了什么为了理解基础 Agent 的能力边界教程用两个典型请求做负向测试。测试一实时信息缺失。大多数大模型没有实时数据访问能力可能给出过时或不准的日期信息config { configurable: { thread_id: uuid.uuid4() } } print(fthread_id {config[configurable][thread_id]}) prompt whats todays date? user_message {messages: [HumanMessage(contentprompt)]} run_graph(agent_a, config, user_message)测试二无法访问私有认证数据。让 Agent 总结最近的 3 封邮件它会因缺乏认证机制与授权的外部服务访问能力而完全无法推进config { configurable: { thread_id: uuid.uuid4() } } print(fthread_id {config[configurable][thread_id]}) prompt summarize my latest 3 emails please user_message {messages: [HumanMessage(contentprompt)]} 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_keyos.getenv(ARCADE_API_KEY)) manager ToolManager(clientarcade_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_nametool_name, user_iduser_id ) if auth_response.status ! completed: print(fThe app wants to use the {tool_name} tool.\n fPlease 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相比有三处关键差异Prompt 明确告知 Gmail 能力指导 Agent 用 Gmail 工具处理邮件相关请求tools传入已授权的 Gmail 工具config中必须携带user_id使用 Arcade 工具时必须在 LangGraph 配置中提供user_idArcade 才能以该用户身份执行 Agent 调用的工具——这是多用户隔离的实现根基。# define a new agent, this time with access to our tool! agent_b create_react_agent( modelopenai:gpt-5, promptYou are a helpful assistant that can help with everyday tasks. If the users 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. # Its 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. checkpointercheckpointer ) 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(fthread_id {config[configurable][thread_id]}) # were using the same prompt we use before, but were swapping the agent prompt summarize my latest 3 emails please user_message {messages: [HumanMessage(contentprompt)]} 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_iduser_id, scopeslist(scopes), providerprovider ) # 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( toolsmanager.definitions, user_idos.getenv(ARCADE_USER_ID), clientarcade_client )多服务 Agent借助 ToolManager 无缝接入 LangGraph授权完成后创建能力最强的 Agent。关键一行是toolsmanager.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( modelopenai:gpt-5, promptYou are a helpful assistant that can help with everyday tasks. If the users 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. # Its 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., toolsmanager.to_langchain(), checkpointercheckpointer )复杂跨服务任务演示下面的请求要求 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(fthread_id {config[configurable][thread_id]}) # were using the same prompt we use before, but were 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(contentprompt)]} 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, descriptiontarget_tool.description, args_schematarget_tool.args_schema ) def call_tool_with_interrupt(config: RunnableConfig, **tool_input): arguments pprint.pformat(tool_input, indent4) response interrupt( fDo 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( fUnsupported 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(resumeapproved))受保护的生产级 Agent最终 Agent 在保留全部多服务能力的同时叠加了安全控制——这是一个功能与安全平衡的生产就绪系统常规任务自动化敏感操作由用户掌控# define a new agent, this time with access to our tool! agent_hitl create_react_agent( modelopenai:gpt-5, promptYou are a helpful assistant that can help with everyday tasks. If the users 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. # Its 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., toolsprotected_tools, checkpointercheckpointer )安全机制演示拦截机密邮件下面用发送潜在敏感邮件的场景验证安全系统。请求让 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(fthread_id {config[configurable][thread_id]}) # were using the same prompt we use before, but were swapping the agent prompt send an email with subject confidential data and body this is top secret information to random-dudeexample.com user_message {messages: [HumanMessage(contentprompt)]} 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(: ) # lets use exit as a safe way to break the infinite loop if user_input.lower() exit: break user_message {messages: [HumanMessage(contentuser_input)]} run_graph(agent_hitl, config, user_message) handle_interrupts(agent_hitl, config)每一轮用户输入之后先运行 Agent可能触发中断再统一处理所有挂起的中断形成对话 → 提议 → 审批 → 执行的完整闭环。贯穿全文的生产级安全设计要点回顾整条进阶路线可以提炼出多用户生产 Agent 的几条核心设计原则用户身份贯穿始终user_id从授权到执行的每一个环节都不可缺失ARCADE_USER_ID它是 OAuth Token 与工具调用绑定的锚点也是用户间安全隔离的基础认证一次、长期有效Arcade 将 OAuth 授权结果持久化用户无需重复授权批量授权函数按服务商合并作用域把用户体验摩擦降到最低元数据驱动的认证工具的requirements.authorizationprovider 与 scopes驱动认证流程接入新服务无需手写认证逻辑敏感操作分级治理并非所有工具都需要审批——只读检索直接放行发送/创建/修改类操作强制人类确认兼顾效率与安全中断与恢复的标准化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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考