连接身份验证提供程序（第三部分/共三部分）¶
在使对话私有教程中，我们添加了资源授权以让用户进行私人对话。然而，我们仍然使用硬编码的令牌进行身份验证，这并不安全。现在我们将这些令牌替换为使用OAuth2的真实用户账户。
我们将保留相同的Auth对象和资源级访问控制，但将身份验证升级为使用Supabase作为我们的身份提供商。虽然本教程中使用Supabase，但这些概念适用于任何OAuth2提供商。您将学习如何：
- 用真实JWT令牌替换测试令牌
- 与OAuth2提供商集成以进行安全用户身份验证
- 处理用户会话和元数据，同时保持现有的授权逻辑
系统要求¶
为了使用 Supabase 的认证服务器，请需要设置一个 Supabase 项目。你可以在这里完成设置：here。
背景¶
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 控制面板中完成此操作：
- 在左侧边栏中，点击“⚙ 项目设置”，然后点击“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' for url 'http://localhost:2024/threads'
✅ 用户 2 被阻止访问用户 1 的线程: 客户端错误 '404 Not Found' for url 'http://localhost:2024/threads/d6af3754-95df-4176-aa10-dbd8dca40f1a'
完美！我们的认证和授权正在协同工作： 1. 用户必须登录才能访问机器人 2. 每个用户只能查看自己的线程
所有用户都由 Supabase 认证提供者管理，因此我们不需要实现任何额外的用户管理逻辑。
恭喜你！🎉¶
你已经成功地为LangGraph应用构建了一个生产就绪的身份验证系统！让我们回顾一下你所完成的工作：
- 设置了一个身份验证提供者（本例中为Supabase）
- 添加了带有电子邮件/密码身份验证的真实用户账户
- 将JWT令牌验证集成到了你的LangGraph服务器中
- 实现了适当的授权，以确保用户只能访问自己的数据
- 创建了一个可以应对下一个身份验证挑战的基础架构 🚀
这标志着我们的身份验证教程系列的完成。你现在拥有了构建一个安全、生产就绪的LangGraph应用所需的构建模块。
下一步是什么？¶
现在你已经有了生产环境的身份验证，可以考虑以下几点：