跳到主要内容
在上一个教程中，你添加了资源授权以向用户提供私有对话。但是，你仍然使用硬编码令牌进行身份验证，这不安全。现在，你将使用OAuth2用真实用户账户替换这些令牌。 你将保留相同的Auth对象和资源级访问控制，但将身份验证升级为使用 Supabase 作为你的身份提供程序。虽然本教程使用 Supabase，但这些概念适用于任何 OAuth2 提供程序。你将学习如何：
  1. 用真实的 JWT 令牌替换测试令牌
  2. 与 OAuth2 提供程序集成以进行安全的用户身份验证
  3. 在保持现有授权逻辑的同时处理用户会话和元数据

​
背景

OAuth2 涉及三个主要角色
  1. 授权服务器：处理用户身份验证和颁发令牌的身份提供程序（例如，Supabase、Auth0、Google）
  2. 应用程序后端：你的 LangGraph 应用程序。它验证令牌并提供受保护的资源（对话数据）
  3. 客户端应用程序：用户与你的服务交互的 Web 或移动应用程序
标准的 OAuth2 流程如下

​
先决条件

在开始本教程之前，请确保你已具备以下条件

​
1. 安装依赖项

安装所需的依赖项。从 custom-auth 目录开始，并确保已安装 langgraph-cli 
cd custom-auth
pip install -U "langgraph-cli[inmem]"

​
2. 设置身份验证提供程序

接下来，获取你的身份验证服务器 URL 和用于身份验证的私钥。由于你正在使用 Supabase，你可以在 Supabase 控制面板中完成此操作
  1. 在左侧边栏中，点击“⚙️ 项目设置”，然后点击“API”
  2. 复制你的项目 URL 并将其添加到你的 .env 文件中
echo "SUPABASE_URL=your-project-url" >> .env
  1. 复制你的服务角色秘密密钥并将其添加到你的 .env 文件中
echo "SUPABASE_SERVICE_KEY=your-service-role-key" >> .env
  1. 复制你的“匿名公共”密钥并记下。这将在稍后设置客户端代码时使用。
SUPABASE_URL=your-project-url
SUPABASE_SERVICE_KEY=your-service-role-key

​
3. 实施令牌验证

在之前的教程中，你使用Auth对象验证硬编码令牌并添加资源所有权。 现在，你将升级身份验证以验证来自 Supabase 的真实 JWT 令牌。主要的更改都将在@auth.authenticate装饰函数中：
  • 你将不再与硬编码的令牌列表进行检查，而是向 Supabase 发送 HTTP 请求来验证令牌。
  • 你将从已验证的令牌中提取真实的用户信息（ID、电子邮件）。
  • 现有的资源授权逻辑保持不变。
更新 src/security/auth.py 以实现此功能
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 项目的私钥，我们可以使用它来验证用户的令牌并提取他们的信息。

​
4. 测试身份验证流程

让我们测试新的身份验证流程。你可以在文件或笔记本中运行以下代码。你需要提供
  • 一个有效的电子邮件地址
  • 一个 Supabase 项目 URL（来自上方）
  • 一个 Supabase 匿名公共密钥（也来自上方）
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)
⚠️ 在继续之前：检查你的电子邮件并点击两个确认链接。在确认用户电子邮件之前，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"]


# Log in as user 1
user1_token = await login(email1, password)
user1_client = get_client(
    url="https://: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="https://: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="https://: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 'https://:2024/threads'
✅ User 2 blocked from User 1's thread: Client error '404 Not Found' for url 'https://:2024/threads/d6af3754-95df-4176-aa10-dbd8dca40f1a'
你的身份验证和授权协同工作
  1. 用户必须登录才能访问机器人
  2. 每个用户只能看到自己的线程
所有用户都由 Supabase 身份验证提供程序管理，因此你无需实现任何额外的用户管理逻辑。

​
后续步骤

你已成功为你的 LangGraph 应用程序构建了一个生产就绪的身份验证系统！让我们回顾一下你所完成的工作
  1. 设置了一个身份验证提供程序（本例中为 Supabase）
  2. 添加了使用电子邮件/密码身份验证的真实用户账户
  3. 将 JWT 令牌验证集成到你的 LangGraph 服务器中
  4. 实施了适当的授权以确保用户只能访问自己的数据
  5. 创建了一个基础，可应对你的下一个身份验证挑战 🚀
既然你拥有了生产身份验证，请考虑
  1. 使用你喜欢的框架构建一个 Web UI（请参阅自定义身份验证模板示例）
  2. 在身份验证概念指南中了解有关身份验证和授权的其他方面的更多信息。
  3. 阅读参考文档后，进一步自定义你的处理程序和设置。
在 GitHub 上编辑此页面源文件。
以编程方式连接这些文档到 Claude、VSCode 等，通过 MCP 获取实时答案。
© . This site is unofficial and not affiliated with LangChain, Inc.