随着AI代理人变得更加自主和有能力,他们的角色正在从被动助手转变为积极参与者,而今天的大型语言模型(LLMs)不仅仅是生成文本 - 他们执行任务,访问API,修改数据库,甚至控制基础设施。 人工智能代理正在采取曾经仅限于人类用户的行动,无论是安排会议、部署服务还是访问敏感文档。 , 单个幻觉的命令、误解的提示或过于广泛的权限可能会导致数据泄露、违规或系统破坏。 When agents operate without guardrails they can inadvertently make harmful or unauthorized decisions 这就是为什么整合 工作流程对于代理人安全和问责性至关重要。 human-in-the-loop (HITL) 是一个框架,旨在使AI代理人能够 敏感的行为,同时允许 . Permit.io’s Access Request MCP request humans to remain the final decision-makers Permit.io 访问请求 MCP 建成 并集成到流行的代理框架中,如 和 ,该系统允许您直接将批准工作流插入到您的LLM驱动的应用程序中。 允许我 LangChain LangGraph 在本教程中,你将学习: 
 
 
 
 为什么向人类授权敏感权限对于值得信赖的AI至关重要 Permit.io 的模型背景协议 (MCP) 如何允许访问请求工作流 如何构建一个真正的系统,将LLM情报与人类监督相结合 - 使用LangGraph的interrupt()功能。 在深入我们的演示应用程序和实施步骤之前,让我们简要讨论将AI权限授权给人类的重要性。 为什么向人类授权AI至关重要 人工智能代理人是强大的,但正如我们都知道的那样,他们不是无误的。 他们遵循指示,但他们不像人类那样理解背景,他们产生响应,但他们无法判断后果,当这些代理人集成到真实的系统中 - 银行工具,内部仪表板,基础设施控制 - 这是一个危险的差距。 在这种情况下,所有可能发生错误的事情都非常清楚: 
 
 
 
 过度允许的代理人:LLM可能被允许访问他们不应该触摸的工具,无论是设计还是意外。 幻觉工具呼叫:代理可以制造从未存在的命令、论点或ID。 缺乏可审计性:没有人体检查点,没有谁批准了什么或为什么的明确记录。 Delegation is the solution. Instead of giving agents unchecked power, we give them a protocol:  “你可以问,但一个人决定。 通过介绍 在关键决策点获得批准,您将获得: human-in-the-loop (HITL) 
 
 
 
 安全性:在发生之前防止不可逆转的行为。 问责:要求明确的人为登记高额交易。 控制:让人们制定规则 - 谁可以批准,什么可以批准,以及何时。 这是一个代理人的区别 一件事情和一个特工 做点什么。 做 请求 而且它正是什么 MCP 访问请求允许。 允许我 Permit.io 访问请求 MCP 是核心的一部分 模型背景协议(MCP) - 一个规格,使AI代理人能够安全、有政策意识地访问工具和资源。 The Access Request MCP 允许我 把它当成一座桥梁 和 . LLMs that want to act humans who need control 它做什么 允许访问请求 MCP 允许 AI 代理人: 
 
 
 
 
 请求访问受限制的资源(例如:“我可以访问这个餐厅吗?”) 请求批准执行敏感操作(例如:“我可以订购这个受限制的菜肴吗?”) 在继续前等待人类输入 - 通过 LangGraph 的 interrupt() 机制 记录审计和合规性的请求和决定 在幕后,它使用 ’s授权能力构建以支持: 允许我 
 
 
 
 ReBAC(基于关系的访问控制)和其他细粒授权(Fine-grained Authorization,FGA)政策。 批准工作流程 通过政策支持的元素,在UI,API和LLM环境中工作 回复 精细谷物许可(Fine Grained Authorization,FGA) 批准工作流程 政策支持的元素 使用 LangChain 和 LangGraph 的 Plug-and-Play Permit 的 MCP 直接集成到 和 生态系统: LangChain MCP Adapter LangGraph 
 
 
 
 您可以将“允许元素”列为兼容 LangGraph 的工具。 当发生敏感操作时,您可以使用 interrupt() 暂停代理。 您可以根据真实的人类决定恢复执行。 这是最容易的方式来 没有必要的定制背后。 inject human judgment into AI behavior 了解实施及其好处,让我们进入我们的演示应用程序。 我们将建立什么 - 演示应用概述 在本教程中,我们将建立一个 在其中,一个 他们。 real-time approval workflow AI agent can request access or perform sensitive actions, but only a human can approve 标签: 家庭饮食系统 要了解 Permit 的 MCP 如何帮助用户应用程序中启用 HITL 工作流程,我们将模型一个 对于一个家庭: food ordering system 
 
 
 
 家长可以访问和管理所有餐厅和菜肴。 儿童可以查看公共物品,但必须要求访问受限制的餐厅或昂贵的菜肴。 当儿童提交请求时,父母会收到请求审查,并必须在采取行动之前明确批准或拒绝。 这个使用案例反映了一个常见的模式: “代理人可以帮助,但人类可以决定。 技术 Stack 我们将使用以下方法构建这个HITL支持的代理: 
 
 
 
 
 
 
 Permit.io - 处理授权、角色、政策和批准 允许 MCP 服务器 - 暴露允许工作流作为代理可以使用的工具 LangChain MCP 适配器 - 将 MCP 工具带入 LangGraph 和 LangChain LangGraph - 通过 interrupt() 支持来组织代理的工作流程 Gemini 2.0 Flash - 轻量级,多式LLM作为代理的推理引擎 Python - 粘合物将所有东西放在一起 您将获得一个工作系统,代理人可以与人类合作,以确保安全,有意的行为 - 使用真实的政策,真实的工具和实时批准。 A repository with the full code for this application is available here. 这个应用程序的完整代码的存储库在这里可用。 步骤式教程 在本节中,我们将探讨如何实施一个完全功能的人类循环代理系统。 还有长篇。 允许我 我们将覆盖: 
 
 
 
 
 
 与许可证的模型许可证 设置许可 MCP 服务器 创建一个 LangGraph + LangChain MCP 客户端 添加 Human-in-the-Loop with interrupt( ) 运行完整的工作流程 让我们进入它 - 与许可证的模型许可证 我们将首先定义系统内部的访问规则。 这允许您模拟哪些用户可以做什么,以及哪些操作应该触发批准流。 允许.io 仪表板 Create a ReBAC Resource 导航至 页面从侧面栏,然后: Policy 
 
 
 
 
 
 点击资源标签 点击创建资源 资源名称:餐厅 Under  , define two roles:
 
 ReBAC Options 
 
 
 parent child-can-order 点击保存 现在,去到 tab 和分配权限: Policy Editor 
 
 
 家长:完全访问(创建、阅读、更新、删除) 儿童阅读:阅读 Set Up Permit Elements 走到The 图标: 图标: 图标: 在 点击,点击 . Elements User Management Create Element 
 
 
 
 
 
 Configure the element as follows: 
 
 
 
 
 : Restaurant Requests Name : ReBAC Resource Roles Configure elements based on : restaurants Resource Type 
 
 Role permission levels 
 
 
 Level 1 – Workspace Owner:  parent Assignable Roles:  child-can-order 点击创建 在新创建的元素卡中,单击“获取代码”,并注意 Config ID: restaurant-requests。 Add Operation Approval Elements 
 
 
 Create a new   element:
 
 Operation Approval 
 
 
 : Dish Approval Name : restaurants Resource Type 点击创建 
 
 
 Then create an   element:
 
 Approval Management 
 
 : Dish Requests Name 点击获取代码并复制 config ID: dish-requests。 Add Test Users & Resource Instances 
 
 
 导航到目录 > 实例 Click  
 
 Add Instance 
 
 
 
 : restaurants Resource Type :  Instance Key pizza-palace : Default Tenant (or your working tenant) Tenant 
 
 
 
 切换到用户标签 Click  
 
 Add User 
 
 
 
 :  Key joe :  Instance Access restaurants:pizza-palace#parent Click  Save Create another user with the key  
 
 henry 
 
 Don’t assign a role 一旦配置了 Permit,我们就可以克隆 MCP 服务器,并将策略连接到工作代理。 设置许可 MCP 服务器 随着您的策略在“允许”仪表板中模型化,是时候通过设置 一个本地服务,将您的访问请求和批准流作为AI代理可以使用的工具。 Permit MCP server Clone and Install the MCP Server 首先,克隆MCP服务器存储库并设置虚拟环境。 git clone <https://github.com/permitio/permit-mcp>
