连接身份验证提供程序¶

在上一个教程中,你添加了资源授权以让用户拥有私密对话。然而,你仍在使用硬编码的令牌进行身份验证,这并不安全。现在你将使用 OAuth2 将这些令牌替换为真实的用户帐户。

Python

你将保留相同的 Auth 对象和资源级访问控制,但将身份验证升级为使用 Supabase 作为你的身份提供者。虽然本教程使用 Supabase,但这些概念适用于任何 OAuth2 提供者。你将学习如何:

JavaScript

你将保留相同的 Auth 对象和资源级访问控制,但将身份验证升级为使用 Supabase 作为你的身份提供者。虽然本教程使用 Supabase,但这些概念适用于任何 OAuth2 提供者。你将学习如何:

用真实的 JWT 令牌替换测试令牌
与 OAuth2 提供者集成以实现安全的用户身份验证
在维护现有授权逻辑的同时处理用户会话和元数据

背景¶

OAuth2 涉及三个主要角色:

授权服务器:处理用户身份验证并颁发令牌的身份提供者(例如 Supabase、Auth0、Google)
应用后端:你的 LangGraph 应用。它验证令牌并提供受保护的资源(对话数据)
客户端应用:用户与你的服务交互的 Web 或移动应用

标准的 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 项目用作你的身份验证服务器。

1. 安装依赖项¶

安装所需的依赖项。从你的 custom-auth 目录开始,确保你已安装 langgraph-cli:

Python

cd custom-auth
pip install -U "langgraph-cli[inmem]"

JavaScript

cd custom-auth
npm install -g @langchain/langgraph-cli

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

接下来,获取你的认证服务器的 URL 和用于身份验证的私钥。由于你在本教程中使用 Supabase,你可以在 Supabase 仪表板中执行此操作:

在左侧边栏中,点击"⚙️ Project Settings",然后点击"API"
复制你的项目 URL 并将其添加到你的 .env 文件
```
echo "SUPABASE_URL=your-project-url" >> .env
```
复制你的 service role secret key 并将其添加到你的 .env 文件:
```
echo "SUPABASE_SERVICE_KEY=your-service-role-key" >> .env
```
复制你的"anon public"密钥并记录下来。这将在稍后设置客户端代码时使用。
```
SUPABASE_URL=your-project-url
SUPABASE_SERVICE_KEY=your-service-role-key
```

3. 实现令牌验证¶

Python

在之前的教程中,你使用 Auth 对象来验证硬编码的令牌并添加资源所有权。

现在你将升级身份验证以验证来自 Supabase 的真实 JWT 令牌。所有主要更改都将在 @auth.authenticate 装饰的函数中:

JavaScript

在之前的教程中,你使用 Auth 对象来验证硬编码的令牌并添加资源所有权。

现在你将升级身份验证以验证来自 Supabase 的真实 JWT 令牌。所有主要更改都将在 auth.authenticate 装饰的函数中:

你将发出 HTTP 请求到 Supabase 来验证令牌,而不是根据硬编码的令牌列表进行检查。
你将从验证的令牌中提取真实的用户信息(ID、电子邮件)。
现有的资源授权逻辑保持不变。

Python

更新 src/security/auth.py 来实现这一点:

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

JavaScript

更新 src/security/auth.ts 来实现这一点:

src/security/auth.ts

import { Auth } from "@langchain/langgraph-sdk";

// 这是从你上面创建的 `.env` 文件中加载的
const SUPABASE_URL = process.env.SUPABASE_URL;
const SUPABASE_SERVICE_KEY = process.env.SUPABASE_SERVICE_KEY;

const auth = new Auth()
  .authenticate(async (request) => {
    // 验证 JWT 令牌并提取用户信息。
    const apiKey = request.headers.get("x-api-key");
    if (!apiKey || !isValidKey(apiKey)) {
      throw new HTTPException(401, "Invalid API key");
    }

    const [scheme, token] = apiKey.split(" ");
    if (scheme.toLowerCase() !== "bearer") {
      throw new Error("Invalid authorization scheme");
    }

    try {
      // 使用认证提供者验证令牌
      const response = await fetch(`${SUPABASE_URL}/auth/v1/user`, {
        headers: {
          Authorization: authorization,
          apiKey: SUPABASE_SERVICE_KEY!,
        },
      });

      if (response.status !== 200) {
        throw new Error("Invalid token");
      }

      const user = await response.json();
      return {
        identity: user.id, // 唯一用户标识符
        email: user.email,
        is_authenticated: true,
      };
    } catch (e) {
      throw new Auth.HTTPException(401, String(e));
    }
  })
  .on(async ({ user, value }) => {
    // 保留我们在上一个教程中的资源授权
    // 使用资源元数据使资源对其创建者私密。
    const filters = { owner: user.identity };
    const metadata = value.metadata || {};
    Object.assign(metadata, filters);
    value.metadata = metadata;
    return filters;
  });

export { auth };

最重要的变化是我们现在使用真实的身份验证服务器验证令牌。我们的身份验证处理程序拥有 Supabase 项目的私钥,我们可以使用它来验证用户的令牌并提取他们的信息。

4. 测试身份验证流程¶

