Team B: A2A Executor 패턴

분석 대상

a2a-connectivity-agent/agent_executer.py — ConnectivityTroubleshootingAgentCoreExecutor

Executor 클래스 구조

class ConnectivityTroubleshootingAgentCoreExecutor(AgentExecutor):
    """A2A 프로토콜 → 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=900):
        self.agent_arn = agent_arn

        # HTTP/2 클라이언트 (15분 타임아웃)
        self.http = httpx.Client(
            timeout=httpx.Timeout(connect=10, read=900, write=30, pool=10),
            http2=True,
            follow_redirects=True
        )

        # OAuth 토큰 캐시
        self._token_cache = _OAuthTokenCache()

OAuth 토큰 캐시

class _OAuthTokenCache:
    """만료 30초 전 자동 갱신"""
    def valid(self, now_ts):
        return self.token and now_ts < (self.expire_epoch - 30)

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

# 사용:
def _bearer(self):
    if self._token_cache.valid(time.time()):
        return self._token_cache.token  # 캐시된 토큰 반환

    tok = get_token(user_pool_id, client_id, client_secret, scope, discovery_url)
    self._token_cache.set(tok["access_token"], int(tok["expires_in"]), time.time())
    return tok["access_token"]

AgentCore Runtime URL 구성

def _build_agent_url(self, qualifier="DEFAULT"):
    endpoint = f"https://bedrock-agentcore.{region}.amazonaws.com"
    escaped_arn = urllib.parse.quote(self.agent_arn, safe='')
    return f"{endpoint}/runtimes/{escaped_arn}/invocations?qualifier={qualifier}"

URL 예시:

https://bedrock-agentcore.us-east-1.amazonaws.com
    /runtimes/arn%3Aaws%3Abedrock-agentcore%3Aus-east-1%3A123456789012%3Aruntime%2Fagent-xxx
    /invocations?qualifier=DEFAULT

HTTP/2 스트리밍 호출

def _invoke_json(self, payload, headers):
    agent_url = self._build_agent_url()
    resp = self.http.post(agent_url, json=payload, headers=headers)
    resp.raise_for_status()

    # JSON 파싱 시도
    try:
        response_data = resp.json()
        return (response_data.get("response") or
                response_data.get("text") or
                response_data.get("output") or
                str(response_data))
    except json.JSONDecodeError:
        return resp.text  # 스트리밍 응답은 텍스트로 처리

execute: A2A 요청 처리 전체 흐름

async def execute(self, context: RequestContext, event_queue: EventQueue):
    updater = TaskUpdater(event_queue, context.task_id, context.context_id)

    # 1. 태스크 상태 관리
    await updater.submit()       # 접수
    await updater.start_work()   # 작업 시작

    # 2. 페이로드 구성
    user_text = context.get_user_input()
    payload = {
        "prompt": user_text,
        "actor_id": f"user-{context.context_id}"
    }

    # 3. OAuth 토큰 + 헤더 구성
    bearer = self._bearer()
    headers = {
        "Authorization": f"Bearer {bearer}",
        "X-Amzn-Bedrock-AgentCore-Runtime-Session-Id": self.agent_session_id,
        "x-correlation-id": f"{context.task_id}:{context.context_id}",
    }

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

    # 5. A2A artifact로 반환
    parts = [Part(root=TextPart(text=content))]
    await updater.add_artifact(parts, name="connectivity_troubleshooting_result")
    await updater.complete()     # 완료

전체 호출 흐름

A2A Client (send_message)
    ↓
A2A Server (ECS)
    ↓
ConnectivityTroubleshootingAgentCoreExecutor.execute()
    ↓
_bearer() → OAuth 토큰 획득/캐시
    ↓
_build_agent_url() → AgentCore URL 구성
    ↓
_invoke_json() → HTTP/2 POST to AgentCore
    ↓
AgentCore Runtime → TroubleshootingAgent
    ↓
Lambda 도구 호출 (dns-resolve, connectivity)
    ↓
응답 → TextPart artifact → A2A Response

토론 질문

1. request_timeout_s=900 (15분)은 왜 이렇게 긴가요?

2. OAuth 토큰 만료 30초 전 갱신 전략의 장단점은?

3. HTTP/2를 사용하는 이유는? HTTP/1.1과의 차이점은?

4. _invoke_json에서 JSON과 텍스트 응답을 모두 처리하는 이유는?