cd permit-mcp

# Create virtual environment, activate it and install dependencies
uv venv
source .venv/bin/activate # For Windows: .venv\\Scripts\\activate
uv pip install -e .
 Add Environment Configuration 创建 a 基于提供的项目的根文件 ,并填充您的允许设置中的正确值: .env .env.example bash
CopyEdit
RESOURCE_KEY=restaurants
ACCESS_ELEMENTS_CONFIG_ID=restaurant-requests
OPERATION_ELEMENTS_CONFIG_ID=dish-requests
TENANT= # e.g. default
LOCAL_PDP_URL=
PERMIT_API_KEY=
PROJECT_ID=
ENV_ID=
 您可以使用以下资源获取这些值: 
 
 
 
 
 地点_PDP_URL 允许_API 密钥 项目ID 简介:ID ️注意:我们正在使用 为此教程支持ReBAC评估和低延迟,离线测试。 允许本地PDP(政策决策点) Start the Server 有了这一切,您现在可以本地运行MCP服务器: uv run -m src.permit_mcp
 一旦服务器运行,它将暴露您配置的允许元素(访问请求,批准管理等)作为代理可以通过MCP协议调用的工具。 创建一个 LangGraph + LangChain MCP 客户端 现在允许MCP服务器已启动并运行,我们将构建一个可以与之交互的AI代理客户端。 
 
 
 
 使用双胞胎驱动的LLM来决定采取哪些行动动态召唤MCP工具,如 request_access,approve_operation_approval等。 完全在 LangGraph 工作流中运行 使用interrupt() 进行人体审查的停机(下一节) 让我们连接点。 Install Required Dependencies 在您的 MCP 项目目录中,安装必要的包: uv add langchain-mcp-adapters langgraph langchain-google-genai
 这给了你: 
 
 
 
 langchain-mcp 适配器:自动将 Permit MCP 工具转换为 LangGraph 兼容的工具 langgraph: 用于组织基于图形的工作流程 langchain-google-genai:为了与双胞胎 2.0 Flash 互动 Add Google API Key 您将需要一个 API 密钥 使用双胞胎。 Google AI Studio 将钥匙添加到您的 檔案: .env GOOGLE_API_KEY=your-key-here
 Build the MCP Client 创建一个名为 在您的 Root 项目中 client.py 我们将这个文件分解成逻辑块: 
 
 
 
 
 
 
 Imports and Setup Start by importing dependencies and loading environment variables: import os
