连接身份验证提供程序（第 3 部分，共 3 部分）¶
在使对话保持私密教程中，我们添加了资源授权，以便用户可以拥有私密对话。然而，我们仍然使用硬编码的令牌进行身份验证，这是不安全的。现在我们将使用OAuth2，用真实用户账户替换这些令牌。
我们将保留相同的Auth对象和资源级别访问控制，但将身份验证升级为使用 Supabase 作为身份提供程序。虽然本教程中使用了 Supabase，但这些概念适用于任何 OAuth2 提供程序。你将学习如何：
- 用真实的JWT 令牌替换测试令牌
- 与 OAuth2 提供程序集成以实现安全用户身份验证
- 在保留现有授权逻辑的同时处理用户会话和元数据
要求¶
你需要设置一个 Supabase 项目，以便在本教程中使用其身份验证服务器。你可以在此处完成此操作。
背景¶
OAuth2 包含三个主要角色：
- 授权服务器：负责用户身份验证和颁发令牌的身份提供程序（例如，Supabase、Auth0、Google）
- 应用程序后端：你的 LangGraph 应用程序。它负责验证令牌并提供受保护资源（对话数据）
- 客户端应用程序：用户与你的服务交互的 Web 或移动应用程序
标准的 OAuth2 流程大致如下：
sequenceDiagram
participant User
participant Client
participant AuthServer
participant LangGraph Backend
User->>Client: Initiate login
User->>AuthServer: Enter credentials
AuthServer->>Client: Send tokens
Client->>LangGraph Backend: Request with token
LangGraph Backend->>AuthServer: Validate token
AuthServer->>LangGraph Backend: Token valid
LangGraph Backend->>Client: Serve request (e.g., run agent or graph)在下面的示例中，我们将使用 Supabase 作为我们的身份验证服务器。LangGraph 应用程序将作为你的应用后端，我们将为客户端应用编写测试代码。让我们开始吧！
设置身份验证提供程序¶
首先，让我们安装所需的依赖项。在你所在的 custom-auth 目录中开始，并确保已安装 langgraph-cli
接下来，我们需要获取身份验证服务器的 URL 和身份验证私钥。由于我们在此使用 Supabase，我们可以在 Supabase 控制面板中完成此操作：
- 在左侧边栏中，点击“⚙ 项目设置”，然后点击“API”
- 复制你的项目 URL 并将其添加到你的
.env文件中
.env 文件中4. 最后，复制你的“匿名公钥（anon public）”并记下来。稍后设置客户端代码时将使用此密钥。
实现令牌验证¶
在之前的教程中，我们使用 Auth 对象来：
现在我们将升级身份验证，以验证来自 Supabase 的真实 JWT 令牌。主要更改将都在 @auth.authenticate 装饰器函数中：
- 不再针对硬编码的令牌列表进行检查，我们将向 Supabase 发起 HTTP 请求来验证令牌
- 我们将从已验证的令牌中提取真实的用户信息（ID、邮箱）
并且我们将保持现有的资源授权逻辑不变
让我们更新 src/security/auth.py 来实现这一点
import os
import httpx
from langgraph_sdk import Auth
auth = Auth()
# This is loaded from the `.env` file you created above
SUPABASE_URL = os.environ["SUPABASE_URL"]
SUPABASE_SERVICE_KEY = os.environ["SUPABASE_SERVICE_KEY"]
@auth.authenticate
async def get_current_user(authorization: str | None):
"""Validate JWT tokens and extract user information."""
assert authorization
scheme, token = authorization.split()
assert scheme.lower() == "bearer"
try:
# Verify token with auth provider
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"], # Unique user identifier
"email": user["email"],
"is_authenticated": True,
}
except Exception as e:
raise Auth.exceptions.HTTPException(status_code=401, detail=str(e))
# ... the rest is the same as before
# Keep our resource authorization from the previous tutorial
@auth.on
async def add_owner(ctx, value):
"""Make resources private to their creator using resource metadata."""
filters = {"owner": ctx.user.identity}
metadata = value.setdefault("metadata", {})
metadata.update(filters)
return filters
最重要的变化是，我们现在使用真实的身份验证服务器来验证令牌。我们的身份验证处理程序拥有 Supabase 项目的私钥，我们可以用它来验证用户的令牌并提取他们的信息。
让我们用真实用户账户来测试一下！
测试身份验证流程¶
让我们测试一下新的身份验证流程。你可以将以下代码在一个文件或 notebook 中运行。你需要提供：
import os
import httpx
from getpass import getpass
from langgraph_sdk import get_client
# Get email from command line
email = getpass("Enter your email: ")
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("Enter your Supabase project URL: ")
# This is your PUBLIC anon key (which is safe to use client-side)
# Do NOT mistake this for the secret service role key
SUPABASE_ANON_KEY = os.environ.get("SUPABASE_ANON_KEY")
if not SUPABASE_ANON_KEY:
SUPABASE_ANON_KEY = getpass("Enter your public Supabase anon key: ")
async def sign_up(email: str, password: str):
"""Create a new user account."""
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()
# Create two test users
print(f"Creating test users: {email1} and {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 控制面板复制的“匿名公钥（anon public）”。
async def login(email: str, password: str):
"""Get an access token for an existing user."""
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"]
# Log in as user 1
user1_token = await login(email1, password)
user1_client = get_client(
url="http://localhost:2024", headers={"Authorization": f"Bearer {user1_token}"}
)
# Create a thread as user 1
thread = await user1_client.threads.create()
print(f"✅ User 1 created thread: {thread['thread_id']}")
# Try to access without a token
unauthenticated_client = get_client(url="http://localhost:2024")
try:
await unauthenticated_client.threads.create()
print("❌ Unauthenticated access should fail!")
except Exception as e:
print("✅ Unauthenticated access blocked:", e)
# Try to access user 1's thread as user 2
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("❌ User 2 shouldn't see User 1's thread!")
except Exception as e:
print("✅ User 2 blocked from User 1's thread:", e)
✅ User 1 created thread: d6af3754-95df-4176-aa10-dbd8dca40f1a
✅ Unauthenticated access blocked: Client error '403 Forbidden' for url 'http://localhost:2024/threads'
✅ User 2 blocked from User 1's thread: Client error '404 Not Found' for url 'http://localhost:2024/threads/d6af3754-95df-4176-aa10-dbd8dca40f1a'
太棒了！我们的身份验证和授权协同工作：1. 用户必须登录才能访问机器人 2. 每个用户只能看到自己的线程
所有用户都由 Supabase 身份验证提供程序管理，因此我们无需实现任何额外的用户管理逻辑。
恭喜！🎉¶
你已成功为你的 LangGraph 应用程序构建了生产级别的身份验证系统！让我们回顾一下你完成的工作：
- 设置了身份验证提供程序（在本例中为 Supabase）
- 添加了使用电子邮件/密码进行身份验证的真实用户账户
- 将 JWT 令牌验证集成到你的 LangGraph 服务器中
- 实现了适当的授权，确保用户只能访问自己的数据
- 构建了一个基础，准备好应对你的下一个身份验证挑战 🚀
这就完成了我们的身份验证教程系列。现在你拥有了构建安全、生产级 LangGraph 应用程序的基础。
接下来是什么？¶
既然你已经有了生产级别的身份验证，请考虑：