인증 및 접근 제어

인증 및 접근 제어

LangSmith는 대부분의 인증 방식과 통합될 수 있는 유연한 인증/권한 부여 시스템을 제공해요. 이 페이지에서 인증과 권한 부여의 핵심 개념, 시스템 아키텍처, 그리고 LangGraph에서 핸들러를 구성하는 방법을 다룹니다.

출처: 문서

본문

LangSmith는 대부분의 인증 방식과 통합될 수 있는 유연한 인증 및 권한 부여 시스템을 제공합니다.

핵심 개념

인증 vs 권한 부여

종종 같은 의미로 쓰이지만, 이 용어들은 서로 다른 보안 개념을 나타냅니다:

  • 인증 (Authentication) ("AuthN")은 당신이 누구인지를 검증합니다. 매 요청마다 미들웨어로 실행됩니다.
  • 권한 부여 (Authorization) ("AuthZ")는 무엇을 할 수 있는지를 결정합니다. 사용자의 권한과 역할을 리소스별로 검증합니다.

LangSmith에서 인증은 @auth.authenticate 핸들러가, 권한 부여는 @auth.on 핸들러가 처리합니다.

기본 보안 모델

LangSmith는 서로 다른 보안 기본값을 제공합니다:

LangSmith

  • 기본적으로 LangSmith API 키 사용
  • x-api-key 헤더에 유효한 API 키 필요
  • 자체 인증 핸들러로 커스터마이즈 가능

참고: 커스텀 인증 LangSmith의 모든 요금제에서 커스텀 인증 이 지원됩니다.

자체 호스팅

  • 기본 인증 없음
  • 보안 모델을 구현할 완전한 유연성
  • 인증과 권한 부여의 모든 측면을 직접 제어

시스템 아키텍처

일반적인 인증 설정은 세 가지 주요 구성 요소를 포함합니다:

  1. 인증 제공자 (Identity Provider/IdP)
    • 사용자 정체성과 자격 증명을 관리하는 전용 서비스
    • 사용자 등록, 로그인, 비밀번호 재설정 등을 처리
    • 성공적인 인증 후 토큰(JWT, 세션 토큰 등)을 발급
    • 예: Auth0, Supabase Auth, Okta, 또는 자체 인증 서버
  2. Agent Server (리소스 서버)
    • 비즈니스 로직과 보호된 리소스를 포함하는 에이전트 또는 LangGraph 애플리케이션
    • 인증 제공자로 토큰 검증
    • 사용자 정체성과 권한에 기반해 접근 제어 적용
    • 사용자 자격 증명을 직접 저장하지 않음
  3. 클라이언트 애플리케이션 (프론트엔드)
    • 웹 앱, 모바일 앱 또는 API 클라이언트
    • 시간에 민감한 사용자 자격 증명을 수집해 인증 제공자로 전송
    • 인증 제공자로부터 토큰 수신
    • 요청에 이 토큰을 포함해 Agent Server로 전송

이 구성 요소들이 일반적으로 상호작용하는 방식:

sequenceDiagram
    participant Client as Client App
    participant Auth as Auth Provider
    participant LG as Agent Server

    Client->>Auth: 1. Login (username/password)
    Auth-->>Client: 2. Return token
    Client->>LG: 3. Request with token
    Note over LG: 4. Validate token (@auth.authenticate)
    LG-->>Auth:  5. Fetch user info
    Auth-->>LG: 6. Confirm validity
    Note over LG: 7. Apply access control (@auth.on.*)
    LG-->>Client: 8. Return resources

LangGraph의 @auth.authenticate 핸들러가 4~6단계를, @auth.on 핸들러가 7단계를 구현합니다.

인증

