title: 连接身份验证提供商 sidebarTitle: 连接身份验证提供商 在上一节教程中，你已经为用户添加了资源级授权，使他们可以拥有私密会话。不过，身份验证仍然依赖硬编码令牌，安全性不足。本教程将演示如何通过 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，可以在其控制台中完成：
  1. 在左侧边栏点击 “⚙ Project Settings”，然后选择 “API”
  2. 复制项目 URL，并写入 .env 文件
echo "SUPABASE_URL=your-project-url" >> .env
  1. 复制你的服务角色密钥（service role key），写入 .env
echo "SUPABASE_SERVICE_KEY=your-service-role-key" >> .env
  1. 复制 “anon public” 密钥备用，稍后配置客户端代码时会用到
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. 测试身份验证流程

接下来验证新的身份验证流程。可以在脚本或 Notebook 中执行以下代码，需要提前准备：
  • 一个有效的邮箱地址
  • Supabase 项目 URL（见上文）
  • Supabase 的 anon public key（同样见上文）
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"创建测试用户：{email1} 和 {email2}")
await sign_up(email1, password)
await sign_up(email2, password)
⚠️ 继续之前，请检查邮箱并点击两封确认邮件中的链接。在确认前，Supabase 会拒绝 /login 请求。 接下来验证用户只能访问自己的数据。继续前请确保服务器已启动（运行 langgraph dev）。下面的代码需要使用你在配置身份提供商时复制的 “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("✅ 未认证请求已被拦截：", 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("✅ 用户 2 无法访问用户 1 的线程：", 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 应用构建了一套可用于生产环境的身份验证系统！回顾一下完成的工作：
  1. 配置了身份验证提供商（本例中为 Supabase）
  2. 新增了基于邮箱/密码的真实用户账号
  3. 将 JWT 令牌校验集成进 Agent Server
  4. 实现了正确的授权逻辑，保证用户只能访问自己的数据
  5. 打下了应对后续身份验证需求的基础 🚀
既然身份验证已经就绪，可以继续：
  1. 使用你偏好的框架构建 Web UI（可参考 Custom Auth 模板）
  2. 在身份验证概念指南中了解更多身份验证与授权的细节
  3. 阅读参考文档，进一步定制处理器与配置
Edit the source of this page on GitHub.
Connect these docs programmatically to Claude, VSCode, and more via MCP for real-time answers.