AgentCore 런타임에 A2A 서버 배포

Amazon Bedrock AgentCore AgentCore 런타임을 사용하면 AgentCore 런타임에서 에이전트 Agent-to-Agent(A2A) 서버를 배포하고 실행할 수 있습니다. 이 가이드에서는 첫 번째 A2A 서버를 생성, 테스트 및 배포하는 방법을 안내합니다.

이 섹션에서는 다음을 배웁니다.

A2A에 대한 자세한 내용은 A2A 프로토콜 계약을 참조하세요.

Amazon Bedrock AgentCore가 A2A를 지원하는 방법

Amazon Bedrock AgentCore의 A2A 프로토콜 지원을 통해 투명한 프록시 계층 역할을 하여 A2A 서버와 원활하게 통합할 수 있습니다. A2A에 대해 구성된 경우 Amazon Bedrock AgentCore는 컨테이너가 기본 A2A 서버 구성과 일치하는 루트 경로(0.0.0.0:9000/)의 포트9000에서 상태 비저장, 스트리밍 가능한 HTTP 서버를 실행할 것으로 예상합니다.

이 서비스는 프로토콜 투명성을 유지하면서 엔터프라이즈급 세션 격리를 제공합니다. InvokeAgentRuntime API의 JSON-RPC 페이로드는 수정 없이 A2A 컨테이너로 직접 전달됩니다. 이 아키텍처는 및 JSON-RPC 통신에서 에이전트 카드를 통한 기본 제공 에이전트 검색/.well-known/agent-card.json과 같은 표준 A2A 프로토콜 기능을 유지하면서 엔터프라이즈 인증(SigV4/OAuth 2.0) 및 확장성을 추가합니다.

다른 프로토콜과의 주요 차별화 요소는 포트(HTTP의 경우 9000 대 8080), 탑재 경로( / 대 /invocations ), 표준화된 에이전트 검색 메커니즘이므로 Amazon Bedrock AgentCore는 프로덕션 환경의 A2A 에이전트에 이상적인 배포 플랫폼입니다.

다른 프로토콜과의 주요 차이점:

자세한 내용은 https://a2a-protocol.org/ 단원을 참조하십시오.

AgentCore 런타임에서 A2A 사용

이 자습서에서는 A2A 서버를 생성, 테스트 및 배포합니다.

사전 조건

1단계: A2A 프로젝트 생성

이 예제에서는 Strands Agents를 사용하지만 AgentCore CLI는 LangChain/LangGraph 및 Google ADK를 사용하는 A2A 프로젝트도 지원합니다.

프로젝트 스캐폴드

다음 명령을 실행하고 메시지가 표시되면 프레임워크로 Strands를 선택합니다.

agentcore create --protocol A2A

CLI는 필요한 모든 종속성과 구성이 포함된 전체 프로젝트를 스캐폴드합니다. 생성된 에는 A2A 서버가 main.py 포함되어 있습니다.

from strands import Agent, tool
from strands.multiagent.a2a.executor import StrandsA2AExecutor
from bedrock_agentcore.runtime import serve_a2a
from model.load import load_model

@tool
def add_numbers(a: int, b: int) -> int:
    """Return the sum of two numbers."""
    return a + b

tools = [add_numbers]

agent = Agent(
    model=load_model(),
    system_prompt="You are a helpful assistant. Use tools when appropriate.",
    tools=tools,
)

if __name__ == "__main__":
    serve_a2a(StrandsA2AExecutor(agent))

코드 이해

이 에이전트를 사용자 지정하려면 add_numbers 도구를 자체 도구로 바꾸고 시스템 프롬프트를 업데이트합니다.

2단계: 로컬에서 A2A 서버 테스트

로컬 개발 환경에서 A2A 서버를 실행하고 테스트합니다.

A2A 서버 시작

AgentCore CLI를 사용하여 로컬에서 A2A 서버를 시작합니다.

agentcore dev

