Files

T

Godopu d4e1246138 refactor: move gRPC_Based_Interface/ into docs/ and add new docs

2026-07-20 09:39:12 +09:00

88 KiB

Raw Permalink Blame History

에이전트에게 프로토콜이 필요한 이유

AI 에이전트 간의 소통이 단절되는 상황을 방지하기 위해 왜 에이전트들이 공통된 프로토콜을 사용해야 하는지 알아봅니다. A2A 프로토콜이 어떻게 상호작용을 표준화하여 다양한 플랫폼에서 개발된 에이전트들이 안전하고 효율적으로 협업할 수 있도록 지원하는지 살펴봅니다. 상호 운용성이 확보된 에이전트 시스템을 구축하기 위해 에이전트 검색, 요청 처리, 권한 부여, 에러 관리 등의 핵심 개념을 학습합니다.

다섯 개의 팀이 있고, 각 팀이 유용한 "로봇 동료"를 개발하고 있다고 가정해 보겠습니다. 한 팀은 Python으로, 다른 팀은 JavaScript로 개발하며, 세 번째 팀은 벤더 클라우드에서 실행합니다. 나머지 두 팀은 CrewAI나 LangGraph 같은 에이전트 프레임워크를 활용하여 실험적인 내부 도구를 만들고 있습니다. 이제 회사에서 이 에이전트들이 협력하여 고객 요청을 처리하도록 지시했습니다. 구체적으로는 여행 계획 수립, 재고 확인, 가격 책정, 그리고 인보이스 발행을 함께 처리해야 합니다. 공통된 규칙이 없다면, 에이전트 쌍마다 커스텀 데이터 구조, 일회성 토큰, 일관되지 않은 에러 처리, 임시방편식 코드처럼 자신들만의 사적인 소통 방식을 만들어내야 합니다. 이는 개발 속도를 늦추고 시스템을 불안정하게 만듭니다. 하나의 서비스가 업데이트되면 연결된 네 개의 링크가 끊어질 수 있으며, 어떤 문제가 발생했는지 추적하기가 어렵고 보안도 취약해집니다.

프로토콜은 이러한 혼란을 방지하는 플레이북입니다. 프로토콜은 에이전트가 자신을 소개하는 방식(기능), 작업을 요청하는 방식(요청), 진행 상황을 공유하는 방식(스트리밍), 안전하게 실패하는 방식(에러), 권한을 증명하는 방식(권한 부여)을 표준화합니다. 프로토콜이 준비되면 팀들은 시스템을 한 번만 구축하여 어디서나 재사용할 수 있으므로, 전체 프로세스가 투명하고 안전하며 예측 가능해집니다. 그리고 이것이 바로 A2A가 해결하고자 하는 핵심 과제입니다.

시작하기에 앞서, 간단하게 생각해 보겠습니다: A2A는 에이전트들이 예기치 못한 문제 없이 서로 도움을 요청하고 결과를 공유할 수 있도록 마련된 공통 규칙의 집합입니다.

A2A란 무엇인가요?

Agent2Agent (A2A) 프로토콜은 본래 구글에서 개발된 오픈 표준으로, 서로 다른 팀이나 프레임워크로 구축된 AI 에이전트들이 안전하고 효율적으로 서로를 검색하고 소통하며 협업할 수 있도록 지원합니다. 이 프로토콜은 공통 언어와 상호작용 모델을 제공하여 에이전트들이 서로의 비공개 메모리, 도구, 또는 내부 정보에 직접 접근하지 않고도 구조화된 데이터, 텍스트, 파일을 교환하며 정보를 공유하고 작업을 관리하며 유저의 목표를 달성할 수 있도록 돕습니다.

A2A는 익숙한 웹 기술(HTTP, JSON-RPC, 그리고 서버 전송 이벤트를 통한 스트리밍)을 기반으로 설계되어 기존 엔터프라이즈 기술 스택에 매끄럽게 연동됩니다. 일반적인 WebSockets이나 gRPC와 달리, A2A는 구조화된 인텐트 스키마를 정의하므로 서로 다른 환경의 에이전트 간에 발견 가능성과 작업 협상이 가능합니다.

또한 엔터프라이즈 환경에 적합한 인증 및 인가 옵션(OAuth, API 키, mTLS)을 지원하며, 실시간 업데이트를 포함하여 단시간 요청부터 장시간 실행되는 작업까지 모두 처리할 수 있습니다. 텍스트뿐만 아니라 에이전트 간에 필요한 파일과 구조화된 데이터도 원활하게 전송할 수 있는 유연한 프로토콜입니다.

상위 개념에서 보면, A2A는 클라이언트 에이전트(요청자)와 원격 에이전트(수행자) 간의 간결한 요청-응답 루틴을 정립합니다. 원격 에이전트는 자신이 수행할 수 있는 작업과 요구 사항을 나열한 작은 JSON 형식의 "Agent Card"를 공개합니다. 클라이언트는 알맞은 카드를 선택하고, 메소드 이름, 입력값, 식별 정보가 포함된 태스크를 전송합니다. 이 태스크는 라이프사이클을 따라 처리되며, 진행 상황이 스트리밍되고 최종 아웃풋으로 산출물이 반환됩니다. 만약 처리 중 문제가 발생하면 원격 에이전트는 구조화된 에러를 전송하고, 모든 과정에서 토큰과 스코프를 활용해 권한이 철저히 검증됩니다.

이러한 용어들이 다소 낯설게 느껴지더라도 괜찮습니다. 학습을 진행하면서 하나씩 자세히 살펴볼 예정입니다. 지금은 핵심 개념에만 집중해 주시기 바랍니다. A2A는 에이전트들이 적합한 파트너를 찾고, 협업을 요청하며, 서로 안전하게 실시간 정보를 업데이트할 수 있는 명확한 단계를 정의합니다. 또한 요청이 모호할 때 에이전트는 임의로 판단하지 않고 누락된 정보를 다시 요청하거나 정중하게 통신을 거절하도록 설계되어 있습니다.

사전 요구 사항은 무엇인가요?

반드시 AI 전문가일 필요는 없으며, 기본적인 소프트웨어 개발 지식과 호기심만 있다면 충분합니다. A2A의 개념들을 파이썬 예제와 함께 단계별로 빌드해 나갈 예정이며, 여기서 배우는 모든 내용은 다른 언어와 프레임워크에도 동일하게 적용할 수 있습니다. 간단한 웹 서비스를 배포해 보았거나, JSON 데이터를 파싱해 보았거나, API를 호출해 본 경험이 있다면 학습을 시작하기에 충분합니다.

도움이 되는 경험:

Python 3(함수, 패키지, venv, uv, poetry, pip 등의 가상 환경 구성, 기본 타입 정의)에 대한 원활한 사용 능력.
HTTP 기본 지식(요청, 응답, 헤더, 상태 코드).
JSON 데이터에 대한 이해(페이로드 읽고 쓰기, 스키마, 에러 포맷).
커맨드라인 환경 사용 능력(스크립트 실행, 환경 변수 설정).
기초 보안 개념(토큰, API 키, 최소 권한 원칙).
로깅 및 디버깅 기본 지식(트레이스백 분석, print 문 또는 로그 출력)
이벤트 및 스트리밍에 대한 이해(SSE 또는 WebSockets), REST/RPC 패턴에 대한 직관적 이해.

코드 예제는 최대한 명료하게 작성되었으며 어려운 용어는 처음 등장할 때 자세히 설명하므로, 초보자도 쉽게 따라오실 수 있습니다.

A2A SDK는 어떻게 사용하나요?

A2A 파이썬 SDK를 설치하여 에이전트를 빌드하고 실행해 볼 수 있습니다. 현재 개발 환경에 맞춰 다음 명령어 중 하나를 입력해 주시기 바랍니다:

# 핵심 SDK
pip install uv
uv pip install a2a-sdk
# 또는
pip install a2a-sdk

선택적 추가 기능을 함께 빌드하려면 다음과 같이 익스텐션을 지정하여 설치할 수 있습니다:

uv add "a2a-sdk[all]"        # 모든 기능 포함
uv add "a2a-sdk[http-server]"  # HTTP 서버 기능
uv add "a2a-sdk[grpc]"         # gRPC 지원 기능
uv add "a2a-sdk[telemetry]"    # OpenTelemetry 트레이싱 지원
uv add "a2a-sdk[encryption]"   # 암호화 지원

설치가 완료되면 이제 첫 번째 에이전트를 정의하고 A2A 프로토콜을 활용해 통신을 시작할 준비가 되었습니다. 아래 터미널 환경에서 원하시는 명령어를 입력하며 테스트를 시작해 보시기 바랍니다.

이렇게 쉽고 간단하게 시스템에 A2A 환경을 구축하고 테스트를 진행할 수 있습니다.

A2A vs. 다른 프로토콜

AI 에이전트 통신에서 A2A와 MCP, ACP 같은 다른 프로토콜의 차이점을 탐구하여 각각의 역할을 이해합니다. A2A가 어떻게 에이전트 간의 협업과 조정에 초점을 맞춰 MCP의 도구 액세스 기능을 보완하는지, 그리고 왜 A2A가 크로스 플랫폼 에이전트 상호작용의 통합 표준으로 자리 잡아가고 있는지 알아봅니다.

A2A의 핵심 구성 요소와 실제 예시를 살펴보기 전에, 광범위한 프로토콜 환경에서 A2A가 어느 위치에 해당하는지 파악해 두면 도움이 됩니다. 각각의 프로토콜은 저마다의 문제를 해결하며, 완성도 높은 시스템들은 단일 접근 방식에 의존하기보다 여러 프로토콜을 결합하여 작동합니다.

A2A는 MCP(Model Context Protocol)와 함께 자주 언급되곤 합니다. 두 프로토콜은 이름은 비슷해 보이지만 서로 다른 레이어를 다룹니다. 크게 보면 오늘날의 프로토콜은 다음 두 가지 카테고리로 분류할 수 있습니다:

**컨텍스트 지향 프로토콜(Context-oriented protocols)**은 에이전트가 외부 도구, API, 데이터와 상호작용하는 방식을 규정합니다. MCP가 여기에 해당합니다. 에이전트나 모델이 외부 함수, 서비스, 정보원에 연결하여 새로운 능력을 얻을 수 있도록 지원합니다.
**에이전트 간 프로토콜(Inter-agent protocols)**은 에이전트들이 서로 소통하고, 구조화된 정보를 교환하며, 역할을 협상하고 협업하는 방식을 규정합니다. A2A가 여기에 해당합니다. 여러 에이전트가 안전하고 예측 가능한 방식으로 작업을 조정할 수 있도록 공통 언어를 제공합니다.

그림: 프로토콜 환경에서 A2A와 MCP의 위치

이제 각각에 대해 더 자세히 살펴보면서, 두 프로토콜이 양자택일의 관계가 아니라 서로 협력하도록 설계되었다는 점을 알아보겠습니다.

