让对话私有化（第2部分/共3部分）¶

这是我们的认证系列的第二部分：

基本认证 - 控制谁可以访问您的机器人
资源授权（您现在的位置）- 让用户拥有私人的对话
生产认证 - 添加真实用户账户并使用OAuth2验证

在本教程中，我们将扩展聊天机器人的功能，为每个用户提供自己的私人对话。我们将添加资源级访问控制，使用户只能看到他们自己的线程。

授权处理器

占位符令牌

正如我们在第一部分中所做的那样，为了说明目的，在这一部分中我们将使用硬编码的令牌。我们将在第三部分中掌握基础知识后，实现一个“生产就绪”的认证方案。

理解资源授权¶

在上一个教程中，我们控制了谁可以访问我们的机器人。但是现在，任何经过身份验证的用户都可以查看其他人的对话！让我们通过添加资源授权来解决这个问题。

首先，请确保您已完成基本认证教程，并且您的安全机器人可以在不报错的情况下运行：

cd custom-auth
pip install -e .
langgraph dev --no-browser

🚀 API: http://127.0.0.1:2024

🎨 Studio UI: https://smith.langchain.com/studio/?baseUrl=http://127.0.0.1:2024

📚 API 文档: http://127.0.0.1:2024/docs

添加资源授权¶

回想一下，在上一个教程中，Auth 对象让我们注册了一个认证函数，LangGraph 平台使用该函数来验证传入请求中的承载令牌。现在我们将使用它来注册一个**授权**处理器。

授权处理器是在认证成功后运行的函数。这些处理器可以向资源（如谁拥有它们）添加元数据，并过滤每个用户可以看到的内容。

让我们更新 src/security/auth.py 并添加一个在每个请求上运行的授权处理器：

src/security/auth.py

from langgraph_sdk import Auth

# 保持我们从上一个教程中使用的测试用户
VALID_TOKENS = {
    "user1-token": {"id": "user1", "name": "Alice"},
    "user2-token": {"id": "user2", "name": "Bob"},
}

auth = Auth()


@auth.authenticate
async def get_current_user(authorization: str | None) -> Auth.types.MinimalUserDict:
    """我们从上一个教程中使用的认证处理器。"""
    assert authorization
    scheme, token = authorization.split()
    assert scheme.lower() == "bearer"

    if token not in VALID_TOKENS:
        raise Auth.exceptions.HTTPException(status_code=401, detail="Invalid token")

    user_data = VALID_TOKENS[token]
    return {
        "identity": user_data["id"],
    }


@auth.on
async def add_owner(
    ctx: Auth.types.AuthContext,  # 包含当前用户的详细信息
    value: dict,  # 正在创建或访问的资源
):
    """使资源对其创建者私有化。"""
    # 示例：
    # ctx: AuthContext(
    #     permissions=[],
    #     user=ProxyUser(
    #         identity='user1',
    #         is_authenticated=True,
    #         display_name='user1'
    #     ),
    #     resource='threads',
    #     action='create_run'
    # )
    # value: 
    # {
    #     'thread_id': UUID('1e1b2733-303f-4dcd-9620-02d370287d72'),
    #     'assistant_id': UUID('fe096781-5601-53d2-b2f6-0d3403f7e9ca'),
    #     'run_id': UUID('1efbe268-1627-66d4-aa8d-b956b0f02a41'),
    #     'status': 'pending',
    #     'metadata': {},
    #     'prevent_insert_if_inflight': True,
    #     'multitask_strategy': 'reject',
    #     'if_not_exists': 'reject',
    #     'after_seconds': 0,
    #     'kwargs': {
    #         'input': {'messages': [{'role': 'user', 'content': 'Hello!'}]},
    #         'command': None,
    #         'config': {
    #             'configurable': {
    #                 'langgraph_auth_user': ... Your user object...
    #                 'langgraph_auth_user_id': 'user1'
    #             }
    #         },
    #         'stream_mode': ['values'],
    #         'interrupt_before': None,
    #         'interrupt_after': None,
    #         'webhook': None,
    #         'feedback_keys': None,
    #         'temporary': False,
    #         'subgraphs': False
    #     }
    # }

    # 做两件事：
    # 1. 将用户的ID添加到资源的元数据中。每个LangGraph资源都有一个持久化的`metadata`字典。
    # 这些元数据在读取和更新操作时非常有用。
    # 2. 返回一个过滤器，让用户只能看到他们自己的资源。
    filters = {"owner": ctx.user.identity}
    metadata = value.setdefault("metadata", {})
    metadata.update(filters)

    # 只让用户看到他们自己的资源
    return filters