그러면 웹 브라우저에서 AgentCore 에이전트 검사기가 열립니다. 대신 터미널 기반 TUI를 사용하려면를 사용합니다agentcore dev --no-browser.

또는 서버를 직접 실행할 수 있습니다.

python main.py

서버가 포트에서 실행 중임을 나타내는 출력이 표시되어야 합니다9000.

에이전트 호출

curl -X POST http://localhost:9000/ \
-H "Content-Type: application/json" \
-d '{
  "jsonrpc": "2.0",
  "id": "req-001",
  "method": "message/send",
  "params": {
    "message": {
      "role": "user",
      "parts": [
        {
          "kind": "text",
          "text": "what is 101 * 11?"
        }
      ],
      "messageId": "12345678-1234-1234-1234-123456789012"
    }
  }
}' | jq .

에이전트 카드 검색 테스트

에이전트 카드 엔드포인트를 로컬에서 테스트할 수 있습니다.

curl http://localhost:9000/.well-known/agent-card.json | jq.

A2A 인스펙터를 사용한 원격 테스트에 설명된 대로 A2A 인스펙터를 사용하여 배포된 서버를 테스트할 수도 있습니다.

3단계: A2A 서버를 Bedrock AgentCore 런타임에 배포

인증을 위한 Cognito 사용자 풀 설정

배포하기 전에 배포된 서버에 대한 보안 액세스를 위한 인증을 구성합니다. 자세한 Cognito 설정 지침은 인증을 위한 Cognito 사용자 풀 설정을 참조하세요. 이렇게 하면 배포된 서버에 대한 보안 액세스에 필요한 OAuth 토큰이 제공됩니다.

에 배포 AWS

에이전트를 배포합니다.

agentcore deploy

이 명령은 다음을 수행합니다.

배포 후 다음과 같은 에이전트 런타임 ARN을 받게 됩니다.

arn:aws:bedrock-agentcore:us-west-2:accountId:runtime/my_a2a_server-xyz123

4단계: 에이전트 카드 가져오기

에이전트 카드는 A2A 서버의 자격 증명, 기능, 기술, 서비스 엔드포인트 및 인증 요구 사항을 설명하는 JSON 메타데이터 문서입니다. A2A 에코시스템에서 자동 에이전트 검색을 활성화합니다.

환경 변수 설정

환경 변수 설정

에이전트 카드 검색

import os
import json
import requests
from uuid import uuid4
from urllib.parse import quote

def fetch_agent_card():
    # Get environment variables
    agent_arn = os.environ.get('AGENT_ARN')
    bearer_token = os.environ.get('BEARER_TOKEN')

    if not agent_arn:
        print("Error: AGENT_ARN environment variable not set")
        return

    if not bearer_token:
        print("Error: BEARER_TOKEN environment variable not set")
        return

    # URL encode the agent ARN
    escaped_agent_arn = quote(agent_arn, safe='')

    # Construct the URL
    url = f"https://bedrock-agentcore.us-west-2.amazonaws.com/runtimes/{escaped_agent_arn}/invocations/.well-known/agent-card.json"

    # Generate a unique session ID
    session_id = str(uuid4())
    print(f"Generated session ID: {session_id}")

    # Set headers
    headers = {
        'Accept': '*/*',
        'Authorization': f'Bearer {bearer_token}',
        'X-Amzn-Bedrock-AgentCore-Runtime-Session-Id': session_id
    }

    try:
        # Make the request
        response = requests.get(url, headers=headers)
        response.raise_for_status()

        # Parse and pretty print JSON
        agent_card = response.json()
        print(json.dumps(agent_card, indent=2))

        return agent_card

    except requests.exceptions.RequestException as e:
        print(f"Error fetching agent card: {e}")
        return None

if __name__ == "__main__":
    fetch_agent_card()

에이전트 카드에서 URL을 가져온 후 환경 변수AGENTCORE_RUNTIME_URL로 내보냅니다.

