인증 관리자로 3-legged OAuth를 사용하여 인증

특정 사용자를 대신하여 에이전트가 외부 도구 및 서비스 (예: Jira 작업 또는 GitHub 저장소)에 액세스하도록 하려면 에이전트 ID 인증 관리자에서 3-legged OAuth 인증 제공업체를 구성해야 합니다.

3-legged OAuth 인증 제공업체는 사용자 리디렉션과 토큰을 관리합니다. 이렇게 하면 복잡한 OAuth 2.0 흐름을 처리하기 위해 맞춤 코드를 작성할 필요가 없습니다.

3-legged OAuth 워크플로

3-legged OAuth 인증 제공자는 에이전트가 사용자를 대신하여 리소스에 액세스하므로 사용자 동의가 필요합니다.

프롬프트 및 리디렉션: 채팅 인터페이스에서 사용자에게 로그인하라는 메시지를 표시한 다음 사용자를 서드 파티 애플리케이션의 동의 페이지로 리디렉션합니다.
동의 및 저장소: 사용자가 권한을 부여하면 결과 OAuth 토큰이 Google 관리 사용자 인증 정보 보관소에 저장됩니다.
삽입: 에이전트 개발 키트 (ADK)를 사용하면 에이전트가 인증 제공업체에서 토큰을 자동으로 가져와 도구 호출 헤더에 삽입합니다.

시작하기 전에

