BytesRouter

主题

LangSmith 文档版本 跟随上游 源快照 7c46f6ca 更新于 2026-10-18

连接身份认证服务商

在上一个教程中，你添加了资源授权以为用户提供私有对话。但是，你仍在使用硬编码的 token 进行身份认证，这并不安全。现在，你将使用 OAuth2 用真实的用户账户替换这些 token。

你将保留相同的 Auth 对象和资源级访问控制，但会升级身份认证，改用 Supabase 作为你的身份提供方。虽然本教程使用 Supabase，但这些概念适用于任何 OAuth2 服务商。你将学习如何：

  1. 将测试 token 替换为真实的 JWT token
  2. 集成 OAuth2 服务商以实现安全的用户身份认证
  3. 在维护现有授权逻辑的同时处理用户会话和元数据

背景

OAuth2 涉及三种主要角色：

  1. 授权服务器：处理用户身份认证并颁发 token 的身份提供方（例如 Supabase、Auth0、Google）
  2. 应用后端：你的 LangGraph 应用。它验证 token 并提供受保护的资源（对话数据）
  3. 客户端应用：用户与你的服务交互所用的 Web 或移动应用

标准的 OAuth2 流程大致如下：

sequenceDiagram
    participant User
    participant Client
    participant AuthServer
    participant Agent Server

    User->>Client: Initiate login
    User->>AuthServer: Enter credentials
    AuthServer->>Client: Send tokens
    Client->>Agent Server: Request with token
    Agent Server->>AuthServer: Validate token
    AuthServer->>Agent Server: Token valid
    Agent Server->>Client: Serve request (e.g., run agent or graph)

前置条件

在开始本教程之前，请确保你已经具备：

1. 安装依赖项

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

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

2. 设置身份认证服务商

接下来，获取你的身份认证服务器的 URL 以及用于身份认证的私钥。 由于你为此使用 Supabase，可以在 Supabase 仪表盘中完成：

  1. 在左侧边栏中，点击 t️⚙ 项目设置"，然后点击 "API"
  2. 复制你的项目 URL 并将其添加到你的 .env file
shell
echo "SUPABASE_URL=your-project-url" >> .env
  1. 复制你的 service role 密钥并将其添加到你的 .env 文件：
shell
echo "SUPABASE_SERVICE_KEY=your-service-role-key" >> .env
  1. 复制你的 "anon public" 密钥并记录下来。稍后设置客户端代码时会用到它。
bash
SUPABASE_URL=your-project-url
SUPABASE_SERVICE_KEY=your-service-role-key

3. 实现 token 校验

在之前的教程中，你使用了 Auth 对象来校验硬编码的 token 并添加资源所有权。

现在，你将升级身份认证，以校验来自 Supabase 的真实 JWT token。主要改动都在 @auth.authenticate 装饰的函数中：

  • 你将向 Supabase 发起 HTTP 请求来校验 token，而不是与硬编码的 token 列表进行比对。
  • 你会从校验通过的 token 中提取真实的用户信息（ID、电子邮件）。
  • 现有的资源授权逻辑保持不变。

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

python
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

最重要的变化是我们现在使用真实的身份认证服务器来校验 token。我们的身份认证处理器拥有 Supabase 项目的私钥，可以用它来校验用户的 token 并提取其信息。

4. 测试身份认证流程

让我们测试新的身份认证流程。你可以在文件或 notebook 中运行以下代码。你需要提供：

  • 一个有效的电子邮件地址
  • 一个 Supabase 项目 URL（来自上文）
  • 一个 Supabase anon 公钥（也来自上文）
python
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 仪表盘复制的 "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)

输出应该类似于：

shell
✅ 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. 用户必须登录才能访问 bot
  2. 每个用户只能看到自己的线程

所有用户都由 Supabase 身份认证服务商管理，因此你无需实现任何额外的用户管理逻辑。

后续步骤

你已成功为你的 LangGraph 应用构建了一个可用于生产环境的身份认证系统！让我们回顾一下你完成了什么：

  1. 设置了一个身份认证服务商（本例中为 Supabase）
  2. 添加了使用电子邮件/密码身份认证的真实用户账户
  3. 将 JWT token 校验集成到你的 Agent Server
  4. 实现了恰当的授权，确保用户只能访问自己的数据
  5. 打下了坚实的基础，已准备好应对下一个身份认证挑战

现在你已拥有生产级身份认证，可以考虑：

  1. 使用你偏好的框架构建一个 Web UI（示例参见 Custom Auth 模板）
  2. 在身份认证概念指南中进一步了解身份认证和授权的其他方面。
  3. 阅读参考文档后，进一步自定义你的处理器和设置。

友情链接

  • Agents.md面向 AI 编码助手的仓库说明文件规范
  • LangChainLangChain 是一个开放的Agent平台
  • LangSmithAgent 工程平台：可观测与评估、部署、Fleet、Engine 等
  • EmbabelEmbabel 是一个JVM平台Agent框架
  • SpringAISpringAI 是一个基于 Spring 的 AI 框架
  • MCPModel Context Protocol，连接 AI 应用与外部数据/工具的开放标准
  • LangChain4jJava 生态的 LLM 应用开发框架
  • LlamaIndex把数据接入 LLM 的框架与文档平台
  • LlamaParse智能体式 OCR 文档解析：把 PDF、扫描件变成 LLM 可用的文本与结构化数据
  • LiteParse本地优先、轻量的文档解析库，不需要 API key
  • CrewAI多智能体编排框架：以 Crew/Flow 组织自主协作的 AI Agent
  • MAFMicrosoft Agent Framework，企业级 .NET/Python 智能体框架

BytesRouter 是社区翻译站点，所有译文仅供参考，一切以官方英文原文为准。

© 2026 BytesRouter · 译文采用 CC BY 4.0 许可