LangGraph의 인증은 매 요청마다 미들웨어로 실행됩니다. @auth.authenticate 핸들러는 요청 정보를 받아 다음을 수행해야 합니다:

  1. 자격 증명 검증
  2. 유효하면 사용자의 정체성과 사용자 정보를 포함하는 사용자 정보 반환
  3. 유효하지 않으면 HTTP 예외 또는 AssertionError 발생
from langgraph_sdk import Auth

auth = Auth()

@auth.authenticate
async def authenticate(headers: dict) -> Auth.types.MinimalUserDict:
    # Validate credentials (e.g., API key, JWT token)
    api_key = headers.get(b"x-api-key")
    if not api_key or not is_valid_key(api_key):
        raise Auth.exceptions.HTTPException(
            status_code=401,
            detail="Invalid API key"
        )

    # Return user info - only identity and is_authenticated are required
    # Add any additional fields you need for authorization
    return {
        "identity": "user-123",        # Required: unique user identifier
        "is_authenticated": True,      # Optional: assumed True by default
        "permissions": ["read", "write"], # Optional: for permission-based auth
        # You can add more custom fields if you want to implement other auth patterns
        "role": "admin",
        "org_id": "org-456"

    }

반환된 사용자 정보는 다음에서 사용할 수 있습니다:

  • ctx.user를 통한 인증 핸들러
  • config["configuration"]["langgraph_auth_user"]를 통한 애플리케이션

지원되는 매개변수

@auth.authenticate 핸들러는 이름으로 다음 매개변수 중 아무거나 받을 수 있습니다:

  • request (Request): 원시 ASGI 요청 객체
  • path (str): 요청 경로, 예: "/threads/abcd-1234-abcd-1234/runs/abcd-1234-abcd-1234/stream"
  • method (str): HTTP 메서드, 예: "GET"
  • path_params (dict[str, str]): URL 경로 매개변수, 예: {"thread_id": "abcd-1234-abcd-1234", "run_id": "abcd-1234-abcd-1234"}
  • query_params (dict[str, str]): URL 쿼리 매개변수, 예: {"stream": "true"}
  • headers (dict[bytes, bytes]): 요청 헤더
  • authorization (str | None): Authorization 헤더 값 (예: "Bearer <token>")

튜토리얼의 여러 곳에서 간결함을 위해 "authorization" 매개변수만 보여주지만, 필요에 따라 커스텀 인증 방식을 구현하기 위해 더 많은 정보를 받도록 선택할 수 있습니다.

에이전트 인증

커스텀 인증은 위임된 접근을 허용합니다. @auth.authenticate에서 반환하는 값은 런 컨텍스트에 추가되며, 에이전트에게 사용자 범위 자격 증명을 부여하면 사용자를 대신해 리소스에 접근할 수 있게 됩니다.

sequenceDiagram
  %% Actors
  participant ClientApp as Client
  participant AuthProv  as Auth Provider
  participant LangGraph as Agent Server
  participant SecretStore as Secret Store
  participant ExternalService as External Service

  %% Platform login / AuthN
  ClientApp  ->> AuthProv: 1. Login (username / password)
  AuthProv   -->> ClientApp: 2. Return token
  ClientApp  ->> LangGraph: 3. Request with token

  Note over LangGraph: 4. Validate token (@auth.authenticate)
  LangGraph  -->> AuthProv: 5. Fetch user info
  AuthProv   -->> LangGraph: 6. Confirm validity

  %% Fetch user tokens from secret store
  LangGraph  ->> SecretStore: 6a. Fetch user tokens
  SecretStore -->> LangGraph: 6b. Return tokens

  Note over LangGraph: 7. Apply access control (@auth.on.*)

  %% External Service round-trip
  LangGraph  ->> ExternalService: 8. Call external service (with header)
  Note over ExternalService: 9. External service validates header and executes action
  ExternalService  -->> LangGraph: 10. Service response

  %% Return to caller
  LangGraph  -->> ClientApp: 11. Return resources