export AGENTCORE_RUNTIME_URL="https://bedrock-agentcore.us-west-2.amazonaws.com/runtimes/<ARN>/invocations/"

5단계: 배포된 A2A 서버 호출

클라이언트 코드를 생성하여 배포된 Amazon Bedrock AgentCore A2A 서버를 호출하고 메시지를 전송하여 기능을 테스트합니다.

배포된 A2A 서버를 호출my_a2a_client_remote.py할 새 파일을 생성합니다.

import asyncio
import logging
import os
from uuid import uuid4

import httpx
from a2a.client import A2ACardResolver, ClientConfig, ClientFactory
from a2a.types import Message, Part, Role, TextPart

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

DEFAULT_TIMEOUT = 300  # set request timeout to 5 minutes

def create_message(*, role: Role = Role.user, text: str) -> Message:
    return Message(
        kind="message",
        role=role,
        parts=[Part(TextPart(kind="text", text=text))],
        message_id=uuid4().hex,
    )

async def send_sync_message(message: str):
    # Get runtime URL from environment variable
    runtime_url = os.environ.get('AGENTCORE_RUNTIME_URL')

    # Generate a unique session ID
    session_id = str(uuid4())
    print(f"Generated session ID: {session_id}")

    # Add authentication headers for Amazon Bedrock AgentCore
    headers = {"Authorization": f"Bearer {os.environ.get('BEARER_TOKEN')}",
        'X-Amzn-Bedrock-AgentCore-Runtime-Session-Id': session_id}

    async with httpx.AsyncClient(timeout=DEFAULT_TIMEOUT, headers=headers) as httpx_client:
        # Get agent card from the runtime URL
        resolver = A2ACardResolver(httpx_client=httpx_client, base_url=runtime_url)
        agent_card = await resolver.get_agent_card()

        # Agent card contains the correct URL (same as runtime_url in this case)
        # No manual override needed - this is the path-based mounting pattern

        # Create client using factory
        config = ClientConfig(
            httpx_client=httpx_client,
            streaming=False,  # Use non-streaming mode for sync response
        )
        factory = ClientFactory(config)
        client = factory.create(agent_card)

        # Create and send message
        msg = create_message(text=message)

        # With streaming=False, this will yield exactly one result
        async for event in client.send_message(msg):
            if isinstance(event, Message):
                logger.info(event.model_dump_json(exclude_none=True, indent=2))
                return event
            elif isinstance(event, tuple) and len(event) == 2:
                # (Task, UpdateEvent) tuple
                task, update_event = event
                logger.info(f"Task: {task.model_dump_json(exclude_none=True, indent=2)}")
                if update_event:
                    logger.info(f"Update: {update_event.model_dump_json(exclude_none=True, indent=2)}")
                return task
            else:
                # Fallback for other response types
                logger.info(f"Response: {str(event)}")
                return event

# Usage - Uses AGENTCORE_RUNTIME_URL environment variable
asyncio.run(send_sync_message("what is 101 * 11"))

부록

인증을 위한 Cognito 사용자 풀 설정

자세한 Cognito 설정 지침은 MCP 설명서의 인증을 위한 Cognito 사용자 풀 설정을 참조하세요.

A2A 검사기를 사용한 원격 테스트

https://github.com/a2aproject/a2a-inspector을(를) 참조하세요.

문제 해결

일반적인 A2A-specific 문제

발생할 수 있는 일반적인 문제는 다음과 같습니다.

예외 처리

오류 처리를 위한 A2A 사양: https://a2a-protocol.org/latest/specification/#81-standard-json-rpc-errors

A2A 서버는 HTTP 200 상태 코드와 함께 오류를 표준 JSON-RPC 오류 응답으로 반환합니다. 내부 런타임 오류는 프로토콜 준수를 유지하기 위해 JSON-RPC 내부 오류로 자동 변환됩니다.

이제 서비스는 표준화된 JSON-RPC 오류 코드와 함께 적절한 A2A-compliant 오류 응답을 제공합니다.