from typing_extensions import TypedDict, Literal, Annotated
from dotenv import load_dotenv
from langchain_google_genai import ChatGoogleGenerativeAI
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command, interrupt
from langgraph.checkpoint.memory import MemorySaver
from langgraph.prebuilt import ToolNode
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from langchain_mcp_adapters.tools import load_mcp_tools
import asyncio
from langgraph.graph.message import add_messages
 Then, load the environment and set up your Gemini LLM: load_dotenv()

global_llm_with_tools = None

llm = ChatGoogleGenerativeAI(
    model="gemini-2.0-flash",
    google_api_key=os.getenv('GOOGLE_API_KEY')
)
 
 
 
 
 
 Tell LangGraph how to communicate with the running MCP server: Configure MCP Server Parameters server_params = StdioServerParameters( command="python", args=["src/permit_mcp/server.py"], )
 定义共享代理状态: class State(TypedDict):
    messages: Annotated[list, add_messages]
 
 
 
 
 
 
  and the  : Define Workflow Nodes graph builder Here’s the logic to route between calling the LLM and invoking tools: async def call_llm(state):
    response = await global_llm_with_tools.ainvoke(state["messages"])
    return {"messages": [response]}

def route_after_llm(state) -> Literal[END, "run_tool"]:
    return END if len(state["messages"][-1].tool_calls) == 0 else "run_tool"