인증 후 플랫폼은 그래프와 모든 노드에 configurable 컨텍스트를 통해 전달되는 특수 구성 객체를 만듭니다. 이 객체는 @auth.authenticate 핸들러에서 반환하는 커스텀 필드를 포함한 현재 사용자에 대한 정보를 담습니다.

에이전트가 사용자를 대신해 행동하도록 하려면 커스텀 인증 미들웨어를 사용하세요. 이를 통해 에이전트가 MCP 서버, 외부 데이터베이스, 심지어 다른 에이전트 같은 외부 시스템과 사용자를 대신해 상호작용할 수 있습니다.

자세한 내용은 커스텀 인증 사용 가이드를 참고하세요.

MCP와 에이전트 인증

에이전트를 MCP 서버에 인증하는 방법은 MCP 개념 가이드를 참고하세요.

권한 부여

인증 후 LangGraph는 @auth.on 핸들러를 호출해 특정 리소스(예: 스레드, 어시스턴트, crons)에 대한 접근을 제어합니다. 이 핸들러들은:

  1. value["metadata"] 딕셔너리를 직접 변경해 리소스 생성 시 저장될 메타데이터를 추가합니다. 각 동작에 대해 value가 가질 수 있는 유형 목록은 지원되는 동작 표를 참고하세요.
  2. 검색/목록 또는 읽기 작업 중에 필터 딕셔너리를 반환해 메타데이터로 리소스를 필터링합니다.
  3. 접근이 거부되면 HTTP 예외를 발생시킵니다.

간단한 사용자 범위 접근 제어만 구현하고 싶다면 모든 리소스와 동작에 단일 @auth.on 핸들러를 사용할 수 있습니다. 리소스와 동작에 따라 다른 제어를 원하면 리소스별 핸들러를 사용할 수 있습니다. 접근 제어를 지원하는 리소스의 전체 목록은 지원되는 리소스 섹션을 참고하세요.

@auth.on
async def add_owner(
    ctx: Auth.types.AuthContext,
    value: dict  # The payload being sent to this access method
) -> dict:  # Returns a filter dict that restricts access to resources
    """Authorize all access to threads, runs, crons, and assistants.

    This handler does two things:
        - Adds a value to resource metadata (to persist with the resource so it can be filtered later)
        - Returns a filter (to restrict access to existing resources)

    Args:
        ctx: Authentication context containing user info, permissions, the path, and
        value: The request payload sent to the endpoint. For creation
              operations, this contains the resource parameters. For read
              operations, this contains the resource being accessed.

    Returns:
        A filter dictionary that LangGraph uses to restrict access to resources.
        See [Filter Operations](#filter-operations) for supported operators.
    """
    # Create filter to restrict access to just this user's resources
    filters = {"owner": ctx.user.identity}

    # Get or create the metadata dictionary in the payload
    # This is where we store persistent info about the resource
    metadata = value.setdefault("metadata", {})

    # Add owner to metadata - if this is a create or update operation,
    # this information will be saved with the resource
    # So we can filter by it later in read operations
    metadata.update(filters)

    # Return filters to restrict access
    # These filters are applied to ALL operations (create, read, update, search, etc.)
    # to ensure users can only access their own resources
    return filters

리소스별 핸들러

@auth.on 데코레이터에 리소스와 동작 이름을 연결하면 특정 리소스와 동작에 대한 핸들러를 등록할 수 있습니다. 요청이 들어오면 해당 리소스와 동작에 일치하는 가장 구체적인 핸들러가 호출됩니다. 특정 리소스와 동작에 대한 핸들러를 등록하는 예는 아래와 같습니다. 다음 설정의 경우:

  1. 인증된 사용자는 스레드를 만들고, 읽고, 스레드에 런을 만들 수 있습니다.
  2. "assistants:create" 권한이 있는 사용자만 새 어시스턴트를 만들 수 있습니다.
  3. 다른 모든 엔드포인트(예: delete assistant, crons, store)는 모든 사용자에게 비활성화됩니다.