处理器接收两个参数：

ctx (AuthContext)：包含有关当前用户的信息、用户的权限、正在执行的操作的资源（“线程”，“定时任务”，“助手”），以及正在执行的动作（“创建”，“读取”，“更新”，“删除”，“搜索”，“创建运行”）
value (dict)：正在创建或访问的数据。此字典的内容取决于所访问的资源和动作。有关如何获取更严格的访问控制，请参阅下方的添加范围授权处理器。

请注意，我们的简单处理器做了两件事：

将用户的ID添加到资源的元数据中。
返回一个元数据过滤器，让用户只能看到他们拥有的资源。

测试私密对话¶

让我们测试一下授权功能。如果设置正确，我们应该会看到所有✅消息。确保你的开发服务器正在运行（运行langgraph dev）：

from langgraph_sdk import get_client

# 创建两个用户的客户端
alice = get_client(
    url="http://localhost:2024",
    headers={"Authorization": "Bearer user1-token"}
)

bob = get_client(
    url="http://localhost:2024",
    headers={"Authorization": "Bearer user2-token"}
)

# 爱丽丝创建了一个助手
alice_assistant = await alice.assistants.create()
print(f"✅ 爱丽丝创建了助手: {alice_assistant['assistant_id']}")

# 爱丽丝创建了一个线程并开始聊天
alice_thread = await alice.threads.create()
print(f"✅ 爱丽丝创建了线程: {alice_thread['thread_id']}")

await alice.runs.create(
    thread_id=alice_thread["thread_id"],
    assistant_id="agent",
    input={"messages": [{"role": "user", "content": "你好，这是爱丽丝的私密对话"}]}
)

# 鲍勃尝试访问爱丽丝的线程
try:
    await bob.threads.get(alice_thread["thread_id"])
    print("❌ 鲍勃不应该看到爱丽丝的线程！")
except Exception as e:
    print("✅ 鲍勃正确地被拒绝访问:", e)

# 鲍勃创建自己的线程
bob_thread = await bob.threads.create()
await bob.runs.create(
    thread_id=bob_thread["thread_id"],
    assistant_id="agent",
    input={"messages": [{"role": "user", "content": "你好，这是鲍勃的私密对话"}]}
)
print(f"✅ 鲍勃创建了自己的线程: {bob_thread['thread_id']}")

# 列出线程 - 每个用户只能看到自己的线程
alice_threads = await alice.threads.search()
bob_threads = await bob.threads.search()
print(f"✅ 爱丽丝看到了 {len(alice_threads)} 个线程")
print(f"✅ 鲍勃看到了 {len(bob_threads)} 个线程")

运行测试代码，你应该会看到类似以下的输出：

✅ 爱丽丝创建了助手: fc50fb08-78da-45a9-93cc-1d3928a3fc37
✅ 爱丽丝创建了线程: 533179b7-05bc-4d48-b47a-a83cbdb5781d
✅ 鲍勃正确地被拒绝访问: 客户端错误 '404 Not Found' 对于 URL 'http://localhost:2024/threads/533179b7-05bc-4d48-b47a-a83cbdb5781d'
更多信息请参阅: https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/404
✅ 鲍勃创建了自己的线程: 437c36ed-dd45-4a1e-b484-28ba6eca8819
✅ 爱丽丝看到了 1 个线程
✅ 鲍勃看到了 1 个线程

这意味着：

每个用户都可以创建并参与自己的线程。
用户不能看到其他人的线程。
列出线程时只会显示自己的线程。

添加范围授权处理程序¶

广义的 @auth.on 处理程序匹配所有授权事件。这种方式简洁明了，但这也意味着 value 字典的内容没有很好地限定范围，并且我们对每个资源都应用相同级别的用户访问控制。如果我们希望更精细地控制，也可以针对特定资源的操作进行控制。

更新 src/security/auth.py 文件以添加特定资源类型的处理程序：

# 保留之前的处理程序...

from langgraph_sdk import Auth

