첫 번째 A2A 호환 에이전트 만들기
작은 에이전트를 서버로 세워 보기
지금까지는 A2A의 구성 요소를 개념으로 살펴보았다. 이번 장에서는 그 개념을 실제 서버 형태로 묶어 본다. 목표는 거창한 멀티 에이전트 시스템이 아니다. 짧은 글을 요약하거나 간단한 질문에 답하는 원격 에이전트 하나를 만들고, 외부 클라이언트가 A2A 방식으로 발견하고 호출할 수 있게 만드는 것이다.
여기서 원격 에이전트는 "다른 팀에 있는 전문가"와 비슷하다. 클라이언트는 그 전문가의 내부 노트, 도구, 작업 방식을 직접 보지 않는다. 대신 명함을 보고 어떤 일을 맡길 수 있는지 확인한 뒤, 표준 양식으로 업무를 요청한다. A2A 서버는 바로 이 명함과 업무 접수 창구를 제공한다.
첫 번째 A2A 호환 에이전트에 필요한 최소 구성은 네 가지다.
실제 일을 처리하는 내부 에이전트 로직
외부에 공개할 Agent Card
A2A 요청을 내부 로직으로 연결하는 실행기
HTTP 서버와 태스크 저장소
Python SDK를 기준으로 설명하되, 이 구조는 JavaScript, Java, Go, .NET SDK로 옮겨도 크게 달라지지 않는다. 언어가 바뀌어도 "카드 작성 → 요청 처리 → 결과 반환 → 서버 노출"이라는 뼈대는 같다.
AI 생성 이미지
예제: 짧은 요약 에이전트
예제로 만들 에이전트는 사용자가 보낸 텍스트를 간단히 요약한다. 실제 서비스라면 LLM, 검색 도구, 사내 문서 저장소, MCP 서버 등을 사용할 수 있다. 하지만 첫 구현에서는 복잡한 모델 연동보다 A2A의 경계를 이해하는 것이 중요하다.
내부 로직은 다음처럼 단순하게 생각할 수 있다.
class SummaryAgent:
def invoke(self, text: str) -> str:
if not text.strip():
return "요약할 텍스트가 비어 있습니다."
sentences = text.replace("n", " ").split(".")
first = sentences[0].strip()
if not first:
return "핵심 문장을 찾지 못했습니다."
return f"요약: {first}."이 코드는 지능적인 요약기는 아니다. 하지만 중요한 점은 따로 있다. A2A는 내부 로직이 규칙 기반인지, LLM 기반인지, LangGraph 그래프인지, ADK 에이전트인지 강제하지 않는다. 외부와 맞닿는 표면만 A2A 규칙에 맞추면 된다. 즉, 기존에 만들어 둔 에이전트도 얇은 어댑터를 씌워 A2A 서버로 노출할 수 있다.
Agent Card: 이 에이전트가 무엇을 할 수 있는가
A2A 클라이언트는 서버를 호출하기 전에 Agent Card를 읽는다. Agent Card는 에이전트의 명함이자 메뉴판이다. 이름, 설명, 서버 주소, 지원 입력·출력 형식, 인증 방식, 제공 스킬을 담는다.
Python SDK의 공식 흐름에서도 먼저 AgentSkill과 AgentCard를 정의한 뒤 서버에 연결한다. SDK 패키지는 a2a-sdk로 배포되며, HTTP 서버 기능을 함께 쓰려면 관련 extra를 설치할 수 있다.1
pip install "a2a-sdk[http-server]" uvicorn요약 에이전트의 Agent Card는 다음과 같은 의미를 가져야 한다. 아래 코드는 Python SDK의 최신 흐름을 기준으로 한 예시이며, 핵심은 "스킬을 선언하고, 그 스킬을 포함한 공개 카드를 만든다"는 점이다.
summary_skill = AgentSkill(
id="summarize_text",
name="Text Summarization",
description="사용자가 보낸 짧은 텍스트를 핵심 문장 중심으로 요약합니다.",
input_modes=["text/plain"],
output_modes=["text/plain"],
tags=["summary", "text"],
examples=["다음 회의록을 요약해 줘", "이 문단의 핵심만 알려 줘"],
)
agent_card = AgentCard(
name="Simple Summary Agent",
description="짧은 텍스트를 간단히 요약하는 A2A 호환 에이전트입니다.",
version="0.1.0",
default_input_modes=["text/plain"],
default_output_modes=["text/plain"],
capabilities=AgentCapabilities(streaming=False),
supported_interfaces=[
AgentInterface(
protocol_binding="JSONRPC",
url="http://127.0.0.1:9999/",
)
],
skills=[summary_skill],
)코드의 정확한 import 경로와 필드 이름은 SDK 버전에 따라 조금씩 달라질 수 있다. 예를 들어 문서나 샘플에 따라 defaultInputModes처럼 JSON 필드 이름을 그대로 보여 주기도 하고, Python 코드에서는 default_input_modes처럼 snake_case 이름을 쓰기도 한다. 그러나 Agent Card에 담아야 하는 의도는 변하지 않는다. 클라이언트는 이 정보를 보고 "이 에이전트는 텍스트 요약을 할 수 있고, 텍스트 입력과 텍스트 출력을 지원하며, 이 URL로 요청하면 된다"고 판단한다.
공개 서버에서는 Agent Card에 적힌 접속 URL이 실제 접근 가능한 주소와 일치해야 한다. 로컬 개발 중에는 http://127.0.0.1:9999/처럼 둘 수 있지만, 운영 환경에서는 HTTPS 주소를 사용해야 한다. 인증이 필요하다면 카드에 인증 방식이나 요구 조건을 설명하되, 토큰이나 비밀 키 자체를 넣어서는 안 된다.
실행기: A2A 요청을 내부 로직으로 연결하기
Agent Card가 "무엇을 할 수 있는가"를 설명한다면, 실행기는 "요청이 들어왔을 때 실제로 무엇을 할 것인가"를 담당한다. Python SDK에서는 서버 측 실행 단위로 AgentExecutor 개념을 사용한다. 일반적으로 execute는 요청을 처리하고, cancel은 취소 요청을 처리한다.2
실행기는 식당의 접수 직원과 비슷하다. 손님이 주문서를 내면 주방에 전달하고, 완성된 음식을 다시 손님에게 내준다. 여기서 주문서는 A2A의 메시지 요청이고, 주방은 내부 에이전트 로직이며, 완성된 음식은 Message나 Artifact다.
단순 요약 에이전트라면 흐름은 다음과 같다.
클라이언트가
message/send또는 스트리밍 메서드로 텍스트를 보낸다.실행기가 요청 컨텍스트에서 사용자 텍스트를 꺼낸다.
내부
SummaryAgent.invoke()를 호출한다.결과를 A2A Message로 감싸 이벤트 큐에 넣는다.
필요하면 공식 산출물인 Artifact로도 저장한다.
예시는 다음처럼 표현할 수 있다.
class SummaryAgentExecutor(AgentExecutor):
def __init__(self):
self.agent = SummaryAgent()
async def execute(self, context, event_queue):
user_text = extract_text_from_context(context)
result = self.agent.invoke(user_text)
message = new_agent_text_message(result)
await event_queue.enqueue_event(message)
async def cancel(self, context, event_queue):
message = new_agent_text_message("요약 작업이 취소되었습니다.")
await event_queue.enqueue_event(message)extract_text_from_context와 new_agent_text_message는 설명을 위한 함수다. 실제 구현에서는 SDK가 제공하는 요청 컨텍스트 구조에서 Message의 Part를 읽고, SDK 유틸리티나 타입을 사용해 에이전트 응답 Message를 만들어야 한다. 앞 장에서 본 것처럼 Message는 여러 Part를 가질 수 있다. 첫 구현에서는 text/plain 하나만 지원하도록 제한하면 이해하기 쉽다.
결과가 짧은 답변이라면 Message만 반환해도 충분하다. 예를 들어 "요약: 이번 회의의 핵심은 출시 일정 조정입니다."라는 응답은 대화 메시지로 자연스럽다. 반면 보고서 파일, 구조화된 JSON, 긴 분석 결과처럼 나중에 저장하고 재사용해야 하는 결과는 Artifact로 다루는 편이 좋다.
요약 에이전트의 경우에도 운영 관점에서는 다음처럼 구분할 수 있다.
Message: "요약이 완료되었습니다." 또는 짧은 요약문
Artifact: 요약 결과 JSON, 원문 일부, 생성 시각, 사용한 요약 정책
이 구분을 해 두면 나중에 오케스트레이터가 결과를 다른 에이전트에 넘기기 쉽다. 다음 장에서 만들 클라이언트와 오케스트레이터는 바로 이 구조를 활용한다.
서버로 묶기: 요청 핸들러, 태스크 저장소, 라우트
이제 Agent Card와 실행기를 HTTP 서버에 연결한다. 공식 튜토리얼의 최소 서버 흐름은 대략 다음 순서다. Agent Card를 만들고, DefaultRequestHandler에 실행기와 태스크 저장소를 연결한 다음, Agent Card 라우트와 JSON-RPC 라우트를 ASGI 앱에 붙이고 Uvicorn으로 실행한다.3
request_handler = DefaultRequestHandler(
agent_executor=SummaryAgentExecutor(),
task_store=InMemoryTaskStore(),
)
routes = []
routes.extend(create_agent_card_routes(agent_card))
routes.extend(create_jsonrpc_routes(request_handler))
server_app = Starlette(routes=routes)
uvicorn.run(
server_app,
host="127.0.0.1",
port=9999,
)튜토리얼이나 SDK 버전에 따라 A2AStarletteApplication처럼 서버 앱을 만들어 주는 헬퍼를 보게 될 수도 있고, 위 예시처럼 라우트 팩토리를 사용해 Starlette나 FastAPI 앱에 직접 붙이는 방식을 보게 될 수도 있다. 어느 쪽이든 외부 클라이언트가 사용하는 주요 엔드포인트는 보통 두 가지다.
/.well-known/agent-card.json: Agent Card 발견/: A2A JSON-RPC 요청 처리
태스크 저장소는 A2A 서버가 작업 상태를 기억하는 공간이다. 개발 중에는 InMemoryTaskStore처럼 메모리 기반 저장소로 충분하다. 하지만 운영 환경에서는 서버 재시작, 장애 복구, 장기 실행 작업을 고려해야 한다. 이때는 SQLite, PostgreSQL, MySQL 같은 영속 저장소를 검토해야 한다. Python SDK는 HTTP 서버, gRPC, OpenTelemetry, SQL 저장소 같은 선택 기능을 제공하는 방향으로 설계되어 있다.1
요청에서 결과까지의 전체 흐름
이제 전체 흐름을 한 번에 정리해 보자.
먼저 클라이언트가 /.well-known/agent-card.json을 요청한다. 클라이언트는 카드에서 summarize_text 스킬을 발견한다. 사용자의 목표가 "이 문단을 요약해 줘"라면 이 에이전트가 적합하다고 판단할 수 있다.
다음으로 클라이언트는 Agent Card의 접속 정보가 가리키는 엔드포인트, 예제에서는 서버 루트 경로 /에 JSON-RPC 요청을 보낸다. 간단한 작업이면 message/send로 충분하다. 오래 걸리는 작업이거나 중간 진행 상황을 보여 주고 싶다면 이전 장에서 본 message/stream을 선택할 수 있다.
서버의 요청 핸들러는 JSON-RPC 요청을 받아 A2A 객체로 해석한다. 실행기는 요청 컨텍스트에서 텍스트를 꺼내 내부 요약 로직에 전달한다. 내부 로직은 결과 문자열을 만든다. 실행기는 이 결과를 Message로 감싸 이벤트 큐에 넣고, 핸들러는 이를 클라이언트가 이해할 수 있는 A2A 응답으로 반환한다.
이 과정에서 핵심은 내부 로직과 외부 프로토콜을 분리하는 것이다. 요약 알고리즘을 바꾸거나 LLM을 교체해도 Agent Card의 계약과 Message·Artifact 반환 방식이 유지된다면 클라이언트는 큰 변경 없이 같은 에이전트를 계속 사용할 수 있다.
다른 언어 SDK로 옮길 때도 같은 구조를 본다
Python으로 첫 서버를 만들었다면 다른 언어 SDK도 낯설지 않다. 공식 샘플 저장소는 Python뿐 아니라 JavaScript, Java, Go, .NET 예제를 언어별 디렉터리로 제공한다.4 파일 이름과 클래스 이름은 달라도 확인할 지점은 거의 같다.
Agent Card를 어디에서 정의하는가
스킬 목록을 어떻게 선언하는가
요청을 받는 실행기 또는 핸들러가 어디에 있는가
Task와 Message를 어떤 타입으로 표현하는가
결과를 Message 또는 Artifact로 어떻게 반환하는가
HTTP 서버가 Agent Card 경로와 JSON-RPC 경로를 어떻게 노출하는가
JavaScript에서는 Node.js 서버와 비동기 핸들러가 중심이 될 가능성이 높다. Java와 .NET에서는 인터페이스 구현과 의존성 주입 형태가 자연스럽다. Go에서는 명시적인 핸들러 함수와 구조체 조합이 자주 보인다. 하지만 A2A 관점에서 보면 모두 같은 일을 한다. 에이전트의 능력을 카드로 공개하고, 표준 요청을 받아, 내부 로직을 실행하고, 표준 결과로 돌려준다.
첫 구현에서 의도적으로 단순화할 것
첫 A2A 서버를 만들 때 모든 기능을 한 번에 넣으려 하지 않는 편이 좋다. 처음에는 다음처럼 범위를 좁히자.
입력은 텍스트 하나만 받는다.
출력도 텍스트 하나로 시작한다.
인증은 로컬 개발 중에는 생략하되, 운영에서는 반드시 추가한다.
태스크 저장소는 메모리로 시작하되, 장기 실행 작업 전에는 영속 저장소로 바꾼다.
스트리밍은 나중에 붙이고, 먼저
message/send흐름을 안정화한다.오류 메시지는 사용자가 이해할 수 있게 반환한다.
이 단순화는 기능을 포기하는 것이 아니다. A2A의 경계를 또렷하게 보기 위한 연습이다. 서버가 정상적으로 Agent Card를 제공하고, 클라이언트 요청을 받아 내부 로직을 실행하고, 결과를 Message로 반환한다면 이미 A2A 호환 에이전트의 핵심을 구현한 것이다.
구현 체크리스트
마지막으로 첫 번째 A2A 호환 에이전트를 만들 때 확인할 항목을 정리한다.
Agent Card의 접속 정보가 실제 서버 주소와 일치하는가?
스킬 설명이 클라이언트가 라우팅에 사용할 만큼 구체적인가?
지원 입력·출력 형식이 실제 구현과 일치하는가?
요청에서 텍스트 Part를 안전하게 추출하는가?
빈 입력, 너무 긴 입력, 지원하지 않는 형식을 처리하는가?
결과를 Message와 Artifact 중 어디에 담을지 기준이 있는가?
취소 요청이 들어왔을 때 최소한의 응답을 반환하는가?
로컬 메모리 저장소를 운영 환경에 그대로 쓰고 있지는 않은가?
Agent Card에는 비밀 값이 들어 있지 않은가?
이 장에서 만든 서버는 작다. 하지만 구조는 실제 시스템과 같다. A2A는 에이전트 내부를 대신 설계해 주지 않는다. 대신 서로 다른 에이전트가 만나는 경계를 안정적으로 정의한다. 이 경계를 한 번 서버로 구현해 보면 Agent Card, Task, Message, Artifact가 왜 필요한지 훨씬 선명해진다.
참고
1Official Python SDK for the Agent2Agent Protocol
이 노트는 보도·비평·교육·연구 목적으로 작성되었습니다. 저작권 문의가 있으시면 에서 알려주세요.