팁: 지원되는 핸들러 지원되는 리소스와 동작의 전체 목록은 아래 지원되는 리소스 섹션을 참고하세요.

# Generic / global handler catches calls that aren't handled by more specific handlers
@auth.on
async def reject_unhandled_requests(ctx: Auth.types.AuthContext, value: Any) -> False:
    print(f"Request to {ctx.path} by {ctx.user.identity}")
    raise Auth.exceptions.HTTPException(
        status_code=403,
        detail="Forbidden"
    )

# Matches the "thread" resource and all actions - create, read, update, delete, search
# Since this is **more specific** than the generic @auth.on handler, it will take precedence
# over the generic handler for all actions on the "threads" resource
@auth.on.threads
async def on_thread(
    ctx: Auth.types.AuthContext,
    value: Auth.types.threads.create.value
):
    # Setting metadata on the thread being created
    # will ensure that the resource contains an "owner" field
    # Then any time a user tries to access this thread or runs within the thread,
    # we can filter by owner
    metadata = value.setdefault("metadata", {})
    metadata["owner"] = ctx.user.identity
    return {"owner": ctx.user.identity}


# Thread creation. This will match only on thread create actions
# Since this is **more specific** than both the generic @auth.on handler and the @auth.on.threads handler,
# it will take precedence for any "create" actions on the "threads" resources
@auth.on.threads.create
async def on_thread_create(
    ctx: Auth.types.AuthContext,
    value: Auth.types.threads.create.value
):
    # Reject if the user does not have write access
    if "write" not in ctx.permissions:
        raise Auth.exceptions.HTTPException(
            status_code=403,
            detail="User lacks the required permissions."
        )
    # Setting metadata on the thread being created
    # will ensure that the resource contains an "owner" field
    # Then any time a user tries to access this thread or runs within the thread,
    # we can filter by owner
    metadata = value.setdefault("metadata", {})
    metadata["owner"] = ctx.user.identity
    return {"owner": ctx.user.identity}

# Reading a thread. Since this is also more specific than the generic @auth.on handler, and the @auth.on.threads handler,
# it will take precedence for any "read" actions on the "threads" resource
@auth.on.threads.read
async def on_thread_read(
    ctx: Auth.types.AuthContext,
    value: Auth.types.threads.read.value
):
    # Since we are reading (and not creating) a thread,
    # we don't need to set metadata. We just need to
    # return a filter to ensure users can only see their own threads
    return {"owner": ctx.user.identity}

# Run creation, streaming, updates, etc.
# This takes precedenceover the generic @auth.on handler and the @auth.on.threads handler
@auth.on.threads.create_run
async def on_run_create(
    ctx: Auth.types.AuthContext,
    value: Auth.types.threads.create_run.value
):
    metadata = value.setdefault("metadata", {})
    metadata["owner"] = ctx.user.identity
    # Inherit thread's access control
    return {"owner": ctx.user.identity}

# Assistant creation
@auth.on.assistants.create
async def on_assistant_create(
    ctx: Auth.types.AuthContext,
    value: Auth.types.assistants.create.value
):
    if "assistants:create" not in ctx.permissions:
        raise Auth.exceptions.HTTPException(
            status_code=403,
            detail="User lacks the required permissions."
        )

위 예시에서 전역 및 리소스별 핸들러를 혼합하고 있습니다. 각 요청은 가장 구체적인 핸들러가 처리하므로, thread를 만드는 요청은 on_thread_create 핸들러와 일치하지만 reject_unhandled_requests 핸들러와는 일치하지 않습니다. 그러나 threadupdate하는 요청은 해당 리소스와 동작에 대한 더 구체적인 핸들러가 없으므로 전역 핸들러가 처리합니다.

필터 작업