async def setup_graph(tools):
    builder = StateGraph(State)
    run_tool = ToolNode(tools)
    builder.add_node(call_llm)
    builder.add_node('run_tool', run_tool)

    builder.add_edge(START, "call_llm")
    builder.add_conditional_edges("call_llm", route_after_llm)
    builder.add_edge("run_tool", "call_llm")

    memory = MemorySaver()
    return builder.compile(checkpointer=memory)
 在上述代码中,我们定义了LLM节点及其条件边缘,该边缘路由到LLM节点。 节点如果状态的消息中有工具调用,或结束图表,我们还定义了一个函数来设置和编译图表,使用内存检查指针。 run_tool 接下来,添加以下代码行,从图表中流响应,并添加交互式聊天循环,直到它明确退出。 
 
 
 
 
 
 
  and an  : Stream Output and Handle Chat Input, infinite loop for user interaction      async def stream_responses(graph, config, invokeWith):
    async for event in graph.astream(invokeWith, config, stream_mode='updates'):
        for key, value in event.items():
            if key == 'call_llm':
                content = value["messages"][-1].content
                if content:
                    print('\\n' + ", ".join(content)
                          if isinstance(content, list) else content)

async def chat_loop(graph):
    while True:
        try:
            user_input = input("\\nQuery: ").strip()
            if user_input in ["quit", "exit", "q"]:
                print("Goodbye!")
                break

            sys_m = """
            Always provide the resource instance key during tool calls, as the ReBAC authorization model is being used. To obtain the resource instance key, use the list_resource_instances tool to view available resource instances.

            Always parse the provided data before displaying it.
            If the user has initially provided their ID, use that for subsequent tool calls without asking them again.
            """

            invokeWith = {"messages": [
                {"role": "user", "content": sys_m + '\\n\\n' + user_input}]}
            config = {"configurable": {"thread_id": "1"}}

            await stream_responses(graph, config, invokeWith)

        except Exception as e:
            print(f"Error: {e}")
 
 
 
 
 Final Assembly Add the main entry point where we will convert the Permit MCP server tool to LangGraph-compatible tools, bind our LLM to the resulting tools, set up the graph, draw it to a file, and fire up the chat loop: python
CopyEdit
async def main():
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            tools = await load_mcp_tools(session)
            llm_with_tools = llm.bind_tools(tools)
            graph = await setup_graph(tools)

            global global_llm_with_tools
            global_llm_with_tools = llm_with_tools

            with open("workflow_graph.png", "wb") as f:
                f.write(graph.get_graph().draw_mermaid_png())

            await chat_loop(graph)

if __name__ == "__main__":
    asyncio.run(main())
 
 
 Lastly,  Run the Client 一旦您保存了所有内容,请启动客户端: uv run client.py
 运行后,一个新的图像文件叫做 将被创建,这表明图表。 workflow_graph.png 有了所有设置,我们现在可以指定这样的查询: Query: My user id is henry, request access to pizza palace with the reason: I am now 18, and the role child-can-order