A2A는 MCP와 어떻게 다른가요?

MCP는 도구 액세스를 표준화합니다. 이를 작업장(workshop)에 비유할 수 있습니다. 에이전트는 도구의 이름, 파라미터, 아웃풋을 학습합니다. 데이터베이스에서 로우(row)를 가져오거나, 날씨 API를 호출하거나, 요약 함수를 실행하는 것처럼 명확하게 정의되고 결정론적인 태스크를 처리하는 데 탁월하며, 개발자가 모든 API를 수동으로 연결할 필요가 없도록 돕습니다.

만약 MCP에 대해 잘 모르신다면, 먼저 MCP 과정을 수강하시는 것을 권장합니다. 에이전트가 도구와 상호작용하는 방식을 이해하는 기초를 마련해 주므로, A2A를 이해하기가 훨씬 수월해집니다.

반면 A2A는 독립된 에이전트들이 서로 소통하고 협력하는 방식에 초점을 맞춥니다. MCP가 작업장이라면, A2A는 협의실(coordination room)과 같습니다. A2A는 여러 에이전트가 서로를 탐색하고, 역할 분담에 합의하며, 컨텍스트를 교환하고, 진행 상황을 스트리밍하며, 서로의 내부 정보를 들여다보지 않고도 안전하게 결과를 도출할 수 있도록 공통 언어를 제공합니다. 이는 전문화된 에이전트들 간의 위임, 협업, 그리고 복잡한 워크플로우 오케스트레이션에 최적화되어 있습니다.

그림: 에이전트 시스템에서 MCP와 A2A의 역할

두 프로토콜은 경쟁 관계가 아니라 상호 보완적인 관계입니다. MCP는 각 에이전트에게 안정적인 도구 액세스를 제공하고, A2A는 이 에이전트들이 안정적으로 협업할 수 있는 수단을 제공합니다. 실무에서 사용되는 대부분의 에이전트 시스템은 에이전트 내부에서는 MCP를 사용하고, 에이전트 간의 소통에는 A2A를 활용합니다.

그림: 에이전트 애플리케이션에서 A2A와 MCP가 함께 작동하는 방식

A2A는 다른 에이전트 간 프로토콜과 어떻게 다른가요?

ACP, Agora, 그리고 Agent Protocol을 포함하여 많은 그룹이 에이전트 간 표준을 개발하고 있습니다. 적용 방식은 다양합니다. 어떤 표준은 웹 기반의 크로스 벤더 상호 운용성을 목표로 하는 반면, 다른 표준은 로컬, 에지, 혹은 특정 도메인 전용 환경에 집중합니다. 현재 가장 두드러진 성과를 보이는 것은 ACP(IBM/BeeAI)와 A2A입니다.

A2A는 플랫폼과 벤더를 초월한 협업을 위해 웹 기반의 방식을 취합니다. HTTP, JSON-RPC, SSE 같은 친숙한 빌딩 블록을 사용하므로 기업의 기존 기술 스택에 부드럽게 연동되며, 프레임워크와 클라우드를 아우르는 분산형 멀티 팀 워크플로우를 지원합니다.

반면 ACP는 IoT, 로보틱스, 또는 개인정보 보호가 중요한 환경처럼 지연 시간이 짧고 대역폭이 제한된 환경에서의 로컬 및 에지 자율성을 위해 발전했습니다. REST, gRPC, ZeroMQ, 로컬 버스 등 다양한 전송 계층을 지원하며 오프라인 검색과 클라우드 독립적인 작동을 선호하고 자원 사용을 최적화합니다. 대역폭이 제한된 환경에서는 강력하지만, 범용적인 크로스 벤더 상호 운용성을 기본 지원하는 데는 덜 집중되어 있습니다.

그림: A2A 통신 메커니즘

반가운 소식은 이제 어떤 프로토콜을 선택해야 할지 고민하지 않아도 된다는 점입니다. ACP가 리눅스 재단(Linux Foundation) 산하의 A2A로 병합되었기 때문입니다. IBM은 A2A의 개발 방향에 기여하고 있으며, ACP 및 BeeAI 사용자를 위한 마이그레이션 경로도 제공됩니다. 핵심 요점은 간단합니다. 업계의 동력이 벤더 중립적인 단일 오픈 협업 레이어(A2A)로 수렴하고 있으며, 이는 도구 액세스를 위한 MCP와 자연스럽게 결합된다는 점입니다.

다음 단계는 무엇인가요?

A2A의 전반적인 맥락을 짚고 다른 주요 프로토콜들과 비교해 보았으니, 이제 직접 실습해 볼 차례입니다. 이어지는 레슨에서는 안전하고 구조화된 에이전트 협업을 가능하게 하는 A2A 프로토콜의 핵심 빌딩 블록들을 살펴보겠습니다.

에이전트 검색, 기능 명시, 요청 전달, 스트리밍, 에러 처리가 어떻게 유기적으로 맞물려 신뢰할 수 있는 에이전트 간 상호작용을 만들어내는지 설명하겠습니다. 마지막으로, 앞서 여러 번 언급했던 유명한 "Agent Card"를 살펴볼 예정입니다. 이 소형 JSON 문서는 다른 에이전트들에게 내가 수행할 수 있는 작업과 나에게 통신하는 방법을 명확하게 알려줍니다.

A2A의 핵심 구성 요소

Agent2Agent 프로토콜의 핵심 구성 요소를 탐구하여 에이전트들이 어떻게 기술을 소개하고, 메시지를 통해 소통하며, 태스크를 관리하고, 산출물을 만드는지 알아봅니다. 이번 레슨에서는 클라이언트-서버 모델, Agent Card를 통한 에이전트 탐색 방식, 그리고 이러한 요소들이 확장 가능하고 동적인 멀티 에이전트 워크플로우를 어떻게 구현하는지 설명합니다.

에이전트들이 협력할 때 모든 것은 구조에 달려 있습니다. 서로를 탐색하고, 통신하며, 작업을 조정하는 공통의 방식이 없다면, 아무리 뛰어난 에이전트라도 서로 의도가 어긋나거나 반대되는 작업을 수행할 수 있습니다. A2A는 에이전트를 누가 만들었는지, 에이전트가 어떤 프레임워크에서 실행되는지에 관계없이 모든 에이전트가 예측 가능한 방식으로 자신을 소개하고, 정보를 교환하고, 작업을 관리할 수 있도록 지원함으로써 이 문제를 해결합니다.

Agent Card, 통신 패턴, 그리고 태스크 및 메시지 모델이라는 세 가지 개념이 A2A 아키텍처의 중추를 형성합니다. 각각은 고유한 역할을 담당하며, 에이전트들이 그 위에서 작동할 수 있는 공통된 기초를 마련해 줍니다. 이러한 빌딩 블록들을 이해하고 나면, 간단한 일회성 호출부터 복잡한 멀티 에이전트 워크플로우에 이르기까지 모든 기능들이 유기적으로 이해되기 시작할 것입니다.

A2A의 핵심 아키텍처를 세분화하여 주요 구성 요소들을 살펴보겠습니다.

A2A는 어떻게 작동하나요?

진행하기에 앞서, 오해를 풀기 위해 A2A 상호작용에서 일어나는 일을 다시 한번 짚어보겠습니다. 아래에 나와 있는 것처럼, A2A에는 두 가지 핵심 역할이 존재합니다.

A2A 클라이언트(A2A client / 클라이언트 에이전트): 유저를 대신하여 요청을 시작하는 에이전트 또는 애플리케이션입니다. 이 "오케스트레이터(orchestrator)" 에이전트는 수행해야 할 작업을 결정하고 해당 작업을 처리하는 데 필요한 기술을 갖춘 다른 에이전트를 식별합니다. 또한 A2A 프로토콜을 사용해 태스크를 공식화하고 요청을 전송합니다.
A2A 서버(A2A server / 원격 에이전트): A2A 프로토콜을 구현하는 HTTP 엔드포인트를 노출하는 에이전트입니다(종종 웹 서비스 형태로 실행됨). 클라이언트로부터 들어오는 요청을 감시하고, 태스크를 처리한 뒤 결과나 상태 업데이트를 반환합니다. 클라이언트는 원격 에이전트의 내부 작동 방식을 알 필요 없이 노출된 기능만 알면 되므로, 원격 에이전트는 하나의 불투명한(opaque) 서비스로 취급할 수 있습니다. 덕분에 다른 에이전트를 블랙박스 형태의 "서비스"나 도구처럼 쉽게 다룰 수 있습니다.

이 아키텍처는 기존의 클라이언트-서버 모델과 유사합니다. 어떤 에이전트든 다른 에이전트(서버 역할)에게 도움을 요청하는 클라이언트 역할을 수행할 수 있습니다. 특히 하나의 에이전트가 두 가지 역할을 동시에 수행할 수도 있습니다. 예를 들어, 코디네이터 에이전트는 유저의 요청을 접수(서버 역할)한 뒤, 세부 태스크를 수행하기 위해 다른 에이전트들을 호출(클라이언트 역할)할 수 있습니다.

그림: A2A 상호작용 라이프사이클

UML 시퀀스 다이어그램에서 "alt" 박스는 내부에 묶인 상호작용들이 조건부이거나 선택적임을 나타냅니다. 즉, 상황에 따라 이 상호작용들이 발생할 수도 있고 발생하지 않을 수도 있습니다.

이해를 돕기 위해 ChatGPT나 Gemini, 혹은 "생각하는" 모델을 탑재한 웹 기반 챗봇을 사용할 때 일어나는 상황을 떠올려 보시기 바랍니다. 모델이 응답하기 전에 추론을 수행하는 동안, MCP 클라이언트-서버 네트워크와 유사하게 단순한 도구나 API뿐만 아니라 네트워크에 사용 가능한 에이전트들까지 스캔할 수 있다고 가정해 보겠습니다. 에이전트는 각 태스크를 가장 잘 수행할 수 있는 파트너 에이전트를 탐색하여 협업을 진행할 수 있습니다. A2A가 바로 이를 가능하게 합니다.

A2A는 노출된 모든 Agent Card를 찾아내어 어떤 에이전트가 작업에 가장 적합한지 식별하고 동적으로 작업을 위임할 수 있습니다. 예를 들어, 모델이 여행 일정을 생성해야 하는 경우 티켓 예매를 위해 항공사 에이전트를 호출하고, 숙소 예약을 위해 호텔 에이전트를 호출하며, 비자 정보를 얻기 위해 정책 에이전트를 호출할 수 있습니다. 이러한 에이전트들은 하드코딩된 통합이 아닌, A2A 탐색을 통해 동적으로 선택됩니다.

에이전트의 기술을 정의하고 A2A에서 탐색 가능하게 만드는 방법은 무엇인가요?

