Executor 패턴

Executor란?

A2A 프로토콜에서 실제 에이전트 로직을 실행하는 컴포넌트입니다. A2A 서버가 메시지를 수신하면 Executor에게 전달하고, Executor가 AgentCore Runtime을 호출합니다.

아키텍처

A2A Client (Collaborator)
    ↓ SendMessageRequest
A2A Server (ECS)
    ↓
Executor (ConnectivityTroubleshootingAgentCoreExecutor)
    ↓ HTTP/2 POST
AgentCore Runtime
    ↓
TroubleshootingAgent → Lambda 도구

ConnectivityTroubleshootingAgentCoreExecutor

class ConnectivityTroubleshootingAgentCoreExecutor(AgentExecutor):
    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):
        # HTTP/2 클라이언트 (확장된 타임아웃)
        self.http = httpx.Client(
            timeout=httpx.Timeout(connect=10, read=900, write=30, pool=10),
            http2=True,
            follow_redirects=True
        )
        self._token_cache = _OAuthTokenCache()

핵심 메서드

_bearer(): OAuth 토큰 관리

def _bearer(self):
    """토큰 캐시, 만료 30초 전 자동 갱신"""
    if self._token_cache.valid(time.time()):
        return self._token_cache.token

    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
    )
    self._token_cache.set(tok["access_token"], int(tok["expires_in"]), time.time())
    return tok["access_token"]

_build_agent_url(): 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}"

execute(): 전체 실행 흐름

async def execute(self, context, event_queue):
    updater = TaskUpdater(event_queue, context.task_id, context.context_id)
    await updater.submit()
    await updater.start_work()

    payload = {"prompt": context.get_user_input(), "actor_id": f"user-{context.context_id}"}
    headers = {
        "Authorization": f"Bearer {self._bearer()}",
        "X-Amzn-Bedrock-AgentCore-Runtime-Session-Id": self.agent_session_id,
    }

    content = self._invoke_json(payload, headers)
    parts = [Part(root=TextPart(text=content))]
    await updater.add_artifact(parts, name="result")
    await updater.complete()

태스크 상태 관리

상태	메서드	시점
submitted	updater.submit()	요청 접수
working	updater.start_work()	처리 시작
complete	updater.complete()	처리 완료
failed	예외 발생 시	에러

Executor 패턴의 장점

1. 관심사 분리: A2A 프로토콜 처리와 비즈니스 로직 분리

2. 토큰 캐싱: OAuth 토큰 재사용으로 인증 오버헤드 최소화

3. HTTP/2: 멀티플렉싱으로 네트워크 효율성 향상

4. 타임아웃 관리: 15분 타임아웃으로 장시간 분석 지원