단계 5: ALB 검증 및 테스트

이 단계에서는 두 Application Load Balancer (ALB)의 적절한 구성과 접근성을 확인하고 A2A 에이전트 엔드포인트를 테스트합니다. 이를 통해 Connectivity와 Performance 에이전트 모두 해당 로드 밸런서를 통해 운영 가능하고 접근 가능한지 확인합니다.

ALB DNS 이름 가져오기

workshop-module-3 디렉토리 내의 a2a-performance-agent 폴더로 이동합니다:

cd /workspace/workshop-module-3/a2a/a2a-performance-agent

ALB DNS 이름 가져오기:

CONNECTIVITY_ALB=$(aws elbv2 describe-load-balancers \
  --names a2a-connectivity-alb \
  --query 'LoadBalancers[0].DNSName' --output text)

PERFORMANCE_ALB=$(aws elbv2 describe-load-balancers \
  --names a2a-performance-alb \
  --query 'LoadBalancers[0].DNSName' --output text)

Health 엔드포인트 테스트

에이전트 상태가 healthy이고 각 에이전트가 특정 스킬과 연결되어 있는지 확인합니다:

printf "=== Connectivity Agent Health ===\n"
curl -s http://$CONNECTIVITY_ALB/health | jq '{status, service}'

printf "\n=== Performance Agent Health ===\n"
curl -s http://$PERFORMANCE_ALB/health | jq '{status, service}'

Agent Card 테스트

printf "\n=== Connectivity Agent Info ===\n"
curl -s http://$CONNECTIVITY_ALB/.well-known/agent-card.json \
  | jq '{name, description, skills: .skills[].name}'

printf "\n=== Performance Agent Info ===\n"
curl -s http://$PERFORMANCE_ALB/.well-known/agent-card.json \
  | jq '{name, description, skills: .skills[].name}'

A2A ALB 확인

양쪽 에이전트의 health 상태가 healthy이고, agent-card에서 에이전트 이름과 skills이 올바르게 표시되어야 합니다.

선택 사항: 콘솔 기반 검증: EC2 > Load Balancers로 이동하여 AWS 콘솔을 통해서도 ALB 구성을 확인할 수 있습니다.

코드 분석: A2A Server 및 Agent Card

A2A Server 아키텍처 (ECS Task)

각 A2A Server는 ECS Task로 실행되며, ALB 뒤에서 두 개의 핵심 엔드포인트를 제공합니다:

엔드포인트	용도	응답
/health	ALB 헬스체크	{"status": "healthy", "service": "..."}
/.well-known/agent-card.json	A2A Agent Card	에이전트 메타데이터
/ (POST)	A2A 메시지 수신	Task 결과

Agent Card 스키마

Agent Card는 A2A 프로토콜의 서비스 디스커버리 메커니즘입니다:

# Connectivity Agent Card (config.yaml)
agent_metadata:
  name: Connectivity_Troubleshooting_Agent
  description: "DNS resolution and connectivity analysis"
  version: "1.0.0"
  capabilities:
    push_notifications: true
    streaming: true

agent_skills:
- id: dns_connectivity_troubleshooting
  name: "DNS Resolution and Connectivity Troubleshooting"
  tags: [dns-resolution, connectivity-analysis, security-groups]
  examples:
  - "Check connectivity between app-frontend and app-backend on port 80"

# Performance Agent Card (config.yaml)
agent_metadata:
  name: Performance_Agent
  description: "AWS network performance analysis"
  version: "1.0.0"

agent_skills:
- id: analyze_network_performance
  name: "Network Performance Analysis"
  tags: [performance, vpc-flow-logs, traffic-mirroring, tcp-analysis]
  examples:
  - "Analyze network performance for application with retransmission issues"

AgentCoreExecutor 핵심 구조 (agent_executer.py)

A2A Server는 AgentExecutor 인터페이스를 구현하여 A2A 메시지를 AgentCore Runtime 호출로 변환합니다.

전체 구조:

# a2a/a2a-connectivity-agent/agent_executer.py
class ConnectivityTroubleshootingAgentCoreExecutor(AgentExecutor):
    """
    A2A executor that fronts a Bedrock AgentCore Runtime
    """

    def __init__(self, *, base_url, agent_arn, agent_session_id,
                 user_pool_id, client_id, client_secret, scope, discovery_url,
                 identity_provider, request_timeout_s: int = 900):
        self.agent_arn = agent_arn
        self.agent_session_id = agent_session_id

        # OAuth 토큰 캐시 (30초 전 갱신)
        self._token_cache = _OAuthTokenCache()

        # HTTP 클라이언트 (15분 타임아웃)
        self.http = httpx.Client(
            timeout=httpx.Timeout(
                connect=10.0,      # 연결 타임아웃
                read=900,          # 읽기 타임아웃 (15분)
                write=30.0,        # 쓰기 타임아웃
                pool=10.0          # 풀 타임아웃
            ),
            http2=True,            # HTTP/2 활성화
            follow_redirects=True
        )