에이전트가 A2A 생태계에서 협업하려면 먼저 자신이 누구이고 무엇을 할 수 있는지 소개해야 합니다. 이는 Agent Card라는 단일 문서를 통해 수행됩니다. Agent Card에는 에이전트의 메타데이터(식별 정보, 엔드포인트, 기능)와 에이전트가 수행할 수 있는 작업들을 기술한 기술(skills) 목록이 포함되어 있습니다.

A2A에서 에이전트의 "기술(skills)"이란 무엇인가요?

**에이전트 기술(agent skill)**은 에이전트가 수행할 수 있는 구체적인 기능이나 행동을 나타냅니다. 잠재적 클라이언트들에게 "내가 이런 일을 잘한다"고 알려주는 역할을 합니다. 각 기술은 a2a.types 클래스의 AgentSkill 타입을 사용해 정의되며, 다음과 같은 핵심 속성들을 포함합니다:

id: 기술의 고유 식별자입니다.
name: 사람이 읽을 수 있는 간결한 이름입니다.
description: 기술이 수행하는 작업에 대한 명확한 설명입니다.
tags: 검색 및 카테고리 분류를 돕는 키워드들입니다.
examples: 사용법을 보여주는 예시 프롬프트나 태스크들입니다.
inputModes/outputModes: 지원하는 미디어 타입(예: "text/plain", "application/json")입니다.

아래는 "hello world" 에이전트의 간단한 예시입니다:

skill = AgentSkill(
    id='helloWorld',
    name='Returns hello world',
    description='Just returns hello world',
    tags=['hello world'],
    examples=['hi', 'hello world'],
 )

AgentSkill 예시

이 예시는 텍스트 기반으로 단순히 "hello world"를 반환하는 가장 기본적인 기술을 보여줍니다. 실제 에이전트들은 데이터 조회용 기술, 텍스트 요약용 기술, 컴플라이언스 규칙 검사용 기술 등 여러 개의 기술을 함께 정의하여 노출합니다.

A2A에서 "Agent Cards"란 무엇인가요?

Agent Card는 에이전트가 자신의 기술을 외부에서 탐색할 수 있도록 공개하는 문서입니다. 보통 .well-known/agent-card.json 경로에서 제공되는 소형 JSON 문서로, A2A 클라이언트는 이를 가져와서 에이전트가 어떤 작업을 하는지, 어디에 호스팅되어 있는지, 어떻게 통신해야 하는지 파악할 수 있습니다.

AgentCard(a2a.types 클래스에 정의됨)는 보통 다음 항목들을 포함합니다:

name, description, version: 기본적인 에이전트 식별 정보 및 버전 정보입니다.
url: 에이전트가 A2A 요청을 수신하는 엔드포인트 주소입니다.
capabilities: 스트리밍이나 푸시 알림 같이 지원하는 기능들입니다.
defaultInputModes/defaultOutputModes: 주로 사용되는 공통 미디어 포맷입니다.
skills: 에이전트가 외부에 노출하는 AgentSkill 객체들의 목록입니다.

아래는 간단한 예시입니다:

public_agent_card = AgentCard(
    name='Hello World Agent',
    description='Just a hello world agent',
    url='http://localhost:9999/',
    version='1.0.0',
    protocolVersion='0.3.0',
    defaultInputModes=['text/plain'],
    defaultOutputModes=['text/plain'],
    capabilities=AgentCapabilities(streaming=True),
    skills=[skill],
    supportsAuthenticatedExtendedCard=True,
  )

AgentCard 예시

이 카드는 에이전트가 localhost:9999에서 작동하고 텍스트 입력과 출력을 지원하며, "hello world"라는 단일 기술을 보유하고 있음을 보여줍니다. 클라이언트는 에이전트가 내부적으로 요청을 어떻게 처리하는지 알 필요가 없습니다. 오직 이 카드만으로 에이전트를 발견하고 안전하게 상호작용하는 데 필요한 모든 정보를 얻을 수 있습니다.

Agent Card는 A2A 탐색 기능의 토대입니다. 에이전트들이 협업할 대상을 찾을 때 비공개 디렉토리를 조회하거나 수동 설정에 의존하지 않고, 단지 상대방의 Agent Card를 가져와서 파싱합니다. 각 카드는 다른 에이전트가 해당 태스크를 도울 수 있는지, 어떤 포맷을 사용해야 하는지, 필요한 경우 어떻게 인증해야 하는지 판단할 수 있는 충분한 정보를 제공합니다.

A2A에서 에이전트들은 어떻게 통신하고 작업을 주고받나요?

에이전트들이 Agent Card를 통해 서로를 소개했다면, 실제로 대화를 나누고 데이터를 공유하며 진행 상황을 추적할 수단이 필요합니다. A2A는 이를 위해 다음 세 가지 핵심 구조를 사용합니다:

메시지(Messages)
태스크(Tasks)
산출물(Artifacts)

이 요소들은 의견이 전달되고, 작업이 수행되며, 결과가 반환되는 방식을 정의합니다. 하나씩 차례대로 살펴보겠습니다.

A2A에서 메시지란 무엇인가요?

**메시지(message)**는 대화에서 한 번의 주고받음(turn), 즉 하나의 요청이나 하나의 응답을 나타냅니다. 모든 메시지에는 다음과 같은 몇 가지 핵심 정보가 포함됩니다:

누가 말하고 있는지 나타내는 role ("user" 또는 "agent").
고유 식별자인 messageId.
실제 콘텐츠를 담고 있는 하나 이상의 Part 객체.

Part 객체는 A2A가 다양한 모달리티를 유연하게 수용할 수 있도록 지원합니다. 덕분에 에이전트들은 동일한 프로토콜 안에서 텍스트뿐만 아니라 파일, 이미지, 구조화된 데이터도 교환할 수 있습니다. Part 객체에는 크게 세 가지 유형이 있습니다:

TextPart: 일반 텍스트 콘텐츠용입니다.
FilePart: 파일 전송용으로, 인라인(Base64 인코딩)으로 전송하거나 URI 형식으로 전송할 수 있으며 filename 및 mimeType 같은 메타데이터를 포함합니다.
DataPart: 구조화된 JSON 데이터용으로, 파라미터, 메타데이터, 혹은 머신 러닝 결과를 전달하는 데 적합합니다.

이러한 계층 구조 덕분에 A2A는 사람이 읽을 수 있는 텍스트와 기계가 처리할 수 있는 데이터를 매끄럽게 결합하여 복합적인 대화를 지원합니다.

A2A에서 태스크란 무엇인가요?

모든 상호작용이 한 번의 메시지 전송으로 완료되는 것은 아닙니다. 때로는 에이전트가 복잡한 연산을 실행하거나, 외부 서비스를 호출하거나, 다른 에이전트들과 작업을 조율하기 위해 시간이 필요할 수 있습니다. 이때 "태스크(tasks)"가 사용됩니다.

클라이언트가 메시지를 보냈을 때 수신하는 에이전트는 아래의 두 가지 방식 중 하나로 응답할 수 있습니다.

상태 비저장(stateless) 메시지로 응답: 빠르고 일회성인 교환에 사용됩니다. 응답이 즉시 반환되며 상호작용이 종료됩니다.
상태 저장(stateful) 태스크 시작: 작업 시간이 길거나 복잡한 업무에 사용됩니다. 에이전트는 수명 주기와 taskId를 지닌 Task 객체로 응답합니다.

태스크는 다음과 같은 상태를 거치며 실행됩니다:

in-progress (진행 중).
input-required (입력 대기) 또는 auth-required (인증 대기).
종료 상태인 completed (완료), canceled (취소), 또는 failed (실패).

태스크가 실행되는 동안 에이전트는 진행 상황을 알리거나 부분적인 아웃풋을 실시간으로 스트리밍하여 업데이트 정보를 전송할 수 있습니다. 덕분에 클라이언트는 작업 진행 상황을 실시간으로 모니터링할 수 있으며, 필요한 경우 작업 중간에 새로운 입력을 제공할 수도 있습니다.

태스크는 작업이 완료될 때까지 대화를 열어두므로, 일종의 "오래 유지되는 메시지 스레드"로 생각할 수 있습니다.

A2A에서 산출물이란 무엇인가요?

마지막으로, 태스크가 완료되면 보통 최종 결과물이라는 실질적인 아웃풋이 도출됩니다. 이를 산출물이라고 합니다. **산출물(artifact)**은 태스크의 최종적인 구조화된 결과물입니다. 이는 다음과 같은 형태일 수 있습니다:

생성된 보고서 (FilePart 형태).
텍스트로 요약된 문서.
구조화된 JSON 데이터 묶음.

메시지와 마찬가지로 산출물도 다음 정보를 갖습니다:

고유 식별자인 artifactId.
사람이 읽을 수 있는 name.
콘텐츠를 담은 하나 이상의 Part 객체.

산출물 역시 실시간으로 스트리밍하여 전송할 수 있으므로, 용량이 크거나 지속적으로 생성되는 결과물(예: 긴 텍스트 응답이나 생성된 비디오 파일)을 파트별로 나누어 전달할 수 있습니다. 메시지가 에이전트의 대화 방식이고 태스크가 에이전트의 작업 방식이라면, 산출물은 에이전트가 생성하는 결과물, 즉 대화가 끝난 후에도 지속되는 결과물입니다.

다음 단계는 무엇인가요?

Agent Card, 메시지, 태스크, 그리고 산출물은 A2A 협업 모델의 뼈대를 구성합니다. 에이전트들이 이러한 요소를 사용해 어떻게 자신의 기술을 정의하고 소통하며 결과물을 완성해 나가는지 이해하고 나면, 코디네이션 패턴부터 스트리밍 아웃풋에 이르는 프로토콜의 다른 모든 세부 기능들도 자연스럽게 그 위에서 빌드될 것입니다.

이어지는 레슨에서는 이 모든 것을 결합하여 Agent Card가 어떻게 작동하는지 더 상세히 알아보고, 여러분만의 Agent Card를 어떻게 설계하고 공개하는지 살펴보겠습니다.

퀴즈: A2A 기초

에이전트 통신 및 협업의 필수 개념들을 다루는 퀴즈를 풀며 A2A 프로토콜에 대한 이해도를 평가합니다. 이번 평가를 통해 A2A 프레임워크 내 클라이언트 및 서버 기초에 대한 지식을 점검하고 다집니다.

지식을 다지고 핵심 개념을 올바르게 이해했는지 확인하기 위해, 잠시 시간을 내어 아래 퀴즈를 풀어보시기 바랍니다. A2A 프로토콜 기초, 클라이언트 개발 및 서버 기반 지식을 명확히 이해하는 데 도움이 될 것입니다. 화이팅입니다!

퀴즈

문제 1: (참 또는 거짓) A2A 프로토콜은 기본 통신 표준으로 JSON-RPC 2.0을 사용합니다.

A. 참 (True)
B. 거짓 (False)