인증 핸들러는 None, 불리언 또는 필터 딕셔너리를 반환할 수 있습니다.

  • NoneTrue는 "기본 리소스 모두에 대한 접근을 허용"을 의미합니다.
  • False는 "기본 리소스 모두에 대한 접근을 거부(403 예외 발생)"를 의미합니다.
  • 메타데이터 필터 딕셔너리는 리소스에 대한 접근을 제한합니다.

필터 딕셔너리는 리소스 메타데이터와 일치하는 키를 가진 딕셔너리입니다. 세 가지 연산자를 지원합니다:

  • 기본값은 정확히 일치("$eq")의 약어입니다. 예: {"owner": user_id}는 메타데이터에 {"owner": user_id}가 포함된 리소스만 포함합니다.
  • $eq: 정확히 일치 (예: {"owner": {"$eq": user_id}}) — 위의 약어 {"owner": user_id}와 동일합니다.
  • $contains: 목록 멤버십 (예: {"allowed_users": {"$contains": user_id}}) 또는 목록 포함 (예: {"allowed_users": {"$contains": [user_id_1, user_id_2]}}). 여기서 값은 각각 목록의 요소이거나 목록 요소의 부분집합이어야 합니다. 저장된 리소스의 메타데이터는 목록/컨테이너 유형이어야 합니다.

여러 키가 있는 딕셔너리는 논리 AND 필터로 처리됩니다. 예: {"owner": org_id, "allowed_users": {"$contains": user_id}}는 메타데이터의 "owner"가 org_id이고 "allowed_users" 목록에 user_id가 포함된 리소스만 일치합니다. 자세한 내용은 레퍼런스 Auth를 참고하세요.

일반적인 접근 패턴

단일 소유자 리소스

이 일반적인 패턴은 모든 스레드, 어시스턴트, crons, 런을 단일 사용자로 범위를 지정하게 합니다. 일반 챗봇 스타일 앱 같은 일반적인 단일 사용자 사용 사례에 유용합니다.

@auth.on
async def owner_only(ctx: Auth.types.AuthContext, value: dict):
    metadata = value.setdefault("metadata", {})
    metadata["owner"] = ctx.user.identity
    return {"owner": ctx.user.identity}

권한 기반 접근

이 패턴은 권한에 기반해 접근을 제어하게 합니다. 특정 역할에 더 넓거나 제한된 리소스 접근을 부여하고 싶을 때 유용합니다.

# In your auth handler:
@auth.authenticate
async def authenticate(headers: dict) -> Auth.types.MinimalUserDict:
    ...
    return {
        "identity": "user-123",
        "is_authenticated": True,
        "permissions": ["threads:write", "threads:read"]  # Define permissions in auth
    }

def _default(ctx: Auth.types.AuthContext, value: dict):
    metadata = value.setdefault("metadata", {})
    metadata["owner"] = ctx.user.identity
    return {"owner": ctx.user.identity}

@auth.on.threads.create
async def create_thread(ctx: Auth.types.AuthContext, value: dict):
    if "threads:write" not in ctx.permissions:
        raise Auth.exceptions.HTTPException(
            status_code=403,
            detail="Unauthorized"
        )
    return _default(ctx, value)


@auth.on.threads.read
async def rbac_create(ctx: Auth.types.AuthContext, value: dict):
    if "threads:read" not in ctx.permissions and "threads:write" not in ctx.permissions:
        raise Auth.exceptions.HTTPException(
            status_code=403,
            detail="Unauthorized"
        )
    return _default(ctx, value)

지원되는 리소스

LangGraph는 가장 일반적인 것부터 가장 구체적인 것까지 세 가지 수준의 인증 핸들러를 제공합니다:

  1. 전역 핸들러 (@auth.on): 모든 리소스와 동작과 일치
  2. 리소스 핸들러 (예: @auth.on.threads, @auth.on.assistants, @auth.on.crons): 특정 리소스의 모든 동작과 일치
  3. 동작 핸들러 (예: @auth.on.threads.create, @auth.on.threads.read): 특정 리소스의 특정 동작과 일치