@auth.on.threads.create
async def on_thread_create(
    ctx: Auth.types.AuthContext,
    value: Auth.types.on.threads.create.value,
):
    """在创建线程时添加所有者。

    此处理程序在创建新线程时运行，并执行以下两项操作：
    1. 在正在创建的线程上设置元数据以跟踪所有权
    2. 返回一个过滤器，确保只有创建者可以访问它
    """
    # 示例值：
    #  {'thread_id': UUID('99b045bc-b90b-41a8-b882-dabc541cf740'), 'metadata': {}, 'if_exists': 'raise'}

    # 将所有者元数据添加到正在创建的线程中
    # 此元数据与线程一起存储并持久化
    metadata = value.setdefault("metadata", {})
    metadata["owner"] = ctx.user.identity

    # 返回过滤器以限制仅创建者可以访问
    return {"owner": ctx.user.identity}

@auth.on.threads.read
async def on_thread_read(
    ctx: Auth.types.AuthContext,
    value: Auth.types.on.threads.read.value,
):
    """只允许用户读取自己的线程。

    此处理程序在读取操作时运行。由于线程已经存在，因此不需要设置元数据——我们只需要返回一个过滤器，以确保用户只能看到自己的线程。
    """
    return {"owner": ctx.user.identity}

@auth.on.assistants
async def on_assistants(
    ctx: Auth.types.AuthContext,
    value: Auth.types.on.assistants.value,
):
    # 为了说明目的，我们将拒绝所有触及助手资源的请求
    # 示例值：
    # {
    #     'assistant_id': UUID('63ba56c3-b074-4212-96e2-cc333bbc4eb4'),
    #     'graph_id': 'agent',
    #     'config': {},
    #     'metadata': {},
    #     'name': 'Untitled'
    # }
    raise Auth.exceptions.HTTPException(
        status_code=403,
        detail="用户缺少所需的权限。",
    )

# 假设您将信息组织在存储中，如(user_id, resource_type, resource_id)
@auth.on.store()
async def authorize_store(ctx: Auth.types.AuthContext, value: dict):
    # 每个存储项的“命名空间”字段是一个你可以将其视为项目目录的元组。
    namespace: tuple = value["namespace"]
    assert namespace[0] == ctx.user.identity, "未授权"

请注意，我们现在有了特定的处理程序而不是一个全局处理程序，这些处理程序包括：

创建线程
读取线程
访问助手

前三个处理程序匹配每个资源上的特定**操作**（参见资源操作），而最后一个处理程序 (@auth.on.assistants) 匹配 assistants 资源上的**任何**操作。对于每个请求，LangGraph 将运行最具体的匹配当前资源和操作的处理程序。这意味着上述四个处理程序将运行，而不是广义的 @auth.on 处理程序。

尝试将以下测试代码添加到您的测试文件中：

# ... 同以前一样
# 尝试创建一个助手。这应该失败
try:
    await alice.assistants.create("agent")
    print("❌ 爱丽丝不应该能够创建助手！")
except Exception as e:
    print("✅ 爱丽丝正确地拒绝了访问：", e)

# 尝试搜索助手。这也应该失败
try:
    await alice.assistants.search()
    print("❌ 爱丽丝不应该能够搜索助手！")
except Exception as e:
    print("✅ 爱丽丝正确地拒绝了搜索助手的访问：", e)

# 爱丽丝仍然可以创建线程
alice_thread = await alice.threads.create()
print(f"✅ 爱丽丝创建了线程：{alice_thread['thread_id']}")
``

然后再次运行测试代码：

```bash
✅ 爱丽丝创建了线程：dcea5cd8-eb70-4a01-a4b6-643b14e8f754
✅ 鲍勃正确地拒绝了访问：客户端错误 '404 Not Found' 对于 url 'http://localhost:2024/threads/dcea5cd8-eb70-4a01-a4b6-643b14e8f754'
更多信息请查看：https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/404
✅ 鲍勃创建了自己的线程：400f8d41-e946-429f-8f93-4fe395bc3eed
✅ 爱丽丝看到了 1 个线程
✅ 鲍勃看到了 1 个线程
✅ 爱丽丝正确地拒绝了访问：
更多信息请查看：https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/500
✅ 爱丽丝正确地拒绝了搜索助手的访问：

恭喜！您已经建立了一个聊天机器人，其中每个用户都有自己的私密对话。虽然此系统使用简单的基于令牌的身份验证，但我们所学习的授权模式将适用于实现任何真实的身份验证系统。在下一个教程中，我们将用真实的用户账户替换测试用户，使用OAuth2。

接下来做什么？¶

现在你可以控制对资源的访问了，你可能还想：

继续阅读生产环境认证，以添加真实用户账户
了解更多关于授权模式的信息
查看API参考，了解本教程中使用的接口和方法的详细信息