참고: 이 퀴즈 페이지는 HTML에서 한 번에 한 문제씩 보여줍니다. 여기서는 첫 번째 문제만 제공됩니다. 전체 퀴즈 문제를 확인하려면 코스 플랫폼을 방문해 주세요.

A2A 준수 클라이언트 및 서버 구축

처음부터 끝까지 A2A 프로토콜을 준수하는 클라이언트와 서버를 설계하는 방법을 탐구합니다. FastAPI와 Pydantic 등의 Python 도구를 활용해 에이전트 탐색, JSON-RPC 통신, 구조화된 메시지 유효성 검사, 태스크 라이프사이클 관리를 구현하는 방법을 학습합니다. 이번 레슨은 상호 운용성이 확보된 AI 상호작용이 가능한 기초 A2A 에이전트를 구축하는 데 큰 도움이 될 것입니다.

A2A의 핵심 개념을 이해했으니 이제 실질적인 결과물을 빌드할 차례입니다. 처음부터 시작해 완전한 A2A 규격을 만족하는 시스템을 개발해 보겠습니다. 구체적으로는 HTTP를 통해 A2A 클라이언트로부터 메시지를 수신하고 수신한 메시지를 그대로 반환하는 에코 에이전트(echo agent)와, 이 에이전트를 탐색하고 통신하는 클라이언트를 함께 구현합니다.

가장 기초적인 단계에서 시작하여 단계별로 A2A 규격에 맞게 살을 붙여가며 코드를 완성해 보겠습니다. 최종적으로는 임의의 A2A 클라이언트가 탐색하여 사용할 수 있는 작동하는 에이전트와, A2A 규격을 준수하는 모든 에이전트와 소통할 수 있는 클라이언트를 확보하게 될 것입니다. 진행 과정에서 관련 라이브러리, 사용된 패턴, 설계 결정 사항을 하나씩 설명하여 어떤 코드를 왜 작성해야 하는지 명확히 짚고 넘어가겠습니다.

무엇을 개발하나요?

우리의 목표는 매우 명확합니다. 가장 단순한 형태의 A2A 상호작용을 구현하는 것입니다. 텍스트 메시지를 수신하여 "You said: [수신한 메시지]" 형태로 응답하는 에코 에이전트를 만들 것입니다. 이는 매우 단순해 보일 수 있지만, 다음과 같은 A2A의 모든 필수 패턴들을 내포하고 있습니다:

표준 엔드포인트를 통한 에이전트 탐색.
JSON-RPC 2.0 통신 프로토콜 준수.
태스크 수명 주기 관리(제출에서 완료까지).
올바른 파트(parts) 구조를 갖춘 메시지 교환.
잘못된 요청 형식에 대한 에러 처리.

이를 에이전트 간 통신의 "hello world"로 생각하셔도 좋습니다. 시스템을 온전히 이해하기 쉬우면서도 한층 더 복잡한 에이전트를 구축할 수 있는 기본 토대가 되어 줄 것입니다.

코드를 작성하기 전에 개발 도구가 갖추어져 있는지 확인해 주시기 바랍니다. 네트워크 처리 및 유효성 검사를 담당해 줄 파이썬 라이브러리들을 다음과 같이 가상 환경에 설치합니다:

# 깨끗한 작업 공간 생성
mkdir a2a-echo-example
cd a2a-echo-example

# 가상 환경 설정
python3 -m venv venv
source venv/bin/activate  # Windows 환경: venv\Scripts\activate

# 핵심 의존성 설치
pip install fastapi uvicorn pydantic requests

패키지 설치

각 라이브러리의 역할은 다음과 같습니다:

FastAPI: API를 쉽게 구축할 수 있도록 돕는 최신 Python 웹 프레임워크입니다. HTTP 라우팅, 요청 유효성 검사, 자동 API 문서 생성을 처리합니다. 입문하기 쉽고 타입 체크 기능이 훌륭하며 A2A의 JSON 기반 통신과 매끄럽게 연동되므로 채택했습니다.
Uvicorn: FastAPI 애플리케이션을 실행해 주는 ASGI 서버입니다. HTTP 요청을 수신하여 작성된 에이전트 코드로 라우팅해 주는 엔진 역할을 수행합니다. 가볍고 빠르며 개발 환경에 매우 적합합니다.
Pydantic: 데이터 유효성을 검증하는 라이브러리로, A2A 메시지가 정확한 데이터 구조를 갖추고 있는지 검증합니다. 사전에 에러를 잡아내고 이상이 있을 시 상세 피드백을 제공하므로 프로토콜 준수에 필수적입니다.
Requests: HTTP 통신을 수행하는 파이썬 표준 라이브러리입니다. 클라이언트가 에이전트를 탐색하고 메시지를 전송하는 데 사용합니다.

이 라이브러리들이 네트워킹, 데이터 검증, HTTP 통신에 따르는 복잡한 처리를 대신해 주므로 우리는 A2A 전용 비즈니스 로직에만 집중할 수 있습니다.

A2A 서버 구축 방법

우선 최소한의 코드로 웹 서버 기능을 생성한 다음 점진적으로 A2A 표준에 맞게 코드를 채워 나가겠습니다. echo_agent.py라는 이름의 파일에 서버 코드를 작성해 주시기 바랍니다.

"""
간단한 A2A 서버 예제 - 1단계: 기본 웹 서버
"""
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def hello():
    return {"message": "Hello from our future A2A agent!"}

if __name__ == "__main__":
    import uvicorn
    print("🚀 Starting basic web server on http://localhost:8000")
    uvicorn.run(app, host="0.0.0.0", port=8000)

FastAPI를 사용한 기본 웹 서버

위 코드의 설명은 다음과 같습니다:

4~6라인: 프레임워크를 불러오고 들어오는 HTTP 요청을 처리할 애플리케이션 인스턴스를 생성합니다.
8라인: / 경로로 GET 요청이 들어올 때 바로 아래 정의된 함수를 호출하도록 FastAPI에 등록합니다.
9~10라인: 딕셔너리를 반환하며, FastAPI가 이를 자동으로 JSON 포맷으로 변환해 줍니다.
12라인: 이 코드가 직접 실행될 때만 작동하고 다른 모듈에서 임포트될 때는 실행되지 않도록 보장합니다.
13~15라인: HTTP 서버를 시작하고 8000 포트에서 대기합니다.

터미널에서 python echo_agent.py 명령어로 서버를 시작한 뒤 http://localhost:8000 주소에 접속하면 다음과 같은 응답을 확인할 수 있습니다:

{"message": "Hello from our future A2A agent!"}

FastAPI는 들어오는 요청을 자동으로 파싱하고, 매핑된 함수를 호출하며, 파이썬 딕셔너리를 JSON 데이터로 변환하고, 올바른 헤더를 추가하여 클라이언트에 반환해 줍니다.

아직 A2A 규격이 완전히 적용되지는 않았지만, 훌륭한 기본 뼈대가 완성되었습니다. 이제 여기에 A2A 기능들을 차례대로 얹어 가겠습니다.

시스템을 A2A 준수 상태로 만드는 방법

개발 중인 에이전트를 외부에서 탐색할 수 있도록 A2A 표준 탐색 엔드포인트를 추가합니다:

# 신규: A2A 탐색용 엔드포인트 추가
@app.get("/.well-known/agent-card.json")
def agent_card():
    """A2A 에이전트 탐색 엔드포인트"""
    return {
        "name": "Echo Agent",
        "description": "A simple agent that echoes your messages back",
        "url": "http://localhost:8000",
        "version": "1.0.0",
        "protocolVersion": "0.3.0",
        "skills": [{
            "id": "echo",
            "name": "Echo Messages", 
            "description": "Repeats whatever you say",
            "examples": ["Hello", "How are you?"]
        }],
        "capabilities": {"streaming": False},
        "defaultInputModes": ["text/plain"],
        "defaultOutputModes": ["text/plain"]
    }

# 업데이트: 헬로 메시지 수정
@app.get("/")
def hello():
    return {"message": "Hello from our A2A agent!"}

A2A 탐색 엔드포인트

위 코드의 설명은 다음과 같습니다:

2~9라인: /.well-known/agent-card.json 경로에 표준 탐색 엔드포인트를 정의합니다.
11~26라인: 에이전트의 메타데이터와 성능 정보를 반환합니다.
13~17라인: 기본적인 식별 정보를 정의합니다.
18~22라인: 예시 구문이 포함된 기술 정의를 명시합니다.
23~25라인: 성능 및 데이터 형식을 명시합니다.

서버를 재시작하면 이제 다음 두 엔드포인트를 모두 테스트해 볼 수 있습니다:

http://localhost:8000/
http://localhost:8000/.well-known/agent-card.json

임의의 A2A 클라이언트는 이제 이 엔드포인트를 호출하여 우리의 에코 에이전트를 탐색할 수 있습니다.

메시지 구조 유효성 검사 추가 방법

이제 A2A 메시지에 대한 구조 유효성 검사 코드를 추가해 보겠습니다. 파일 상단 부근에 다음 임포트 문과 모델 정의들을 입력합니다:

from pydantic import BaseModel
from typing import List
import uuid
from datetime import datetime

class Part(BaseModel):
    kind: str
    text: str = None

class Message(BaseModel):
    kind: str = "message"
    messageId: str
    role: str
    parts: List[Part]

class MessageParams(BaseModel):
    message: Message

class JSONRPCRequest(BaseModel):
    jsonrpc: str = "2.0"
    id: str
    method: str
    params: MessageParams

메시지 유효성 검증 모델 추가

위 코드의 설명은 다음과 같습니다:

1~4라인: 데이터 유효성 검사와 고유 ID 생성을 위한 모듈들을 불러옵니다.
6~8라인: 메시지 조각들을 정의할 Part 모델을 선언합니다.
10~18라인: ID, 역할, 콘텐츠 목록을 포함하는 Message 모델을 선언합니다.
19~23라인: 메시지를 JSON-RPC 2.0 요청 봉투로 감싸서 구조화합니다.

그다음 아래의 POST 엔드포인트를 추가합니다:

@app.post("/")
def handle_message(request: JSONRPCRequest):
    """A2A 메시지 핸들러 - 곧 추가 예정!"""
    return {"jsonrpc": "2.0", "id": request.id, "result": "Handler coming soon!"}

POST 엔드포인트 추가

FastAPI가 들어오는 메시지의 규격을 백그라운드에서 자동으로 검증해 주기 시작합니다.

에이전트 로직 구현 방법

이제 POST 핸들러 내부에 에이전트가 실제로 메시지를 가공해 반환하는 비즈니스 로직을 구성합니다:

@app.post("/")
def handle_message(request: JSONRPCRequest):
    """A2A 메시지 핸들러"""
    # 메소드 유효성 검사
    if request.method != "message/send":
        return {
            "jsonrpc": "2.0", 
            "id": request.id, 
            "error": {
                "code": -32601, 
                "message": "Method not found"
            }
        }
    
    # 유저 메시지 추출
    user_message = request.params.message
    user_text = user_message.parts[0].text if user_message.parts else "No text"
    
    # 완료된 태스크로 A2A 응답 생성
    task = {
        "kind": "task",
        "id": str(uuid.uuid4()),
        "contextId": str(uuid.uuid4()),
        "status": {
            "state": "completed", 
            "timestamp": datetime.utcnow().isoformat() + "Z"
        },
        "history": [
            # 원본 유저 메시지 포함
            user_message.dict(),
            # 에이전트 응답 추가
            {
                "kind": "message",
                "messageId": str(uuid.uuid4()),
                "role": "agent",
                "parts": [{"kind": "text", "text": f"You said: '{user_text}'"}]
            }
        ]
    }
    
    return {"jsonrpc": "2.0", "id": request.id, "result": task}

에이전트용 POST 핸들러

위 코드의 설명은 다음과 같습니다:

5~13라인: 요청받은 메소드가 "message/send"가 맞는지 유효성을 검사하고, 다를 경우 규격에 맞는 JSON-RPC 에러 객체(-32601)를 반환합니다.
16~17라인: 요청 파라미터에서 유저 메시지를 추출하고, 텍스트가 있을 시 첫 번째 파트의 내용을 읽어옵니다.
20라인: ID값들과 상태 정보, 대화 내역이 들어간 A2A 태스크 객체를 최종 정의합니다.
21~24라인: UUID4 표준을 이용해 고유한 태스크 ID와 컨텍스트 ID를 발행합니다.
25라인: 태스크 상태를 "completed"로 마킹하고 ISO 8601 국제 규격 타임스탬프를 기재합니다.
28~38라인: 원본 유저 메시지와 에이전트의 답변으로 대화 히스토리를 구성합니다.
41라인: 태스크 객체를 JSON-RPC 2.0 결과 봉투에 감싸서 반환합니다.

이로써 A2A 규격을 완벽하게 준수하는 에코 에이전트가 완성되었습니다. A2A 에이전트는 일반적인 단순 텍스트를 응답으로 돌려주지 않고, 구조화된 '태스크(task)' 객체를 반환한다는 사실이 핵심입니다. 이 방식을 통해 소통 기록이 확실히 보존되고 향후 복잡한 상호작용도 구조적으로 처리할 수 있게 됩니다.

아래는 모든 단계를 누적하여 구현을 완료한 전체 echo_agent.py 코드입니다:

"""
간단한 A2A 서버 예제 - 전체 구현
"""
from fastapi import FastAPI
from pydantic import BaseModel
from typing import List
import uuid
from datetime import datetime

app = FastAPI()

# A2A 메시지 구조 모델
class Part(BaseModel):
    kind: str
    text: str = None

class Message(BaseModel):
    kind: str = "message"
    messageId: str
    role: str
    parts: List[Part]

class MessageParams(BaseModel):
    message: Message

class JSONRPCRequest(BaseModel):
    jsonrpc: str = "2.0"
    id: str
    method: str
    params: MessageParams

@app.get("/.well-known/agent-card.json")
def agent_card():
    """A2A 에이전트 탐색 엔드포인트"""
    return {
        "name": "Echo Agent",
        "description": "A simple agent that echoes your messages back",
        "url": "http://localhost:8000",
        "version": "1.0.0",
        "protocolVersion": "0.3.0",
        "skills": [{
            "id": "echo",
            "name": "Echo Messages",
            "description": "Repeats whatever you say",
            "examples": ["Hello", "How are you?"]
        }],
        "capabilities": {"streaming": False},
        "defaultInputModes": ["text/plain"],
        "defaultOutputModes": ["text/plain"]
    }

@app.get("/")
def hello():
    return {"message": "Hello from our A2A agent! Visit /.well-known/agent-card.json to see my capabilities."}

@app.post("/")
def handle_message(request: JSONRPCRequest):
    """A2A 메시지 핸들러"""
    # 메소드 유효성 검사
    if request.method != "message/send":
        return {
            "jsonrpc": "2.0", 
            "id": request.id, 
            "error": {
                "code": -32601, 
                "message": "Method not found"
            }
        }
    
    # 유저 메시지 추출
    user_message = request.params.message
    user_text = user_message.parts[0].text if user_message.parts else "No text"
    
    # 완료된 태스크로 A2A 응답 생성
    task = {
        "kind": "task",
        "id": str(uuid.uuid4()),
        "contextId": str(uuid.uuid4()),
        "status": {
            "state": "completed", 
            "timestamp": datetime.utcnow().isoformat() + "Z"
        },
        "history": [
            # 원본 유저 메시지 포함
            user_message.dict(),
            # 에이전트 응답 추가
            {
                "kind": "message",
                "messageId": str(uuid.uuid4()),
                "role": "agent",
                "parts": [{"kind": "text", "text": f"You said: '{user_text}'"}]
            }
        ]
    }
    
    return {"jsonrpc": "2.0", "id": request.id, "result": task}

if __name__ == "__main__":
    import uvicorn
    print("🤖 Starting Simple A2A Echo Agent on http://localhost:8000")
    uvicorn.run(app, host="0.0.0.0", port=8000)

에코 에이전트 전체 소스 코드

A2A 클라이언트 구축 방법

이제 개발한 에이전트를 탐색하여 메시지를 보낼 수 있는 클라이언트를 만들어보겠습니다. simple_a2a_client.py 파일에 다음 코드를 작성합니다:

import requests
import uuid
import json

def main():
    # 1단계: 에이전트 탐색
    base_url = "http://localhost:8000"
    print("🔍 Discovering A2A agent...")
    
    try:
        agent_card = requests.get(f"{base_url}/.well-known/agent-card.json").json()
        print(f"✅ Found: {agent_card['name']} - {agent_card['description']}")
    except requests.RequestException as e:
        print(f"❌ Discovery failed: {e}")
        return
    
    # 2단계: A2A JSON-RPC를 사용해 메시지 전송
    print("\n💬 Sending message...")
    
    message = {
        "jsonrpc": "2.0",
        "id": str(uuid.uuid4()),
        "method": "message/send",
        "params": {
            "message": {
                "kind": "message",
                "messageId": str(uuid.uuid4()),
                "role": "user",
                "parts": [{"kind": "text", "text": "Hello A2A world!"}]
            }
        }
    }
    
    try:
        response = requests.post(base_url, json=message).json()
    except requests.RequestException as e:
        print(f"❌ Communication failed: {e}")
        return
    
    # 3단계: 응답 처리
    if "result" in response:
        task = response["result"] 
        print(f"📋 Task Status: {task['status']['state']}")
        
        # 히스토리에서 에이전트 답변 검색
        for msg in task.get("history", []):
            if msg.get("role") == "agent":
                agent_reply = msg["parts"][0]["text"]
                print(f"🤖 Agent: {agent_reply}")
                break
    else:
        error = response.get("error", {})
        print(f"❌ Error: {error.get('message', 'Unknown error')}")
    
    print("\n✨ A2A communication complete!")

if __name__ == "__main__":
    main()

클라이언트 전체 소스 코드

위 코드의 설명은 다음과 같습니다:

6~8라인: HTTP 요청 수행, 고유 식별자(UUID) 생성, JSON 파싱을 위한 도구를 임포트합니다.
12~18라인: 표준 .well-known/agent-card.json 엔드포인트를 호출하여 에이전트 정보를 성공적으로 파싱합니다.
21~34라인: A2A 메시지 형식을 준수하는 JSON-RPC 2.0 요청 규격 데이터를 생성합니다.
25~26라인: 프로토콜 명세 버전 및 통신 세션 추적용 고유 ID가 들어간 헤더 레이어를 구성합니다.
29~33라인: 작성 주체(user), 메시지 고유 식별자, 텍스트 데이터 파트가 포함된 바디 레이어를 구성합니다.
36~40라인: 지정한 에이전트의 서버 URL로 POST 요청 데이터를 전송하고 응답 결과를 해석합니다.
43~54라인: 수신한 태스크의 작업 상태를 읽어오고 대화 히스토리에서 에이전트 답변 데이터 영역을 가져옵니다.

이 클라이언트는 표준 탐색 및 통신 프로토콜을 정확하게 준수하므로 A2A 규격을 만족하는 전 세계의 모든 에이전트와도 호환됩니다.

A2A 시스템 테스트 진행

서버와 클라이언트가 실제로 유기적으로 연동하여 동작하는지 테스트를 통해 관측해 보겠습니다.

터미널 창을 하나 켜고 에이전트 서버를 실행합니다:

python echo_agent.py

다른 터미널 창을 켜고 클라이언트를 작동시킵니다:

cd app
python simple_a2a_client.py

성공 시 다음과 같은 출력을 확인할 수 있습니다:

🔍 Discovering A2A agent...
✅ Found: Echo Agent - A simple agent that echoes your messages back

💬 Sending message...
📋 Task Status: completed
🤖 Agent: You said: 'Hello A2A world!'

✨ A2A communication complete!

이외에도 다음 curl 명령어를 통해 탐색 데이터를 직접 덤프해 보는 테스트도 가능합니다:

curl http://0.0.0.0:8000/.well-known/agent-card.json

우리의 단순한 에코 에이전트는 다음과 같은 핵심 A2A 상호작용 패턴을 명확히 증명합니다:

프로토콜 준수: JSON-RPC 2.0 표준, 고정된 탐색 주소, 타입 체크가 완료된 메시지 구조를 준수합니다.
탐색 가능성: 어떤 A2A 클라이언트도 Agent Card를 파싱하여 우리 에이전트의 스펙을 파악하고 실행을 지시할 수 있습니다.
구조화된 통신: 모든 통신 메시지는 작성 역할(role), 고유 ID, 명세된 콘텐츠 조각 리스트를 명확히 가집니다.
태스크 기반 처리: 짧고 직관적인 요청 조차도 규정된 수명 주기 관리가 가능한 태스크 개체에 싸여서 반환됩니다.
규격화된 에러 대응: 비정상 요청을 받았을 때 약속된 JSON-RPC 표준 규격 에러 객체 형태로 응답을 돌려줍니다.
확장 용이성: 향후 실시간 스트리밍 지원, 파일 송수신, 대화 세션 누적 및 보다 복합적인 도구 결합 등으로 쉽게 코드를 확장해 나갈 수 있는 견고한 뼈대가 되어 줍니다.

이제 아래 마련된 터미널 창을 조작하여 실제로 출력되는 결과를 직접 관측해 보시기 바랍니다.

터미널이 연결되면 Terminal 1 탭 우측에 "+" 모양의 버튼을 눌러 추가 탭을 생성할 수 있으며, 이를 통해 두 개 터미널의 출력을 동시에 확인하실 수 있습니다.

향후 실무형 고급 에이전트를 설계할 때도 다음 핵심 수칙들을 기억하고 준수해 주시기 바랍니다.