Query: My user id is joe, list all access requests
 Your agent is now able to call MCP tools dynamically! 添加人性循环与 interrupt() 随着 LangGraph 驱动的 MCP 客户端的启动和运行,允许工具现在可以自动调用。 例如,允许访问有限资源或批准高风险操作? sensitive, 这就是LangGraph的位置。 变得有用。 interrupt() 现在我们将添加一个 每当代理人试图召唤关键工具时,都会拦截和暂停工作流程,例如: human approval node 
 
 
 批准_访问_请求 批准_操作_批准 人类将被要求 在代理人继续前,工具呼叫。 manually approve or deny Define the Human Review Node 在你的顶部 檔案(以前) )添加以下函数: client.py setup_graph async def human_review_node(state) -> Command[Literal["call_llm", "run_tool"]]:
    """Handle human review process."""
    last_message = state["messages"][-1]
    tool_call = last_message.tool_calls[-1]

    high_risk_tools = ['approve_access_request', 'approve_operation_approval']
    if tool_call["name"] not in high_risk_tools:
        return Command(goto="run_tool")

    human_review = interrupt({
        "question": "Do you approve this tool call? (yes/no)",
        "tool_call": tool_call,
    })

    review_action = human_review["action"]

    if review_action == "yes":
        return Command(goto="run_tool")

    return Command(goto="call_llm", update={"messages": [{
        "role": "tool",
        "content": f"The user declined your request to execute the {tool_call.get('name', 'Unknown')} tool, with arguments {tool_call.get('args', 'N/A')}",
        "name": tool_call["name"],
        "tool_call_id": tool_call["id"],
    }]})
 此节点检查是否被调用的工具被认为是“高风险”。如果是,则图表被中断,请求人类确认。 Update Graph Routing 更改其 功能,使工具呼叫路由到人类审查节点,而不是立即运行: route_after_llm def route_after_llm(state) -> Literal[END, "human_review_node"]:
    """Route logic after LLM processing."""
    return END if len(state["messages"][-1].tool_calls) == 0 else "human_review_node"
 Wire in the HITL Node 更新中 功能 添加 作为图表中的一个节点: setup_graph human_review_node async def setup_graph(tools):
    builder = StateGraph(State)
    run_tool = ToolNode(tools)
    builder.add_node(call_llm)
    builder.add_node('run_tool', run_tool)
    builder.add_node(human_review_node)  # Add the interrupt node here

    builder.add_edge(START, "call_llm")
    builder.add_conditional_edges("call_llm", route_after_llm)
    builder.add_edge("run_tool", "call_llm")

    memory = MemorySaver()
    return builder.compile(checkpointer=memory)
 Handle Human Input During Runtime 最后,让我们增强你的 功能检测图表中断时,提示作出决定,并使用人类输入恢复 . stream_responses Command(resume={"action": user_input}) 运行客户端后,图表不应该像这样: 在运行客户端后,您的图形图表( )现在将包括在LLM和工具执行阶段之间的人类审查节点: workflow_graph.png 这确保了 当代理人试图做出可能改变权限或绕过限制的决定时。 you remain in control 通过这种方式,您已经成功地将人类的监督添加到您的AI代理,而不需要重新编写您的工具或后端逻辑。 结论 在本教程中,我们建立了一个安全,人为意识的AI代理,使用 . Permit.io 访问请求 MCP , 长篇 ,和 LangChain MCP 适配器 Permit.io 访问请求 MCP 长篇 LangChain MCP 适配器 相反,我们让代理人不受控制地运作,我们给了它权力。 访问和 就像一个负责任的团队成员一样。 request defer critical decisions to human users, We covered: 
 
 
 
 
 如何使用 Permit Elements 和 ReBAC 建模权限和批准流 如何通过允许MCP服务器曝光这些流 如何构建一个由LangGraph驱动的客户端,自然地调用这些工具 如何使用 interrupt() 实时插入人体循环(HITL)检查 想看到完整的演示在行动? 查看 . GitHub 回复 Further Reading - 
 
 
 
 
 
 通过允许网关安全 AI 协作 允许 MCP GitHub Repo LangChain MCP 适配器 允许Rebac政策 LangGraph interrupt() 引用

Walkthroughs, tutorials, guides, and tips. This story will teach you how to do something new or how to do something better.

該音頻是用故事的原始語言製作的！

通过 Permit.io 访问请求 MCP 向人类用户授权 AI 权限

About Author

註釋

標籤

这篇文章刊登在

Related Stories

成功云迁移的完整指南：策略和最佳实践

如何将您的工作流程提高 10 倍：17 个必备应用程序

创建以用户为中心的加密产品：客户反馈的重要性

架构师指南：构建 AI/ML 数据湖参考架构

成功云迁移的完整指南：策略和最佳实践

如何将您的工作流程提高 10 倍：17 个必备应用程序

创建以用户为中心的加密产品：客户反馈的重要性

架构师指南：构建 AI/ML 数据湖参考架构

Light-Mode

Classic

Newspaper

Minty

Dark-Mode

Neon Noir

Minty

HN StartUps