连接身份验证提供程序（第3部分/共3部分）¶
在使对话私有化教程中，我们添加了资源授权，以让用户拥有私有对话。然而，我们仍然使用硬编码的令牌进行认证，这并不安全。现在我们将这些令牌替换为使用OAuth2的真实用户账户。
我们将继续使用相同的Auth对象和资源级访问控制，但升级我们的认证以使用Supabase作为我们的身份提供商。虽然本教程中使用Supabase，但这些概念适用于任何OAuth2提供商。您将学习如何：
- 将测试令牌替换为真实的JWT令牌
- 与OAuth2提供商集成，以实现安全的用户认证
- 在保持现有授权逻辑的同时处理用户会话和元数据
需求¶
为了使用本教程中的Supabase认证服务器，您需要设置一个Supabase项目。您可以在此处完成设置：此处。
背景¶
OAuth2 涉及三个主要角色：
- 授权服务器：处理用户身份验证并发放令牌的身份提供商（例如，Supabase、Auth0、Google）
- 应用后端：您的 LangGraph 应用程序。它验证令牌并提供受保护的资源（对话数据）
- 客户端应用程序：用户与您的服务交互的网页或移动应用
标准的 OAuth2 流程大致如下所示：
sequenceDiagram
participant User
participant Client
participant AuthServer
participant LangGraph Backend
User->>Client: 初始化登录
User->>AuthServer: 输入凭据
AuthServer->>Client: 发送令牌
Client->>LangGraph Backend: 使用令牌请求
LangGraph Backend->>AuthServer: 验证令牌
AuthServer->>LangGraph Backend: 令牌有效
LangGraph Backend->>Client: 提供请求（例如运行代理或图）在接下来的例子中，我们将使用 Supabase 作为我们的授权服务器。LangGraph 应用程序将为您的应用提供后端，并且我们将编写测试代码用于客户端应用。 让我们开始吧！
设置身份验证提供者¶
首先，让我们安装所需的依赖项。在您的 custom-auth 目录中开始操作，并确保已安装 langgraph-cli：
接下来，我们需要获取身份验证服务器的URL以及私钥。 由于我们使用的是Supabase，因此可以在Supabase仪表板中完成此操作：
- 在左侧边栏中，点击“项目设置”（⚙ Project Settings），然后点击“API”
- 复制您的项目URL并将其添加到您的
.env文件中
-
接下来，复制您的服务角色密钥并将其添加到您的
.env文件中 -
最后，复制您的“匿名公共”密钥并记下它。稍后在设置客户端代码时会用到。
实现令牌验证¶
在之前的教程中，我们使用了Auth对象来：
现在我们将升级我们的认证以验证来自Supabase的真实JWT令牌。所有的关键更改都将发生在由@auth.authenticate装饰的函数中：
- 不再检查硬编码的令牌列表，而是向Supabase发送一个HTTP请求来验证令牌。
- 从验证后的令牌中提取真实用户信息（ID、邮箱）。
同时，我们将保持现有的资源授权逻辑不变。
让我们更新src/security/auth.py以实现这一点：
import os
import httpx
from langgraph_sdk import Auth
auth = Auth()
# 这是从上面创建的`.env`文件中加载的
SUPABASE_URL = os.environ["SUPABASE_URL"]
SUPABASE_SERVICE_KEY = os.environ["SUPABASE_SERVICE_KEY"]
@auth.authenticate
async def get_current_user(authorization: str | None):
"""验证JWT令牌并提取用户信息。"""
assert authorization
scheme, token = authorization.split()
assert scheme.lower() == "bearer"
try:
# 使用身份提供者验证令牌
async with httpx.AsyncClient() as client:
response = await client.get(
f"{SUPABASE_URL}/auth/v1/user",
headers={
"Authorization": authorization,
"apiKey": SUPABASE_SERVICE_KEY,
},
)
assert response.status_code == 200
user = response.json()
return {
"identity": user["id"], # 唯一的用户标识符
"email": user["email"],
"is_authenticated": True,
}
except Exception as e:
raise Auth.exceptions.HTTPException(status_code=401, detail=str(e))
# ... 其余部分与之前相同
# 保持我们从上一个教程中使用的资源授权逻辑
@auth.on
async def add_owner(ctx, value):
"""通过资源元数据使资源对其创建者私有。"""
filters = {"owner": ctx.user.identity}
metadata = value.setdefault("metadata", {})
metadata.update(filters)
return filters
最重要的变化是我们现在使用真实的认证服务器来验证令牌。我们的认证处理器拥有Supabase项目的私钥，可以用来验证用户的令牌并提取其信息。
让我们用一个真实的用户账户来测试这个功能！
测试认证流程¶
让我们测试一下新的认证流程。你可以在一个文件或笔记本中运行以下代码。你需要提供：
import os
import httpx
from getpass import getpass
from langgraph_sdk import get_client
# 从命令行获取电子邮件地址
email = getpass("请输入您的电子邮件：")
base_email = email.split("@")
password = "secure-password" # CHANGEME
email1 = f"{base_email[0]}+1@{base_email[1]}"
email2 = f"{base_email[0]}+2@{base_email[1]}"
SUPABASE_URL = os.environ.get("SUPABASE_URL")
if not SUPABASE_URL:
SUPABASE_URL = getpass("请输入您的 Supabase 项目 URL：")
# 这是您的公共匿名密钥（在客户端使用时是安全的）
# 不要将其与秘密服务角色密钥混淆
SUPABASE_ANON_KEY = os.environ.get("SUPABASE_ANON_KEY")
if not SUPABASE_ANON_KEY:
SUPABASE_ANON_KEY = getpass("请输入您的公共 Supabase 匿名密钥：")
async def sign_up(email: str, password: str):
"""创建一个新的用户账户。"""
async with httpx.AsyncClient() as client:
response = await client.post(
f"{SUPABASE_URL}/auth/v1/signup",
json={"email": email, "password": password},
headers={"apiKey": SUPABASE_ANON_KEY},
)
assert response.status_code == 200
return response.json()
# 创建两个测试用户
print(f"正在创建测试用户：{email1} 和 {email2}")
await sign_up(email1, password)
await sign_up(email2, password)
然后运行代码。
关于测试电子邮件
我们将通过在您的电子邮件地址后面添加“+1”和“+2”来创建两个测试账户。例如，如果您使用的是 "myemail@gmail.com"，我们将创建 "myemail+1@gmail.com" 和 "myemail+2@gmail.com"。所有电子邮件都将发送到您的原始地址。
⚠️ 在继续之前，请检查您的电子邮件并点击两个确认链接。Supabase 将会拒绝 /login 请求，直到您确认了用户的电子邮件。
现在让我们测试用户只能看到自己的数据。确保服务器正在运行（运行 langgraph dev）后再继续。以下片段需要您在 设置身份验证提供者 时从 Supabase 控制台复制的“匿名公共”密钥。
async def login(email: str, password: str):
"""为现有用户获取访问令牌。"""
async with httpx.AsyncClient() as client:
response = await client.post(
f"{SUPABASE_URL}/auth/v1/token?grant_type=password",
json={
"email": email,
"password": password
},
headers={
"apikey": SUPABASE_ANON_KEY,
"Content-Type": "application/json"
},
)
assert response.status_code == 200
return response.json()["access_token"]
# 以用户 1 身份登录
user1_token = await login(email1, password)
user1_client = get_client(
url="http://localhost:2024", headers={"Authorization": f"Bearer {user1_token}"}
)
# 用户 1 创建一个线程
thread = await user1_client.threads.create()
print(f"✅ 用户 1 创建了线程：{thread['thread_id']}")
# 尝试不带令牌访问
unauthenticated_client = get_client(url="http://localhost:2024")
try:
await unauthenticated_client.threads.create()
print("❌ 未授权访问应该失败！")
except Exception as e:
print("✅ 未授权访问被阻止：", e)
# 尝试作为用户 2 访问用户 1 的线程
user2_token = await login(email2, password)
user2_client = get_client(
url="http://localhost:2024", headers={"Authorization": f"Bearer {user2_token}"}
)
try:
await user2_client.threads.get(thread["thread_id"])
print("❌ 用户 2 不应看到用户 1 的线程！")
except Exception as e:
print("✅ 用户 2 被阻止访问用户 1 的线程：", e)
输出应该如下所示：
✅ 用户 1 创建了线程：d6af3754-95df-4176-aa10-dbd8dca40f1a
✅ 未授权访问被阻止：客户端错误 '403 Forbidden' 对于 URL 'http://localhost:2024/threads'
✅ 用户 2 被阻止访问用户 1 的线程：客户端错误 '404 Not Found' 对于 URL 'http://localhost:2024/threads/d6af3754-95df-4176-aa10-dbd8dca40f1a'
完美！我们的认证和授权一起工作： 1. 用户必须登录才能访问机器人 2. 每个用户只能看到自己的线程
我们所有的用户都由 Supabase 认证提供者管理，因此我们不需要实现任何额外的用户管理逻辑。
恭喜！🎉¶
您已经成功为LangGraph应用构建了一个生产就绪的身份验证系统！让我们回顾一下您所完成的工作：
- 设置了一个身份验证提供者（本例中使用的是Supabase）
- 添加了带有电子邮件/密码认证的真实用户账户
- 将JWT令牌验证集成到您的LangGraph服务器中
- 实现了适当的授权机制，以确保用户只能访问自己的数据
- 建立了一个可以应对下一个身份验证挑战的基础 🚀
这标志着我们的身份验证教程系列已经完成。现在您拥有了构建一个安全、生产就绪的LangGraph应用所需的组件。
下一步是什么？¶
现在你已经有了生产环境的身份验证，可以考虑以下几点：