모든 요소에 고유 ID 배포: 추적 및 디버깅을 원활하게 하기 위해 태스크, 메시지, 컨텍스트 전체에 UUID 등의 고유 ID를 부여해야 합니다.
안정적인 예외 처리: 서버 다운이나 비정상 실행 시 생으로 에러를 뿜게 두지 말고, 클라이언트가 식별할 수 있도록 규격에 맞춰 에러 객체를 리턴해야 합니다.
스펙 완벽 준수: A2A 프로토콜의 진정한 매력은 강력한 규칙 일관성에서 나옵니다. 조그마한 스펙 일탈도 에이전트 간 상호 호환성을 파괴할 수 있습니다.
태스크 중심 설계: 단순 연산조차 태스크 구조에 감싸서 리턴해 버릇하면 로깅, 모니터링 및 향후 대규모 확장 시 시스템 통합이 매우 쉬워집니다.
선 검증 후 로직 처리: Pydantic 등의 모델을 사용해 최우선으로 들어오는 데이터 형식이 적합한지 검사한 뒤 비즈니스 로직에 전달하십시오.

이제 여러분은 완성되어 작동하는 A2A 기초 시스템을 완전히 소유하게 되었습니다. 하지만 이것은 위대한 시작일 뿐입니다. 오늘 구축한 가벼운 에코 에이전트는 향후 마주할 복잡한 설계 기법들의 단단한 기반이 되어 줄 것입니다. 핵심 소통 패턴을 확실하게 익혔으니, 앞으로 마주하게 될 고급 기법들도 이미 알고 있는 지식을 바탕으로 어렵지 않게 쌓아 올리실 수 있습니다. 더 중요한 것은 여러분이 글로벌 A2A 생태계에 성공적으로 합류했다는 사실입니다. 여러분의 에이전트는 이제 규격을 준수하는 타 에이전트들과 자유롭게 통신할 수 있으며, 클라이언트는 세계 각지의 팀들이 공개한 유용한 에이전트들을 자유롭게 검색하여 협업을 요청할 수 있습니다. 이것이 바로 서로 다른 전문적인 에이전트들이 상호 작용하며 복잡한 사회적 문제를 해결해 나가는 A2A 프로토콜이 그리는 미래상입니다.

A2A SDK 사용법

FastAPI 등을 이용한 직접 구현 방식에서 공식 A2A SDK를 사용한 에이전트 개발 방식으로 업그레이드하는 방법을 알아봅니다. AgentExecutor, DefaultRequestHandler 등 SDK의 핵심 컴포넌트들과 태스크 라이프사이클 관리법을 이해합니다. 향상된 프로토콜 처리 기능과 상용화 스펙을 갖춘 확장 가능하고 유지 관리가 용이한 에이전트를 구축하는 방법을 학습합니다.

우리는 앞서 FastAPI와 원시 JSON-RPC 처리를 이용해 처음부터 끝까지 완전한 A2A 에이전트를 구축해 보았습니다. 이를 통해 탐색 엔드포인트, 메시지 구조, 태스크 수명 주기, 프로토콜 작동 구조 등 백그라운드에서 A2A가 어떻게 동작하는지 깊이 이해할 수 있었습니다. 이제 공식 A2A SDK를 활용하여 이 기능들을 어떻게 한층 더 간결하고, 유지 관리가 쉬우며, 상용화 수준에 걸맞은 코드로 업그레이드할 수 있는지 살펴보겠습니다.

이전의 직접 구현 과정이 수동 변속기 차량의 운전법(엔진, 클러치, 기어가 서로 맞물려 작동하는 원리)을 배운 것이라면, 이번 과정은 이 복잡한 제어 로직을 알아서 처리해 주는 자동 변속기 차량으로 업그레이드하여 목적지로 가는 비즈니스 로직에만 온전히 집중하는 것과 같습니다.

동일하게 동작하는 에코 에이전트를 다시 만들되, 이번에는 공식 A2A Python SDK를 적용하겠습니다. 메시지를 수신하여 그대로 반환하는 기능은 완전히 같지만, SDK를 사용함으로써 얼마나 많은 코드 복잡성이 걷히는지 확인하실 수 있습니다. 더 중요한 것은 이 SDK 기반 버전이 내장 에러 처리, 관측 가능성, 그리고 뛰어난 확장성을 기본 제공하여 실제 프로덕션 환경에 즉시 투입될 수 있다는 점입니다.

A2A SDK 사용 전에 알아야 할 사항은 무엇인가요?

코드를 살펴보기 전에, A2A SDK를 강력하게 만드는 핵심 컴포넌트들의 역할을 짚고 넘어가겠습니다.

AgentExecutor: 에이전트의 비즈니스 로직이 구현되는 클래스입니다. 개발자가 직접 JSON-RPC 요청을 파싱하고 태스크 개체를 선언할 필요 없이, 검증된 컨텍스트 데이터를 수신하여 이벤트를 발행하는 execute()와 cancel() 메소드만 상속받아 구현하면 됩니다.
DefaultRequestHandler: 요청 라이프사이클의 전 과정을 조율합니다. 태스크 생성을 관리하고, 요청 유효성을 검증하며, 작성한 AgentExecutor를 호출하고, 실시간 스트리밍을 제어하며, 올바른 A2A 규격 응답을 빌드합니다.
A2AFastAPIApplication: 수동으로 작성했던 FastAPI 설정을 대체합니다. 탐색 엔드포인트 제공, JSON-RPC 라우팅, 유효성 검사, 에러 제어 및 응답 데이터 포맷팅 등 A2A 프로토콜 관련 처리를 전담합니다.
InMemoryTaskStore: 태스크의 지속성(persistence)과 상태를 관리합니다. SDK가 태스크의 생명 주기와 소통 히스토리를 추적하는 데 필요합니다.
이벤트 큐(Event queue): AgentExecutor가 이벤트를 발행하여 클라이언트에 결과를 전달할 때 사용하는 통로입니다. SDK가 이 데이터를 감싸서 JSON-RPC 규격의 태스크로 구성해 줍니다.

이 아키텍처의 장점은 관심사 분리(separation of concerns)에 있습니다. 에이전트가 무엇을 할지(비즈니스 로직)에만 집중하면, 클라이언트와 어떻게 소통할지(프로토콜 명세)는 SDK가 알아서 처리해 줍니다.

A2A SDK 기반 서버 구축 방법

앞서 그랬듯이 이번에도 단계별로 코드를 구성하여 각 파트의 역할을 확실히 보여드리겠습니다. sdk_echo_agent.py라는 파일을 생성하고 순서대로 개발을 진행해 보겠습니다. 먼저 핵심 임포트 문을 추가하고 각 모듈의 성격을 파악합니다:

# 에이전트 능력 정의를 위한 핵심 A2A 타입
from a2a.types import AgentCard, AgentSkill, AgentCapabilities

# 서버 프레임워크 컴포넌트
from a2a.server.apps.jsonrpc.fastapi_app import A2AFastAPIApplication
from a2a.server.request_handlers.default_request_handler import DefaultRequestHandler
from a2a.server.agent_execution.agent_executor import AgentExecutor
from a2a.server.tasks.inmemory_task_store import InMemoryTaskStore

# 서버 실행용
import uvicorn
import asyncio

A2A SDK 임포트 구성

1~2라인: 에이전트의 성능 명세 및 메타데이터 정의에 쓰이는 A2A 코어 데이터 구조를 불러옵니다.
4~8라인: 에이전트 구동에 필요한 실제 SDK 서버 컴포넌트들을 불러옵니다.
11~12라인: HTTP 서버를 구동하고 비동기 제어를 수행하기 위해 uvicorn과 asyncio를 가져옵니다.

실제 소스 코드에서 쓰이는 강력한 형식의 전용 클래스들을 그대로 제공받을 수 있습니다. 이제 필드명에 유의하여 안정적인 형식으로 에이전트 규격을 선언해 줍니다:

# 임포트 아래에 에이전트 정의 추가

# 실제 SDK 클래스를 이용해 에이전트 기술 정의
echo_skill = AgentSkill(
    id="echo_messages",
    name="Echo Messages", 
    description="Repeats whatever you say back to you",
    tags=["echo", "simple", "demo"],
    examples=[
        "Hello there!",
        "How are you doing?", 
        "Echo this message back to me"
    ],
    input_modes=["text/plain"],
    output_modes=["text/plain"]
)

# 에이전트 성능 정의
capabilities = AgentCapabilities(
    streaming=False,
    push_notifications=False,
    state_transition_history=True
)

# 실제 SDK 클래스를 이용해 에이전트 카드 생성
agent_card = AgentCard(
    name="SDK Echo Agent",
    description="An echo agent built with the official A2A SDK to demonstrate best practices",
    url="http://localhost:8000",
    version="1.0.0",
    protocol_version="0.3.0",
    skills=[echo_skill],
    default_input_modes=["text/plain"],
    default_output_modes=["text/plain"],
    capabilities=capabilities
)

A2A SDK를 활용한 AgentSkill 및 AgentCard 정의

4~16라인: 언더바 표기법을 준수하여 에이전트의 구체적 기술을 정의합니다. (예: input_modes, output_modes)
18~22라인: AgentCapabilities 객체를 선언하고 상세 성능 플래그를 할당합니다. (예: push_notifications, state_transition_history)
25~36라인: 에이전트 카드를 구성하고 규격 버전 등을 명시합니다. (예: default_input_modes, default_output_modes, protocol_version)

이제 핵심 작동 로직을 구현하기 위해 AgentExecutor 인터페이스를 상속받습니다:

# 에이전트 카드 아래에 실행기 클래스 추가

class EchoAgentExecutor(AgentExecutor):
    """
    실제 A2A SDK AgentExecutor를 사용한 비즈니스 로직 구현.
    SDK가 모든 프로토콜 복잡성을 대신 처리함.
    """
    
    async def execute(self, context, event_queue):
        """
        실제 SDK 패턴을 사용해 에코 로직 실행.
        
        인수:
            context: 유저 입력 및 메타데이터를 담은 RequestContext
            event_queue: 응답 이벤트를 발행할 EventQueue
        """
        try:
            # 유저 입력 획득 - 모든 텍스트 파트가 결합된 문자열 반환
            user_text = context.get_user_input()
            
            # 텍스트를 찾을 수 없는 경우 처리
            if not user_text or user_text.strip() == "":
                # SDK 유틸리티를 사용해 에러 메시지 생성
                error_message = new_agent_text_message(
                    "I didn't receive any text to echo. Please send me a message with text content."
                )
                await event_queue.enqueue_event(error_message)
                return
            
            # SDK 유틸리티를 사용해 에코 응답 생성 및 큐에 삽입
            echo_message = new_agent_text_message(f"You said: '{user_text.strip()}'")
            await event_queue.enqueue_event(echo_message)
            
        except Exception as e:
            # SDK가 구조화된 에러 처리를 지원함
            error_message = new_agent_text_message(f"Error processing your message: {str(e)}")
            await event_queue.enqueue_event(error_message)
    
    async def cancel(self, task_id, event_queue):
        """
        실제 SDK 인터페이스를 사용해 태스크 취소 처리.
        
        인수:
            task_id: 취소 대상 태스크 ID
            event_queue: 취소 이벤트를 발행할 EventQueue
        """
        cancel_message = new_agent_text_message(f"Echo task {task_id} has been canceled.")
        await event_queue.enqueue_event(cancel_message)