让我们测试新的身份验证流程。你可以在文件或笔记本中运行以下代码。你需要提供:

一个有效的电子邮件地址
Supabase 项目 URL(来自上面)
Supabase anon 公钥(也来自上面)

Python

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"  # 修改这里
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: ")

# 这是你的公共 anon 密钥(在客户端使用是安全的)
# 不要将其与秘密 service role 密钥混淆
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):
    """创建新的用户帐户。"""
    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)

JavaScript

import { Client } from "@langchain/langgraph-sdk";

// 从命令行获取电子邮件
const email = process.env.TEST_EMAIL || "your-email@example.com";
const baseEmail = email.split("@");
const password = "secure-password"; // 修改这里
const email1 = `${baseEmail[0]}+1@${baseEmail[1]}`;
const email2 = `${baseEmail[0]}+2@${baseEmail[1]}`;

const SUPABASE_URL = process.env.SUPABASE_URL;
if (!SUPABASE_URL) {
  throw new Error("SUPABASE_URL environment variable is required");
}

// 这是你的公共 anon 密钥（在客户端使用是安全的）
// 不要将其与秘密 service role 密钥混淆
const SUPABASE_ANON_KEY = process.env.SUPABASE_ANON_KEY;
if (!SUPABASE_ANON_KEY) {
  throw new Error("SUPABASE_ANON_KEY environment variable is required");
}

async function signUp(email: string, password: string) {
  /**创建新的用户帐户。*/
  const response = await fetch(`${SUPABASE_URL}/auth/v1/signup`, {
    method: "POST",
    headers: {
      apiKey: SUPABASE_ANON_KEY,
      "Content-Type": "application/json",
    },
    body: JSON.stringify({ email, password }),
  });

  if (response.status !== 200) {
    throw new Error(`Failed to sign up: ${response.statusText}`);
  }

  return response.json();
}

// 创建两个测试用户
console.log(`Creating test users: ${email1} and ${email2}`);
await signUp(email1, password);
await signUp(email2, password);

⚠️ 继续之前：检查你的电子邮件并点击两个确认链接。Supabase 将拒绝 /login 请求，直到你确认了用户的电子邮件。

现在测试用户只能看到自己的数据。在继续之前，请确保服务器正在运行（运行 langgraph dev）。以下代码片段需要你之前在设置身份验证提供程序时从 Supabase 仪表板复制的"anon public"密钥。

Python

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)

JavaScript

async function login(email: string, password: string): Promise<string> {
  /**获取现有用户的访问令牌。*/
  const response = await fetch(
    `${SUPABASE_URL}/auth/v1/token?grant_type=password`,
    {
      method: "POST",
      headers: {
        apikey: SUPABASE_ANON_KEY,
        "Content-Type": "application/json",
      },
      body: JSON.stringify({ email, password }),
    }
  );

  if (response.status !== 200) {
    throw new Error(`Failed to login: ${response.statusText}`);
  }

  const data = await response.json();
  return data.access_token;
}

// 以用户 1 的身份登录
const user1Token = await login(email1, password);
const user1Client = new Client({
  apiUrl: "http://localhost:2024",
  headers: { Authorization: `Bearer ${user1Token}` },
});

// 以用户 1 的身份创建线程
const thread = await user1Client.threads.create();
console.log(`✅ User 1 created thread: ${thread.thread_id}`);

// 尝试在没有令牌的情况下访问
const unauthenticatedClient = new Client({ apiUrl: "http://localhost:2024" });
try {
  await unauthenticatedClient.threads.create();
  console.log("❌ Unauthenticated access should fail!");
} catch (e) {
  console.log("✅ Unauthenticated access blocked:", e.message);
}

// 尝试以用户 2 的身份访问用户 1 的线程
const user2Token = await login(email2, password);
const user2Client = new Client({
  apiUrl: "http://localhost:2024",
  headers: { Authorization: `Bearer ${user2Token}` },
});

try {
  await user2Client.threads.get(thread.thread_id);
  console.log("❌ User 2 shouldn't see User 1's thread!");
} catch (e) {
  console.log("✅ User 2 blocked from User 1's thread:", e.message);
}

输出应该如下所示:

✅ 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'

你的身份验证和授权正在协同工作:

用户必须登录才能访问机器人
每个用户只能看到自己的线程

所有用户都由 Supabase 认证提供者管理,因此你不需要实现任何额外的用户管理逻辑。

下一步¶

你已成功为 LangGraph 应用构建了生产就绪的身份验证系统!让我们回顾一下你完成的工作:

设置了身份验证提供者(在本例中为 Supabase)
添加了具有电子邮件/密码身份验证的真实用户帐户
将 JWT 令牌验证集成到 LangGraph 服务器中
实现了适当的授权以确保用户只能访问自己的数据
创建了准备好应对下一个身份验证挑战的基础 🚀

既然你已经有了生产身份验证,可以考虑:

使用你喜欢的框架构建 Web UI(参见 Custom Auth 模板示例)
在身份验证概念指南中了解有关身份验证和授权其他方面的更多信息。

Python

阅读参考文档后进一步自定义你的处理程序和设置。

JavaScript

阅读参考文档后进一步自定义你的处理程序和设置。