在本教程中，您将扩展 在上一教程中创建的聊天机器人 为每个用户提供自己的私密对话。您将添加 资源级访问控制 使用户只能看到自己的线程。
!授权流程：认证后，授权处理程序用 owner=user id 标记每个资源，并返回过滤器，使用户只能看到自己的线程。
前提条件
在开始本教程之前，请确保您已具备 第一个教程中的机器人 正常运行无错误。
1. 添加资源授权
回顾上一教程，Auth 对象允许您注册一个 认证函数，LangSmith 使用它来验证传入请求中的 bearer token。现在您将使用它来注册一个 **授权** handler.
授权处理程序是运行的函数 **在** 认证成功之后。这些处理程序可以添加 元数据 到资源上（如资源的所有者），并过滤每个用户可以查看的内容。
更新您的 src/security/auth.py 并添加一个授权处理程序，使其在每个请求上运行：
from langgraph_sdk import Auth
# Keep our test users from the previous tutorial
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:
"""Our authentication handler from the previous tutorial."""
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, # Contains info about the current user
value: dict, # The resource being created/accessed
):
"""Make resources private to their creator."""
# Examples:
# 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
# }
# }
# Does 2 things:
# 1. Add the user's ID to the resource's metadata. Each LangGraph resource has a `metadata` dict that persists with the resource.
# this metadata is useful for filtering in read and update operations
# 2. Return a filter that lets users only see their own resources
filters = {"owner": ctx.user.identity}
metadata = value.setdefault("metadata", {})
metadata.update(filters)
# Only let users see their own resources
return filters该处理程序接收两个参数：
ctx(AuthContext)：包含当前user、用户的permissions、resource（"threads"、"crons"、"assistants"）以及action正在执行的操作（"create"、"read"、"update"、"delete"、"search"、"create_run"）value(dict：正在创建或访问的数据。此字典的内容取决于所访问的资源类型和操作。请参阅下面的 添加作用域授权处理程序 以获取有关如何实现更精确的访问控制的信息。
请注意，简单的处理程序执行两个操作：
- 将用户的 ID 添加到资源的元数据中。
- 返回一个元数据过滤器，使用户只能看到自己拥有的资源。
2. 测试私密对话
测试您的授权。如果您正确设置了所有内容，您将看到所有 ✅ 消息。请确保您的开发服务器正在运行（运行 langgraph dev):
from langgraph_sdk import get_client
# Create clients for both users
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 creates an assistant
alice_assistant = await alice.assistants.create()
print(f"✅ Alice created assistant: {alice_assistant['assistant_id']}")
# Alice creates a thread and chats
alice_thread = await alice.threads.create()
print(f"✅ Alice created thread: {alice_thread['thread_id']}")
await alice.runs.create(
thread_id=alice_thread["thread_id"],
assistant_id="agent",
input={"messages": [{"role": "user", "content": "Hi, this is Alice's private chat"}]}
)
# Bob tries to access Alice's thread
try:
await bob.threads.get(alice_thread["thread_id"])
print("❌ Bob shouldn't see Alice's thread!")
except Exception as e:
print("✅ Bob correctly denied access:", e)
# Bob creates his own thread
bob_thread = await bob.threads.create()
await bob.runs.create(
thread_id=bob_thread["thread_id"],
assistant_id="agent",
input={"messages": [{"role": "user", "content": "Hi, this is Bob's private chat"}]}
)
print(f"✅ Bob created his own thread: {bob_thread['thread_id']}")
# List threads - each user only sees their own
alice_threads = await alice.threads.search()
bob_threads = await bob.threads.search()
print(f"✅ Alice sees {len(alice_threads)} thread")
print(f"✅ Bob sees {len(bob_threads)} thread")Output:
✅ Alice created assistant: fc50fb08-78da-45a9-93cc-1d3928a3fc37
✅ Alice created thread: 533179b7-05bc-4d48-b47a-a83cbdb5781d
✅ Bob correctly denied access: Client error '404 Not Found' for url 'http://localhost:2024/threads/533179b7-05bc-4d48-b47a-a83cbdb5781d'
For more information check: https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/404
✅ Bob created his own thread: 437c36ed-dd45-4a1e-b484-28ba6eca8819
✅ Alice sees 1 thread
✅ Bob sees 1 thread这意味着：
- 每个用户可以在自己的线程中创建和聊天
- 用户无法看到彼此的线程
- 列出线程只显示您自己的
<a id="scoped-authorization"></a> ## 3. 添加作用域授权处理程序
宽泛的 @auth.on 处理程序与所有 授权事件相匹配。这很简洁，但这意味着 value 字典的作用域不够细致，相同的用户级访问控制被应用于每个资源。如果您想要更细粒度的控制，也可以对特定资源操作进行控制。
更新 src/security/auth.py 以为特定资源类型添加处理程序：
# Keep our previous handlers...
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,
):
"""Add owner when creating threads.
This handler runs when creating new threads and does two things:
1. Sets metadata on the thread being created to track ownership
2. Returns a filter that ensures only the creator can access it
"""
# Example value:
# {'thread_id': UUID('99b045bc-b90b-41a8-b882-dabc541cf740'), 'metadata': {}, 'if_exists': 'raise'}
# Add owner metadata to the thread being created
# This metadata is stored with the thread and persists
metadata = value.setdefault("metadata", {})
metadata["owner"] = ctx.user.identity
# Return filter to restrict access to just the creator
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,
):
"""Only let users read their own threads.
This handler runs on read operations. We don't need to set
metadata since the thread already exists - we just need to
return a filter to ensure users can only see their own threads.
"""
return {"owner": ctx.user.identity}
@auth.on.assistants
async def on_assistants(
ctx: Auth.types.AuthContext,
value: Auth.types.on.assistants.value,
):
# For illustration purposes, we will deny all requests
# that touch the assistants resource
# Example 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 lacks the required permissions.",
)
# Assumes you organize information in store like (user_id, resource_type, resource_id)
@auth.on.store()
async def authorize_store(ctx: Auth.types.AuthContext, value: dict):
# The "namespace" field for each store item is a tuple you can think of as the directory of an item.
namespace: tuple = value["namespace"]
assert namespace[0] == ctx.user.identity, "Not authorized"请注意，您现在拥有针对以下内容的特定处理程序，而不是一个全局处理程序：
- 创建线程
- 读取线程
- 访问助手
前三个对应特定的 **操作** 每个资源（参见 资源操作），而最后两个（@auth.on.assistants 和 @auth.on.store）匹配 _任何_ 资源上的操作 assistants 和 store 资源上的任何操作。对于每个请求，LangGraph 将运行最具体的匹配处理程序。这意味着上述四个处理程序将优先执行，而不是使用更广泛的作用域的"@auth.on"处理程序。
存储授权的工作方式与threads和assistants不同。关于存储隔离模式和权衡，参见 为每个用户隔离存储.
尝试将以下测试代码添加到测试文件中：
# ... Same as before
# Try creating an assistant. This should fail
try:
await alice.assistants.create("agent")
print("❌ Alice shouldn't be able to create assistants!")
except Exception as e:
print("✅ Alice correctly denied access:", e)
# Try searching for assistants. This also should fail
try:
await alice.assistants.search()
print("❌ Alice shouldn't be able to search assistants!")
except Exception as e:
print("✅ Alice correctly denied access to searching assistants:", e)
# Alice can still create threads
alice_thread = await alice.threads.create()
print(f"✅ Alice created thread: {alice_thread['thread_id']}")Output:
✅ Alice created thread: dcea5cd8-eb70-4a01-a4b6-643b14e8f754
✅ Bob correctly denied access: Client error '404 Not Found' for url 'http://localhost:2024/threads/dcea5cd8-eb70-4a01-a4b6-643b14e8f754'
For more information check: https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/404
✅ Bob created his own thread: 400f8d41-e946-429f-8f93-4fe395bc3eed
✅ Alice sees 1 thread
✅ Bob sees 1 thread
✅ Alice correctly denied access:
For more information check: https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/500
✅ Alice correctly denied access to searching assistants:恭喜！您已构建了一个聊天机器人，每个用户都能进行私人对话。虽然此系统使用简单的基于令牌的身份验证，但这些授权模式可与任何真实身份验证系统配合使用。在下一个教程中，您将使用 OAuth2 将测试用户替换为真实用户账户。
后续步骤
既然您可以控制对资源的访问，您可能想要：
- 继续进行 连接身份验证提供程序 以添加真实用户账户。
- 了解更多关于 授权模式.
- 查看 API 参考 了解本教程中使用的接口和方法的详细信息。