AgentExecutor 구현

4~8라인: 실제 SDK 베이스 클래스인 AgentExecutor를 올바르게 상속받습니다.
10~16라인: 정의에 정해진 인수 리스트대로 비동기 execute() 메소드를 구성합니다.
18~19라인: context 객체로부터 유저 입력 텍스트 문자열을 가져옵니다. (get_user_input()은 직접 구현 방식과 달리 메시지 객체가 아닌 조합된 문자열을 바로 반환함)
21~28라인: 입력값이 비어 있을 때의 예외를 처리하고 new_agent_text_message() 헬퍼로 에러 메시지를 큐에 밀어 넣습니다.
30~32라인: 에코 응답 객체를 만들어 비동기 enqueue_event()를 호출해 적재합니다.
34~37라인: 기타 예외 상황 시에도 에러 메시지를 구조화하여 돌려줍니다.
39~47라인: 취소 신호가 들어왔을 때 수행할 비동기 cancel() 동작을 구성합니다.

규정된 함수 인자와 클래스 명세를 온전히 유지하여 에이전트 실행기를 작성했습니다.

마지막으로 이 컴포넌트들을 A2A SDK 기본 앱 객체와 연결하여 실행시킵니다:

# 실행기 클래스 아래에 애플리케이션 설정 추가

def create_app():
    """실제 SDK 컴포넌트를 사용해 A2A 애플리케이션 생성 및 설정"""
    
    # 에이전트 실행기 생성
    executor = EchoAgentExecutor()
    
    # 태스크 상태 관리를 위한 태스크 저장소 생성
    task_store = InMemoryTaskStore()
    
    # 전체 프로세스를 조율할 요청 핸들러 생성
    request_handler = DefaultRequestHandler(
        agent_executor=executor,
        task_store=task_store
    )
    
    # A2A FastAPI 애플리케이션 생성
    app = A2AFastAPIApplication(
        agent_card=agent_card,
        http_handler=request_handler
    )
    
    # 설정된 FastAPI 앱 빌드 및 반환
    return app.build()

# 메인 영역 작성 완료
if __name__ == "__main__":
    print("🚀 Starting A2A SDK Echo Agent...")
    print("📡 Agent Card: http://localhost:8000/.well-known/agent-card.json")
    print("🔗 A2A Endpoint: http://localhost:8000/")
    
    app = create_app()
    
    uvicorn.run(
        app,
        host="0.0.0.0",
        port=8000,
        log_level="info"
    )

전체 에이전트 서버 구성 완료

4~7라인: 앞서 개발한 커스텀 AgentExecutor 인스턴스를 확보합니다.
9~10라인: 조율 핸들러에 필수적인 인메모리 저장소 InMemoryTaskStore를 생성합니다.
12~16라인: 실행기와 저장소를 연결하여 공통 조율기 DefaultRequestHandler를 생성합니다.
18~22라인: 에이전트 카드 스펙과 조율기를 태워 최종 A2A 응용 객체 A2AFastAPIApplication을 준비합니다.
24~25라인: .build() 메소드를 실행하여 표준 FastAPI 객체로 변환해 리턴합니다.
29~31라인: 서버 구동 정보를 로그에 기록하고 준비된 응용 객체를 확보합니다.
33~38라인: uvicorn 포트 설정을 맞춘 뒤 웹 서비스를 시작합니다.

SDK 내부에서 제어 흐름은 어떻게 돌아가나요? 외부 클라이언트의 요청은 우선 A2AFastAPIApplication에 인입된 다음, 핵심 조율기인 DefaultRequestHandler로 전달됩니다. 핸들러는 데이터를 검증하고 InMemoryTaskStore를 활용해 태스크 상태를 기록한 다음, 개발자가 작성한 AgentExecutor.execute() 비동기 프로세스를 호출합니다. 실행기에서 큐에 집어넣은 결과 및 실시간 진행 이벤트들은 SDK가 모아서 완벽한 JSON-RPC 규격의 데이터로 가공해 클라이언트로 스트리밍해 줍니다.

이를 통해 다음과 같은 백그라운드 작업들을 SDK가 대신 처리해 줍니다:

/.well-known/agent-card.json 표준 경로에 에이전트 카드 호스팅.
JSON-RPC 2.0 기반의 요청 구문 해석 및 유효성 진단.
태스크 생명 주기 전 단계 관리 및 고유 식별값 할당.
이벤트 수집 통로(큐) 관리 및 통신 응답 패키징.
비정상 구문 수신 시 규격화된 에러 전송.

아래는 공식 SDK 컴포넌트들을 온전히 활용한 완성된 sdk_echo_agent.py 전체 소스 코드입니다. 이를 구동하여 이전 레슨에서 만들었던 클라이언트 프로그램과 정상적으로 작동하는지 확인해 보겠습니다. "실행" 버튼을 눌러 테스트를 수행해 보십시오.

# 에이전트 능력 정의를 위한 핵심 A2A 타입
from a2a.types import AgentCard, AgentSkill, AgentCapabilities

# 서버 프레임워크 컴포넌트
from a2a.server.apps.jsonrpc.fastapi_app import A2AFastAPIApplication
from a2a.server.request_handlers.default_request_handler import DefaultRequestHandler
from a2a.server.agent_execution.agent_executor import AgentExecutor
from a2a.server.tasks.inmemory_task_store import InMemoryTaskStore

# 메시지 유틸리티
from a2a.utils.message import new_agent_text_message

# 서버 실행용
import uvicorn

# 실제 SDK 클래스를 이용해 에이전트 기술 정의
echo_skill = AgentSkill(
    id="echo_messages",
    name="Echo Messages",
    description="Repeats whatever you say back to you.",
    tags=["echo", "simple", "demo"],
    examples=[
        "Hello there!",
        "How are you doing?",
        "Echo this message back to me."
    ],
    input_modes=["text/plain"],
    output_modes=["text/plain"]
)

# 에이전트 성능 정의
capabilities = AgentCapabilities(
    streaming=False,
    push_notifications=False,
    state_transition_history=True
)

# 실제 SDK 클래스를 이용해 에이전트 카드 생성
agent_card = AgentCard(
    name="SDK Echo Agent",
    description="An echo agent built with the official A2A SDK to demonstrate best practices.",
    url="http://localhost:8000",
    version="1.0.0",
    protocol_version="0.3.0",
    skills=[echo_skill],
    default_input_modes=["text/plain"],
    default_output_modes=["text/plain"],
    capabilities=capabilities
)

class EchoAgentExecutor(AgentExecutor):
    """
    실제 A2A SDK AgentExecutor를 사용한 비즈니스 로직 구현.
    SDK가 모든 프로토콜 복잡성을 대신 처리함.
    """
    
    async def execute(self, context, event_queue):
        """
        실제 SDK 패턴을 사용해 에코 로직 실행.
        
        인수:
            context: 유저 입력 및 메타데이터를 담은 RequestContext.
            event_queue: 응답 이벤트를 발행할 EventQueue.
        """
        try:
            # 유저 입력 획득 - 모든 텍스트 파트가 결합된 문자열 반환
            user_text = context.get_user_input()
            
            # 텍스트를 찾을 수 없는 경우 처리
            if not user_text or user_text.strip() == "":
                # SDK 유틸리티를 사용해 에러 메시지 생성
                error_message = new_agent_text_message(
                    "I didn't receive any text to echo. Please send me a message with text content."
                )
                await event_queue.enqueue_event(error_message)
                return
            
            # SDK 유틸리티를 사용해 에코 응답 생성 및 큐에 삽입
            echo_message = new_agent_text_message(f"You said: '{user_text.strip()}'")
            await event_queue.enqueue_event(echo_message)
            
        except Exception as e:
            # SDK가 구조화된 에러 처리를 지원함
            error_message = new_agent_text_message(f"Error processing your message: {str(e)}")
            await event_queue.enqueue_event(error_message)
    
    async def cancel(self, task_id, event_queue):
        """
        실제 SDK 인터페이스를 사용해 태스크 취소 처리.
        
        인수:
            task_id: 취소 대상 태스크 ID.
            event_queue: 취소 이벤트를 발행할 EventQueue.
        """
        cancel_message = new_agent_text_message(f"Echo task {task_id} has been canceled.")
        await event_queue.enqueue_event(cancel_message)

def create_app():
    """실제 SDK 컴포넌트를 사용해 A2A 애플리케이션 생성 및 설정."""
    
    # 에이전트 실행기 생성
    executor = EchoAgentExecutor()
    
    # 태스크 상태 관리를 위한 태스크 저장소 생성
    task_store = InMemoryTaskStore()
    
    # 전체 프로세스를 조율할 요청 핸들러 생성
    request_handler = DefaultRequestHandler(
        agent_executor=executor,
        task_store=task_store
    )
    
    # A2A FastAPI 애플리케이션 생성
    app = A2AFastAPIApplication(
        agent_card=agent_card,
        http_handler=request_handler
    )
    
    # 설정된 FastAPI 앱 빌드 및 반환
    return app.build()

if __name__ == "__main__":
    print("🚀 Starting A2A SDK Echo Agent...")
    print("📡 Agent Card: http://localhost:8000/.well-known/agent-card.json")
    print("🔗 A2A Endpoint: http://localhost:8000/")
    
    app = create_app()
    
    uvicorn.run(
        app,
        host="0.0.0.0",
        port=8000,
        log_level="info"
    )

이전에 구현한 클라이언트를 활용한 동작 테스트

이전에 만들어 놓은 오리지널 수동 클라이언트(simple_a2a_client.py)를 가동 중인 SDK 서버에 붙여서 작동시켜 보면 오류가 발생하는 것을 알 수 있습니다. 직접 구현 방식은 응답으로 status와 history 등이 들어간 Task 구조의 결과를 리턴해 주길 기대한 반면, 정식 SDK 환경은 결과 봉투에 메시지를 곧바로 담아 보내는 방식이기 때문입니다. 이러한 직접 응답 구조는 불필요한 레이어가 없어 효율적이지만, 클라이언트 입장에서도 이에 맞게 향상된 포맷을 받아들이도록 갱신되어야 함을 시사합니다. 비록 두 버전 다 동일한 A2A 규약 하에 탐색과 소통을 수행하고 있지만 결과 전송 구조에 차이가 존재합니다.

SDK의 직접 메시지 포맷은 확실히 데이터 전달 효율이 뛰어납니다.

