AWS 기술 블로그
Amazon Bedrock AgentCore를 활용한 멀티에이전트 운영과 접근제어
AI 에이전트를 처음 구축할 때 가장 단순한 접근 방식은 하나의 에이전트가 외부 서비스(API, MCP)를 직접 호출하도록 구성하는 것 입니다. 이러한 구조는 초기 PoC 단계에서는 구현이 간단하고, 빠르게 아이디어를 검증하는 데 효과적입니다. 그러나 에이전트 기반 시스템을 엔터프라이즈 환경으로 확장하기 시작하면, 이러한 접근 방식은 곧 한계에 부딪히게 됩니다. 에이전트의 수가 증가하고 외부 API, MCP 내부 서비스가 지속적으로 추가되면서 각 에이전트가 개별적으로 인증과 호출 로직을 관리하는 구조는 운영 복잡성을 빠르게 증가시킵니다.
엔터프라이즈 환경의 현실
- 에이전트 수가 1개에서 여러 개로 확장되며, 호출해야 할 외부 API, MCP, 내부 서비스의 수가 지속적으로 증가합니다.
- 각 시스템은 API Key, OAuth, JWT 등 서로 다른 인증 방식을 사용합니다.
- 각 에이전트는 사전에 정의된 범위 내에서만 도구를 호출할 수 있어야 하며, 이에 대한 명확한 권한 제어가 필요합니다.
이 단계에서 문제의 핵심은 더 이상 모델의 성능이나 프롬프트 설계가 아니라 에이전트를 어떻게 ‘운영 가능한 시스템’으로 설계하고 관리할 것인가입니다. 이 글에서는 이러한 AI Agent 개발 시에 당면할 문제들을 바탕으로, Amazon Bedrock AgentCore가 왜 필요한지를 설명합니다.
1. 멀티 에이전트 운영의 현실적인 Pain Point
Pain Point 1. N×N 연결 복잡도의 증가
에이전트마다 외부 API나 MCP와 같은 호출해야 할 도구가 늘어나면서, 에이전트와 도구 간의 연결은 자연스럽게 N×N 구조로 확장됩니다. 이 구조에서는 에이전트 수와 도구 수에 비례하여 연결 지점이 기하급수적으로 증가합니다.
Figure 1. NxN 연결 복잡도
그 결과, 외부 API의 엔드포인트 변경이나 인증 방식 변경과 같은 운영상 변경 사항이 발생할 때마다, 관련된 모든 에이전트 코드를 수정하고 재배포해야 하는 상황이 반복됩니다.
엔터프라이즈 환경에서 비용이 크게 발생하는 지점은 초기 구현 단계가 아니라, 변경 사항을 반영하고 이를 안정적으로 운영·유지하는 과정입니다.
Pain Point 2. 인증관리의 분산
엔터프라이즈 환경에서 에이전트가 호출해야하는 도구와 API 는 서로 다른 보안 표준을 가지고 있습니다. 예를 들어 Amadeus API는 전세계 항공사와 여행사가 사용하는 글로벌 항공 예약 표준 API로 OAuth Client Credentials Flow를 필수로 요구합니다.
이와 함께 기업 내부에서는 Okta, Microsoft Entra ID, Amazon Cognito 등 서로 다른 IdP를 사용하고, Slack, Salesforce와 같은 서드파티 API 역시 각기 다른 인증 방식을 요구하는 경우가 일반적입니다. 이러한 환경에서 인증 로직을 에이전트 코드에 직접 구현하기 시작하면 다음과 같은 문제가 발생합니다.
- 토큰 발급·갱신 로직이 에이전트마다 중복됩니다.
- 보안 설정이 코드 곳곳에 흩어지며, 이 인증 정보는 어디에서 관리되고 있는가? 확인하기 어렵습니다.
- 무엇보다 어떤 Agent가 어떤 API에 접근할 수 있는지 중앙에서 통제하거나 추적하기 어려운 상태가 됩니다.
Pain Point 3. Agent Access Control의 부재
멀티 에이전트 환경에서는 에이전트별로 수행할 수 있는 작업을 명확히 구분하고, 이에 대한 권한 제어(Access Control)가 반드시 필요합니다. 운영 과정에서 다음과 같은 요구사항이 자연스럽게 등장합니다.
- 이 에이전트는 조회만 가능해야 하지 않는가?
- 특정 에이전트가 특정 MCP만 호출하도록 제한해야 하지 않는가?
- 프롬프트 오류나 인젝션으로 권한 밖 도구 호출이 발생하면 어떻게 막을 것인가?
프롬프트나 애플리케이션 코드 수준의 제어는 엔터프라이즈 기준에서는 충분한 보안 장치가 되기 어렵습니다.
Pain Point 4. 실행 환경과 관측(Observability)이 분산
에이전트는 Amazon Elastic Container Service(Amazon ECS), Amazon Elastic Kubernetes Service(Amazon EKS), Amazon EC2, AWS Lambda 등 서로 다른 환경에서 실행되는 경우가 많습니다. 이때 흔히 겪는 문제는 다음과 같습니다.
- 로그와 메트릭이 여러 서비스에 분산되어 수집됩니다.
- 멀티 에이전트 호출 흐름을 한눈에 파악하기 어렵습니다.
- 장애 발생 시 어느지점에서 병목이 발생했는지 추적하기 어렵습니다.
결과적으로 에이전트 수가 늘어날수록 시스템의 전체 동작을 이해하고 문제를 진단하는 데 점점 더 많은 시간을 소모하게 됩니다.
2. Pain Points를 해결하는 방법: Amazon Bedrock AgentCore
Amazon Bedrock AgentCore는 이러한 문제를 모델이나 프롬프트가 아닌 ‘플랫폼 레이어’에서 해결합니다.
AgentCore는 멀티 에이전트 시스템을 운영 가능한 구조로 만들기 위한 세 가지 핵심 구성 요소로 이루어져 있습니다.
- AgentCore Gateway: 대규모 도구를 관리하기 위한 중앙 집중식 서버 역할로 에이전트가 도구를 검색, 액세스 및 호출할 수 있는 통합 인터페이스를 제공합니다. 이 글에서는 모든 도구 및 외부 API 호출을 AgentCore Gateway로 통합하고 단일 진입점으로 활용합니다.
- AgentCore Identity: 에이전트의 신원을 안전하게 관리하고, 인증과 권한 위임을 표준화하는 레이어입니다. 사용자 마이그레이션이나 인증 흐름을 재구축하지 않고도, 에이전트별 인증과 접근 제어를 일관된 방식으로 적용할 수 있습니다. 이 글에서는 에이전트 인증을 중앙화하는 역할로 활용합니다.
- AgentCore Runtime: 인기 있는 오픈 소스 프레임워크, 도구, 모델을 비롯하여 에이전트 프레임워크를 지원하고, 멀티모달 워크로드와 장기 실행 에이전트를 처리하는 샌드박스형 저지연 서버리스 환경을 제공합니다. 이 글에서는 에이전트 실행과 운영을 하나의 환경으로 통합하여, 로그와 실행 흐름을 일관되게 관측할 수 있도록 활용합니다.
Figure 2. Bedrock AgentCore 사용한 전체 아키텍처
이 조합이 어떻게 Pain Point를 해결하는지, 아래 멀티 에이전트를 활용한 항공사 예약 서비스를 기반으로 설명하겠습니다.
3. 비즈니스 시나리오 : 멀티 에이전트를 활용한 항공사 예약 서비스
항공사 환경에서 자주 발생하는 문의 유형을 기준으로, 네 가지 역할의 에이전트를 구성하였습니다.
- Orchestrator Agent – 사용자 요청 분석 및 에이전트 조율합니다.
- Booking Agent – 항공편 예약 및 고객 정보를 조회합니다.
- Loyalty Agent – 마일리지 및 회원 등급을 관리 합니다.
- Flight Policy Agent – 수하물 정책 및 취소 정책을 안내합니다.
Figure 3. 항공사 예약서비스 Flow Chart
이 네 가지 에이전트는 협력하여 고객의 복합적인 요청을 처리합니다.
예를 들어, 고객이 “예약을 변경하고 싶은데 수수료가 얼마나 드나요?”라고 문의하면, Booking Agent가 현재 예약 정보를 조회하고, Loyalty Agent가 고객의 회원 등급을 확인하며, Flight Policy Agent가 해당 항공편의 변경 정책을 기반으로 수수료 여부를 판단합니다.
각 에이전트는 서로 다른 데이터 소스(Customer DB, Loyalty DB, Flight Policy DB)와 MCP에 접근하며, 항공편 검색의 경우에는 Amadeus API를 통해 관련 정보를 조회합니다.
- [코드 예시] 항공권 검색 및 예약을 담당하는 Booking Agent의 코드입니다.
"""
Booking Agent
AgentCore Gateway 연동 및 항공편 검색/예약
"""
import os
import json
import time
import logging
import asyncio
import requests
from datetime import datetime, timedelta
from requests.auth import HTTPBasicAuth
from strands import Agent
from strands.models.bedrock import BedrockModel
from strands.tools.mcp.mcp_client import MCPClient
from mcp.client.streamable_http import streamablehttp_client
from bedrock_agentcore.runtime import BedrockAgentCoreApp
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("booking_agent")
app = BedrockAgentCoreApp()
# ============================================================================
# 1. Cognito OAuth2로 Agent JWT 발급
# ============================================================================
class AuthManager:
"""Cognito Client Credentials로 JWT 토큰 관리"""
def __init__(self, cognito_domain: str, client_id: str, client_secret: str, region: str):
self.cognito_domain = cognito_domain
self.client_id = client_id
self.client_secret = client_secret
self.region = region
self.access_token = None
self.token_expires_at = 0
self._refresh_token()
def _refresh_token(self):
"""Cognito에서 새 JWT 발급"""
token_url = f"https://{self.cognito_domain}.auth.{self.region}.amazoncognito.com/oauth2/token"
res = requests.post(
token_url,
data={"grant_type": "client_credentials"},
auth=HTTPBasicAuth(self.client_id, self.client_secret),
headers={"Content-Type": "application/x-www-form-urlencoded"},
timeout=10,
)
res.raise_for_status()
body = res.json()
self.access_token = body["access_token"]
expires_in = int(body.get("expires_in", 3600))
self.token_expires_at = time.time() + expires_in - 30 # 30초 여유
logger.info(f"✅ JWT acquired (expires_in={expires_in}s)")
def get_valid_token(self) -> str:
"""만료 임박 시 갱신 후 토큰 반환"""
if time.time() >= self.token_expires_at:
logger.info("🔁 Refreshing JWT")
self._refresh_token()
return self.access_token
# ============================================================================
# 2. AgentCore Gateway 연결 및 Tools 로드
# ============================================================================
class BookingAgent:
"""Booking Agent - Gateway 연동"""
def __init__(self, gateway_url: str, auth_manager: AuthManager):
self.gateway_url = gateway_url
self.auth_manager = auth_manager
self._init_mcp_client()
self._create_agent()
def _init_mcp_client(self):
"""Gateway에 인증해 MCP 클라이언트 초기화"""
logger.info(f"🔌 Connecting to Gateway: {self.gateway_url}")
def client_factory():
# 요청 시마다 최신 토큰 사용
token = self.auth_manager.get_valid_token()
return streamablehttp_client(
self.gateway_url,
headers={"Authorization": f"Bearer {token}"},
)
self.mcp_client = MCPClient(client_factory)
logger.info("✅ MCP client initialized")
def _create_agent(self):
"""Gateway에서 Tools 로드 후 Agent 생성"""
logger.info("🤖 Loading tools from Gateway...")
with self.mcp_client as client:
tools = client.list_tools_sync()
tool_names = [t.tool_name for t in tools]
logger.info(f"✅ Loaded {len(tools)} tools: {tool_names}")
self.agent = Agent(
name="booking_agent",
model=BedrockModel(
model_id="us.anthropic.claude-haiku-4-5-20251001-v1:0",
temperature=0.3,
max_tokens=4000,
),
system_prompt=self._get_prompt(),
tools=tools,
)
def _get_prompt(self) -> str:
"""Agent System Prompt"""
now = datetime.now()
date_context = (
f"현재 날짜: {now.strftime('%Y년 %m월 %d일')}\n"
f"오늘: {now.strftime('%Y-%m-%d')}\n"
f"내일: {(now + timedelta(days=1)).strftime('%Y-%m-%d')}"
)
return f"""당신은 항공편 예약 어시스턴트입니다.
{date_context}
[핵심 규칙]
1) 항공편 검색 시 currencyCode=KRW 기준으로 응답합니다.
2) 가격은 한화(KRW)로 표시합니다.
3) Korean Air(KE) 항공편만 추천합니다.
4) 최대 3개 옵션만 제공합니다.
[사용 가능한 Tool]
- 고객 정보 조회: get_customer_info(customer_id)
- 예약 내역 조회: get_customer_bookings(customer_id)
- 항공편 검색: search_flights(originLocationCode, destinationLocationCode, departureDate, adults)
- 예약 생성: create_booking(customer_id, flight_offer)
[응답 형식]
[옵션 1] KE703 09:55 ICN → 12:25 NRT ₩180,000
[옵션 2] KE711 13:25 ICN → 15:55 NRT ₩180,000
[옵션 3] KE713 16:55 ICN → 19:30 NRT ₩180,000"""
async def stream(self, query: str):
"""스트리밍 응답"""
try:
with self.mcp_client:
async for event in self.agent.stream_async(query):
yield event
except requests.HTTPError as e:
# 401 Unauthorized 시 토큰 갱신 후 재시도
if getattr(e.response, "status_code", None) == 401:
logger.warning("⚠️ 401 Unauthorized. Refreshing JWT and retrying...")
self.auth_manager._refresh_token()
with self.mcp_client:
async for event in self.agent.stream_async(query):
yield event
else:
raise
# ============================================================================
# 3. AgentCore Runtime Entrypoint
# ============================================================================
_booking_agent = None
_booking_agent_lock = asyncio.Lock()
@app.entrypoint
async def invoke(payload: dict):
"""AgentCore Runtime 진입점"""
global _booking_agent
logger.info(f"🚀 Runtime invoked: {json.dumps(payload or {}, ensure_ascii=False)}")
# Agent 싱글톤 초기화 (첫 요청 시만)
if _booking_agent is None:
async with _booking_agent_lock:
if _booking_agent is None:
auth_manager = AuthManager(
cognito_domain=os.environ["COGNITO_DOMAIN"],
client_id=os.environ["BOOKING_CLIENT_ID"],
client_secret=os.environ["BOOKING_CLIENT_SECRET"],
region=os.getenv("AWS_REGION", "us-east-2"),
)
_booking_agent = BookingAgent(
gateway_url=os.environ["MCP_GATEWAY_URL"],
auth_manager=auth_manager,
)
query = (payload or {}).get("prompt", "")
if not query:
yield {"type": "error", "error": "No prompt provided"}
return
# 스트리밍 응답
async for event in _booking_agent.stream(query):
yield event
if __name__ == "__main__":
app.run()
- [코드 예시] 각각의 에이전트로 라우팅 역할을 하는 Orchestrator Agent 코드입니다.
"""
Multi-Agent Orchestrator
AI 기반 의도 분석 및 Agent 라우팅
"""
import os
import json
import boto3
from typing import Any, Dict, Optional
from strands import Agent, tool
from strands.models.bedrock import BedrockModel
from bedrock_agentcore.runtime import BedrockAgentCoreApp
app = BedrockAgentCoreApp()
# ============================================================================
# 1. AgentCore Runtime에 배포된 Agent 호출
# ============================================================================
def invoke_agent_runtime(
agent_name: str,
user_message: str,
session_id: Optional[str],
) -> Dict[str, Any]:
"""
AgentCore Runtime에 배포된 Agent를 boto3로 호출
"""
agent_arn = os.getenv(f"{agent_name.upper()}_ARN")
if not agent_arn:
return {"ok": False, "error": f"{agent_name} ARN not found"}
client = boto3.client("bedrock-agentcore", region_name="us-east-2")
try:
invoke_params = {
"agentRuntimeArn": agent_arn,
"qualifier": "DEFAULT",
"payload": json.dumps({"prompt": user_message}, ensure_ascii=False),
}
if session_id:
invoke_params["runtimeSessionId"] = session_id
resp = client.invoke_agent_runtime(**invoke_params)
# SSE(text/event-stream) 또는 JSON 응답 처리
content_type = resp.get("contentType", "")
body = resp.get("response")
if "text/event-stream" in content_type and body:
chunks = []
for raw in body.iter_lines(chunk_size=1024):
line = raw.decode("utf-8", errors="ignore").strip()
if line.startswith("data:"):
chunks.append(line[5:].strip())
return {"ok": True, "result": "".join(chunks)}
return {"ok": True, "result": ""}
except Exception as e:
return {"ok": False, "error": str(e)}
# ============================================================================
# 2. Orchestrator Entrypoint
# ============================================================================
@app.entrypoint
async def invoke(payload: dict, session_id: str = None):
"""
Orchestrator 진입점
- 사용자 쿼리를 받아 적절한 Agent로 라우팅
- session_id로 대화 컨텍스트 유지
"""
query = (payload or {}).get("prompt", "")
if not query:
yield {"type": "error", "error": "No prompt provided"}
return
# ========================================================================
# 3. Tool 정의 (session_id는 클로저로 캡쳐)
# ========================================================================
@tool
def call_booking_agent(customer_id: str, task: str):
"""항공편 검색/예약/고객정보 조회"""
msg = f"고객 {customer_id} 관련 요청: {task}"
res = invoke_agent_runtime("booking_agent", msg, session_id)
if not res["ok"]:
raise RuntimeError(res["error"])
return res["result"]
@tool
def call_loyalty_agent(customer_id: str, task: str):
"""마일리지/등급/포인트 조회"""
msg = f"고객 {customer_id} 관련 요청: {task}"
res = invoke_agent_runtime("loyalty_agent", msg, session_id)
if not res["ok"]:
raise RuntimeError(res["error"])
return res["result"]
@tool
def call_policy_agent(task: str):
"""정책/수하물/취소/환불 규정"""
res = invoke_agent_runtime("policy_agent", task, session_id)
if not res["ok"]:
raise RuntimeError(res["error"])
return res["result"]
# ========================================================================
# 4. Orchestrator Agent 생성
# ========================================================================
orchestrator = Agent(
model=BedrockModel(
model_id="us.anthropic.claude-haiku-4-5-20251001-v1:0",
temperature=0.3,
max_tokens=8000,
),
system_prompt="""You are the orchestrator for AnyAirline customer service.
You coordinate between specialized agents and MUST use tools.
[Tool 사용 규칙]
- 정책(수하물/취소/변경/환불) → call_policy_agent
- 마일리지/등급/포인트 → call_loyalty_agent(customer_id, task)
- 예약/항공편/고객정보 → call_booking_agent(customer_id, task)
절대 추측으로 답하지 말고, 반드시 적절한 Tool을 호출한 뒤 그 결과를 기반으로 응답하세요.
Always provide clear, helpful responses in Korean.""",
tools=[call_booking_agent, call_loyalty_agent, call_policy_agent],
)
# ========================================================================
# 5. Streaming 응답
# ========================================================================
try:
async for event in orchestrator.stream_async(query):
if "data" in event:
yield {"type": "text_chunk", "data": event["data"]}
if "result" in event:
yield {"type": "streaming_complete", "result": str(event["result"])}
except Exception as e:
yield {"type": "error", "error": str(e)}
if __name__ == "__main__":
app.run()
3.1 모든 도구 호출의 단일 진입점을 위한 AgentCore Gateway
각 에이전트가 외부 API나 MCP에 직접 연결하는 구조에서는 API endpoint나 tool을 추가할때 마다 기하급수적으로 늘어납니다. 하지만 AgentCore Gateway는 모든 tool 호출을 단일 진입점으로 수렴시킵니다. 
Figure 4. AgentCore Gateway와 연결된 Targets
AgentCore Gateway에서는 JSON 형태로 MCP 도구를 정의하고 Target으로 등록함으로써, 에이전트와 도구 간의 직접적인 연결을 제거하고 아키텍처를 보다 간결하게 구성할 수 있습니다. 이제 모든 에이전트는 하나의 AgentCore Gateway만 인지하면 됩니다.
- 인증 로직 중앙화: Tool의 인증은 AgentCore identity와 함께 AgentCore Gateway에서만 관리할 수 있습니다.
- API 변경 영향 최소화: API 엔드포인트 변경 시 AgentCore Gateway에서 수정할 수 있습니다.
- Tool 확장의 용이성: 새로운 API나 MCP를 추가할 때, Gateway에만 등록함으로써 손쉽게 확장할 수 있습니다.
3.2 통합 인증 거버넌스를 위한 AgentCore Identity
에이전트마다 서로 다른 인증방식을 직접 구현하기 시작하면, 보안 설정이 코드에 분산되고 토큰 관리가 복잡해집니다. AgentCore Identity는 IdP(Okta, Microsoft Entra ID, Amazon Cognito 등)와 통합되며, OAuth2/OIDC 흐름을 지원하여 모든 인증을 중앙에서 관리합니다. 이 글에서는 Amazon Cognito를 활용하여 인증을 구성하였습니다.
Figure 5. AgentCore Identity 인증 관계도
- 인바운드 인증 (Inbound Authentication)
인바운드 인증에서는 각 에이전트가 Amazon Cognito를 통해 JWT를 발급받습니다. 에이전트는 이 JWT를 포함해 Gateway에 요청을 전달하며, Gateway는 토큰을 검증하여 요청 주체가 어떤 에이전트인지 식별합니다.
- 아웃바운드 인증 (Outbound Authentication)
아웃바운드 인증에서는 Gateway가 실제 도구(API, MCP)를 호출할 때 대상에 맞는 인증 방식을 자동으로 처리합니다.
-
- AWS 내부 리소스 접근 시에는 AWS IAM Role 기반 인증을 사용합니다.
- 외부 API 서비스(예: Amadeus API)의 경우, OAuth 토큰 발급 및 갱신을 AgentCore Identity가 자동으로 수행합니다.
이러한 구조를 통해 개발자는 토큰 저장이나 갱신 로직을 직접 구현할 필요 없이 에이전트의 비즈니스 로직에만 집중할 수 있습니다. 인증과 권한 관리는 AgentCore Identity와 Gateway가 플랫폼 레벨에서 일관되게 처리합니다.
3.3 Agent 세밀한 권한 제어를 위한 AgentCore Gateway Interceptor
기본적으로 모든 에이전트가 모든 도구에 접근할 수 있다면, 이는 보안 위험과 동시에 에이전트의 역할 분리(Separation of Concerns)를 위반합니다. Amazon Bedrock AgentCore의 Gateway Request Interceptor는 MCP 요청이 실제 도구에 도달하기 전에 가로채서 권한을 검사할 수 있는 기능입니다.
Figure 6. AgentCore Gateway Interceptor 중심의 서비스 관계도
- 인증: 각 에이전트는 Cognito에서 고유한 JWT 토큰을 발급받습니다.
- 요청: 에이전트가 MCP 도구 호출 시 JWT 토큰을 Authorization 헤더에 포함합니다.
- 가로채기: Gateway의 Request Interceptor가 요청을 가로챕니다.
- 식별: Interceptor Lambda가 JWT에서
client_id를 추출하여 에이전트를 식별합니다. - 권한 검사: 에이전트의 권한 정책을 확인하고 도구 호출을 허용하거나 거부합니다.
- [코드 예시] 아래는 Lambda interceptor 예제로, 에이전트별 권한을 검증하는 핵심 로직입니다.
import json
import base64
import logging
import os
logger = logging.getLogger()
logger.setLevel(logging.INFO)
# ============================================================================
# 1. Agent별 허용 Tool 정의
# ============================================================================
AGENT_PERMISSIONS = {
"booking": {
"allowed_tools": [
"customer-mcp___get_customer_info",
"customer-mcp___create_booking",
"customer-mcp___get_customer_bookings"
]
},
"loyalty": {
"allowed_tools": [
"loyalty-mcp___get_member_loyalty",
"loyalty-mcp___get_points_balance",
"loyalty-mcp___get_tier_benefits"
]
},
"policy": {
"allowed_tools": [
"policy-mcp___get_baggage_policy",
"policy-mcp___get_cancellation_policy",
"policy-mcp___get_change_policy"
]
},
"orchestrator": {
"allowed_tools": ["*"] # 모든 Tool 허용
}
}
# Client ID → Agent Type 매핑
CLIENT_ID_TO_AGENT = {
os.environ.get("BOOKING_CLIENT_ID", ""): "booking",
os.environ.get("LOYALTY_CLIENT_ID", ""): "loyalty",
os.environ.get("POLICY_CLIENT_ID", ""): "policy",
os.environ.get("ORCHESTRATOR_CLIENT_ID", ""): "orchestrator",
}
# ============================================================================
# 2. JWT에서 Agent 타입 추출
# ============================================================================
def decode_jwt_payload(token: str) -> dict:
"""JWT 토큰에서 payload 추출"""
try:
if token.startswith("Bearer "):
token = token[7:]
# JWT: header.payload.signature
parts = token.split(".")
if len(parts) != 3:
return {}
# payload 디코딩 (base64url)
payload = parts[1]
padding = 4 - len(payload) % 4
if padding != 4:
payload += "=" * padding
decoded = base64.urlsafe_b64decode(payload)
return json.loads(decoded)
except Exception as e:
logger.error(f"JWT decode error: {e}")
return {}
def get_agent_type_from_jwt(headers: dict) -> str:
"""JWT에서 Agent 타입 추출"""
auth_header = headers.get("Authorization", headers.get("authorization", ""))
if not auth_header:
logger.warning("No Authorization header")
return "unknown"
payload = decode_jwt_payload(auth_header)
client_id = payload.get("client_id", "")
agent_type = CLIENT_ID_TO_AGENT.get(client_id, "unknown")
logger.info(f"JWT client_id: {client_id} → Agent: {agent_type}")
return agent_type
# ============================================================================
# 3. Tool 접근 권한 검사
# ============================================================================
def is_tool_allowed(agent_type: str, tool_name: str) -> bool:
"""Agent가 해당 Tool을 호출할 수 있는지 확인"""
if agent_type not in AGENT_PERMISSIONS:
return False
allowed_tools = AGENT_PERMISSIONS[agent_type]["allowed_tools"]
# "*"는 모든 Tool 허용
if "*" in allowed_tools:
return True
return tool_name in allowed_tools
# ============================================================================
# 4. Lambda Handler
# ============================================================================
def lambda_handler(event, context):
"""
Gateway Request Interceptor
1. JWT에서 Agent 타입 식별
2. tools/call 요청 시 권한 검사
3. 허용되지 않은 Tool 호출 시 403 반환
"""
logger.info(f"Interceptor event: {json.dumps(event, default=str)}")
try:
mcp_data = event.get("mcp", {})
gateway_request = mcp_data.get("gatewayRequest", {})
headers = gateway_request.get("headers", {})
body = gateway_request.get("body", {})
# MCP 메서드 확인
mcp_method = body.get("method", "")
logger.info(f"MCP method: {mcp_method}")
# ====================================================================
# tools/call 요청만 권한 검사
# ====================================================================
if mcp_method == "tools/call":
# Agent 타입 확인
agent_type = get_agent_type_from_jwt(headers)
if agent_type == "unknown":
logger.warning("Unknown agent type - denying access")
return {
"interceptorOutputVersion": "1.0",
"mcp": {
"transformedGatewayResponse": {
"statusCode": 401,
"body": {
"jsonrpc": "2.0",
"error": {
"code": -32600,
"message": "Unauthorized: Unknown agent identity"
}
}
}
}
}
# 호출하려는 Tool 확인
params = body.get("params", {})
tool_name = params.get("name", "")
logger.info(f"Agent '{agent_type}' attempting to call tool '{tool_name}'")
# 권한 검사
if not is_tool_allowed(agent_type, tool_name):
logger.warning(f"Access denied: {agent_type} → {tool_name}")
allowed = AGENT_PERMISSIONS.get(agent_type, {}).get("allowed_tools", [])
return {
"interceptorOutputVersion": "1.0",
"mcp": {
"transformedGatewayResponse": {
"statusCode": 403,
"body": {
"jsonrpc": "2.0",
"error": {
"code": -32600,
"message": f"Access Denied: {agent_type} cannot access '{tool_name}'. Allowed: {allowed}"
}
}
}
}
}
logger.info(f"✅ Access granted: {agent_type} → {tool_name}")
# tools/list, initialize 등은 통과
elif mcp_method in ["tools/list", "initialize", "ping"]:
logger.info(f"Passing through {mcp_method} request")
# 요청 통과
return {
"interceptorOutputVersion": "1.0",
"mcp": {
"transformedGatewayRequest": {
"body": body
}
}
}
except Exception as e:
logger.error(f"Interceptor error: {e}", exc_info=True)
return {
"interceptorOutputVersion": "1.0",
"mcp": {
"transformedGatewayResponse": {
"statusCode": 500,
"body": {
"jsonrpc": "2.0",
"error": {
"code": -32600,
"message": f"Internal error: {str(e)}"
}
}
}
}
}
3.4 일관된 실행· 운영 환경 제공을 위한 Amazon Bedrock AgentCore Runtime
에이전트가 여러 환경(ECS, EKS, EC2, Lambda)에서 실행할 경우 로그가 분산됩니다. AgentCore Runtime은 agentcore configure + agentcore launch 두 명령으로 복잡한 인프라 설정 없이 에이전트를 즉시 프로덕션에 배포할 수 있습니다. 또한 이 모든 에이전트를 일원화된 환경에서 실행하고 CloudWatch 대시보드를 구성하여 한눈에 확인하실 수 있습니다.
Figure 7. 실행환경 모니터 CloudWatch 대시보드 예시
- 통합 로깅: 모든 에이전트 로그가 한 곳에 모입니다.
- 분산 추적: 멀티 에이전트 호출 흐름을 한 눈에 파악할 수 있습니다.
- 통합 모니터링: 에이전트별 성능, 에러율 등을 대시보드에서 확인하실 수 있습니다.
4. Sequence Diagram 및 Demo
4.1. 항공권 검색 Sequence Diagram
다음 Sequence Diagram은 사용자가 항공권 검색을 했을때 요청이 여러 에이전트를 거쳐 어떻게 처리되는지를 단계별로 보여줍니다.
Figure 8. 항공권 검색 Sequence Diagram
- 사용자 요청 입력 : 사용자가 프론트엔드에서 ‘항공편 검색 요청’을 입력합니다.
- 프론트엔드 → Orchestrator Agent 호출 및 의도분석: 입력된 요청이 Orchestrator Agent로 전달되어, 항공편 검색 의도로 분류하고 필요한 파라미터를 추출합니다.
- Booking Agent 선택: Orchestrator가 항공편 검색을 담당하는 Booking Agent를 호출합니다.
- AgentCore Gateway 요청: Booking Agent가 JWT 토큰을 포함해 AgentCore Gateway로 MCP 요청을 전송합니다.
- 인증 및 외부 API 호출: AgentCore Gateway가 에이전트 권한을 검증하고 Amadeus API를 호출 합니다.
- 검색 결과 수신: Amadeus API가 항공편 검색 결과를 반환합니다.
- 결과 반환: AgentCore Gateway가 결과를 Booking Agent에 전달합니다.
- Orchestrator Agent로 전달: Booking Agent가 검색 결과를 Orchestrator Agent에 전달합니다.
- 프론트엔드 응답: Orchestrator Agent가 최종 결과를 프론트엔드로 반환합니다.
- 결과 표시: 사용자가 프론트엔드에서 항공편 목록을 확인합니다.
4.2. 데모영상
아래는 사용자의 요청에 따라 여러 에이전트와 통신하여 다양한 시나리오 (항공권 예약, 마일리지 조회, 항공권 정책 문의등)을 처리하는 흐름을 보여줍니다.
- 항공권 검색 데모 영상 – https://youtu.be/S66TMAca8c4
5. 마무리하며
엔터프라이즈 환경에서 에이전트의 가치는 모델 성능이나 프롬프트의 설계를 넘어서, 운영 단계에서 발생하는 복잡성을 어떻게 관리하느냐에 의해 결정됩니다. 이를 해결하기 위해서 인증, 배포, 확장을 일관된 방식으로 중앙화된 운영 모델을 먼저 정립하는것이 필수적입니다.
Amazon Bedrock AgentCore를 활용해 구조와 체계를 갖춘다면, 다음과 같은 이점을 얻을 수 있습니다.
- 개발 생산성 향상: 개발자는 비즈니스 로직에만 집중함으로써 신규 에이전트나 Tool을 추가하는 속도가 빨라집니다.
- 거버넌스 구축 : 중앙화된 인증과 세밀한 권한 제어를 통해 에이전트 기반 시스템에서도 엔터프라이즈 수준의 보안 기준을 유지할 수 있습니다.
- 운영 효율성과 확장성: 멀티 에이전트 시스템을 통합 관측하고 관리함으로써, 서비스 규모가 확장되더라도 안정적인 운영이 가능합니다.
결국 Amazon Bedrock AgentCore는 이러한 운영의 복잡성을 코드가 아닌 플랫폼 레벨에서 해결함으로써, 안전하고 신뢰할 수 있게 AI 기반의 비즈니스 확장을 가속화할 수 있습니다.