가장 구체적으로 일치하는 핸들러가 사용됩니다. 예를 들어 스레드 생성에는 @auth.on.threads.create@auth.on.threads보다 우선합니다. 더 구체적인 핸들러가 등록되면 해당 리소스와 동작에 대해 더 일반적인 핸들러는 호출되지 않습니다.

팁: 타입 안전성 각 핸들러는 Auth.types.on.<resource>.<action>.value에서 value 매개변수에 대한 타입 힌트가 있습니다. 예:

@auth.on.threads.create
async def on_thread_create(
ctx: Auth.types.AuthContext,
value: Auth.types.on.threads.create.value  # Specific type for thread creation
):
...

@auth.on.threads
async def on_threads(
ctx: Auth.types.AuthContext,
value: Auth.types.on.threads.value  # Union type of all thread actions
):
...

@auth.on
async def on_all(
ctx: Auth.types.AuthContext,
value: dict  # Union type of all possible actions
):
...

더 구체적인 핸들러는 처리하는 동작 유형이 적으므로 더 나은 타입 힌트를 제공합니다.

지원되는 동작과 유형

지원되는 모든 동작 핸들러는 다음과 같습니다:

리소스 핸들러 설명 값 유형
Threads @auth.on.threads.create 스레드 생성 ThreadsCreate
@auth.on.threads.read 스레드 검색 ThreadsRead
@auth.on.threads.update 스레드 업데이트 ThreadsUpdate
@auth.on.threads.delete 스레드 삭제 ThreadsDelete
@auth.on.threads.search 스레드 목록 ThreadsSearch
@auth.on.threads.create_run 런 생성 또는 업데이트 RunsCreate
Assistants @auth.on.assistants.create 어시스턴트 생성 AssistantsCreate
@auth.on.assistants.read 어시스턴트 검색 AssistantsRead
@auth.on.assistants.update 어시스턴트 업데이트 AssistantsUpdate
@auth.on.assistants.delete 어시스턴트 삭제 AssistantsDelete
@auth.on.assistants.search 어시스턴트 목록 AssistantsSearch
Crons @auth.on.crons.create Cron 작업 생성 CronsCreate
@auth.on.crons.read Cron 작업 검색 CronsRead
@auth.on.crons.update Cron 작업 업데이트 CronsUpdate
@auth.on.crons.delete Cron 작업 삭제 CronsDelete
@auth.on.crons.search Cron 작업 목록 CronsSearch
Store @auth.on.store 모든 store 작업 Auth.types.on.store.value
@auth.on.store.put 항목 저장 Auth.types.on.store.put.value
@auth.on.store.get 항목 검색 Auth.types.on.store.get.value
@auth.on.store.search 검색 항목 Auth.types.on.store.search.value
@auth.on.store.delete 항목 삭제 Auth.types.on.store.delete.value
@auth.on.store.list_namespaces 네임스페이스 나열 Auth.types.on.store.list_namespaces.value

Store 인증은 스레드와 어시스턴트와 다릅니다. 핸들러는 메타데이터 필터를 반환하는 대신 value의 변경 가능한 namespace 필드를 다시 작성해 데이터를 사용자별로 범위를 지정해야 합니다. 연습은 사용자별 store 격리를 참고하세요.

참고: 런에 대하여

런은 접근 제어를 위해 상위 스레드에 범위가 지정됩니다. 즉 권한은 보통 데이터 모델의 대화적 특성을 반영해 스레드에서 상속됩니다. 생성 외의 모든 런 작업(읽기, 목록)은 스레드의 핸들러가 제어합니다. 핸들러에서 볼 수 있는 인자가 더 많기 때문에 새 런 생성에는 특정 create_run 핸들러가 있습니다.

다음 단계

구현 세부 정보:

더 알아보기