직접 구현 방식의 응답: {"result": {"status": {...}, "history": [...]}}
SDK 활용 방식의 응답: {"result": {"kind": "message", "parts": [...], "role": "agent"}}

이와 같이 SDK 기반에서는 군더더기 없는 직접적인 통신 모델을 채택하여 설계 효율을 향상했습니다. 완결성 있는 작동을 보장하기 위해 아래와 같이 두 유형의 응답 포맷을 모두 매끄럽게 흡수하여 처리해 주는 SDK 호환 클라이언트 코드(simple_a2a_client_sdk.py)를 구현해 사용합니다:

import requests
import uuid
import json

def main():
    # 1단계: 에이전트 탐색
    base_url = "http://localhost:8000"
    print("🔍 Discovering A2A agent...")
    
    try:
        agent_card = requests.get(f"{base_url}/.well-known/agent-card.json").json()
        print(f"✅ Found: {agent_card['name']} - {agent_card['description']}")
    except requests.RequestException as e:
        print(f"❌ Discovery failed: {e}")
        return
    
    # 2단계: A2A JSON-RPC를 사용해 메시지 전송
    print("\n💬 Sending message...")
    
    message = {
        "jsonrpc": "2.0",
        "id": str(uuid.uuid4()),
        "method": "message/send",
        "params": {
            "message": {
                "kind": "message",
                "messageId": str(uuid.uuid4()),
                "role": "user",
                "parts": [{"kind": "text", "text": "Hello A2A world!"}]
            }
        }
    }
    
    try:
        response = requests.post(base_url, json=message).json()
    except requests.RequestException as e:
        print(f"❌ Communication failed: {e}")
        return
    
    # 3단계: 응답 처리
    # SDK는 직접 구현 방식처럼 히스토리가 포함된 Task를 주지 않고 Message를 직접 반환함
    if "result" in response:
        result = response["result"] 
        if isinstance(result, dict) and "parts" in result:
            # SDK에서 수신한 직접 Message 응답
            agent_reply = result["parts"][0]["text"]
            print(f"🤖 Agent: {agent_reply}")
        else:
            # Task 응답인 경우 (호환성 유지용)
            print(f"📋 Task Status: {result.get('status', {}).get('state', 'completed')}")
            
            # 히스토리에서 에이전트 답변 검색
            for msg in result.get("history", []):
                if msg.get("role") == "agent":
                    agent_reply = msg["parts"][0]["text"]
                    print(f"🤖 Agent: {agent_reply}")
                    break
    else:
        error = response.get("error", {})
        print(f"❌ Error: {error.get('message', 'Unknown error')}")
    
    print("\n✨ A2A communication complete!")

if __name__ == "__main__":
    main()

A2A SDK 호환 클라이언트를 통한 테스트

이 클라이언트 프로그램은 두 가지 반환 규격을 상황에 따라 알아서 매끄럽게 처리하므로, 직접 구현한 원격 서버나 정식 SDK로 띄운 서버 양측 모두와 정상적으로 호환되어 연동합니다.

두 가지 구현 방법에는 어떤 차이점이 있나요?

SDK 구현 방식이 여러 내장 클래스를 갖춰야 하고 비동기 이벤트 기반 구조를 띠고 있어 처음에는 직접 작성하는 것보다 복잡해 보일 수 있으나, 시스템 규모가 커질수록 월등히 효율적이고 편리합니다. FastAPI를 단독으로 써서 직접 개발하는 형태는 프로토콜 명세에 기재된 온갖 전송 상세 규칙을 개발자가 한 땀 한 땀 구현해야 하므로 에이전트의 기능이 늘어날수록 심각한 부담으로 돌아옵니다. SDK는 이러한 귀찮은 소통 인프라 구축 작업을 감춰 주므로 여러분은 에이전트 본연의 차별화된 비즈니스 로직 설계에만 오롯이 집중할 수 있으며, 가벼운 에코 에이전트부터 고도화된 상호 작용 서버에 이르기까지 시스템을 안전하게 스케일 아웃할 수 있는 튼튼한 토대가 되어 줍니다.

비교 항목	직접 구현 방식	SDK 활용 방식
JSON-RPC 처리	수동 파싱 및 포맷팅	자동 처리
메시지 유효성 검사	커스텀 Pydantic 모델 설계	내장 검증 기능
태스크 라이프사이클	수동 상태 관리	자동 처리
에러 처리	커스텀 JSON-RPC 에러	SDK 유틸리티 함수
에러 탐색	수동 엔드포인트 구성	자동 처리
응답 데이터 빌드	수동 JSON 빌드	이벤트 큐 추상화
타입 안전성	제한적 유효성 검사	SDK 전체 타입 검증
테스트	Mock HTTP 요청	SDK 테스트 유틸리티
관측 가능성	커스텀 로깅	내장 트레이싱 훅
인증	미구현	SDK 프레임워크 자체 지원

SDK의 진정한 위력은 단순 에코 기능을 넘어서 대규모 비즈니스 오케스트레이션을 짤 때 여실히 드러납니다. 항공권 예약 서비스, 호텔 제휴 시스템, 현지 관광 상품 연동 서버들과 유기적으로 상호작용하여 움직이는 종합 여행 플래너 에이전트를 설계해야 한다고 상상해 보십시오. 만약 직접 구현 방식으로 이를 개발하려면 실시간 가격 현황을 밀어주기 위해 스트리밍 제어 로직을 직접 짜야 하고, 협력 서버별로 서로 다른 다회차(multi-turn) 대화 세션 컨텍스트 데이터를 수동으로 쌓아서 들고 있어야 합니다. 제휴사별로 각종 API 인증 규격(OAuth 등)을 수동 개발해 붙여야 하며 통신 장애 시 작동할 자가 복구 회로도 직접 짜서 붙여야 합니다. 반면 정식 SDK를 사용하면 이 모든 고난도 로직이 백그라운드에서 완전히 자동 처리됩니다. 데이터 스트리밍은 비동기 이벤트 큐로 제어되고, 복잡한 대화 상태는 내장된 태스크 저장소에 의해 추적되며, 다채로운 인증 규격은 제공되는 플러그인 형식의 핸들러에 의해 적용되고, 에러 감지 및 복구는 약속된 예외 처리 흐름을 통해 간편하게 제어됩니다.

이러한 견고한 기초 아키텍처 덕분에 향후 파일 송수신, 다국어 번역, 제3사 외부 API 통합 등의 신규 비즈니스 능력을 탑재할 때 프로토콜 레이어 자체를 뒤엎을 필요 없이, 구현된 AgentExecutor에 살만 붙여서 손쉽게 기능을 이식해 나갈 수 있게 됩니다. 가벼운 에코 에이전트를 빌드할 때 쓰인 동일한 개발 패턴이 엔터프라이즈급 대규모 에이전트 군을 구축할 때도 고스란히 재사용되어, 채널 전반에서 일어나는 수천 건의 동시 대화 세션도 흐트러짐 없이 관제해 줍니다.

에이전트에게 프로토콜이 필요한 이유

A2A이란 무엇인가요?

사전 요구 사항은 무엇인가요?

도움이 되는 경험:

Python 3(함수, 패키지, venv, uv, poetry, pip 등의 가상 환경 구성, 기본 타입 정의)에 대한 원활한 사용 능력.
HTTP 기본 지식(요청, 응답, 헤더, 상태 코드).
JSON 데이터에 대한 이해(페이로드 읽고 쓰기, 스키마, 에러 포맷).
커맨드라인 환경 사용 능력(스크립트 실행, 환경 변수 설정).
기초 보안 개념(토큰, API 키, 최소 권한 원칙).
로깅 및 디버깅 기본 지식(트레이스백 분석, print 문 또는 로그 출력).
이벤트 및 스트리밍에 대한 이해(SSE 또는 WebSockets), REST/RPC 패턴에 대한 직관적 이해.

코드 예제는 최대한 명료하게 작성되었으며 어려운 용어는 처음 등장할 때 자세히 설명하므로, 초보자도 쉽게 따라오실 수 있습니다.

A2A SDK는 어떻게 사용하나요?

A2A 파이썬 SDK를 설치하여 에이전트를 빌드하고 실행해 볼 수 있습니다. 현재 개발 환경에 맞춰 다음 명령어 중 하나를 입력해 주시기 바랍니다:

# 핵심 SDK
pip install uv
uv pip install a2a-sdk
# 또는
pip install a2a-sdk

선택적 추가 기능을 함께 빌드하려면 다음과 같이 익스텐션을 지정하여 설치할 수 있습니다:

uv add "a2a-sdk[all]"        # 모든 기능 포함
uv add "a2a-sdk[http-server]"  # HTTP 서버 기능
uv add "a2a-sdk[grpc]"         # gRPC 지원 기능
uv add "a2a-sdk[telemetry]"    # OpenTelemetry 트레이싱 지원
uv add "a2a-sdk[encryption]"   # 암호화 지원

이렇게 쉽고 간단하게 시스템에 A2A 환경을 구축하고 테스트를 진행할 수 있습니다.

88 KiB Raw Permalink Blame History

에이전트에게 프로토콜이 필요한 이유

A2A란 무엇인가요?

사전 요구 사항은 무엇인가요?

A2A SDK는 어떻게 사용하나요?

A2A vs. 다른 프로토콜

A2A는 MCP와 어떻게 다른가요?

A2A는 다른 에이전트 간 프로토콜과 어떻게 다른가요?

다음 단계는 무엇인가요?

A2A의 핵심 구성 요소

A2A는 어떻게 작동하나요?

에이전트의 기술을 정의하고 A2A에서 탐색 가능하게 만드는 방법은 무엇인가요?

A2A에서 에이전트의 "기술(skills)"이란 무엇인가요?

A2A에서 "Agent Cards"란 무엇인가요?

A2A에서 에이전트들은 어떻게 통신하고 작업을 주고받나요?

A2A에서 메시지란 무엇인가요?

A2A에서 태스크란 무엇인가요?

A2A에서 산출물이란 무엇인가요?

다음 단계는 무엇인가요?

퀴즈: A2A 기초

퀴즈

A2A 준수 클라이언트 및 서버 구축

무엇을 개발하나요?

A2A 서버 구축 방법

시스템을 A2A 준수 상태로 만드는 방법

메시지 구조 유효성 검사 추가 방법

에이전트 로직 구현 방법

A2A 클라이언트 구축 방법

A2A 시스템 테스트 진행

A2A SDK 사용법

A2A SDK 사용 전에 알아야 할 사항은 무엇인가요?

A2A SDK 기반 서버 구축 방법

두 가지 구현 방법에는 어떤 차이점이 있나요?

에이전트에게 프로토콜이 필요한 이유

A2A이란 무엇인가요?

사전 요구 사항은 무엇인가요?

A2A SDK는 어떻게 사용하나요?

88 KiB

Raw Permalink Blame History