agent_arn이란?: AgentCore Runtime의 **ARN (Amazon Resource Name)**으로, 각 에이전트 런타임의 고유 식별자입니다. config.yaml의 agent_card_info.agent_arn 필드에서 로드되며, 형식은 arn:aws:bedrock-agentcore:{region}:{account_id}:runtime/{runtime_name} 입니다. 아래 _build_agent_url()에서 이 ARN을 사용하여 AgentCore Runtime API 호출 URL을 구성합니다.

1. AgentCore Runtime URL 구성:

def _build_agent_url(self, qualifier: str = "DEFAULT") -> str:
    """Build the AgentCore invocation URL"""
    import urllib.parse

    # AgentCore 엔드포인트 (리전별)
    region = boto3.session.Session().region_name
    endpoint = f"https://bedrock-agentcore.{region}.amazonaws.com"

    # ARN을 URL-safe하게 인코딩
    escaped_agent_arn = urllib.parse.quote(self.agent_arn, safe='')

    # 최종 URL: /runtimes/{arn}/invocations?qualifier=DEFAULT
    return f"{endpoint}/runtimes/{escaped_agent_arn}/invocations?qualifier={qualifier}"

2. OAuth 토큰 관리 (캐싱):

def _bearer(self) -> str:
    """Get OAuth token with 30s pre-expiry refresh"""
    now_ts = time.time()

    # 캐시된 토큰이 유효하면 재사용
    if self._token_cache.valid(now_ts):
        return self._token_cache.token

    # 만료되었으면 새 토큰 발급 (Cognito)
    tok = get_token(
        user_pool_id=self.user_pool_id,
        client_id=self.client_id,
        client_secret=self.client_secret,
        scope_string=self.scope,
        discovery_url=self.discovery_url
    )

    # 캐시에 저장 (expires_in - 30초)
    self._token_cache.set(tok["access_token"], int(tok.get("expires_in", 1800)), now_ts)
    return tok["access_token"]

3. AgentCore Runtime HTTP 호출:

def _invoke_json(self, payload: Dict[str, Any], headers: Dict[str, str]) -> str:
    """POST request to AgentCore Runtime"""
    agent_url = self._build_agent_url()

    # HTTP POST 요청
    resp = self.http.post(agent_url, json=payload, headers=headers)
    resp.raise_for_status()

    # 응답 파싱 (JSON 또는 텍스트)
    response_text = resp.text.strip()

    try:
        response_data = resp.json()
        # JSON 응답에서 의미있는 필드 추출
        content = (
            response_data.get("response") or
            response_data.get("text") or
            response_data.get("output") or
            str(response_data)
        )
        return content
    except json.JSONDecodeError:
        # JSON이 아니면 텍스트로 처리
        return response_text

4. A2A 메시지 처리 (execute 메서드):

async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
    """A2A 메시지 → AgentCore Runtime 호출 → A2A 응답"""

    # TaskUpdater 생성
    updater = TaskUpdater(event_queue, context.task_id, context.context_id)
    await updater.submit()
    await updater.start_work()

    # A2A 메시지에서 사용자 입력 추출
    user_text = context.get_user_input()
    payload = {
        "prompt": user_text,
        "actor_id": f"user-{context.context_id}"
    }

    try:
        # OAuth 토큰 가져오기 (캐시 활용)
        bearer = self._bearer()

        # AgentCore Runtime 호출 헤더
        headers = {
            "Authorization": f"Bearer {bearer}",
            "Content-Type": "application/json",
            "X-Amzn-Bedrock-AgentCore-Runtime-Session-Id": self.agent_session_id,
            "X-Amzn-Trace-Id": f"a2a-task-{context.task_id}",
            "x-correlation-id": f"{context.task_id}:{context.context_id}",
        }

        # AgentCore Runtime 호출 (동기)
        content = self._invoke_json(payload, headers)

        # A2A 응답 생성
        parts = [Part(root=TextPart(text=content))]
        await updater.add_artifact(parts, name="connectivity_troubleshooting_result")
        await updater.complete()

    except httpx.HTTPStatusError as e:
        # HTTP 오류 처리
        status = e.response.status_code
        if 400 <= status < 500:
            raise ServerError(error=UnsupportedOperationError()) from e
        raise ServerError(error=InternalError()) from e

OAuth 토큰 캐시 메커니즘

class _OAuthTokenCache:
    """만료 30초 전 자동 갱신하는 토큰 캐시"""
    def valid(self, now_ts: float) -> bool:
        return self.token is not None and now_ts < (self.expire_epoch - 30)

    def set(self, token: str, expires_in: int, now_ts: float):
        self.expire_epoch = now_ts + max(1, int(expires_in))

A2A 통신 흐름

핵심 포인트: A2A Server는 프로토콜 브릿지 역할을 합니다. A2A 프로토콜의 SendMessage 요청을 받아 AgentCore Runtime의 HTTP API로 변환합니다. OAuth 토큰 캐시가 Cognito 호출을 최소화하고, 15분 타임아웃이 장시간 에이전트 실행을 지원합니다.

이 단계가 완료되었습니다. 다음 단계로 진행하세요.