连接身份验证提供程序（第 3 部分/共 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. 最后，复制你的 “匿名公共” 密钥并记录下来。在我们设置客户端代码时会用到它。
实现令牌验证¶
在之前的教程中，我们使用 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):
"""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))
# ... 其余部分与之前相同
# 保留上一教程中的资源授权
@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 项目的私钥，我们可以使用该私钥来验证用户的令牌并提取他们的信息。
让我们使用一个真实的用户账户来测试这个功能！
测试认证流程¶
让我们来测试一下新的认证流程。你可以在文件或笔记本中运行以下代码。你需要提供：
import os
import httpx
from getpass import getpass
from langgraph_sdk import get_client
# 从命令行获取电子邮件
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: ")
# 这是你的公共匿名密钥（可以在客户端安全使用）
# 不要将其与秘密服务角色密钥混淆
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()
# 创建两个测试用户
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 仪表板复制的“匿名公钥”。
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"]
# 以用户 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"✅ User 1 created thread: {thread['thread_id']}")
# 尝试在没有令牌的情况下访问
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)
# 尝试以用户 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("❌ 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 应用程序的基础组件。
下一步做什么？¶
既然你已经实现了生产环境认证，不妨考虑以下事项：