AWS 기술 블로그
Amazon Bedrock AgentCore Identity로 안전한 기업형 에이전트 구현하기
“자율 AI”의 등장으로 인한 새로운 보안 과제
AI 에이전트는 단순한 대화형 챗봇을 넘어 실제 업무를 수행하는 단계로 진화하고 있습니다. 에이전트들은 API 호출, 코드 실행, 외부 시스템 제어 등 실제 행동(action)을 수행하며, 다수의 전문화된 에이전트가 협업하는 멀티 에이전트 패턴의 경우 에이전트-도구(tool) 연계를 넘어 에이전트 간 협업까지도 가능하게 하고 있습니다. 그러나 에이전트의 자율성이 높아질수록 새로운 보안 위협의 가능성 또한 증가합니다. 실제로 OWASP Agentic Security Initiative가 발행한 ‘Agentic AI – Threats and Mitigations’는 AI 에이전트에 대한 위협 모델로서 인증과 권한 관리와 관련된 다수의 보안 위협을 명시하고 있습니다.
- 권한 침해(Privilege Compromise): 권한 관리의 취약점을 악용하여 동작을 수행하는 경우로, 권한이 잘못 설정되거나 지나치게 많은 경우, 의도하지 않은 동적 권한 상속이 발생하는 경우 등이 포함됩니다.
- 신원 위장 및 사칭(Identity Spoofing & Impersonation): 공격자가 인증 메커니즘을 악용하여 AI 에이전트나 사용자를 사칭하고, 허위 신원으로 악의적인 동작을 수행합니다.
- 멀티 에이전트 시스템 대상 인간 공격(Human Attacks on Multi-Agent Systems): 공격자가 에이전트 간의 권한 위임 관계, 신뢰 관계, 혹은 워크플로우 종속성을 악용하여 권한을 상승시키거나 동작을 조작합니다.
이와 같은 새로운 보안 위협에 대응하는 과정에서 AI 에이전트 어플리케이션을 구축하는 고객들은 여러 도전적인 과제에 직면하게 됩니다.
- 복잡한 인증/인가 체계: 에이전트가 다양한 도구에 접근할수록 인증/인가 체계가 복잡해지고, 잠재적인 보안 위협과의 접점이 증가합니다.
- 인증 정보 관리의 복잡성: 인증 정보가 여러 워크로드에 분산될 경우 관리 리스크 및 인증 구현의 복잡성이 증가합니다.
- 동적 권한 관리의 어려움: 에이전트가 도구 실행 시 사용자를 대신해 권한 행사를 해야할 경우 해당 권한이 필요한 시점에, 필요한 권한만 부여하는 동적 권한 설정이 필수적입니다. 이때 획득된 인증 정보는 요건에 따라 유효시간 동안 유지하도록 실행 주체를 구분하여 저장하거나 1회성으로만 사용되도록 관리해야 합니다.
본 포스팅에서는 AI 에이전트를 위한 인증 관리 서비스인 Amazon Bedrock AgentCore Identity가 이러한 문제들을 어떻게 해결하는지, 그리고 이를 통해 기업형 AI 에이전트의 안전한 보안 환경을 어떻게 구현할 수 있는지 살펴봅니다.
Amazon Bedrock AgentCore Identity 소개
Amazon Bedrock AgentCore Identity는 AI 에이전트 워크로드의 안전하고 간편한 인증 구현을 위한 관리형 아이덴티티 및 자격 증명(credential) 관리 서비스로, 주요 특징은 다음과 같습니다.
- 중앙화된 인증 정보 관리 및 자격 증명 격리 – 에이전트가 AWS 리소스와 외부 도구에 접근하는 데 필요한 인증 정보를 중앙에서 관리하고, 발급된 자격 증명을 암호화하여 안전하게 저장합니다. 이때 각 에이전트에는 고유한 식별자가 할당되어 서로 다른 에이전트 간 구분이 가능하며, 자격 증명 저장 시에는 에이전트와 해당 에이전트를 호출한 사용자의 조합에 따라 구분하여 자체 토큰 저장소(token vault)에 저장함으로써 권한 격리를 가능하게 합니다.
- 개발 및 운영 복잡도 감소 – 토큰 저장소에 유효한 자격 증명이 있을 경우 해당 정보를 활용하여 도구 호출 시마다 반복적인 인증 과정을 거치지 않아도 되며, 인증 절차 구현을 단순화시키는 AgentCore SDK를 이용하여 개발자가 보안 구현의 복잡성에서 벗어나 비즈니스 로직에 집중할 수 있도록 합니다. 또한 AWS CloudTrail을 통한 감사 로그를 제공하여 에이전트의 자격 증명 접근 활동을 추적할 수 있습니다.
- 사용자 직접 인가를 통한 동적 권한 부여 – Amazon Bedrock AgentCore Identity는 OAuth 2.0 client credentials grant 방식의 two-legged OAuth(2LO) 뿐만 아니라 authorization code grant 방식의 three-legged OAuth(3LO) 역시 네이티브하게 지원합니다. 때문에 해당 사용자 인가가 필요한 도구의 실행 시점에 인증 정보를 요청하여 동적으로 에이전트에게 권한을 부여할 수 있습니다.
- AgentCore Runtime 및 Gateway 인증 설정 간소화 – AgentCore의 세션 단위 에이전트 런타임 서비스인 AgentCore Runtime과 중앙화된 에이전트 도구 엔드포인트를 제공하는 AgentCore Gateway의 경우 호출 시 인증을 요구합니다. 이때 AgentCore Identity를 이용하여 인증 설정을 간소화함으로써 에이전트 기반 환경 구축 및 개발 시에 소요되는 시간을 최소화할 수 있습니다.
또한 AgentCore Identity는 인증/인가 수행을 위한 중계자 역할만 수행하기 때문에 AgentCore Identity에 새로운 인증서버를 구성하거나 사용자 정보를 마이그레이션하는 작업을 요구하지 않습니다. 대신 Google, Okta, Amazon Cognito와 같은 기존 IdP(Identity Provider) 정보를 AgentCore Identity에 등록하여 기존 인증 체계를 유지하며 에이전트 보안을 구현할 수 있습니다.
AgentCore Identity를 이용한 에이전트 인증 구성
AgentCore Identity가 지원하는 기능들을 자세히 살펴보기 위해 예시 시나리오를 기반으로 에이전트 인증을 직접 구현해보도록 하겠습니다.
구축 대상 에이전트 설계하기
기업 사용자들은 업무 시 다양한 소스에서 데이터를 검색하고 필요한 정보만 식별해내는 데에 많은 시간을 투자합니다. 가령 사내 정책이나 표준 프로세스를 확인하기 위해 업무 포털에서 문서를 찾기도 하고, 프로젝트 팀원 간 정보 공유를 위해 구축한 솔루션에서 필요한 문서를 검색하고 그 결과를 정리하는 작업도 자주 하게 됩니다. 이렇듯 업무 시 반복되는 작업을 효율적으로 수행하기 위해 업무 지원 에이전트를 구축해보도록 하겠습니다. 해당 에이전트의 아키텍처는 다음과 같이 구성됩니다.
에이전트는 필요에 따라 사용자가 요청하는 정보를 찾기 위해 세가지 도구를 활용합니다.
- 정책서 전체 목록 조회 도구(
list_company_policies): 공용 MCP 엔드포인트인 AgentCore Gateway의 target으로 등록된 AWS Lambda function에 구현된 도구로, 전체 사내 정책서 목록을 조회하여 문서명과 ID를 반환합니다. - 정책서 내용 조회 도구(
get_company_policy): 공용 MCP 엔드포인트인 AgentCore Gateway의 target으로 등록된 AWS Lambda function에 구현된 도구로, 정책서 ID를 이용하여 해당 문서의 내용을 반환합니다. - 커스텀 Confluence 검색 도구(
search_confluence_page): Strands SDK을 이용해 개발한 커스텀 도구로, 기업에서 자주 사용되는 협업 솔루션 중 하나인 Atlassian Confluence의 API를 이용해 사용자가 요청한 정보와 관련된 문서를 조회하고 각 문서의 내용을 반환합니다.
각 도구 호출 시나 에이전트 접근 시 필요한 인증은 AgentCore Identity를 이용합니다. 해당 아키텍처에서는 백엔드 에이전트 서버인 AgentCore Runtime이나 단일 MCP 도구 엔드포인트 역할을 할 AgentCore Gateway로 접근 시 필요한 인증의 IdP로 Amazon Cognito를 사용하였으나, 기존에 사용 중인 OAuth 2.0 IdP가 있다면 해당 IdP를 AgentCore Identity에 등록하여 사용할 수도 있습니다.
AgentCore Identity에 인증 정보 등록하기
AgentCore Identity로 인증 작업을 수행하기 위해서는 먼저 IdP나 리소스 서버 접근에 필요한 API 키나 OAuth 클라이언트 정보를 자격 증명 공급자(credential provider)로 등록해야합니다. 같은 인증 정보를 사용하는 에이전트 및 AgentCore Gateway 리소스가 있을 경우 자격 증명 공급자를 한번만 등록해주면 각 리소스 혹은 에이전트 서버 별로 인증 정보를 설정하고 유지보수할 필요없이 중앙에서 일관되게 관리할 수 있습니다. Amazon Cognito client 정보 발급을 위한 user pool 생성은 이 문서를, Atlassian client 정보 발급을 위한 app 생성은 이 문서를 참고하시길 바랍니다.
[Amazon Cognito 자격 증명 공급자 등록]
import boto3
import json
agentcore_client = boto3.client('bedrock-agentcore-control')
cognito_cp_response = agentcore_client.create_oauth2_credential_provider(
name='cognito-cred-provider',
credentialProviderVendor='CognitoOauth2',
oauth2ProviderConfigInput={
"includedOauth2ProviderConfig": {
"clientId": COGNITO_CLIENT_ID,
"clientSecret": COGNITO_CLIENT_SECRET,
"issuer": f"https://cognito-idp.{REGION}.amazonaws.com/{COGNITO_USERPOOL_ID}",
"authorizationEndpoint": f"https://{COGNITO_DOMAIN}.auth.{REGION}.amazoncognito.com/oauth2/authorize",
"tokenEndpoint": f"https://{COGNITO_DOMAIN}.auth.{REGION}.amazoncognito.com/oauth2/token",
}
}
)[Atlassian 자격 증명 공급자 등록]
confluence_cp_response = agentcore_client.create_oauth2_credential_provider(
name='confluence-cred-provider',
credentialProviderVendor='AtlassianOauth2',
oauth2ProviderConfigInput={
"atlassianOauth2ProviderConfig": {
"clientId": ATLASSIAN_CLIENT_ID,
"clientSecret": ATLASSIAN_CLIENT_SECRET,
}
}
)등록된 자격 증명 공급자의 API 키나 client secret 등 민감한 정보는 AWS Secrets Manager에 저장되어 안전하게 보관됩니다.
OAuth 자격 증명 공급자의 경우 생성 시 IdP 인증 후 AgentCore Identity로 리다이렉트될 때 필요한 callback URL이 발급됩니다. 해당 callback URL을 Atlassian app의 authorization callback URL로 추가합니다. Amazon Cognito의 경우 이번 워크로드에서는 에이전트가 OAuth access token을 직접 발급받는 시나리오가 없기 때문에 별도 callback URL 설정이 필요하지 않습니다.
print(f"Callback URL: {confluence_cp_response['callbackUrl']}")
# 예시 Callback URL
# https://bedrock-agentcore.{REGION}.amazonaws.com/identities/oauth2/callback/xxx-xxx-xxx
Strands Agents SDK 기반 에이전트 구성
이제 인증을 수행할 에이전트를 생성해보도록 하겠습니다. AgentCore Runtime에 배포될 Strands Agents SDK 기반 에이전트를 선언합니다.
# 예시 main.py
import jwt
from datetime import datetime
from zoneinfo import ZoneInfo
from bedrock_agentcore import BedrockAgentCoreApp
from bedrock_agentcore.memory.integrations.strands.config import AgentCoreMemoryConfig
from bedrock_agentcore.memory.integrations.strands.session_manager import AgentCoreMemorySessionManager
from strands import Agent
from strands.hooks import HookProvider
app = BedrockAgentCoreApp()
...
@app.entrypoint
async def agent_entrypoint(payload, context):
# 현재 사용자 session_id 추출
session_id = getattr(context, "session_id")
# 사용자 질문 추출
user_message = payload.get("prompt")
# Cognito 인증 시 사용된 token 및 token 내 user 정보 추출
cognito_auth = context.request_headers.get('Authorization')
cognito_token = cognito_auth.replace('Bearer ', '')
claims = jwt.decode(cognito_token, options={"verify_signature": False})
user_id = claims.get('username')
# AgentCore Memory config 생성
agentcore_memory_config = AgentCoreMemoryConfig(
memory_id=AGENTCORE_MEMORY_ID,
session_id=session_id,
actor_id=user_id
)
# 대화이력 관리를 위한 AgentCore Memory session manager 생성
session_manager = AgentCoreMemorySessionManager(
agentcore_memory_config=agentcore_memory_config,
region_name=REGION
)
...
# agent 초기화
agent = Agent(
model=MODEL_ID,
system_prompt=f"""당신은 기업 사용자를 위한 친절한 개인 비서입니다.
현재 날짜 및 시간: {datetime.now(ZoneInfo("Asia/Seoul")).strftime("%Y.%m.%d %H:%M")}
친근하면서도 전문적인 톤으로 답변해주세요.""",
session_manager=session_manager
)
# agent 호출
response = agent(user_message)
# agent 답변 반환
return response.message['content'][0]['text']
if __name__ == "__main__":
app.run()인증 시나리오 구현 #1: AgentCore Runtime 접근 시
구성된 기본 에이전트를 AgentCore Runtime에 배포해보겠습니다. AgentCore Runtime 접근 시 인증은 별도 인증 모듈 개발이나 서버 구성없이 AgentCore Identity를 이용하여 간편하게 구성할 수 있습니다. 이번 워크로드에서는 Runtime 인증에 Cognito를 사용할 예정이기 때문에 Runtime 배포 시 Cognito의 app client ID와 Discovery URL을 이용해 authorizer configuration을 추가해줍니다. 또한 예시 main.py와 같이 JWT 토큰을 활용할 경우, request_header_configuration을 이용해 에이전트로 전달을 허용할 request header로 ‘Authorization’을 추가합니다.
# 간단한 AgentCore Runtime 배포를 위해 Starter Toolkit 사용
from bedrock_agentcore_starter_toolkit import Runtime
# AgentCore Runtime client 생성
agentcore_runtime = Runtime()
# Cognito 인증 정보 설정
auth_config = {
"customJWTAuthorizer": {
"allowedClients": [
COGNITO_APPCLIENT_ID
],
"discoveryUrl": COGNITO_DISCOVERY_URL
}
}
# AgentCore Runtime 설정 파일 생성
response = agentcore_runtime.configure(
entrypoint="main.py", # 작성한 에이전트 코드 파일 위치
auto_create_execution_role=True,
auto_create_ecr=True,
requirements_file="requirements.txt", # requirements.txt 파일 위치
authorizer_configuration=auth_config,
request_header_configuration={"requestHeaderAllowlist": ["Authorization"]},
region=REGION,
agent_name="biz_assist_agent",
)
# AgentCore Runtime 배포
launch_result = agentcore_runtime.launch()배포된 Runtime에 접근해보도록 하겠습니다. 해당 에이전트 Runtime은 JWT 인증을 요구하기 때문에 어플리케이션에서 에이전트 호출 전 Cognito access token을 발급 받은 후 ‘Authorization’ header를 추가합니다.
# boto3 Cognito client 생성
cognito_client = boto3.client('cognito-idp', region_name=REGION)
# 유저 정보로 인증 후 access token 획득
auth_response = cognito_client.initiate_auth(
ClientId=client_id,
AuthFlow='USER_PASSWORD_AUTH',
AuthParameters={
'USERNAME': USERNAME,
'PASSWORD': PASSWORD
}
)
bearer_token = auth_response['AuthenticationResult']['AccessToken']
...
# Runtime ARN을 이용해 에이전트 호출 URL 생성
escaped_agent_arn = urllib.parse.quote(agent_arn, safe='')
url = f"https://bedrock-agentcore.{REGION}.amazonaws.com/runtimes/{escaped_agent_arn}/invocations?qualifier=DEFAULT"
# Set up headers
headers = {
"Authorization": f"Bearer {bearer_token}", # Cognito access token
"Content-Type": "application/json",
"X-Amzn-Bedrock-AgentCore-Runtime-Session-Id": SESSION_ID # 사용자 세션 ID
}
invoke_response = requests.post(
url,
headers=headers,
data=json.dumps({"prompt": PROMPT}), # 사용자 질문을 payload로 전달
stream=True
)에이전트 호출 시 AgentCore Identity를 통해 성공적으로 인증되어 에이전트가 실행되고, Runtime 내에서 ‘Authorization’ header로 전달된 토큰도 확인할 수 있습니다.
인증 시나리오 구현 #2: AgentCore Gateway 도구 접근 시
이번엔 AgentCore Runtime과 동일한 IdP를 공유하는 AgentCore Gateway의 인증을 구현해보겠습니다. AgentCore Gateway가 제공하는 정책서 전체 목록 조회 도구 및 정책서 내용 조회 도구의 경우 사내 구성원 전체가 사용 가능한 공용 MCP 도구입니다. 때문에 동일하게 사내 임직원들이 사용하는 업무 지원 에이전트와 같은 IdP 설정을 공유하며, 해당 IdP인 Cognito의 인증 정보는 이미 자격 증명 공급자로 등록되어 있기 때문에 추가 등록이 필요없습니다. 때문에 Runtime과 유사하게 AgentCore Gateway ‘customJWTAuthorizer’ 설정만 추가한 후 gateway 리소스를 생성합니다.
# Cognito 인증 정보 설정
auth_config = {
"customJWTAuthorizer": {
"allowedClients": [
COGNITO_APPCLIENT_ID
],
"discoveryUrl": COGNITO_DISCOVERY_URL
}
}
# AgentCore Gateway 생성
gw_response = agentcore_client.create_gateway(
name='CommonMCPGateway',
roleArn = GATEWAY_ROLE_ARN,
protocolType='MCP',
authorizerType='CUSTOM_JWT',
authorizerConfiguration=auth_config,
description='AgentCore Gateway for common MCP tools'
)생성된 Gateway에 Lambda target을 추가합니다.
lambda_target_config = {
"mcp": {
"lambda": {
"lambdaArn": LAMBDA_FUNCTION_ARN, # 도구 코드가 구현된 Lambda function의 ARN
"toolSchema": {
"inlinePayload": [
{
"description": "Tool that lists all company policy document IDs and titles.",
"inputSchema": {
"properties": {},
"type": "object"
},
"name": "list_company_policies"
},
{
"description": "Tool that retrieves a company policy document by its ID, returning both title and content.",
"inputSchema": {
"properties": {
"docId": {
"description": "The document ID of the company policy.",
"type": "string"
}
},
"required": [
"docId"
],
"type": "object"
},
"name": "get_company_policy"
}
]
}
}
}
}
credential_config = [
{
"credentialProviderType" : "GATEWAY_IAM_ROLE"
}
]
targetname='LambdaToolTarget'
response = agentcore_client.create_gateway_target(
gatewayIdentifier=gw_response['gatewayId'],
name=targetname,
description='Lambda Target for AgentCore Gateway',
targetConfiguration=lambda_target_config,
credentialProviderConfigurations=credential_config
)MCP 도구 구성이 완료되어 에이전트에서 도구를 호출할 수 있도록 설정해보겠습니다. 도구 호출 시 동일한 인증을 중복으로 수행할 필요없이 기존에 에이전트 호출 시 사용된 인증 정보를 재사용(identity propagation)할 수 있습니다. 앞서 작성한 main.py 코드를 수정하여 Gateway 엔드포인트 URL과 Cognito access token을 이용해 MCP 클라이언트를 생성하고 Gateway에 등록된 도구들을 에이전트에 추가하겠습니다.
# 예시 main.py
...
from mcp.client.streamable_http import streamablehttp_client
from strands.tools.mcp import MCPClient
# Gateway URL과 access token으로 MCP client 생성
async def get_mcp_client(access_token: str, gateway_url: str):
mcp_client = MCPClient(
lambda: streamablehttp_client(
f"{gateway_url}", headers={"Authorization": f"Bearer {access_token}"}
)
)
return mcp_client
...
@app.entrypoint
async def agent_entrypoint(payload, context):
...
# Cognito access token 추출
cognito_auth = context.request_headers.get('Authorization')
cognito_token = cognito_auth.replace('Bearer ', '')
...
# MCP Client 생성
mcp_client = await get_mcp_client(cognito_token, GATEWAY_URL)
# 전체 Gateway 도구 목록 가져오기
tool_list = []
mcp_client.__enter__()
gateway_tools = mcp_client.list_tools_sync()
tool_list.extend(gateway_tools)
# agent 초기화
agent = Agent(
model=MODEL_ID,
# 시스템 프롬프트 수정: 가용 도구 안내
system_prompt=f"""당신은 기업 사용자를 위한 친절한 개인 비서입니다.
당신은 다음의 작업을 할 수 있습니다.
- 회사 사내 정책서 목록 및 내용 조회
현재 날짜 및 시간: {datetime.now(ZoneInfo("Asia/Seoul")).strftime("%Y.%m.%d %H:%M")}
친근하면서도 전문적인 톤으로 답변해주세요.""",
session_manager=session_manager,
tools=tool_list, # 에이전트에 도구 등록
)
...등록된 도구를 이용해 에이전트에게 사내 정책서 조회를 요청해보겠습니다. 별도 인증 과정없이 기존 인증정보를 이용해 바로 도구를 호출하여 관련 문서를 조회하고 정책 내용을 안내해주는 것을 확인할 수 있습니다.
AgentCore Gateway에는 Lambda function 외에도 MCP 서버나 REST API, 사전 구성된 템플릿을 통해 등록된 3rd party 솔루션 역시 도구 대상(tool target)으로 추가할 수 있습니다. 이때 해당 대상 유형의 도구에 접근하기 위해 필요한 AgentCore Gateway와 대상 간 인증 구현 역시 AgentCore Identity 자격 증명 공급자 설정을 이용해 간소화할 수 있습니다. 자세한 내용은 이 문서에서 확인하시길 바랍니다.
인증 시나리오 구현 #3: 3LO를 통한 커스텀 Confluence 도구 호출 시
마지막으로 Confluence에서 문서 조회를 위해 사용자 인증/인가 수행 후 Atlassian API로 검색 및 문서 내용 조회를 수행하는 도구를 구현해보겠습니다. 해당 인증 유형은 앞서 다룬 인증 시나리오와는 몇 가지 차이점이 있습니다. 먼저 AgentCore Runtime 및 Gateway 인증의 경우 각 리소스로 들어오는 inbound 요청에 대한 유효성 및 적절성을 확인하는 과정이었다면, 이번엔 Confluence 리소스 서버에 에이전트가 접근하기 위해 OAuth 토큰을 IdP로부터 발급받아야하는 outbound 인증 요청 유형입니다(시나리오 #2의 에이전트 -> AgentCore Gateway로의 접근 또한 outbound 요청에 해당하나 identity propagation으로 인해 추가 인증이 진행되지 않아 outbound 인증은 수행되지 않았습니다). 또한 해당 에이전트 도구의 특성을 고려할 때 Confluence에서 사용자 별로 권한을 가진 컨텐츠에만 접근해야하기 때문에 일반적인 machine to machine 인증(2LO)이 아닌 사용자 단위의 직접 인증(3LO)이 필요합니다. 두 인증 패턴 중 어떤 유형이 더 적절할 지 검토할 때는 이 문서를 참고하시길 바랍니다.
AgentCore Identity가 outbound 인증을 어떻게 지원하는 지 알아보기 위해서는 몇 가지 용어에 대한 이해가 필요합니다.
| 용어 | 설명 |
|---|---|
| 에이전트 아이덴티티
(Agent identity) |
각 에이전트의 고유 식별자 및 관련 메타데이터
AgentCore Identity에서는 ‘워크로드 아이덴티티(Workload Identity)’라는 명칭으로 구현 |
| 에이전트 아이덴티티 디렉토리
(Agent identity directory) |
에이전트 아이덴티티에 대해 single source of truth 역할을 하는 중앙 디렉토리 서비스
각 에이전트 아이덴티티와 관련된 메타데이터, 접근 정책을 저장 |
| 토큰 저장소
(Token vault) |
OAuth 토큰이나 API 키를 포함한 자격 증명들의 저장소
최초에 해당 자격 증명을 획득한 에이전트+사용자 조합일 경우에만 접근 가능하도록 관리 |
| 워크로드 access token
(Workload access token) |
워크로드 아이덴티티와 사용자 아이덴티티 정보(ex. user ID, JWT 토큰)를 모두 포함하는 토큰으로,
두 아이덴티티의 조합으로 토큰 저장소 접근을 통제하기 위해 사용 |
해당 기능들을 이용해 3LO를 수행하게 되며, 이때 인증 과정은 크게 네 단계로 나눌 수 있습니다.
- 에이전트 호출: 사용자가 어플리케이션에 접근하여 에이전트가 호출되고, 3LO 수행이 필요한 도구가 실행됩니다.
- Authorization URL 발급: OAuth access token 발급을 위해 사용자 인증/인가를 수행할 authorization URL을 전달합니다.
- Access token 발급: 사용자 인증/인가 후 access token을 발급받아 AgentCore Identity의 토큰 저장소에 저장합니다.
- 도구 실행 및 응답 반환: 발급 받은 access token을 이용해 외부 API를 호출한 후 도구 실행 및 에이전트 응답 생성을 완료합니다.
해당 인증 과정을 세분화하면 다음과 같이 도식화할 수 있습니다. 이제부터는 실제로 AgentCore Identity를 이용해 아래의 3LO outbound 인증을 단계적으로 구현해보겠습니다.
1. (최초 1회) 워크로드 아이덴티티 생성
AgentCore Identity를 이용해 outbound 인증을 수행하려면 먼저 에이전트 ID와도 같은 워크로드 아이덴티티를 생성해야합니다. 워크로드 아이덴티티는 에이전트 등 인증을 요청하는 주체를 식별하기 위한 고유 식별자로, 다음과 같은 포맷으로 생성됩니다. 워크로드 아이덴티티는 CreateWorkloadIdentity API로 생성 가능하며, AgentCore Runtime 혹은 Gateway 리소스를 배포할 경우 자동으로 함께 생성됩니다.
[워크로드 아이덴티티 예시]
2. 워크로드 access token 획득
워크로드 아이덴티티가 생성되었다면 토큰 저장소에 자격 증명을 저장하고 조회하기 위해 워크로드 access token을 획득해야합니다. AgentCore Identity는 워크로드 아이덴티티와 사용자 아이덴티티 정보의 조합으로 워크로드 access token을 생성함으로써 동일한 에이전트, 동일한 사용자의 요청일 때만 해당 토큰 저장소에 접근이 가능하도록 설계되었습니다. 워크로드 access token은 GetWorkloadAccessTokenForJWT, GetWorkloadAccessTokenForUserId, GetWorkloadAccessToken(예외적으로 사용자 대신 권한 행사를 하지 않을 경우 워크로드 아이덴티티만 이용하여 획득) API로 발급 및 조회가 가능합니다. AgentCore Runtime 혹은 Gateway의 경우 워크로드 access token 역시 자동으로 획득합니다.
[워크로드 access token 예시]
워크로드 access token은 opaque token으로 생성됩니다.
3. AgentCore SDK를 이용한 인증 흐름 구현
이렇게 획득한 워크로드 access token을 이용해 에이전트는 outbound 인증에 필요한 자격 증명 정보를 토큰 저장소에 저장하고 조회할 수 있습니다. 자격 증명 획득 시에는 AgentCore Python SDK를 이용해 인증 절차를 간소화할 수 있습니다. AgentCore Identity는 @requires_api_key, @requires_access_token 두 개의 decorator를 제공합니다. 각 decorator를 API 키, OAuth access token을 요청하는 함수에 추가하면 AgentCore Identity가 자동으로 자격 증명 공급자 정보를 이용해 자격 증명을 획득합니다.
[Decorator 사용 예시]
이번 Confluence 도구 접근 시 필요한 자격 증명도 해당 decorator를 이용해서 발급받아 보겠습니다.
from typing import Dict, Any
import requests
import json
import asyncio
from strands import tool
from bedrock_agentcore.identity.auth import requires_access_token
...
class MessageQueue:
"""3LO 수행 시 인증 URL을 event loop 중단없이 출력하기 위한 비동기 큐"""
def __init__(self):
self._queue = asyncio.Queue()
self._finished = False
async def put(self, item: dict) -> None:
"""큐에 item 추가"""
await self._queue.put(item)
async def finish(self) -> None:
"""큐 종료"""
self._finished = True
await self._queue.put(None)
async def stream(self):
"""큐에 있는 item들을 반환"""
while True:
item = await self._queue.get()
if item is None and self._finished:
break
yield item
queue = MessageQueue()
@tool
async def search_confluence_page(
search_query: str,
) -> Dict[str, Any]:
"""
Get all found Confluence pages with search_query included in the content
Args:
search_query: query to use when searching Confluence page contents
Returns:
Dict[str, Any]: A dictionary containing Confluence contents
"""
# 인증 시작 안내 메세지
response_data = {
"type": "info",
"content": "🔐 Confluence 인증을 시작합니다..."
}
await queue.put(response_data)
# Authorization URL 발급 후 사용자 전달을 위한 callback 함수
async def on_auth_url(url: str) -> None:
await queue.put({"type": "auth_url", "content": url})
# AgentCore SDK를 이용한 OAuth 토큰 발급 요청
@requires_access_token(
provider_name=CRED_PROVIDER_NAME,
scopes=["search:confluence", "read:page:confluence", "offline_access"],
auth_flow='USER_FEDERATION',
on_auth_url=on_auth_url,
into="access_token",
callback_url=CALLBACK_URL,
force_authentication=False,
)
async def request_confluence_token(access_token: str):
return access_token
# AgentCore Identity에 access token 요청 시작
confluence_token = await request_confluence_token(access_token='')
response_data = {
"type": "info",
"content": "✅ Confluence 인증이 완료되었습니다!"
}
await queue.put(response_data)
# 획득한 access token으로 Confluence에서 page 컨텐츠 검색
search_url = f"https://api.atlassian.com/ex/confluence/{CONFLUENCE_CLOUD_ID}/wiki/rest/api/search"
headers = {
"Accept": "application/json",
"Authorization": f"Bearer {confluence_token}"
}
# Confluence 쿼리 문법 작성
cql = f"siteSearch ~ '{search_query}' order by created"
query = {
'cql': cql,
'limit': 5
}
response = requests.get(
search_url,
headers=headers,
params=query
)
response_json = json.loads(response.text)
response_results = response_json.get('results')
content_xml = ""
# 검색된 Confluence page들의 전체 컨텐츠 조회 후 도구 실행 결과 반환
if response_results and len(response_results) > 0:
for idx, result in enumerate(response_results):
content_id = result['content']['id']
read_content_url = f"https://api.atlassian.com/ex/confluence/{CONFLUENCE_CLOUD_ID}/wiki/api/v2/pages/{content_id}?body-format=storage"
response = requests.get(
read_content_url,
headers=headers
)
response_json = json.loads(response.text)
page_title = response_json['title']
page_body = response_json['body']['storage']['value']
content_xml += f"""<{idx+1}번째 Page>"""
content_xml += f"""<Page Title>{page_title}</Page Title>"""
content_xml += f"""<Page Content Body>{page_body}</Page Content Body>"""
content_xml += f"""</{idx+1}번째 Page>"""
return f"""<message>{len(response_results)}개의 검색 결과가 있습니다.</message><confluence_search_result>{content_xml}</confluence_search_result>"""
...
# 에이전트 코드 수정
@app.entrypoint
async def agent_entrypoint(payload, context):
...
# 커스텀 도구 추가
tool_list.append(search_confluence_page)
# agent 초기화
agent = Agent(
model=MODEL_ID,
# 시스템 프롬프트 수정: 추가 가용 도구 안내
system_prompt=f"""당신은 기업 사용자를 위한 친절한 개인 비서입니다.
당신은 다음의 작업을 할 수 있습니다.
- 회사 사내 정책서 목록 및 내용 조회
- Confluence에서 프로젝트 관련 문서 검색
현재 날짜 및 시간: {datetime.now(ZoneInfo("Asia/Seoul")).strftime("%Y.%m.%d %H:%M")}
친근하면서도 전문적인 톤으로 답변해주세요.""",
session_manager=session_manager,
tools=tool_list,
)
async def process_user_message(user_message: str):
stream_response = agent.stream_async(user_message)
chunk_buffer = ""
async for event in stream_response:
# 생성된 Assistant 메세지 큐에 쌓기
if "data" in event:
chunk = event["data"]
chunk_buffer += chunk
response_data = {
"type": "generated_text",
"content": chunk
}
await queue.put(response_data)
...
task = asyncio.create_task(process_user_message(user_message))
async def stream_with_task():
"""큐에 쌓인 메세지 출력"""
async for item in queue.stream():
yield item
await task
return stream_with_task()상기 코드는 서비스의 기능 이해를 돕기 위한 예시로, 세션 격리 등은 고려되지 않았습니다.
requires_access_token decorator의 인자들을 자세히 살펴보겠습니다.
provider_name: 자격 증명 공급자명scopes: OAuth 2.0 scope. “offline_access”는 refresh token 발급을 위해 추가(참고).auth_flow: 인증 flow 유형. ‘USER_FEDERATION'(3LO)과 ‘M2M'(2LO) 중에 선택.on_auth_url: Authorization URL 발급 후 사용자 전달을 위한 callback 함수into: 토큰 정보를 주입시킬 파라미터명(기본값: access_token)callback_url: 인증 완료 후 이동할 어플리케이션 callback URLforce_authentication: 재인증 강제 여부, 해당 도구 실행 건에서만 1회성으로 자격 증명을 사용할 경우 활성화
이때 ‘callback_url’로 설정되는 어플리케이션 callback URL은 3LO 인증 시 authorization URL이 유출된 상황에서 악의적인 사용자가 인증을 대신 수행하는 상황을 방지하기 위해 인증 절차 완료 전 해당 인증을 요청한 사용자가 맞는 지 검증하고 완료하기 위해 호출됩니다. 이를 세션 바인딩(session binding)이라고 합니다. 세션 바인딩 시에는 로그인된 사용자 정보의 유효성을 검증한 후 CompleteResourceTokenAuth API를 호출하여 해당 자격 증명이 정상적으로 발급되었음을 AgentCore Identity에 알립니다. 이때 사용되는 로그인된 사용자 정보는 사용자 토큰 또는 사용자 ID 값이며, 워크로드 access token 획득 시 사용된 사용자 정보와 동일해야합니다.
# 워크로드 access token 획득 시 사용자 토큰을 사용한 경우의 예시 코드
from bedrock_agentcore.services.identity import IdentityClient, UserTokenIdentifier
...
def _handle_3lo_callback(self, request: Request) -> JSONResponse:
# Query parameter로 전달되는 session_id(session URI) 값 추출
session_id = request.query_params.get("session_id")
if not session_id:
return JSONResponse(status_code=400, content={"message": "session_id 파라미터가 존재하지 않습니다."})
# 현재 로그인된 사용자 세션 유효성 검증
# 원격 세션 저장소가 아닌 브라우저 쿠키 등에서 직접 로그인된 사용자 정보 추출
session_details = validate_session_cookies(request.cookies.get('my-application-cookie'))
...
# 워크로드 access token 획득 시 사용된 사용자 토큰 추출
user_token = session_details.get(USER_TOKEN_PARAM_NAME)
if not user_token:
return JSONResponse(status_code=500, content={"message": "유효하지 않은 사용자 정보입니다."})
# 유효한 사용자 정보와 session URI에 대해 CompleteResourceTokenAuth API 호출
identity_client = IdentityClient(region=REGION)
identity_client.complete_resource_token_auth(
session_uri=session_id, user_identifier=UserTokenIdentifier(user_token=user_token) # 사용자 ID 유형의 경우 'UserIdIdentifier' 사용
)
return JSONResponse(status_code=200, content={"message": "OAuth2 3LO 인증이 완료되었습니다."})어플리케이션 callback 엔드포인트 구성이 완료되었다면 워크로드 아이덴티티에 해당 엔드포인트를 승인된 return URL로 등록해야합니다.
from bedrock_agentcore.services.identity import IdentityClient
identity_client = IdentityClient(region=REGION)
# 워크로드 아이덴티티 정보 조회
workload_identity = identity_client.get_workload_identity(name=WORKLOAD_NAME)
# 기존에 등록된 OAuth return URL이 있을 경우 추출
allowed_return_urls = workload_identity.get("allowedResourceOauth2ReturnUrls") or []
# Session binding을 위한 callback URL 설정
new_return_url = CALLBACK_URL
# callback URL 등록 혹은 추가
updated_workload_identity = identity_client.update_workload_identity(
name=WORKLOAD_NAME,
allowed_oauth2_return_urls=[*allowed_return_urls, new_return_url],
)Callback 설정까지 완료되었다면 3LO 인증을 통해 자격 증명을 발급받은 후 Confluence 검색 도구를 실행해보겠습니다. 이번에 인증을 수행해볼 첫번째 사용자는 다음 Confluence 문서에 대해 조회 권한을 갖고 있습니다. 
이중 회의록과 관련된 문서만 조회하여 요약해보겠습니다.
성공적으로 authorization URL 발급 및 사용자 인증/인가 후 도구가 실행되어 회의록 내용을 요약해주는 결과를 볼 수 있습니다. 이번엔 두번째 사용자로 로그인해서 기존 자격 증명을 재사용하진 않는 지 확인해보겠습니다. 두번째 사용자는 Confluence 역시 다른 계정을 사용하며, 회의록 문서의 경우 아래와 같이 Project A가 아닌 Project B에 대해서만 권한을 보유하고 있습니다.
새로운 사용자가 동일한 에이전트에서 동일한 도구를 수행하더라도 에이전트+사용자 별로 자격 증명 정보를 구분하여 관리하기 때문에 별도 자격 증명으로 도구를 실행함을 확인할 수 있습니다. 또한 동일한 도구를 실행함에도 이번엔 다른 Confluence 사용자로 로그인하였기 때문에 Project A 회의록을 제외한 Project B의 회의록만 조회된 것처럼 사용자에 따라 다른 권한으로 도구를 실행할 수 있다는 점도 관찰할 수 있습니다. 이처럼 AgentCore Identity를 이용해 에이전트에게 미리, 광범위한 권한을 부여하는 대신 필요한 시점에, 사용자 단위로 인증/인가를 수행하여 허용된 동작만 수행할 수 있도록 구성하는 것이 가능함을 확인할 수 있습니다.
마무리
AI 에이전트의 자율성은 비즈니스 혁신을 가속화하지만, 동시에 보안 아키텍처에 새로운 접근 방식을 요구합니다. 에이전트가 실제 동작을 수행하고 다양한 시스템과 상호작용하는 환경에서는 전통적인 애플리케이션 보안 모델만으로는 효율적인 인증정보 관리 및 권한 격리, 사용자 컨텍스트 기반 접근 제어와 같은 에이전트 특화된 보안 요구사항을 충족하기 어렵습니다.
Amazon Bedrock AgentCore Identity는 이러한 과제에 대한 해답을 제시합니다. 에이전트와 사용자 조합별 자격 증명 격리를 통해 권한 오남용 위험을 최소화하면서도 인증정보 관리 중앙화와 SDK 활용을 통해 개발 및 운영 복잡도를 크게 줄입니다. 무엇보다 기존 IdP와의 원활한 통합을 통해 기업이 구축한 아이덴티티 체계를 활용하면서도 에이전트 특화된 보안을 빠르게 구현할 수 있습니다. 이외에도 AgentCore Identity가 제공하는 다양한 기능들과 구체적인 활용 사례가 궁금하시다면 실제 데모와 함께 진행되는 저자의 강의 영상을 포함해 아래에 제공되는 레퍼런스들을 확인해보시길 바랍니다.
AI 에이전트 기술이 더욱 정교해지고 비즈니스 크리티컬한 업무로 확장됨에 따라, 안전하고 신뢰 가능한 에이전트 실행 환경 구축은 선택이 아닌 필수가 되고 있습니다. Amazon Bedrock AgentCore Identity를 통해 보안과 혁신의 균형을 유지하면서 차세대 AI 에이전트 애플리케이션을 구축하시기 바랍니다.
References
- Amazon Bedrock AgentCore 강의 시리즈
- Amazon Bedrock AgentCore Identity 문서
- Amazon Bedrock AgentCore Starter Toolkit GitHub
- Amazon Bedrock AgentCore Samples 리포지토리