[Agent] AI 에이전트 프로토콜, 구글 A2A 개념부터 원리 실습하기
이 글은 [요즘 AI 에이전트 개발, LLM RAG ADK MCP LangChain A2A LangGraph]에서 발췌했습니다.
수많은 회사, 조직, 팀에서 AI 에이전트를 이야기하고 에이전트를 만들고 있습니다. 프레젠테이션을 만들어주는 에이전트, 요약을 하는 에이전트, 수많은 자료를 검색한 후 보고서를 만들어주는 에이전트, 코딩을 도와주는 에이전트 등 이미 상용화된 에이전트들이 있고 사람들이 이미 사용하고 있습니다. 그렇다면 여기서 한 단계 더 나아가서 다른 곳에서 만든 에이전트를 SaaS 형태로 사용할 수 있다면 어떨까요? 이런 고민에 대한 해결책으로 구글은 A2A라는 프로토콜을 내놓았습니다.
A2A란?
A2A(Agent to Agent)는 2025년 4월, 라스베이가스에서 열린 구글 클라우드 넥스트Google Cloud Next에서 처음 발표되었습니다. 구글이 50개 이상의 기술 파트너와 함께 만든 프로토콜입니다. 개방형 프로토콜이므로 누구나 의견을 제시할 수 있습니다. 깃허브에 저장소*도 있습니다. A2A 소개의 내용을 살펴보면 다음과 같은 말이 있습니다. “에이전트 AI의 이점을 극대화하려면 에이전트들이 분산된 데이터 시스템과 애플리케이션 전반에서 역동적인 다중 에이전트 생태계에서 협업할 수 있어야 합니다.”
핵심 키워드는 분산된 환경에서 다중 에이전트가 협업할 수 있어야 한다는 겁니다. 로컬에 혹은 한 대의 머신에 코드로 생성한 논리적인 에이전트가 아니라 물리적으로 떨어진 네트워크 공간에 있는 다른 에이전트와 네트워크를 넘어서 협업을 하고 싶다는 의도를 담아냈습니다. 사람으로 생각하면 서로 이메일을 주고 받으며 업무를 진행하는 것과 유사합니다.
물론 사람이 아니기에 에이전트들은 사람의 언어로 통신할 수 없습니다. A2A라는 프로토콜이 있기 전에는 에이전트를 만드는 곳이 제각각이므로 통신을 하는 방법을 먼저 생각을 해야만 했습니다. A2A를 사용하면 에이전트 간의 통신을 표준화할 수 있게 된다는 큰 장점이 있습니다.
A2A 프로토콜 스펙**은 웹에 공개되어 있습니다. A2A 스펙을 출력하면 64장이나 됩니다. 이는 프로토콜이 꽤 많은 기능을 담고 있고, 프로토콜을 어떻게 구현해야 하는지 상세히 정의해놓았다는 의미가 됩니다. 스펙에 나와 있는 핵심 목표는 다음 6가지입니다.
상호운용성(Interoperability) : 서로 다른 에이전트 시스템이 통신할 수 있게 함
협업(Collaboration) : 에이전트가 작업을 위임, 맥락을 전달, 복잡한 사용자의 요청에 대해 함께 작업
발견(Discovery) : 다른 에이전트를 동적으로 발견할 수 있음
유연성(Flexibility) : 동기 요청/응답, 리얼 타임 업데이트를 위한 스트리밍, 오래 걸리는 작업을 위한 비동기 모드를 지원
보안(Security) : 기업 환경을 위한 보안 통신을 지원
비동기성(Asynchronicity) : 오래 걸리는 작업과 상호작용을 지원
A2A의 핵심 개념 및 용어들
A2A가 무엇인지 그리고 어떤 것을 목표로 하는지 알았습니다. A2A 프로토콜을 사용해서 개발을 하려면 A2A에서 사용하는 개념과 용어들을 알아야 할 필요가 있습니다. 다음 그림에는 A2A에 등장하는 핵심 주체들이 있습니다.
사용자는 에이전트에게 요청 혹은 일감을 주는 사용자를 의미합니다. 사용자는 인간이거나 혹은 자동화된 서비스가 될 수도 있습니다. 클라이언트는 A2A 클라이언트를 의미합니다. 사용자를 대신하여 원격 에이전트에게 작업을 할당하거나 정보를 요청합니다. 클라이언트는 A2A 프로토콜을 사용하여 통신을 합니다. 여기서 원격 에이전트는 A2A 서버를 의미합니다. A2A 서버는 HTTP 엔드포인트를 가지고 있는 에이전트이며, 클라이언트의 요청을 수신하고, 작업을 처리하고 결과물을 반환하거나 작업의 상태를 업데이트합니다.
에이전트 카드
클라이언트가 원격에 있는 에이전트에게 작업을 요청하려면, 에이전트를 찾아야 합니다. 에이전트를 찾을 때 필요한 것이 바로 에이전트 카드입니다. 에이전트 카드는 A2A 서버를 설명하는 JSON 데이터 문서입니다. 일반적으로 .well-known/agent.json 경로에 위치합니다.
에이전트 카드에는 에이전트의 이름, 서비스의 엔드포인트 URL, 버전, 지원되는 A2A의 기능, 기본 입출력 방식, 인증 요구사항 등이 나와 있습니다. 정확한 내용은 프로토콜 스펙의 에이전트 카드를 참고하면 됩니다.
태스크
A2A에서는 보통 클라이언트가 에이전트에게 수행해야 하는 업무를 요청하는데, 이를 태스크라고 합니다. 스펙에 나오는 용어로 설명을 하면 A2A 클라이언트를 위해 A2A 서버에서 처리되는 상태를 가진 작업 단위라고 할 수 있습니다. 태스크는 다음의 형태를 가지는 객체입니다.
Use a different Browser
export interface Task { id: string; contextId: string; status: TaskStatus; history?: Message[]; artifacts?: Artifact[]; metadata?: { [key: string]: any; }; kind: “task”; }
id는 서버에서 생성된 식별자이며, contextId는 서버에서 생성한 ID입니다. status는 태스크의 상태를 의미하고, artifacts는 에이전트가 생성한 결과물 리스트를 의미합니다. history는 최근에 클라이언트와 주고받은 메시지의 리스트이고, metadata는 태스크와 관련된 값을 임의로 딕셔너리 형태로 추가할 수 있도록 해둔 파라미터입니다. kind는 데이터 유형의 구분자이며 task로 고정값이 들어갑니다.
TaskStatus는 다음과 같은 형태로 되어 있는 데이터입니다.
Use a different Browser
export interface TaskStatus {
state: TaskState;
message?: Message;
timestamp?: string;
}
status는 작업의 상태를 의미하며, 상태는 submitted(서버가 작업을 수신), working(작업 중), input-required(클라이언트의 추가 입력 필요), completed(완료), canceled(작업 취소) 값을 가지고 있습니다. message는 현재 상태에 대한 정보를 제공하는 메시지입니다. timestamp는 상태가 기록된 시간(UTC 권장)입니다.
메시지
클라이언트와 에이전트 간의 의사소통 과정 혹은 단위를 나타냅니다. 메시지Message는 객체이며 데이터는 다음과 같은 형태로 되어 있습니다.
Use a different Browser
export interface Message { role: “user” | “agent”; parts: Part[]; metadata?: { [key: string]: any; }; extensions?: string[]; referenceTaskIds?: string[]; messageId: string; taskId?: string; contextId?: string; kind: “message”; }
role은 메시지를 보낸 주체(user 또는 agent)를 나타냅니다. parts는 메시지 콘텐츠의 기본 단위인 파트(part) 객체들의 목록입니다. 파트는 텍스트, 이미지 등 다양한 데이터 구조를 가질 수 있는 유니온 타입입니다. 메시지에는 텍스트, 파일, JSON 등의 다양한 타입의 메시지가 한 번에 올 수 있습니다. 이를 표현하기 위해 A2A에서는 메시지의 일부분이라는 의미로 파트라는 용어를 사용했습니다. 예를 들어 사용자가 텍스트와 함께 PDF 파일을 함께 전송했다면, TextPart와 FilePart가 parts 리스트에 들어가게 됩니다. 2025년 6월 23일 기준 스펙에 명시된 파트의 타입은 TextPart, FilePart, DataPart가 있습니다. 각각 일반 텍스트, base64 인코딩 혹은 URI를 통해 참조할 수 있는 파일, JSON 데이터를 의미합니다.
아티팩트
아티팩트Artifact는 원격 에이전트가 생성한 작업 결과물을 말합니다. 텍스트, 이미지, 문서, 스프레드시트, JSON 데이터 등 다양한 데이터가 될 수 있습니다. 아티팩트는 하나 이상의 파트 객체로 이루어져 있으며 스트리밍 형식으로 생성되기도 합니다.
MCP와 A2A는 뭐가 다른가?
MCP는 AI 모델이 도구 및 외부 리소스를 사용하는 방법을 표준화한 프로토콜입니다. 반면 A2A는 AI 에이전트들이 서로 소통하고 협업하는 방식을 표준화하는 프로토콜입니다. 그런데 A2A를 공부하고 있으면 다음과 같은 의문이 생기게 됩니다. MCP의 클라이언트에서 원격에 있는 에이전트를 도구로 사용한다면, A2A와 같은 것 아닌가? 이에 대해서는 MCP와 A2A를 구분하여 사용해야 한다는 입장과 A2A까지 적용하는 것은 실제적으로 어렵다는 입장이 동시에 있습니다. 그리고 실제로 MCP의 로드맵에 에이전트 워크플로를 지원하는 기능을 추가하는 내용이 들어 있기도 합니다.
AI 업계는 매우 빠르게 변하고 발전하기에 앞으로 어찌 될지 예측하는 것은 매우 어렵습니다만, 저 개인적으로도 MCP와 A2A를 동시에 사용하는 것은 구현해야 할 부분들이 매우 많아지고 애플리케이션이 복잡해지기에 두 프로토콜이 모두 안정화될 때까지는 지켜볼 필요가 있다고 생각합니다.
다만 이미 구현되어 있는 SDK의 완성도 측면에서 보자면, 6개월 먼저 나온 MCP 쪽이 더 높다고 할 수 있습니다. 그리고 적용하는 것이 매우 쉽고, 기존의 LSP 프로토콜과 매우 비슷하기에 기존에 LSP 프로토콜을 쓰고 있던 IDE 쪽에서는 최소한의 노력으로 바로 적용할 수 있기도 합니다.
실제로 커서나 클라인cline 등 코딩 에이전트들이 MCP를 빠르게 도입하기도 했습니다. 또한 최근에는 윈도우 11에 MCP 레지스트리가 탑재되기도 했습니다. MCP는 이미 사용할 수 있는 프로토콜인 반면, A2A는 구현체가 완성되길 조금 기다려야 하는 상황입니다.
2025년 연말에 A2A의 파이썬 SDK를 거의 완성할 것이라는 이야기가 있으니, MCP는 적극적으로 도입을 검토해보되 A2A는 동향을 계속 살펴보는 것이 좋겠습니다. 언젠가는 두 프로토콜을 자유자재로 섞어 쓸 수 있는 상황이 될 수도 있고, 두 프로토콜이 극적인 대통합을 이룰 수도 있고, 아니면 둘 중에 하나의 프로토콜이 사실상 시장의 표준이 되는 상황이 될 수도 있습니다. 구글에서 작성한 AI 프로토콜 진화의 승자 예측****이라는 글도 있으니, 함께 보면 두 프로토콜의 차이점에 대해 더 잘 알 수 있게 됩니다. 미리 말씀드리면 구글에서 작성했기에 A2A에 더 호의적인 입장의 글입니다.
A2A 동작 원리 실습해보기 : AI 비서 서버와 클라이언트
A2A 프로토콜의 동작 원리를 보여주는 간단한 AI 비서 서버와 클라이언트를 만드는 실습용 프로그램을 만들겠습니다. AI 에이전트들이 서로 통신하고 협업하는 표준 방식인 A2A 프로토콜을 직접 구현하고 테스트하겠습니다. A2A는 파이썬*****과 타입스크립트****** 구현체가 있습니다. 이 책에서는 파이썬 구현체를 기반으로 설명드리겠습니다. 각 SDK 버전이 아직 매우 낮은 상태이므로 추후에는 코드가 동작하지 않을 수도 있습니다. 그런 경우 책의 공식 깃허브 저장소*******를 참고하시거나, 공식 SDK의 문서를 참고해주세요.
시작은 역시나 패키지 설치부터입니다. 패키지명은 a2a-sdk입니다. pip 혹은 uv 명령을 사용하여 설치해주세요.
Use a different Browser
pip install a2a-sdk==0.3.0
예제 작성 순서
A2A는 MCP보다 조금 더 많은 준비가 필요합니다. 코드를 작성하면서 길을 잃지 않도록 그림을 준비했습니다. 하나씩 만들면서 추가해나가겠습니다. 코드 작성 순서는 다음과 같습니다.
Agent Card를 만들어보고 테스트
Agent Executor 생성 및 테스트
Http 핸들러를 만들고 서버 코드를 작성 후 기동
A2A 인스펙터*******를 사용하여 에이전트 테스트
A2A 클라이언트 코드 작성
A2A에도 에이전트를 테스트할 수 있는 A2A 인스펙터(Inspector)가 추가되었기에 서버 쪽 코드가 완
성되면 A2A 인스펙터를 사용하여 서버 기능을 테스트하겠습니다. 서버가 잘 동작하면 다음으로
클라이언트 코드를 작성하고 테스트하겠습니다. 서버의 스킬로는 LLM과 연동하여 인사말을 나
누는 함수를 추가하려고 합니다.
Agent Card 작성
에이전트 카드를 먼저 작성해봅시다.
Use a different Browser
from a2a.types import AgentCapabilities, AgentCard, AgentSkill def create_agent_card() -> AgentCard: “”” ❶ 에이전트 카드를 만드는 함수””” # ❷ 인사를 하는 간단한 스킬 greeting_skill = AgentSkill( id=”basic_greeting”, name=”Basic Greeting”, description=”간단한 인사와 기본적인 대화를 제공합니다”, tags=[“greeting”, “hello”, “basic”], examples=[“안녕하세요”, “hello”, “hi”, “고마워요”], input_modes=[“text”], output_modes=[“text”], ) # ❸ 에이전트 카드 객체 생성 agent_card = AgentCard( name=”Basic Hello World Agent”, description=”A2A 프로토콜을 학습하기 위한 기본적인 Hello World 에이전트입니다”, url=”http://localhost:9999/”, version=”1.0.0”, default_input_modes=[“text”], default_output_modes=[“text”], capabilities=AgentCapabilities(streaming=True), skills=[greeting_skill], supports_authenticated_extended_card=False, ) return agent_card def main(): agent_card = create_agent_card() print(agent_card.model_dump_json()) if __name__ == “__main__”: main()
❶ create_agent_card( ) 함수는 에이전트 카드를 생성하여 반환하는 함수입니다. AgentSkill
과 AgentCard 클래스의 객체를 생성하고 반환합니다.
❷ AgentSkill은 pydantic의 BaseModel을 상속받은 클래스입니다. id, name, description, tags가 필수값이고, examples, inputModes, outputModes는 선택값입니다. 뒤에 s로 끝나는 변수들은 문자열의 리스트인 list[str] 형태인 데이터를 넣으면 됩니다. 나머지는 모두 문자열을 넣으면 됩니다. id는 에이전트의 스킬을 구분하기 위한 유니크 값이며, name은 스킬명, description은 스킬 설명입니다. 스킬 설명은 클라이언트가 해당 스킬을 언제 사용하는지 알려주는 용도입니다. examples는 스킬을 실행했을 때 나오는 예시들입니다. inputModes와 outputModes는 각각 입력과 출력 시 어떤 미디어 타입을 사용할 수 있는지 나타내는 겁니다.
❸ AgentCard는 에이전트의 메타데이터와 스킬을 정의하는 클래스입니다. 에이전트를 사람에 비유한다면 AgentCard는 명함이라고 생각하면 편합니다(스킬이 있다는 부분에서는 포켓몬 카드와 유사한 부분도 있습니다). 명함에는 이름, 직함, 연락처, 소속, 하는 일 등 다양한 정보가 들어 있습니다. 마찬가지로 AgentCard에는 에이전트의 이름이 무엇인지(name), 무슨 역할을 하는지(description), 어디가면 찾을 수 있는지 (url) 같은 정보가 들어 있습니다.
url은 에이전트가 실행되고 있는 엔드포인트 주소를 나타내는 문자열로, 클라이언트가 에이전트와 통신할 때 사용합니다.
version은 에이전트의 버전을 나타내는 문자열입니다. defaultInputModes와 defaultOutputModes는 각각 에이전트가 기본적으로 지원하는 입력과 출력 미디어 타입을 나타내는 문자열 리스트입니다. 예시에서는 둘 다 [“text”]로 설정되어 있어 텍스트 형식의 입출력을 기본으로 지원함을 나타냅니다.
capabilities는 AgentCapabilities 객체로, 에이전트가 지원하는 추가 기능을 정의합니다. 예제에서는 streaming=True로 설정되어 스트리밍을 지원함을 나타냅니다. skills는 에이전트가 제공하는 스킬들의 리스트입니다. 인사하는 greeting_skill을 스킬로 넣어주었습니다.
supportsAuthenticatedExtendedCard는 불리언 값으로, 에이전트가 인증된 확장 카드를 지원하는지 여부를 나타냅니다. False로 설정되어 있으면 인증된 확장 카드를 지원하지 않음을 의미합니다.
코드의 실행 결과는 다음과 같습니다.
Use a different Browser
{“capabilities”:{“extensions”:null,”pushNotifications”:null, ”stateTransitionHistory”:null,”streaming”:true},”defaultInputModes”:[“text”],”defaultOutputModes”:[“text”],”description”:”A2A 프로토콜을 학습하기 위한 기본적인 Hello World 에이전트입니다”,”documentationUrl”:null,”iconUrl”:null,”name”:”Basic Hello World Agent”,”provider”:null,”security”:null,”securitySchemes”:null,”skills”:[{“description”:”간단한 인사와 기본적인 대화를 제공합니다”,”examples”:[“안녕하세요”,”hello”,”hi”,”고마워요”],”id”:”basic_greeting”,”inputModes”:[“text”],”name”:”Basic Greeting”,”outputModes”:[“text”],”tags”:[“greeting”,”hello”,”basic”]}],”supportsAuthenticatedExtendedCard”:false,”url”:”http://localhost:9999/”,”version”:”1.0.0”}
https://jsonformatter.org/json-pretty-print 같은 웹사이트에서 붙여넣고 정렬이 된 형태로 출력하면 데이터를 조금 더 편하게 확인하실 수 있습니다. 에이전트 카드에 꽤나 많은 정보가 있다는 것을 알게 되셨을 겁니다.
데이터가 너무 많다고 판단된다면, 필수 값만 넣고 테스트해보시는 것도 좋을 것 같습니다. 다음으로 에이전트를 실행하는 Agent Executor 관련 코드를 작성해봅시다.
Agent Executor 작성
Agent Executor는 A2A 서버에서 받은 유저의 메시지를 에이전트에 전달 및 실행하고 결괏값을 돌려주는 역할을 하는 클래스입니다. 메시지를 전달하고 에이전트를 호출하는 execute 메서드와 에이전트가 처리할 수 없는 요청인 경우에 받은 요청을 취소하는 cancel( ) 메서드가 있습니다. 예제는 단순하게 인사를 하는 에이전트이므로 취소 기능은 구현하지 않습니다. HelloAgent를 만들고 HelloAgent 실행을 위한 HelloAgentExecutor에 관한 코드를 작성해봅시다.
Use a different Browser
from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate from a2a.server.agent_execution import AgentExecutor, RequestContext from a2a.server.events import EventQueue from a2a.types import Message from a2a.utils import new_agent_text_message class HelloAgent: “”” ❶ 랭체인과 OpenAI를 사용한 간단한 Hello World 에이전트.””” def __init__(self): self.chat = ChatOpenAI( model=”gpt-4.1-mini”, temperature=0.7, ) self.prompt = ChatPromptTemplate.from_messages( [ ( “system”, “””당신은 친절한 Hello World 에이전트입니다. 사용자와 간단한 대화를 나누고, 인사와 기본적인 질문에 답변합니다. 당신의 목표는 사용자에게 친근하고 도움이 되는 경험을 제공하는 겁니다.”””, ), (“user”, “{message}”), ] ) async def invoke(self, user_message: str) -> str: “””❷ 유저 메시지를 처리하고 응답을 생성합니다.””” chain = self.prompt | self.chat response = await chain.ainvoke({“message”: user_message}) return response.content class HelloAgentExecutor(AgentExecutor): “””❸ 간단한 Hello World 에이전트의 Executor””” def __init__(self): self.agent = HelloAgent() async def execute( self, context: RequestContext, event_queue: EventQueue, ) “””❹ 요청을 처리하고 응답을 생성합니다.””” # 유저 메시지를 추출 message = context.message for part in message.parts: if part.root.text: user_message = part.root.text # 에이전트 실행 result = await self.agent.invoke(user_message) # ❺ 응답 메시지를 생성하고 이벤트 큐에 추가 await event_queue.enqueue_event(new_agent_text_message(result)) async def cancel( self, context: RequestContext, event_queue: EventQueue, ) “””요청을 취소””” # 취소 기능은 지원하지 않음 error_msg = “취소 기능은 지원되지 않습니다. Hello 에이전트는 즉시 응답합니다.” error_message = Message( role=”agent”, parts=[{“type”: “text”, “text”: error_msg}], messageId=”cancel_error”, ) event_queue.enqueue_event(error_message)
이 코드는 A2A 프로토콜을 사용하여 랭체인과 OpenAI를 활용한 간단한 대화형 에이전트를 구현한 겁니다.
❶ HelloAgent 클래스는 랭체인과 OpenAI를 활용하여 실제 대화 로직을 처리하는 핵심 에이전트입니다. 이 클래스는 독립적으로 동작하며, A2A 프로토콜과는 직접적인 의존성이 없습니다. 생성자에서 오픈AI의 GPT-4.1-mini 모델을 사용하는 ChatOpenAI 인스턴스를 생성하고, temperature를 0.7로 설정하여 적당한 창의성을 가진 응답을 생성하도록 합니다. 또한 시스템 프롬프트를 통해 에이전트의 페르소나와 행동 지침을 정의합니다.
❷ invoke( ) 메서드는 비동기 메서드로, 사용자의 메시지를 받아 AI 응답을 생성하는 실제 처리 로직을 담당합니다. 랭체인의 체인 구조를 사용하여 프롬프트 템플릿과 ChatOpenAI 모델을 연결하고, 사용자 메시지를 처리합니다. 이 메서드는 순수하게 문자열을 입력받아 문자열을 반환하는 단순한 인터페이스를 가지고 있어, A2A 프로토콜과 독립적으로 테스트하거나 사용할 수 있습니다.
❸ HelloAgentExecutor 클래스는 AgentExecutor를 상속받아 A2A 프로토콜과 HelloAgent를 연결하는 어댑터 역할을 합니다. 이 클래스는 A2A 서버와의 통신을 처리하고, HelloAgent의 기능을 A2A 프로토콜에 맞게 래핑합니다. 생성자에서 HelloAgent 인스턴스를 생성하여 내부적으로 관리합니다.
❹ execute 메서드는 A2A 프로토콜의 요청을 처리하는 핵심 메서드입니다. RequestContext에서 사용자 메시지를 추출하고, 이를 HelloAgent의 invoke( ) 메서드에 전달하여 응답을 생성합니다. 메시지는 복잡한 구조를 가질 수 있으므로, message.parts를 순회하면서 텍스트 타입의 파트를 찾아 추출합니다. 이는 A2A 프로토콜이 멀티모달 메시지를 지원하기 때문입니다.
❺ 응답 메시지 생성 및 이벤트 큐 처리 부분에서는 HelloAgent로부터 받은 응답을 A2A 프로토콜에 맞는 메시지 형식으로 변환합니다. new_agent_text_message 헬퍼 함수를 사용하여 텍스트 응답을 적절한 Message 객체로 변환하고, 이를 EventQueue에 추가합니다. EventQueue는 비동기적으로 클라이언트에게 이벤트를 전달하는 메커니즘으로, 스트리밍이나 다중 응답을 지원할 수 있게 합니다.
A2A 서버 코드 작성 및 기동
A2A 서버를 시작하려면 에이전트, Agent Executor, 요청 핸들러(RequestHandler), 태스크 저장소(Task Store), 에이전트 카드가 필요합니다. 요청 핸들러와 태스크 저장소는 각각 A2A SDK에서 작성한 DefaultRequestHandler와 InMemoryTaskStore를 사용하겠습니다. 또한 FastAPI 서버를 생성하는 A2AFastAPIApplication도 SDK를 활용하여 작성했습니다. 코드를 작성하겠습니다.
Use a different Browser
import sys from pathlib import Path import uvicorn sys.path.append(str(Path(__file__).parent.parent)) from a2a.server.apps import A2AFastAPIApplication from a2a.server.request_handlers import DefaultRequestHandler from a2a.server.tasks import InMemoryTaskStore from a2a.types import AgentCapabilities, AgentCard, AgentSkill from basic_agent.agent_executor import HelloAgentExecutor # ❶ 임포트 def create_agent_card() -> AgentCard: ... 생략 return agent_card def main(): # ❷ 에이전트 카드 생성 agent_card = create_agent_card() port = 9999 host = “0.0.0.0” print(“Hello World 에이전트 서버 시작 중...”) print(f”서버 구동 : http://{host}:{port}”) print(f”Agent Card: http://{host}:{port}/.well-known/agent.json”) print(“이것은 A2A 프로토콜 학습을 위한 기본 예제입니다”) # ❸ 기본 요청 핸들러 생성 request_handler = DefaultRequestHandler( agent_executor=HelloAgentExecutor(), task_store=InMemoryTaskStore(), ) # ❹ A2A FastAPI 애플리케이션 생성 server = A2AFastAPIApplication( agent_card=agent_card, http_handler=request_handler, ) # ❺ 서버 빌드 및 실행 app = server.build() uvicorn.run(app, host=host, port=port) if __name__ == “__main__”: main()
❶ server.py에서는 agent_executor.py에 있는 HelloAgentExecutor 클래스를 임포트해서 사용합니다. from import 구문을 사용할 때 클래스의 상대 경로 혹은 절대 경로를 지정해야 합니다. 상대 경로는 . 기호 혹은 .. 기호로 현재 패키지 혹은 상위 패키지를 설정할 수 있습니다. 상대 경로는 실수하기 쉬워서 제 개인적으로는 절대 경로를 선호합니다. 절대 경로를 사용하려면 기준이 있어야 하는데 이때 기준이 되는 것이 PYTHONPATH입니다. 파이썬에서는 PYTHONPATH에 있는 디렉터리를 기준으로 파이썬 파일을 절대 경로로 찾을 수 있습니다. 예제에서는 ‘sys.path.append(str(Path(__file__).parent.parent))’ 코드를 사용하여 PYTHONPATH에 chapter8/a2a 패키지 경로를 추가했습니다. HelloAgentExecutor 클래스는 chapter8/a2a/basic_agent/agent_executor.py에 있으니 여기서 chapter8/a2a를 빼고 basic_agent.agent_executor로 임포트할 수 있습니다. PYTHONPATH를 .env 파일에 설정해두는 것도 방법입니다.
❷ 에이전트 카드 생성 부분에서는 create_agent_card 함수를 호출하여 AgentCard 인스턴스를 생성합니다. AgentCard가 에이전트의 메타데이터와 기능을 정의하는 중요한 객체라는 것을 이제 알 수 있습니다. 에이전트 카드의 정보는 나중에 /.well-known/agent.json 엔드포인트를 통해 외부에 노출되어, A2A 프로토콜을 지원하는 클라이언트가 에이전트를 자동으로 발견하고 연결할 수 있게 합니다.
❸ 기본 요청 핸들러 생성 부분은 DefaultRequestHandler 인스턴스를 생성합니다. 이 핸들러는 HTTP 요청을 받아서 HelloAgentExecutor에게 전달하는 중간 계층 역할을 합니다. InMemoryTaskStore는 비동기 작업들을 메모리에서 관리하는 저장소로, 각 요청의 상태를 추적하고 취소 기능을 지원하는 데 사용됩니다. DefaultRequestHandler는 A2A 프로토콜의 표준 요청/응답 패턴을 구현하며, 에러 처리와 작업 관리를 자동으로 처리합니다.
❹ A2A FastAPI 애플리케이션 생성 부분에서는 A2AFastAPIApplication 인스턴스를 생성합니다. 이 클래스는 FastAPI 프레임워크를 기반으로 A2A 프로토콜에 필요한 모든 엔드포인트를 자동으로 설정합니다. agent_card를 전달하여 에이전트 정보를 노출하고, http_handler를 통해 실제 요청 처리 로직을 연결합니다. 이 애플리케이션은 A2A 프로토콜의 표준 엔드포인트들을 자동으로 생성하므로, 개발자가 직접 라우팅을 구현할 필요가 없습니다. 소스 코드에는 엔드포인트가 딱 2개 있습니다. 하나는 RPC 요청을 처리하는 엔드포인트로 POST “/”입니다. 다른 하나는 에이전트 카드를 확인하는 엔드포인트로 GET “/.well-known/agent.json”이 됩니다. RPC 요청은 다름 아닌 JSON-RPC 요청을 하기 위한 엔드포인트입니다.
❺ 서버 빌드 및 실행 부분은 설정된 애플리케이션을 실제로 구동합니다. build( ) 메서드를 호출하여 FastAPI 애플리케이션 인스턴스를 생성하고, uvicorn을 사용하여 ASGI 서버를 실행합니다. 서버는 0.0.0.0 호스트와 9999 포트에서 실행되어 모든 네트워크 인터페이스에서 접속을 받을 수 있게 설정됩니다. 이렇게 구동된 서버는 A2A 프로토콜을 완벽하게 지원하며, 클라이언트의 요청을 받아 HelloAgent를 통해 응답을 생성할 준비가 완료됩니다.
서버를 실행하고 브라우저에서 http://0.0.0.0:9999/.well-known/agent.json으로 접속해봅시다. 다음과 같은 에이전트 카드 정보를 확인할 수 있습니다.
A2A 인스펙터를 활용한 테스트
A2A 서버가 있으니 이제 클라이언트가 있으면 테스트를 해볼 수 있습니다. MCP 인스펙터처럼 A2A도 A2A 서버를 테스트해볼 수 있는 인스펙터가 있습니다. 인스펙터를 활용하여 에이전트를 테스트해보려면 파이썬, uv, Node.js 3가지 환경이 준비되어 있어야 합니다. uv의 설치 및 간단한 사용은 부록에서 다루고 있으니 혹시나 설치하지 않았다면 확인해보시길 바랍니다. Node.js는 https://nodejs.org/ko/download에서 패키지를 내려받아서 설치하시거나, brew(맥OS), Chocolatey(윈도우)로 설치할 수도 있고, nvm, fnm, Volta(맥OS//윈도우) 등의 패키지 매니저를 사용하여 설치할 수도 있습니다.
https://github.com/a2aproject/a2a-inspector에 접속하셔서 저장소 파일을 내려받아주세요. 가장 추천하는 방법은 터미널에서 ‘git clone https://github.com/a2aproject/a2a-inspector.git’으로 저장소를 복사해오는 겁니다.
저장소 복사 혹은 다운로드 후에는 의존성을 설치해야 합니다. 다음의 명령을 터미널에서 차례대
로 입력하여 의존성을 설치합시다.
Use a different Browser
cd a2a-inspector uv sync cd frontend npm install
uv sync는 파이썬 의존성을 파이썬 가상 환경과 동기화하는 명령어입니다. uv에 관한 설명은 부록에 있으니 확인해보시길 바랍니다. npm install은 Node.js 의존성을 설치하는 명령어입니다.
의존성 패키지 설치가 끝났다면, 프론트엔드 파일을 빌드하고 서버를 시작해야 합니다. 터미널에 접속하고 다음의 명령을 차례대로 입력합시다.
Use a different Browser
cd frontend npm run build # 프론트엔드 파일 빌드 cd .. cd backend uv run app.py
npm run build는 프론트엔드에서 사용할 파일을 빌드하는 데 사용합니다. 실행하면 public/script.js 파일이 생성됩니다. uv run app.py는 app.py를 실행하고 서버가 실행됩니다. 다음과 같은 내용이 터미널에 나타나게 될 겁니다.
Use a different Browser
INFO: Will watch for changes in these directories: [‘/Users/ gyus/VSCode/a2a-inspector/backend’] INFO: Uvicorn running on http://127.0.0.1:5001 (Press CTRL+C to quit) INFO: Started reloader process [6773] using StatReload INFO: Started server process [6776] INFO: Waiting for application startup. INFO: Application startup complete. INFO: 127.0.0.1:57513 - “GET /socket.io/?EIO=4&transport=polling&t=55m7n3 4v HTTP/1.1” 200 OK
브라우저를 켜서 http://127.0.0.1:5001에 접속합시다. 그러면 다음과 같은 화면이 나오고 테스트를 해볼 수 있습니다.
Enter Agent URL이라고 되어 있는 곳에 우리가 만든 서버의 URL인 http://0.0.0.0:9999를 넣고 [Connect]를 눌러보세요. 그러면 다음과 같이 Agent Card 정보가 나옵니다.
Connect를 눌러서 에이전트와 연결되었으므로 Chat을 사용하여 메시지를 남겨 볼 수 있습니다.
메모리 기능이 없기에 이전 대화를 기억하지 못하는 것을 알 수 있습니다. 여기까지 A2A 인스펙터를 활용하여 A2A를 테스트해보았습니다.
A2A 클라이언트 코드 작성 및 테스트
A2A 인스펙터를 사용하면 A2A 서버를 빠르게 테스트해볼 수 있습니다. 이번에는 직접 클라이언트 코드를 작성하여 A2A를 테스트해봅시다. 스트리밍을 사용하지 않는 요청과 스트리밍을 사용하는 요청 두 가지 모두 테스트합니다. 코드는 다음과 같습니다.
Use a different Browser
import asyncio from uuid import uuid4 from typing import Any, Optional import httpx from a2a.client import A2ACardResolver from a2a.client.client_factory import ClientFactory from a2a.client.client import ClientConfig from a2a.types import Message from a2a.utils import get_message_text def create_user_message(text: str, message_id: Optional[str] = None) -> Message: “””A2A 사용자 메시지 생성 함수.””” return Message( role=”user”, parts=[{“kind”: “text”, “text”: text}], messageId=message_id or uuid4().hex, ) async def test_basic_agent(): “””헬로 월드 A2A 에이전트 테스트 함수.””” base_url = “http://localhost:9999” print(“Basic Hello World A2A Agent 테스트 시작...”) print(f”서버 URL: {base_url}”) print(“-” * 50) async with httpx.AsyncClient() as httpx_client: try: # ❶ A2A 카드 리졸버 생성 resolver = A2ACardResolver( httpx_client=httpx_client, base_url=base_url, ) # ❷ 에이전트 카드 가져오기 print(“에이전트 카드를 가져오는 중...”) agent_card = await resolver.get_agent_card() print(f”에이전트 이름: {agent_card.name}”) print(f”에이전트 설명: {agent_card.description}”) print(f”지원 스킬: {[skill.name for skill in agent_card.skills]}”) print() # ❸ A2A 클라이언트 생성 non_streaming_config = ClientConfig(httpx_client=httpx_client, streaming=False) non_streaming_factory = ClientFactory(non_streaming_config) non_streaming_client = non_streaming_factory.create(agent_card) streaming_config = ClientConfig(httpx_client=httpx_client, streaming=True) streaming_factory = ClientFactory(streaming_config) streaming_client = streaming_factory.create(agent_card) # ❹ 테스트 메시지 목록 test_messages = [ “안녕하세요”, “날씨가 어때요?”, “고마워요”, “이름이 뭔가요?”, “오늘 기분이 어때요?”, ] # ❺ 비스트리밍 메시지 테스트 print(“=== 비스트리밍 메시지 테스트 ===”) for i, message_text in enumerate(test_messages, 1): print(f”\n{i}. 사용자: {message_text}”) # ❻ 사용자 메시지 생성 user_message = create_user_message(message_text) request = SendMessageRequest( id=str(uuid4()), params=MessageSendParams(message=user_message) ) # ❼ 비스트리밍 메시지 전송 async for event in non_streaming_client.send_message(user_message): if isinstance(event, Message): response_text = get_message_text(event) print(response_text) break # 첫 번째 Message 응답만 처리 print(“\n” + “=” * 50) # ❽ 스트리밍 메시지 테스트 print(“=== 스트리밍 메시지 테스트 ===”) for i, message_text in enumerate( test_messages[:3], 1 ): # 3개의 메시지만 스트리밍 테스트 print(f”\n{i}. 사용자: {message_text}”) # ❾ 사용자 메시지 생성 user_message = create_user_message(message_text) # ❿ 스트리밍 메시지 전송 print(“ 에이전트 (스트리밍): “, end=””, flush=True) async for event in streaming_client.send_message(user_message): if isinstance(event, Message): response_text = get_message_text(event) print(response_text, end=””, flush=True) print() print(“\n테스트 완료!”) except Exception as e: print(f”테스트 중 오류 발생: {e}”) print(“서버가 실행 중인지 확인해주세요.”) print(“서버 실행: python basic_agent/__main__.py”) async def main(): “””Main function to run the test.””” await test_basic_agent() if __name__ == “__main__”: asyncio.run(main())
❶ A2A 카드 리졸버 생성 부분에서는 A2ACardResolver 인스턴스를 생성합니다. 이 리졸버는 에이전트의 카드 정보를 가져오는 역할을 담당합니다. httpx_client를 전달하여 비동기 HTTP 통신을 처리하고, base_url을 통해 에이전트 서버의 주소를 지정합니다. 리졸버는 표준화된 경로인 /.well-known/agent.json에서 에이전트 카드를 자동으로 가져올 수 있게 해줍니다.
❷ 에이전트 카드 가져오기 부분은 get_agent_card( ) 메서드를 호출하여 실제로 서버에서 에이전트의 메타데이터를 가져옵니다. 이를 통해 에이전트의 이름, 설명, 지원하는 스킬 목록 등의 정보를 확인할 수 있습니다. 이 정보는 클라이언트가 에이전트의 기능을 이해하고 적절한 요청을 구성하는 데 필요합니다.
❸ A2A 클라이언트 생성 부분에서는 ClientConfig, ClientFactory 클래스를 사용하여 스트리밍을 위한 클라이언트와 비스트리밍을 위한 클라이언트 인스턴스를 각각 생성합니다. 이 클라이언트는 에이전트와의 모든 통신을 담당하며, 내부적으로 에이전트 카드를 자동으로 가져와서 설정합니다. 이렇게 생성된 클라이언트는 메시지 전송, 스트리밍, 작업 취소 등 A2A 프로토콜의 모든 기능을 사용할 수 있습니다.
❹ 테스트 메시지 목록 부분은 에이전트를 테스트하기 위한 다양한 메시지들을 정의합니다. 인사말부터 날씨, 감사 인사, 이름 질문, 기분 질문 등 다양한 유형의 대화를 테스트하여 에이전트가 올바르게 응답하는지 확인합니다.
❺ 비스트리밍 메시지 테스트 섹션에서는 일반적인 요청-응답 패턴을 테스트합니다. 비스트리밍 방식은 에이전트가 전체 응답을 생성한 후 한 번에 클라이언트에게 전송하는 방식으로, 간단한 대화나 짧은 응답에 적합합니다.
❻ 사용자 메시지 생성 부분은 create_user_message( ) 헬퍼 함수를 사용하여 A2A 프로토콜에 맞는 메시지 구조를 생성합니다. 메시지는 role, parts, messageId로 구성되며, parts는 텍스트, 이미지 등 다양한 콘텐츠를 담을 수 있는 배열입니다. 각 메시지에는 고유한 ID가 부여되어 추적할 수 있습니다.
❼ 메시지 전송 부분에서는 SendMessageRequest 객체를 생성하고 client.send_message( ) 메서드를 호출하여 실제로 메시지를 전송합니다. 응답은 SendMessageResponse 타입으로 받
으며, get_message_text 유틸리티 함수를 사용하여 응답에서 텍스트를 추출합니다.
❽ 스트리밍 메시지 테스트 섹션은 실시간 스트리밍 응답을 테스트합니다. 스트리밍 방식은 에이전트가 응답을 생성하는 동안 부분적으로 클라이언트에게 전송하는 방식으로, 긴 응답이나 실시간 상호작용이 필요한 때 유용합니다.
❾ 스트리밍용 사용자 메시지 생성도 동일하게 create_user_message() 함수를 사용하여 메시지를 생성합니다.
❿ 스트리밍 메시지 전송 부분에서는 send_message( ) 메서드를 호출하여 스트리밍 응답을 받습니다. async for 루프를 사용하여 스트림에서 각 부분 응답을 순차적으로 받아 처리합니다. 각 스트림 응답이 도착할 때마다 즉시 화면에 출력하여 실시간으로 응답이 생성되는 것을 확인할 수 있습니다.
서버가 구동된 상태에서 클라이언트 코드를 실행하면 다음과 같은 결괏값을 얻을 수 있습니다. 화려한 UI는 없는 투박한 클라이언트이지만, 직접 만든 클라이언트로 한 번 테스트해보시길 바랍니다.
Use a different Browser
Basic Hello World A2A Agent 테스트 시작... 서버 URL: http://localhost:9999 -------------------------------------------------- 에이전트 카드를 가져오는 중... 에이전트 이름: Basic Hello World Agent 에이전트 설명: A2A 프로토콜을 학습하기 위한 기본적인 Hello World 에이전트입니다 지원 스킬: [‘Basic Greeting’] === 비스트리밍 메시지 테스트 === 1. 사용자: 안녕하세요 안녕하세요! 만나서 반가워요. 오늘 기분은 어떠세요? 2. 사용자: 날씨가 어때요? 안녕하세요! 저는 실시간 날씨 정보를 제공하지는 못하지만 현재 계신 지역의 날씨를 알려주면 도움이 될 만한 팁이나 정보를 드릴 수 있어요. 오늘 기분 좋은 하루 보내시길 바랄게요! 3. 사용자: 고마워요 천만에요! 도움이 필요하면 언제든지 말씀해주세요. 좋은 하루 보내세요! 4. 사용자: 이름이 뭔가요? 안녕하세요! 저는 Hello World 에이전트입니다. 만나서 반가워요! 당신의 이름은 무엇인가요? 5. 사용자: 오늘 기분이 어때요? 안녕하세요! 저는 항상 기분이 좋아요. 오늘은 어떻게 지내고 계신가요? ================================================== === 스트리밍 메시지 테스트 === 1. 사용자: 안녕하세요 에이전트 (스트리밍): 안녕하세요! 만나서 반가워요. 어떻게 도와드릴까요? 2. 사용자: 날씨가 어때요? 에이전트 (스트리밍): 안녕하세요! 현재 날씨 정보는 실시간으로 확인할 수 없지만, 제가 도 와드릴 수 있는 다른 것이 있나요? 예를 들어 간단한 인사나 다른 질문에 답해드릴 수 있어요! 3. 사용자: 고마워요 에이전트 (스트리밍): 천만에요! 도움이 필요하면 언제든지 말씀해주세요. 좋은 하루 보내 세요! 테스트 완료!
A2A 프로토콜의 요청-응답 처리 흐름
예제에 나온 A2A 프로토콜의 흐름을 다이어그램으로 그리면 다음과 같습니다.
1-2단계 : 클라이언트 요청 및 요청 전달
TestClient가 HTTP POST 요청으로 유저가 작성한 메시지를 서버에 전송합니다. test_client.py에서 A2AClient를 통해 요청이 시작되며, 서버의 A2AFastAPIApplication이 이를 받아 DefaultRequestHandler로 요청을 라우팅합니다.
3-4단계 : 에이전트 실행 계층
DefaultRequestHandler가 요청을 처리하여 HelloAgentExecutor의 execute( ) 메서드를 호출합니다. HelloAgentExecutor는 요청 처리 시작을 담당하며, 실제 비즈니스 로직을 위해 HelloAgent의 invoke( ) 메서드를 호출합니다. 이 과정에서 사용자 메시지가 추출되어 전달됩니다.
5-6단계 : AI 모델 호출
HelloAgent가 ChatPromptTemplate과 사용자 메시지를 결합하여 OpenAI API를 호출합니다. 시스템 프롬프트와 사용자 메시지가 적절히 포맷팅되어 GPT-4 모델에 전달되고, AI가 생성한 응답이 반환됩니다.
7-8단계 : 응답 처리
OpenAI로부터 받은 AI 응답은 역순으로 전달됩니다. HelloAgent가 응답 텍스트를 반환하면, HelloAgentExecutor는 이를 A2A 메시지 형식으로 변환하여 EventQueue에 추가합니다. 이는 new_agent_text_message 헬퍼 함수를 통해 적절한 포맷으로 변환됩니다.
9-10단계 : 클라이언트 응답
EventQueue에 있는 응답이 DefaultRequestHandler를 거쳐 서버로 전달되고, 최종적으로 JSON 형식의 HTTP 응답으로 TestClient에게 반환됩니다. 클라이언트는 SendMessageResponse 객체로 응답을 받아 처리합니다.
A2A 예제 자체는 매우 간단한 예제였습니다만, A2A 프로토콜 자체가 구현해야 하는 내용이 많아서 다소 복잡한 부분이 있습니다. 이를 그림으로 표현하면 다음과 같습니다.
마무리
구글이 50개 이상의 기술 파트너와 함께 개발한 A2AAgent-to-Agent 프로토콜을 자세히 살펴보았습니다. A2A는 AI 에이전트들이 분산된 환경에서 서로를 발견하고, 표준화된 방식으로 통신하며, 복잡한 작업을 협업하여 수행할 수 있도록 하는 개방형 프로토콜입니다.
A2A의 핵심은 에이전트 간의 상호운용성을 보장하는 겁니다. 에이전트 카드를 통해 각 에이전트는 자신의 기능과 인터페이스를 표준화된 방식으로 노출하고, 클라이언트는 이를 통해 에이전트를 자동으로 발견하고 연결할 수 있습니다. 태스크 기반의 비동기 처리 메커니즘은 오래 걸리는 작업도 효율적으로 관리할 수 있게 하며, 스트리밍 지원을 통해 실시간 상호작용도 가능합니다.
실습을 통해 우리는 A2A 서버와 클라이언트를 직접 구현해보았습니다. 간단한 인사 에이전트였지만, 에이전트 카드 작성부터 에이전트 실행자Agent Executor 구현, FastAPI 기반 서버 구축, 그리고 클라이언트를 통한 비스트리밍 및 스트리밍 통신까지 A2A 프로토콜의 전체 흐름을 경험할 수 있었습니다. A2A 인스펙터를 활용한 테스트 과정에서는 프로토콜이 실제로 어떻게 동작하는지 시각적으로 확인할 수 있었습니다.
MCP와 A2A를 비교해보면, 두 프로토콜은 서로 다른 문제를 해결하면서도 AI 애플리케이션의 복잡도를 줄이고 상호운용성을 높인다는 공통된 목표를 가지고 있습니다. MCP가 AI 모델과 도구 간의 연결을 표준화한다면, A2A는 AI 에이전트 간의 협업을 표준화합니다. 현재 MCP는 이미 실무에서 사용 가능한 수준으로 안정화되었지만, A2A는 아직 초기 단계에 있어 2025년 말 SDK 완성을 목표로 개발이 진행 중입니다.
앞으로 AI 에이전트 생태계가 성숙해지면서 이러한 표준 프로토콜의 중요성은 더욱 커질 겁니다. 개발자들은 당장은 MCP의 도입을 적극 검토하면서도 A2A의 발전 상황을 지속적으로 모니터링할 필요가 있습니다. 언젠가는 두 프로토콜이 통합되거나 상호 보완적으로 사용되는 날이 올 수도 있고, 둘 중 하나가 사실상의 표준이 될 수도 있습니다. 어떤 방향으로 발전하든, 이러한 프로토콜을 이해하고 활용할 수 있는 개발자들은 더욱 강력하고 유연한 AI 애플리케이션을 구축할 수 있을 겁니다.
- A2A GitHub 저장소: https://github.com/google-a2a/A2A
- A2A 프로토콜 스펙: https://google-a2a.github.io/A2A/latest/specification/
- 에이전트 카드: https://google-a2a.github.io/A2A/latest/specification/#5-agent-discovery-the-agent-card
- A2A MCP: AI 프로토콜 진화의 승자 예측 https://a2aprotocol.ai/blog/a2a-mcp-ai-protocol-winner-ko
- A2A 파이썬 SDK : https://github.com/google-a2a/a2a-python
- A2A 타입스크립트 SDK : https://github.com/google-a2a/a2a-js
- 요즘 AI 에이저트 깃허브 저장소 : https://github.com/wapj/yozm-ai-agent/
- A2A 인스펙터 : https://github.com/a2aproject/a2a-inspector