[올바른 인증 방법을 선택했는지 확인하세요](/iam/docs/agent-identity-overview#auth-models).
Agent Identity Connector API를 사용 설정합니다.
API 사용 설정에 필요한 역할
API를 사용 설정하려면 serviceusage.services.enable 권한이 포함된 서비스 사용량 관리자 IAM 역할(roles/serviceusage.serviceUsageAdmin)이 필요합니다. 역할 부여 방법 알아보기
API 사용 설정하기
에이전트 만들기 및 배포
사용자 로그인 프롬프트를 처리하고 서드 파티 동의 페이지로 리디렉션하는 프런트엔드 애플리케이션이 있어야 합니다.
이 작업을 완료하는 데 필요한 역할이 있는지 확인합니다.

필요한 역할

3단계 인증 프로바이더를 만들고 사용하는 데 필요한 권한을 얻으려면 운영자에게 프로젝트에 대한 다음 IAM 역할을 부여해 달라고 요청하세요.

인증 제공업체를 만들려면 다음 단계를 따르세요.
- IAM 커넥터 관리자 (roles/iamconnectors.admin)
- IAM 커넥터 편집자 (roles/iamconnectors.editor)
인증 제공업체를 사용하려면 다음 단계를 따르세요.
- IAM 커넥터 사용자 (roles/iamconnectors.user)
- 에이전트 기본 액세스 (roles/aiplatform.agentDefaultAccess)
- 에이전트 컨텍스트 편집자 (roles/aiplatform.agentContextEditor)
- Vertex AI 사용자(roles/aiplatform.user)
- 서비스 사용량 소비자(roles/serviceusage.serviceUsageConsumer)

역할 부여에 대한 자세한 내용은 프로젝트, 폴더, 조직에 대한 액세스 관리를 참조하세요.

이러한 사전 정의된 역할에는 3레그 인증 프로바이더를 만들고 사용하는 데 필요한 권한이 포함되어 있습니다. 필요한 정확한 권한을 보려면 필수 권한 섹션을 펼치세요.

필수 권한

3단계 인증 프로바이더를 만들고 사용하려면 다음 권한이 필요합니다.

인증 제공업체를 만들려면 iamconnectors.connectors.create 권한이 필요합니다.
인증 제공업체를 사용하려면 다음 단계를 따르세요.
- iamconnectors.connectors.retrieveCredentials
- aiplatform.endpoints.predict
- aiplatform.sessions.create

커스텀 역할이나 다른 사전 정의된 역할을 사용하여 이 권한을 부여받을 수도 있습니다.

3레그 인증 제공업체 만들기

인증 제공자를 만들어 서드 파티 애플리케이션의 구성과 사용자 인증 정보를 정의합니다.

3단계 인증 프로바이더를 만들려면 Google Cloud 콘솔 또는 Google Cloud CLI를 사용하세요.

콘솔

Google Cloud 콘솔에서 Agent Registry 페이지로 이동합니다.
에이전트 레지스트리로 이동
인증 제공자를 만들 에이전트의 이름을 클릭합니다.
ID를 클릭합니다.
인증 제공업체 섹션에서 인증 제공업체 추가를 클릭합니다.
인증 제공업체 추가 창에서 이름과 설명을 입력합니다.

이름에는 소문자, 숫자 또는 하이픈만 포함할 수 있으며, 하이픈으로 끝나면 안 되고 소문자로 시작해야 합니다.
OAuth 유형 목록에서 OAuth (3 legged)를 선택합니다.
만들고 계속하기를 클릭합니다.
에이전트 ID에 인증 제공자를 사용할 권한을 부여하려면 액세스 권한 부여를 클릭합니다.
이렇게 하면 인증 제공업체 리소스의 에이전트 ID에 커넥터 사용자(roles/iamconnectors.user) 역할이 자동으로 할당됩니다.
콜백 URL을 복사합니다.
별도의 탭에서 서드 파티 애플리케이션에 콜백 URL을 등록합니다.
서드 파티 애플리케이션에서 클라이언트 ID, 클라이언트 보안 비밀번호, 토큰 URL, 승인 URL을 가져옵니다.
인증 제공업체 사용자 인증 정보 섹션에서 다음 정보를 입력합니다.
- 클라이언트 ID
- 클라이언트 보안 비밀번호
- 토큰 URL
- 승인 URL
공급업체 구성 추가를 클릭합니다.

새로 생성된 인증 제공자가 인증 제공자 목록에 표시됩니다.

Google Cloud CLI

승인 및 토큰 URL을 사용하여 인증 제공업체를 만듭니다.

gcloud alpha agent-identity connectors create AUTH_PROVIDER_NAME \
    --location="LOCATION" \
    --three-legged-oauth-authorization-url="AUTHORIZATION_URL" \
    --three-legged-oauth-token-url="TOKEN_URL"

인증 제공업체 세부정보에서 리디렉션 URI를 가져옵니다.
```
gcloud alpha agent-identity connectors describe AUTH_PROVIDER_NAME \
    --location="LOCATION"
```
이 명령어는 redirectUrl 필드에 리디렉션 URI를 반환합니다.
서드 파티 애플리케이션에 리디렉션 URI를 등록하여 클라이언트 ID와 클라이언트 보안 비밀번호를 가져옵니다.

클라이언트 ID와 클라이언트 보안 비밀번호로 인증 제공업체를 업데이트합니다.

gcloud alpha agent-identity connectors update AUTH_PROVIDER_NAME \
    --location="LOCATION" \
    --three-legged-oauth-client-id="CLIENT_ID" \
    --three-legged-oauth-client-secret="CLIENT_SECRET"

인증 제공업체를 사용할 수 있는 권한을 에이전트 ID에 부여하려면 프로젝트 또는 특정 인증 제공업체의 IAM 허용 정책을 업데이트하고 커넥터 사용자 (roles/iamconnectors.user) 역할을 에이전트 보안 주체에 부여하세요.

상담사 ID는 업계 표준 SPIFFE ID 형식을 기반으로 합니다. IAM 허용 정책에서 에이전트 ID는 주 구성원 식별자를 사용하여 참조됩니다.
프로젝트 수준 (gcloud)

프로젝트 수준에서 역할을 부여하면 에이전트가 해당 프로젝트의 모든 인증 제공업체를 사용할 수 있습니다.
- 단일 에이전트에게 프로젝트의 인증 제공업체에 대한 액세스 권한을 부여하려면 다음 명령어를 실행합니다.
  gcloud projects add-iam-policy-binding PROJECT_ID \ --role='roles/iamconnectors.user' \ --member="principal://agents.global.org-ORGANIZATION_ID.system.id.goog/resources/aiplatform/projects/PROJECT_NUMBER/locations/LOCATION/reasoningEngines/ENGINE_ID"
- 프로젝트의 모든 에이전트에게 인증 제공업체에 대한 액세스 권한을 부여하려면 다음 명령어를 실행하세요.
  gcloud projects add-iam-policy-binding PROJECT_ID \ --role='roles/iamconnectors.user' \ --member="principalSet://agents.global.org-ORGANIZATION_ID.system.id.goog/attribute.platformContainer/aiplatform/projects/PROJECT_NUMBER"
커넥터 수준 (curl)

단일 에이전트에게 특정 인증 제공업체에 대한 액세스 권한을 부여하려면 setIamPolicy API를 사용합니다. 이 명령어는 리소스의 기존 허용 정책을 덮어씁니다.
```
curl -X POST \
    -H "Authorization: Bearer $(gcloud auth print-access-token)" \
    -H "Content-Type: application/json" \
    -d '{
  "policy": {
    "bindings": [
      {
        "role": "roles/iamconnectors.user",
        "members": ["principal://agents.global.org-ORGANIZATION_ID.system.id.goog/resources/aiplatform/projects/PROJECT_NUMBER/locations/LOCATION/reasoningEngines/ENGINE_ID"]
      }
    ]
  }
}' \
    "https://iamconnectors.googleapis.com/v1alpha/projects/PROJECT_ID/locations/LOCATION/connectors/AUTH_PROVIDER_NAME:setIamPolicy"
```
다음을 바꿉니다.
- PROJECT_ID: Google Cloud 프로젝트 ID입니다.
- AUTH_PROVIDER_NAME: 인증 제공업체의 이름입니다.
- ORGANIZATION_ID: Google Cloud 조직 ID입니다.
- PROJECT_NUMBER: Google Cloud 프로젝트 번호.
- LOCATION: 에이전트의 위치입니다 (예: us-central1).
- ENGINE_ID: 추론 엔진의 ID입니다.

에이전트 코드에서 인증

에이전트를 인증하려면 ADK를 사용하거나 에이전트 ID API를 직접 호출하면 됩니다.

ADK

ADK의 MCP 도구 세트를 사용하여 에이전트의 코드에서 인증 제공자를 참조합니다.

from google.adk.agents.llm_agent import LlmAgent
from google.adk.auth.credential_manager import CredentialManager
from google.adk.integrations.agent_identity import GcpAuthProvider, GcpAuthProviderScheme
from google.adk.tools.mcp_tool.mcp_session_manager import StreamableHTTPConnectionParams
from google.adk.tools.mcp_tool.mcp_toolset import McpToolset
from google.adk.auth.auth_tool import AuthConfig

# Register the Google Cloud Auth Provider so the CredentialManager can use it.
CredentialManager.register_auth_provider(GcpAuthProvider())

# The URI to redirect the user to after consent is granted and the
# callback is received by the auth provider.
CONTINUE_URI = "https://YOUR_FRONTEND_URL/validateUserId"

# Create the Auth Provider scheme using the auth provider's full resource name.
auth_scheme = GcpAuthProviderScheme(
    name="projects/PROJECT_ID/locations/LOCATION/connectors/AUTH_PROVIDER_NAME",
    continue_uri=CONTINUE_URI
)

# Configure an MCP tool with the authentication scheme.
toolset = McpToolset(
    connection_params=StreamableHTTPConnectionParams(url="https://YOUR_MCP_SERVER_URL"),
    auth_scheme=auth_scheme,
)

# Initialize the agent with the authenticated tools.
agent = LlmAgent(
    name="YOUR_AGENT_NAME",
    model="gemini-2.0-flash",
    instruction="YOUR_AGENT_INSTRUCTIONS",
    tools=[toolset],
)

ADK

ADK에서 인증된 함수 도구를 사용하여 에이전트의 코드에서 인증 제공업체를 참조합니다.

import httpx
from google.adk.agents.llm_agent import LlmAgent
from google.adk.auth.credential_manager import CredentialManager
from google.adk.integrations.agent_identity import GcpAuthProvider
from google.adk.integrations.agent_identity import GcpAuthProviderScheme
from google.adk.apps import App
from google.adk.auth.auth_credential import AuthCredential
from google.adk.auth.auth_tool import AuthConfig
from google.adk.tools.authenticated_function_tool import AuthenticatedFunctionTool
from vertexai import agent_engines

# First, register Google Cloud auth provider
CredentialManager.register_auth_provider(GcpAuthProvider())

# The URI to redirect the user to after consent is completed.
CONTINUE_URI = "WEB_APP_VALIDATE_USER_URI"

# Create Auth Config
spotify_auth_config = AuthConfig(
    auth_scheme=GcpAuthProviderScheme(
        name="projects/PROJECT_ID/locations/LOCATION/connectors/AUTH_PROVIDER_NAME",
        continue_uri=CONTINUE_URI
    )
)

# Use the Auth Config in Authenticated Function Tool
spotify_search_track_tool = AuthenticatedFunctionTool(
    func=spotify_search_track, auth_config=spotify_auth_config
)

# Sample function tool
async def spotify_search_track(credential: AuthCredential, query: str) -> str | list:
    token = None
    if credential.http and credential.http.credentials:
        token = credential.http.credentials.token

    if not token:
        return "Error: No authentication token available."

    async with httpx.AsyncClient() as client:
        response = await client.get(
            "https://api.spotify.com/v1/search",
            headers={"Authorization": f"Bearer {token}"},
            params={"q": query, "type": "track", "limit": 1},
        )
        # Add your own logic here

agent = LlmAgent(
    name="YOUR_AGENT_NAME",
    model="YOUR_MODEL_NAME",
    instruction="YOUR_AGENT_INSTRUCTIONS",
    tools=[spotify_search_track_tool],
)

app = App(
    name="YOUR_APP_NAME",
    root_agent=agent,
)

vertex_app = agent_engines.AdkApp(app_name=app)

ADK

ADK의 에이전트 레지스트리 MCP 도구 세트를 사용하여 에이전트 코드에서 인증 제공업체를 참조합니다.

from google.adk.agents.llm_agent import LlmAgent
from google.adk.auth.credential_manager import CredentialManager
from google.adk.integrations.agent_identity import GcpAuthProvider
from google.adk.integrations.agent_identity import GcpAuthProviderScheme
from google.adk.tools.mcp_tool.mcp_session_manager import StreamableHTTPConnectionParams
from google.adk.tools.mcp_tool.mcp_toolset import McpToolset
from google.adk.auth.auth_tool import AuthConfig
from google.adk.integrations.agent_registry import AgentRegistry

# First, register Google Cloud auth provider
CredentialManager.register_auth_provider(GcpAuthProvider())

# The URI to redirect the user to after consent is completed.
CONTINUE_URI="WEB_APP_VALIDATE_USER_URI"

# Create Google Cloud auth provider by providing auth provider full resource name
auth_scheme=GcpAuthProviderScheme(
name="projects/GOOGLE_PROJECT/locations/LOCATION/connectors/AUTH_PROVIDER_NAME",
continue_uri=CONTINUE_URI)

# Set Agent Registry
registry = AgentRegistry(project_id="GOOGLE_PROJECT", location="global")

toolset = registry.get_mcp_toolset(mcp_server_name="projects/GOOGLE_PROJECT/locations/global/mcpServers/agentregistry-00000000-0000-0000-0000-000000000000", auth_scheme=auth_scheme )

# Example MCP tool
toolset = McpToolset(
    connection_params=StreamableHTTPConnectionParams(url="MCP_URL"),
    auth_scheme=auth_scheme,
)

agent = LlmAgent(
    name="YOUR_AGENT_NAME",
    model="YOUR_MODEL_NAME",
    instruction="YOUR_AGENT_INSTRUCTIONS",
    tools=[toolset],
)

API 직접 호출

ADK를 사용하지 않는 경우 에이전트가 iamconnectorcredentials.retrieveCredentials API를 호출하여 토큰을 가져와야 합니다.

이는 다단계 OAuth 흐름이므로 API는 장기 실행 작업 (LRO)을 반환합니다. 에이전트는 작업의 수명 주기를 처리해야 합니다.

초기 요청: 에이전트가 retrieveCredentials를 호출합니다.
동의 필요: 사용자가 동의를 부여하지 않은 경우 API는 메타데이터에 auth_uri 및 consent_nonce이 포함된 LRO를 반환합니다.
프런트엔드 리디렉션: 애플리케이션이 사용자를 auth_uri로 리디렉션해야 합니다.
완료: 사용자가 동의를 부여한 후 consent_nonce를 사용하여 FinalizeCredential를 호출하여 흐름을 완료하고 토큰을 가져옵니다.

클라이언트 측 애플리케이션 업데이트

3단계 OAuth의 사용자 로그인 및 리디렉션을 처리하려면 클라이언트 측 애플리케이션에서 다음 단계를 구현하여 사용자 동의를 관리하고 대화를 재개해야 합니다.

승인 트리거 처리
사용자 확인 엔드포인트 구현
에이전트 대화 재개하기

전체 구현 예는 ValidateUserId 프런트엔드 샘플을 참고하세요.

승인 트리거 처리

상담사에게 사용자 동의가 필요한 경우 adk_request_credential 함수 호출이 반환됩니다. 애플리케이션은 이 호출을 가로채 사용자 승인 대화상자를 시작하거나 리디렉션해야 합니다.

인증 제공업체에서 제공한 consent_nonce를 기록하여 세션 컨텍스트를 관리합니다. 이 nonce는 검증 단계에서 사용자를 확인하는 데 필요합니다. 동의가 부여된 후 흐름을 다시 시작할 수 있도록 세션 내에 auth_config 및 auth_request_function_call_id을 저장합니다.

if (auth_request_function_call := get_auth_request_function_call(event_data)):
    print("--> Authentication required by agent.")
    try:
        auth_config = get_auth_config(auth_request_function_call)
        auth_uri, consent_nonce = handle_adk_request_credential(
            auth_config, auth_provider_name, request.user_id
        )
        if auth_uri:
            event_data['popup_auth_uri'] = auth_uri
            fc_id = auth_request_function_call.get('id') if isinstance(auth_request_function_call, dict) else getattr(auth_request_function_call, 'id', None)
            event_data['auth_request_function_call_id'] = fc_id
            event_data['auth_config'] = auth_config.model_dump()

            # Store session state
            if session_id:
                consent_sessions[session_id] = {
                    "user_id": request.user_id,
                    "consent_nonce": consent_nonce
                }
    except Exception as e:
        print(f"Error handling adk_request_credential: {e}")
        # Optionally, add logic to inform the user about the error.

def handle_adk_request_credential(auth_config, auth_provider_name, user_id):
    if auth_config.exchanged_auth_credential and auth_config.exchanged_auth_credential.oauth2:
        oauth2 = auth_config.exchanged_auth_credential.oauth2
        return oauth2.auth_uri, oauth2.nonce
    return None, None

사용자 검증 엔드포인트 구현

웹 서버에 검증 엔드포인트를 구현합니다 (구성 중에 continue_uri로 제공된 URI와 동일). 이 엔드포인트는 다음을 수행해야 합니다.

user_id_validation_state 및 auth_provider_name을 쿼리 매개변수로 수신합니다.
세션 컨텍스트에서 user_id 및 consent_nonce을 가져옵니다.
다음 매개변수를 사용하여 인증 제공업체의 FinalizeCredentials API를 호출합니다.
성공 응답을 받으면 승인 창을 닫습니다.

@app.api_route("/validateUserId", methods=["GET"])
async def validate_user(request: Request):
  auth_provider_name = request.query_params.get("auth_provider_name")
  session_id = request.cookies.get("session_id")
  session = consent_sessions.get(session_id, {})

  payload = {
      "userId": session.get("user_id"),
      "userIdValidationState": request.query_params.get(
          "user_id_validation_state"
      ),
      "consentNonce": session.get("consent_nonce"),
  }

  finalize_url = f"https://iamconnectorcredentials.googleapis.com/v1alpha/{auth_provider_name}/credentials:finalize"

  try:
    async with httpx.AsyncClient(timeout=30.0) as client:
      resp = await client.post(finalize_url, json=payload)
      resp.raise_for_status()
  except httpx.HTTPError as e:
    err_text = e.response.text if hasattr(e, "response") else str(e)
    status = e.response.status_code if hasattr(e, "response") else 500
    return HTMLResponse(err_text, status_code=status)

  return HTMLResponse("""
      <script>
          window.close();
      </script>
      <p>Success. You can close this window.</p>
  """)

상담사와의 대화 재개

사용자가 동의를 부여하고 승인 창이 닫히면 세션 데이터에서 auth_config 및 auth_request_function_call_id를 가져옵니다. 대화를 계속하려면 상담사에게 보내는 새 요청에 이러한 세부정보를 function_response로 포함하세요.

if request.is_auth_resume and session.auth_request_function_call_id and session.auth_config:
    auth_content = types.Content(
        role='user',
        parts=[
            types.Part(
                function_response=types.FunctionResponse(
                    id=session.auth_request_function_call_id,
                    name='adk_request_credential',
                    response=session.auth_config
                )
            )
        ],
    )
    # Send message to agent
    async for event in agent.async_stream_query(
        user_id=request.user_id,
        message=auth_content,
        session_id=session_id,
    ):
        # ...

에이전트 배포

에이전트를 Google Cloud에 배포할 때는 에이전트 ID가 사용 설정되어 있어야 합니다.

Agent Runtime에 배포하는 경우 identity_type=AGENT_IDENTITY 플래그를 사용합니다.

import vertexai
from vertexai import types
from vertexai.agent_engines import AdkApp

# Initialize the Vertex AI client with v1beta1 API for Agent Identity support
client = vertexai.Client(
    project="PROJECT_ID",
    location="LOCATION",
    http_options=dict(api_version="v1beta1")
)

# Use the proper wrapper class for your Agent Framework (e.g., AdkApp)
app = AdkApp(agent=agent)

# Deploy the agent with Agent Identity enabled
remote_app = client.agent_engines.create(
    agent=app,
    config={
        "identity_type": types.IdentityType.AGENT_IDENTITY,
        "requirements": ["google-cloud-aiplatform[agent_engines,adk]", "google-adk[agent-identity]"],
    },
)