AGENTIC AI DESIGN PATTERNS · 15
내 에이전트가 남의 에이전트와 일하게 하라 — 에이전트 간 통신(A2A 프로토콜)
Agent Card·Discovery·Task 생명주기·JSON-RPC 2.0·SSE 스트리밍, FastAPI 서버와 httpx 클라이언트 구현, A2A vs MCP 분업 구조, CRM·ERP·Groupware 엔터프라이즈 적용까지
🟢 입문~중급 ⏱️ 약 22분 🔧 Python · FastAPI · httpx · SSE · Gemini 🏢 엔터프라이즈 🗓️ 최종 검토 2026-06-11
TL;DR — LangGraph로 만든 우리 에이전트와 CrewAI로 만든 파트너사 에이전트는 서로 말이 통하지 않습니다. A2A(Agent2Agent Protocol)는 서로 다른 프레임워크로 구축된 에이전트들이 표준화된 방식으로 통신·협업하게 하는 개방형 프로토콜입니다(HTTP + JSON-RPC 2.0). 핵심 흐름은 Agent Card 발견 → Task 전송 → 상태 추적 → Artifact 수신 이고, 통신 방식은 동기·폴링·SSE 스트리밍·웹훅 네 가지 중 작업 길이에 맞춰 고릅니다. MCP가 에이전트와 도구를 잇는 표준이라면, A2A는 에이전트와 에이전트를 잇는 표준 — 경쟁이 아니라 분업 관계입니다.
1. 왜 A2A인가 — 혼자 잘하는 에이전트의 다음 문제
지난 편에서 에이전트는 환각 대신 근거로 말하는 법(RAG)을 배웠습니다. 검색하고, 검증하고, 출처를 달고 — 혼자서는 꽤 똑똑해졌죠. 그런데 실제 업무는 한 에이전트로 끝나지 않습니다. 우리 회사의 LangGraph 영업 에이전트가 파트너사의 CrewAI 재고 에이전트에게 납기를 물어야 하고, 사내의 Google ADK 일정 에이전트에게 미팅을 잡아 달라고 해야 합니다. 그 순간 깨닫게 됩니다 — 이 에이전트들은 서로 말이 통하지 않는다는 것을요.
교재는 이 단절을 네 가지 문제로 정리합니다. 프레임워크 고립(LangGraph·CrewAI·ADK 각자 별도 통신 방식), 통합 비용(에이전트 연동마다 커스텀 개발), 발견 어려움(어떤 에이전트가 무엇을 할 수 있는지 모름), 보안 분산(시스템마다 다른 인증 방식). 에이전트가 N개, 연결할 상대가 M개면 커스텀 연동은 N×M개가 필요해집니다 — 시스템 통합(SI)의 오래된 악몽이 에이전트 세계에서 재현되는 거죠.
🔌 포인트-투-포인트 커스텀 연동
❌ 프레임워크 고립 — LangGraph·CrewAI·ADK 각자 별도 통신 방식
❌ 통합 비용 폭증 — 에이전트 쌍마다 커스텀 개발 (N×M)
❌ 발견 불가 — 상대가 무엇을 할 수 있는지 코드를 까봐야 안다
❌ 보안 파편화 — 연동마다 다른 인증 체계
🤝 A2A 표준 프로토콜
✅ 표준 프로토콜로 통합 — HTTP + JSON-RPC 2.0 하나로 통일
✅ 한 번 구현, 모두 연결 — A2A를 말하는 에이전트끼리는 즉시 협업
✅ Agent Card로 능력 명시 — 상대의 스킬·기능을 JSON으로 발견
✅ 표준 인증 체계 — OAuth 2.0 · API Key · mTLS 선언적 관리
그림 1. 커스텀 연동 vs A2A — 기존 4대 문제와 A2A의 해법이 1:1로 대응한다
교재의 비유가 이 패턴의 본질을 정확히 잡아줍니다. A2A는 국제 전화 표준입니다. 각 나라의 전화 시스템이 각 프레임워크(LangGraph·CrewAI·ADK)라면, 국제 전화 번호 체계가 A2A 프로토콜이고, 전화번호부가 Agent Card(누가 어떤 서비스를 제공하는지), 전화 교환원이 A2A 서버/클라이언트입니다. 국제 표준이 없었다면 나라끼리 일일이 전용선을 깔아야 했겠죠 — A2A가 없으면 에이전트 프레임워크 쌍마다 별도 연동을 개발해야 하는 것과 같습니다.
A2A(Agent2Agent Protocol)는 2025년 4월 Google이 공개하고 같은 해 6월 Linux Foundation 산하 프로젝트로 이관된 개방형 표준입니다. 교재 기준으로 Salesforce·SAP·Microsoft(클라우드/플랫폼), Atlassian·Box·ServiceNow(협업/문서), LangChain·Google·MongoDB(AI/개발) 등 주요 기업이 지원하고 있고, Microsoft는 Azure AI Foundry·Copilot Studio 통합 계획을, SAP과 Auth0는 플랫폼·에이전트에 A2A 지원 통합을 밝혔습니다. "에이전트의 공용어"가 되겠다는 야심이 허풍만은 아닌 셈입니다.
2. 개념과 오해 — A2A는 "HTTP API 하나 더"가 아니다
📌 한 줄 정의 — A2A(Agent2Agent Protocol)는 서로 다른 프레임워크로 구축된 AI 에이전트들이 표준화된 방식으로 통신하고 협업할 수 있게 해주는 개방형 프로토콜입니다. 핵심 구성요소는 다섯 가지 — Agent Card(에이전트의 디지털 신원) · Agent Discovery(발견 메커니즘) · Task(작업 요청·상태 관리) · Message(통신 메시지) · Artifact(작업 결과물)입니다.
통신 구조에는 세 참여자(Core Actors)가 있습니다. User(요청을 시작하는 사용자 또는 시스템), A2A Client(사용자를 대신해 요청을 전송하는 클라이언트 에이전트, 예: Orchestrator), A2A Server(HTTP 엔드포인트로 요청을 처리하는 원격 에이전트, 예: Specialist). 여기서 중요한 설계 철학 하나 — 원격 에이전트는 "Opaque"(불투명) 시스템입니다. 클라이언트는 상대가 LangGraph인지 CrewAI인지, 내부에서 어떤 모델을 쓰는지 알 필요가 없습니다. Agent Card에 적힌 능력과 프로토콜 규약만 믿고 일을 맡기면 됩니다.
👤 User
사용자/시스템
에이전트 지원 요청 시작
→
📞 A2A Client
Client Agent
사용자 대신 요청 전송 (예: Orchestrator)
A2A Protocol
⇄
HTTP(S) + JSON-RPC 2.0
🤖 A2A Server
Remote Agent
요청 처리·결과 반환 (예: Specialist)
Remote Agent는 Opaque — 클라이언트는 상대의 내부 구현(프레임워크·모델)을 알 필요가 없다
그림 2. A2A 핵심 참여자 — User · A2A Client · A2A Server
시작 전에, 자주 만나는 오해 세 가지를 짚습니다.
오해 ① "A2A는 MCP의 경쟁자다"
둘은 층이 다릅니다. A2A는 에이전트 "간"(Agent ↔ Agent) 고수준 작업 조율, MCP는 에이전트와 "도구" 간(Agent ↔ Tools/Data) 컨텍스트 구조화입니다. 교재의 결론은 명확합니다 — 상호 보완적 관계. A2A로 에이전트끼리 협업하고, 각 에이전트는 MCP로 자기 도구를 씁니다(10절).
오해 ② "멀티 에이전트 프레임워크(7편)가 있는데 왜 또 필요한가"
Supervisor·Crew는 같은 프로세스, 같은 프레임워크 안의 협업입니다 — 객체 참조를 주고받는 한 지붕 아래 팀이죠. A2A는 네트워크·조직·프레임워크 경계를 넘는 원격 협업입니다. 우리 회사 에이전트가 파트너사 에이전트와 일하려면 객체 참조가 아니라 프로토콜이 필요합니다.
오해 ③ "그냥 REST API 하나 만들면 되는 것 아닌가"
커스텀 API에는 표준 발견 메커니즘이 없습니다. A2A가 더해주는 건 셋 — ① 상대 능력을 기계가 읽는 Agent Card, ② 비동기 장기 작업을 다루는 Task 상태 모델(submitted→working→completed), ③ 동기·폴링·스트리밍·푸시를 아우르는 표준 메서드 규약. 이게 없으면 연동 N개마다 이 셋을 다시 발명하게 됩니다.
3. Agent Card — 에이전트의 디지털 명함
A2A의 출발점은 Agent Card — 에이전트의 디지털 신원을 정의하는 JSON 문서입니다. 사람의 명함에 이름·직함·연락처가 있듯, Agent Card에는 이름·능력(skills)·엔드포인트·인증 요구사항이 들어갑니다. 이 카드 하나로 클라이언트는 "이 에이전트가 무엇을 할 수 있고, 어떻게 말을 걸어야 하는지"를 코드를 보지 않고 파악합니다.
JSON · Agent Card — 구현 예제(01_a2a_server.py)의 KnowledgeAssistant 명함
{ "name": "KnowledgeAssistant", "description": "지식 기반 질문 응답 및 분석을 제공하는 AI Agent입니다.", "url": "http://localhost:8000", // A2A 엔드포인트 "version": "1.0.0", "capabilities": { "streaming": true, // SSE 스트리밍 지원 "pushNotifications": false, // Webhook 미지원 "stateTransitionHistory": true // 상태 변경 이력 제공 }, "authentication": { "schemes": ["apiKey"] }, "defaultInputModes": ["text"], "defaultOutputModes": ["text"], "skills": [ { "id": "explain", "name": "개념 설명", "tags": ["education"], "examples": ["머신러닝의 기본 개념을 설명해주세요"] }, { "id": "summarize", "name": "텍스트 요약", "tags": ["summary"] }, { "id": "analyze", "name": "데이터/텍스트 분석", "tags": ["analysis"] }, { "id": "translate", "name": "번역", "tags": ["translation"] } ] }
| 필드 |
설명 |
목적 |
| name · description |
에이전트 식별 정보 |
사람·시스템이 이해 가능한 설명 |
| url |
A2A 엔드포인트 주소 |
통신 대상 지정 |
| version |
버전 정보 |
호환성 관리 |
| capabilities |
지원 기능 (streaming · pushNotifications · stateTransitionHistory) |
클라이언트가 통신 방식을 고르는 근거 |
| authentication |
인증 요구사항 (schemes) |
보안 연결 방식 선언 |
| skills |
제공 기능 목록 (id · name · description · tags · examples) |
에이전트가 할 수 있는 일의 카탈로그 |
그림 3. Agent Card 구성요소 — 능력·엔드포인트·인증을 한 장의 JSON으로
눈여겨볼 디테일은 skills의 examples 필드입니다. "파리 날씨 어때?" 같은 예시 문장은 사람을 위한 문서이기도 하지만, 호출하는 쪽 LLM이 "이 스킬에 이 일을 맡겨도 되는가"를 판단하는 힌트이기도 합니다. 도구 사용 패턴(5편)에서 docstring이 LLM의 도구 선택 기준이 됐던 것과 정확히 같은 원리가, 에이전트 단위로 확장된 셈입니다.
4. Agent Discovery — 명함을 어디서 받을 것인가
명함이 있어도 받을 방법이 없으면 소용없습니다. 클라이언트가 사용 가능한 에이전트를 찾는 Discovery 전략은 세 가지이고, 각자 어울리는 환경이 다릅니다.
1
Well-Known URI
표준 경로
표준 Well-Known 경로에 Agent Card 호스팅 — 현행 스펙은 /.well-known/agent-card.json, 구현 예제는 구버전 경로 agent.json 사용. 자동 발견 가능 — 공개·도메인 특화 사용에 적합
2
Curated Registry
중앙 등록소
카탈로그에 Agent Card를 등록하고 조건 기반 검색. 중앙 관리·접근 제어 가능 — 엔터프라이즈 환경에 적합(12절의 Agent Registry)
3
Direct Configuration
직접 설정
config.json·환경 변수에 상대 주소를 직접 정의. 동적 발견이 불필요한 긴밀히 결합된 private 시스템에 적합
그림 4. Agent Discovery 3전략 — 공개 웹은 Well-Known URI, 사내는 Registry, 고정 연동은 Direct
한 가지 주의 — Agent Card는 "공개 명함"이지만 아무에게나 뿌릴 명함은 아닙니다. 교재는 Agent Card 엔드포인트에도 접근 제어·mTLS·네트워크 제한을 권고합니다. 카드에는 엔드포인트와 능력 정보가 담기므로, 그 자체가 공격 표면의 지도가 될 수 있기 때문입니다(11절에서 보안 전체를 다룹니다).
5. Task 생명주기 — 통신의 기본 단위는 "작업"이다
A2A 통신의 기본 단위는 요청-응답이 아니라 Task(작업)입니다. 이 선택이 프로토콜의 성격을 결정합니다 — 에이전트의 일은 몇 초짜리 조회부터 몇 시간짜리 분석까지 길이가 제각각이라, 상태를 가진 작업 모델이 필요하기 때문입니다. 모든 Task는 고유 ID를 가지며(병렬 처리 지원), 다음 상태 흐름을 따릅니다.
submitted
제출됨 — Task 접수·ID 발급
→
working
처리 중 — 에이전트가 작업 수행
→
completed
완료 — Artifact 반환
input-required
추가 입력 필요 → (재개) → working
canceled
취소됨 — tasks/cancel 요청 시
기본 흐름 submitted → working → completed · working에서 failed/input-required로 분기 · 완료/실패 전이면 canceled 가능
그림 5. Task 상태 흐름 — 구현 코드의 TaskState 6종 (v0.2 이후 스펙에는 rejected·auth-required 등 상태 추가 — 버전에 따라 표기 변동)
특히 input-required 상태가 흥미롭습니다. 에이전트가 "추가 정보가 있어야 진행할 수 있다"고 멈춰 서는 상태인데, 13편(휴먼 인 더 루프)의 interrupt()가 프로토콜 차원의 일급 상태로 승격된 모양새입니다. 사람이든 호출자 에이전트든, 입력을 보태면 작업이 재개됩니다.
Task 안을 오가는 데이터는 두 종류입니다. Message는 통신 메시지(발신자 role + 내용 parts), Artifact는 작업 결과물(파일·데이터, 점진적 스트리밍 가능)입니다. parts가 리스트인 이유는 멀티모달 — 텍스트뿐 아니라 이미지·JSON도 한 메시지에 담을 수 있게 한 설계입니다.
JSON · Message 구조 — role + parts(멀티모달 리스트) + attributes(메타데이터)
{ "role": "user", // 발신자: user 또는 agent "parts": [ { "type": "text", "text": "USD에서 EUR로 환율이 얼마인가요?" } ], "attributes": { "priority": "high", "created_at": "2025-01-04T10:00:00Z" } }
6. 통신 메커니즘 4종 — 작업 길이에 맞는 채널을 골라라
A2A는 한 가지 통신 방식을 강요하지 않습니다. 작업이 얼마나 오래 걸리는가에 따라 네 가지 상호작용 방식을 제공합니다 — 이 챕터에서 가장 실무적인 결정 지점입니다.
1️⃣ Synchronous (동기)
요청 → 대기 → 완료 응답. 빠른 즉시 작업에 적합.
tasks/send
2️⃣ Async Polling (폴링)
task_id 받고 상태를 반복 조회. 오래 걸리는 작업 — 클라이언트가 그동안 딴 일 가능.
tasks/get
3️⃣ SSE Streaming (스트리밍)
구독하면 서버가 실시간 업데이트를 밀어줌. 점진적 결과·UX 개선에 적합.
tasks/sendSubscribe
4️⃣ Push Notifications (웹훅)
요청에 webhook URL 동봉, 완료 시 서버가 호출. 매우 긴 배치 작업 — 연결 유지가 없어 리소스 효율적.
pushNotifications
그림 6. A2A 통신 메커니즘 4종 — 메서드 표기는 구현 예제 기준(프로토콜 버전에 따라 명칭이 다를 수 있음)
| 상황 |
권장 방식 |
이유 |
| 간단한 조회 |
동기 (Sync) |
빠른 응답, 단순 구현 |
| 분석/처리 작업 |
폴링 (Polling) |
클라이언트 자유도 |
| 실시간 생성 |
스트리밍 (SSE) |
UX 개선, 점진적 결과 |
| 배치 작업 |
푸시 (Webhook) |
리소스 효율 |
전송 형식은 전부 JSON-RPC 2.0입니다. HTTP POST 본문에 jsonrpc · id · method · params 네 필드를 담는 단순한 규약이라, REST 엔드포인트를 늘리는 대신 메서드 이름으로 라우팅합니다. 동기 요청은 이렇게 생겼습니다.
JSON · JSON-RPC 2.0 동기 요청 — method 하나로 무엇을 할지 말한다
{ "jsonrpc": "2.0", "id": "1", "method": "tasks/send", // 동기 Task 실행 "params": { "id": "task-001", "sessionId": "session-001", // 연속 대화 유지용 "message": { "role": "user", "parts": [{ "type": "text", "text": "USD에서 EUR로 환율이 얼마인가요?" }] } } }
💡 명칭 주의 — 학습 자료의 개념 설명(패턴 정의)에는 sendTask · getTask · sendTaskSubscribe로, 구현 예제의 JSON-RPC 메서드 문자열은 tasks/send · tasks/get · tasks/sendSubscribe · tasks/cancel로 표기됩니다. 본 글은 구현 기준으로 통일합니다. A2A 스펙은 활발히 개정 중이라 명칭이 프로토콜 버전에 따라 달라집니다 — v0.3.0은 message/send · message/stream, 현행 v1.0(2026-03 릴리스)은 SendMessage · SendStreamingMessage로 재명명했고, Well-Known 경로는 /.well-known/agent-card.json(v0.3.0에서 변경)입니다. 실제 연동 시 공식 스펙(a2a-protocol.org)의 현행 버전을 확인하세요.
7. 구현 ① A2A 서버 — FastAPI + JSON-RPC + SSE
이제 직접 만들어 봅니다. 첫 번째 구현(01_a2a_server.py)은 A2A 프로토콜을 준수하는 서버 — FastAPI 위에 Agent Card 엔드포인트와 JSON-RPC Task API를 올리고, 실제 작업 처리는 Gemini가 담당합니다. 의존성은 다섯 개 — fastapi(웹 서버) · sse-starlette(SSE 스트리밍) · google-generativeai(Gemini) · pydantic(데이터 검증) · uvicorn(ASGI)입니다.
서버의 뼈대는 단 두 개의 엔드포인트입니다. GET 하나로 명함을 내밀고, POST 하나로 모든 작업을 받습니다.
PYTHON · 서버 뼈대 — Agent Card 엔드포인트 + JSON-RPC 메서드 라우팅
app = FastAPI(title="A2A Knowledge Assistant Server") task_store: dict[str, Task] = {} # 메모리 저장소 (프로덕션은 Redis/DB) model = genai.GenerativeModel(MODEL_ID) # Gemini — 실제 Task 처리 담당 @app.get("/.well-known/agent.json") # Well-Known 경로(구현 예제 기준) — Discovery의 출발점 async def get_agent_card(): return JSONResponse(content=AGENT_CARD) @app.post("/") # JSON-RPC 단일 엔드포인트 async def handle_jsonrpc(request: Request): body = await request.json() rpc_request = JSONRPCRequest(**body) if rpc_request.method == "tasks/send": # 동기 return await handle_send_task(rpc_request) elif rpc_request.method == "tasks/sendSubscribe": # SSE 스트리밍 return await handle_send_task_subscribe(rpc_request) elif rpc_request.method == "tasks/get": # 상태 조회(폴링) return await handle_get_task(rpc_request) elif rpc_request.method == "tasks/cancel": # 취소 return await handle_cancel_task(rpc_request) else: return create_error_response(rpc_request.id, -32601, f"Method not found: {rpc_request.method}")
동기 처리(tasks/send)의 핵심은 상태 전이를 정직하게 밟는 것입니다. Task를 submitted로 만들고, working으로 올리고, Gemini 호출이 성공하면 completed + Artifact, 실패하면 failed + 오류 메시지. 5절의 상태 다이어그램이 그대로 코드가 됩니다.
PYTHON · handle_send_task — submitted → working → completed/failed 상태 전이
task = Task(id=task_id, sessionId=params.sessionId, status=TaskStatus(state=TaskState.SUBMITTED), # 1) 접수 history=[params.message]) task_store[task_id] = task task.status = TaskStatus(state=TaskState.WORKING) # 2) 처리 중 try: response = await asyncio.to_thread( # 동기 API를 비동기로 래핑 model.generate_content, f"당신은 지식 기반 AI 어시스턴트입니다. ...\n\n{user_text}") agent_message = Message.text("agent", response.text) task.history.append(agent_message) # 대화 이력 보존 task.artifacts = [Artifact.text(response.text)] # 3) 결과물 저장 task.status = TaskStatus(state=TaskState.COMPLETED, message=agent_message) except Exception as e: task.status = TaskStatus(state=TaskState.FAILED, # 4) 실패도 상태로 message=Message.text("agent", f"처리 중 오류 발생: {str(e)}")) return JSONResponse(content={"jsonrpc": "2.0", "id": rpc_request.id, "result": task.model_dump()})
스트리밍(tasks/sendSubscribe)은 sse-starlette의 EventSourceResponse로 구현합니다 — 비동기 제너레이터가 yield하는 이벤트를 SSE 스트림으로 흘려보내는 구조입니다. 이벤트는 두 종류 — task/status(상태 변화, final 플래그로 종료 표시)와 task/artifact(Gemini 스트리밍 청크를 그대로 중계).
PYTHON · SSE 이벤트 생성기 — Gemini 청크를 task/artifact로 실시간 중계
async def event_generator(): yield {"event": "task/status", # 1) submitted 알림 "data": json.dumps({"id": task_id, "status": task.status.model_dump(), "final": False})} task.status = TaskStatus(state=TaskState.WORKING) yield {"event": "task/status", "data": ...} # 2) working 알림 response = await asyncio.to_thread( model.generate_content, prompt, stream=True) # 3) Gemini 스트리밍 for chunk in response: if chunk.text: yield {"event": "task/artifact", # 4) 청크 단위 중계 "data": json.dumps({"id": task_id, "artifact": { "parts": [{"text": chunk.text}], "index": chunk_count, "append": True}})} yield {"event": "task/status", # 5) completed (final=True) "data": json.dumps({"id": task_id, "status": task.status.model_dump(), "final": True})} return EventSourceResponse(event_generator())
마지막으로 에러 처리의 디테일 하나 — JSON-RPC는 에러도 HTTP 200으로 반환합니다. 프로토콜 레벨의 성공/실패는 HTTP 상태 코드가 아니라 응답 본문의 result/error 필드로 구분합니다. 표준 에러 코드는 다섯 개입니다.
| 코드 |
의미 |
예시 상황 |
| -32700 |
Parse error |
본문이 유효한 JSON이 아님 |
| -32600 |
Invalid request |
완료된 Task에 취소 요청 |
| -32601 |
Method not found |
지원하지 않는 메서드 |
| -32602 |
Invalid params |
존재하지 않는 task_id 조회 |
| -32603 |
Internal error |
서버 내부 오류 |
그림 7. JSON-RPC 2.0 표준 에러 코드 — 에러도 HTTP 200 본문의 error 필드로 반환된다
8. 구현 ② A2A 클라이언트 — Discovery부터 스트리밍 수신까지
반대편 구현(02_a2a_client.py)은 httpx 기반 비동기 클라이언트입니다. requests가 아니라 httpx인 이유는 셋 — async/await 지원, SSE 스트리밍 지원(stream()), 연결 재사용. 클라이언트는 컨텍스트 매니저로 감싸 연결 풀이 예외 상황에도 정리되게 합니다.
PYTHON · A2AClient — Discovery: 명함부터 받는다
class A2AClient: def __init__(self, base_url: str): self.base_url = base_url.rstrip("/") self.agent_card: Optional[AgentCard] = None self._client = httpx.AsyncClient(timeout=60.0) async def discover(self) -> AgentCard: # /.well-known/agent.json 에서 Agent Card 조회 — 능력·기능·인증 파악 url = f"{self.base_url}/.well-known/agent.json" response = await self._client.get(url) response.raise_for_status() self.agent_card = AgentCard.from_dict(response.json()) return self.agent_card # 사용 — 컨텍스트 매니저로 리소스 자동 해제 async with A2AClient("http://localhost:8000") as client: card = await client.discover() if card.capabilities.get("streaming"): # 카드 보고 통신 방식 결정 ... # send_task_subscribe (SSE) else: ... # send_task (동기)
이 짧은 사용 예에 A2A의 핵심 흐름이 다 들어 있습니다. 먼저 명함을 받고(discover), 명함에 적힌 능력(capabilities)을 보고 통신 방식을 고른다 — 하드코딩된 가정 대신 상대의 자기 기술(Self-Description)에 맞춰 동작하는 클라이언트입니다. 동기 전송은 JSON-RPC 요청을 만들어 POST 한 번이면 끝입니다.
PYTHON · send_task — 동기 전송과 결과 꺼내기
async def send_task(self, message: str, session_id=None) -> dict: params = {"id": str(uuid.uuid4()), "message": {"role": "user", "parts": [{"text": message}]}} if session_id: params["sessionId"] = session_id # 연속 대화 유지 request = {"jsonrpc": "2.0", "id": str(uuid.uuid4()), "method": "tasks/send", "params": params} response = await self._client.post(self.base_url, json=request) result = response.json() if "error" in result: # JSON-RPC 에러는 본문으로 온다 raise Exception(f"RPC Error: {result['error']}") return result["result"] # 사용 result = await client.send_task("A2A Protocol이란 무엇인지 간단히 설명해주세요.") print(result["status"]["state"]) # "completed" print(result["artifacts"][0]["parts"][0]["text"]) # 응답 텍스트
스트리밍 수신은 SSE의 텍스트 규약을 직접 파싱합니다. SSE는 event: 줄과 data: 줄, 그리고 빈 줄(이벤트 종료)로 이뤄진 단순한 형식 — 줄 단위로 읽다가 빈 줄을 만나면 이벤트 하나를 yield하는 비동기 제너레이터로 구현합니다.
PYTHON · send_task_subscribe — SSE 라인 파싱 (event: / data: / 빈 줄)
async with self._client.stream("POST", self.base_url, json=request, headers={"Accept": "text/event-stream"}) as response: event_type, event_data = None, "" async for line in response.aiter_lines(): line = line.strip() if not line: # 빈 줄 = 이벤트 1개 완성 if event_type and event_data: yield {"event": event_type, "data": json.loads(event_data)} event_type, event_data = None, "" continue if line.startswith("event:"): event_type = line[6:].strip() elif line.startswith("data:"): event_data = line[5:].strip() # 사용 — 청크가 오는 대로 실시간 출력 async for event in client.send_task_subscribe("Python의 장점을 3가지 알려주세요."): if event["event"] == "task/artifact": print(event["data"]["artifact"]["parts"][0]["text"], end="", flush=True) elif event["event"] == "task/status": if event["data"]["status"]["state"] == "completed": print("\n[완료]")
폴링 패턴(tasks/get)은 task_id로 상태를 반복 조회하다 completed/failed에서 빠져나오는 while 루프입니다. 해설 문서는 여기에 tenacity 재시도(지수 백오프), 병렬 에이전트 호출(asyncio.gather), 세션 매니저 같은 확장 패턴도 제시합니다 — 12편(예외 처리)과 3편(병렬화)의 기법이 A2A 클라이언트에 그대로 합류하는 지점입니다.
9. 멀티 에이전트 시나리오 — 조사시키고, 요약시키고, 조율한다
부품이 모였으니 협업을 만들어 봅니다. 구현 예제의 시나리오는 셋 — Client Agent(Orchestrator, 작업 분배·조율) · Research Agent(자료 조사) · Summary Agent(요약 작성)입니다. 조율자가 조사를 맡기고, 그 결과를 요약자에게 넘기는 2단계 핸드오프죠.
1
조사 요청
Orchestrator → Research
tasks/send "핵심 특징 3가지 조사"
→
2
조사 결과
Research → Orchestrator
completed + Artifact(조사 텍스트)
→
3
요약 요청
Orchestrator → Summary
조사 결과를 포함해 tasks/send
→
4
최종 요약
Summary → Orchestrator
3줄 요약 Artifact 반환
각 에이전트는 서로의 내부를 모른 채(Opaque) A2A 프로토콜로만 협업한다 — 보라=조율자의 위임, 초록=최종 결과
그림 8. 멀티 에이전트 핸드오프 — Orchestrator가 Research·Summary 에이전트에 차례로 위임
PYTHON · 멀티 에이전트 핸드오프 — 앞 결과가 뒤 요청의 입력이 된다
# 실제 환경 — 서로 다른 URL의 에이전트 서버들 research_client = A2AClient("http://research-agent.example.com") summary_client = A2AClient("http://summary-agent.example.com") async with research_client, summary_client: # Step 1: Research Agent에 조사 위임 research = await research_client.send_task( "A2A Protocol의 핵심 특징 3가지를 조사해주세요.") if research["status"]["state"] == "completed": research_text = research["artifacts"][0]["parts"][0]["text"] # Step 2: 조사 결과를 Summary Agent에 넘겨 요약 위임 summary = await summary_client.send_task( f"다음 내용을 3줄로 요약해주세요:\n{research_text}") print(summary["artifacts"][0]["parts"][0]["text"])
구조가 낯익지 않나요? 7편(멀티 에이전트 협업)의 Supervisor 패턴과 논리적으로 동일합니다 — 차이는 경계뿐입니다. 7편의 핸드오프가 한 프로세스 안의 함수 호출이었다면, A2A의 핸드오프는 네트워크 너머의 표준 프로토콜 호출입니다. 그래서 교재의 다른 시나리오들도 같은 골격으로 확장됩니다 — 파리 출장 일정을 LangGraph 조율자가 CrewAI 날씨·ADK 캘린더·LangGraph 교통 에이전트에 나눠 맡기는 다중 프레임워크 협업, 고객 문의가 Intake→Analysis→Resolution→Response 에이전트를 차례로 통과하는 워크플로우 오케스트레이션, Primary Agent가 주가·뉴스 에이전트를 병렬로 부르는 동적 정보 검색까지.
10. A2A vs MCP — 경쟁이 아니라 분업
10편(MCP)을 읽으셨다면 자연스러운 질문이 떠오를 겁니다 — "연결 표준이라면 MCP가 이미 있잖아?" 둘의 차이는 누구와 누구를 잇느냐로 갈립니다.
🤝 A2A (Agent2Agent)
Agent ↔ Agent — 에이전트 간 통신·작업 위임·협업
에이전트 "간" 고수준 작업 조율 — Task 위임과 상태 관리
발견: Agent Card · 프로토콜: HTTP + JSON-RPC 2.0
주도: Google 공개 → Linux Foundation 이관
🔧 MCP (Model Context Protocol)
Agent ↔ Tools/Data — 브라우저·파일·DB·API 같은 외부 리소스 연결
에이전트와 "도구" 간 컨텍스트 구조화 — 도구 호출
발견: Tool Manifest · 프로토콜: JSON-RPC over stdio/HTTP
주도: Anthropic 공개
그림 9. A2A vs MCP — 같은 "연결 표준"이지만 잇는 대상이 다르다 (상호 보완적)
그래서 실전 아키텍처에서는 둘을 겹쳐 씁니다. 수평 방향(에이전트끼리)은 A2A, 수직 방향(에이전트→도구)은 MCP — 교재의 통합 아키텍처가 이 분업을 그림 한 장으로 보여줍니다.
🧑💼 사용자 → Orchestrator Agent
↓ A2A (작업 위임·협업) ↓
Research Agent
Analysis Agent
Report Agent
↓ MCP (도구·데이터 접근) ↓
🌐 Web API Tool
🗄️ Database Tool
📄 Docs Tool
수평(보라)은 A2A로 에이전트 간 위임 · 수직(슬레이트)은 MCP로 각 에이전트가 자기 도구 사용
그림 10. A2A + MCP 통합 아키텍처 — 두 표준이 한 시스템에서 만난다
💡 한 줄 정리 — MCP가 에이전트에게 손(도구)을 달아주는 표준이라면, A2A는 에이전트에게 입과 귀(동료와의 대화)를 달아주는 표준입니다. 손이 있어야 일을 하고, 입이 있어야 일을 나눕니다.
11. 보안 — 모르는 에이전트와 일하려면 신뢰 장치가 필요하다
A2A의 전제는 "내부 구현을 모르는 상대(Opaque)와의 협업"입니다. 모르는 상대와 일하려면 신뢰를 프로토콜이 보장해야 하죠. 교재가 정리한 보안 메커니즘은 네 겹입니다.
1️⃣ Mutual TLS (mTLS)
클라이언트·서버가 양방향으로 인증서를 검증. 무단 접근과 데이터 가로채기를 차단
2️⃣ Comprehensive Audit Logs
모든 에이전트 간 통신 기록 — 정보 흐름·참여 에이전트·수행 작업. 책임 추적·문제 해결·보안 분석의 근거
3️⃣ Agent Card 인증 선언
"schemes": ["apiKey", "oauth2"] — 인증 요구사항을 카드에 명시적으로 선언, 중앙화된 인증 관리
4️⃣ Credential Handling
자격증명은 HTTP 헤더로만 전달 (Authorization: Bearer <token> · X-API-Key) — URL·본문 노출 방지
그림 11. A2A 보안 4중 장치 — 암호화·기록·선언·전달 규약
엔터프라이즈에서는 여기에 거버넌스 층이 더해집니다. 적용 전략 문서의 권고는 — Identity Provider(Azure AD·Okta·Keycloak)로 에이전트에게도 신원을 부여하고(OAuth 2.0 + OpenID Connect), 에이전트별 권한을 scope로 쪼개고(crm.read · erp.write처럼), mTLS·API Rate Limiting·Request Signing·IP Allowlisting을 겹쳐 거는 것. 사람 직원에게 적용하던 최소 권한 원칙을 에이전트 직원에게 그대로 적용하는 셈입니다. 민감 데이터(주민번호·카드번호·이메일)는 에이전트 간 전송 전에 마스킹·암호화하는 패턴도 함께 제시됩니다.
JSON · 감사 로그 — "누가, 누구에게, 무엇을, 어떤 결과로"가 한 줄에 남는다
{ "timestamp": "2024-01-15T10:30:00Z", "source_agent": "CRM Agent", // 요청한 에이전트 "target_agent": "ERP Agent", // 처리한 에이전트 "method": "tasks/send", "action": "order_processing", "user_context": { "user_id": "user-123", "department": "영업팀" }, "data_classification": "confidential", // 데이터 등급 "result": "success", "response_time_ms": 245 }
12. 엔터프라이즈 — CRM·ERP·Groupware가 서로 대화할 때
엔터프라이즈에서 A2A의 그림은 명확합니다. 시스템마다 전담 에이전트를 세우고, 중앙 Agent Registry(Agent Card 저장소 + Discovery + Health Monitoring)가 이들을 중개하는 구조입니다. CRM Agent는 Salesforce·HubSpot을, ERP Agent는 SAP·Oracle을, GW Agent는 Microsoft 365·Slack을 등에 업고, 서로는 A2A로 대화합니다. 포인트-투-포인트 연동이 "Agent 기반 자율 협업"으로, 수동 데이터 전달이 "Agent 간 자동 핸드오프"로 바뀌는 거죠.
| Agent |
핵심 역할 |
제공 Skills |
| CRM Agent |
고객 중심 정보 관리 |
고객조회 · 영업기회관리 · 서비스이력 · 고객 360 뷰 |
| ERP Agent |
비즈니스 리소스 관리 |
재고확인 · 주문처리 · 생산계획 · 구매 · 재무거래 |
| GW Agent |
업무 협업 지원 |
결재 워크플로우 · 일정관리 · 알림발송 · 문서·업무관리 |
| Analytics Agent |
데이터 분석 |
리포트생성 · 예측분석 · 대시보드 |
| Compliance Agent |
규정 준수 |
감사로그 · 정책검증 · 리스크평가 |
그림 12. 엔터프라이즈 A2A 생태계의 에이전트 역할 분담 (적용 전략 문서 기준)
움직이는 모습은 이렇습니다. 재고가 임계점에 닿으면 Inventory Monitor가 ERP Agent에 알리고, ERP Agent가 수요를 예측한 뒤 Procurement Agent에 공급업체 선정을, CRM Agent에 고객 주문 예정 확인을 동시에 맡깁니다. 구매 승인은 GW Agent의 결재 라우팅을 타는데 — 100만원 이하는 자동 승인, 초과는 결재 라인으로 가는 금액 기준 분기가 들어 있죠. 마찬가지로 Order-to-Cash 자동화에서는 1천만원 초과 대형 주문에만 결재 요청이 붙습니다. 13편(휴먼 인 더 루프)의 금액 기준 에스컬레이션이 A2A 멀티 에이전트 위에 그대로 올라간 모양입니다.
⚠️ 아래 수치는 적용 전략 문서의 도입 기대효과(목표치)이며 특정 기업의 실측 결과가 아닙니다. 개선율은 Before→After 값으로 재계산해 표기했습니다.
| 시스템 |
지표 |
Before → After |
개선 |
| CRM |
리드 응답 시간 |
4시간 → 15분 |
약 94%↓ |
| 영업 기회 전환율 |
12% → 28% |
+133% |
| 고객 정보 조회 시간 |
8분 → 30초 |
약 94%↓ |
| 크로스셀 성공률 |
5% → 18% |
+260% |
| ERP |
주문 처리 시간 |
2일 → 2시간 |
약 96%↓ |
| 재고 부족 발생률 |
8% → 1.5% |
약 81%↓ |
| 구매 승인 사이클 |
3일 → 4시간 |
약 94%↓ |
| 데이터 입력 오류 |
5% → 0.3% |
약 94%↓ |
| Groupware |
결재 처리 시간 |
2.5일 → 6시간 |
90%↓ |
| 회의 조율 시간 |
45분 → 3분 |
약 93%↓ |
| 업무 할당 누락 |
15% → 2% |
약 87%↓ |
| 알림 전달률 |
75% → 99% |
+32% |
그림 13. CRM·ERP·Groupware 도입 기대효과(목표치) — 개선율은 Before→After 재계산 기준
성공 지표(KPI)는 세 층으로 잡습니다. 운영 — 프로세스 자동화율 70%, 평균 처리 시간 -80%, 오류율 <1%, 시스템 가용성 99.9%. 비즈니스 — 고객 응답 시간 -90%, 주문 처리 속도 -85%, 결재 사이클 -80%, 직원 생산성 +40%. 기술 — Agent 응답 시간 P95 <500ms, 동시 처리량 1000 TPS, Agent 가용성 99.95%, 통합 성공률 >99% (모두 목표치).
1
Foundation
1~2개월
Registry·인증 인프라 + 기본 Agent 3종 + 통합 테스트
→
2
Core Integration
2~3개월
CRM-ERP·ERP-GW 연동 + 워크플로우 3~5개 + 대시보드
→
3
Advanced
3~4개월
3개 이상 Agent 복합 워크플로우 + AI 의사결정 + 자동 복구
→
4
Optimization
지속
성능 최적화 + 신규 Use Case + 외부 시스템 확장
그림 14. 4단계 구현 로드맵 — 인프라부터 깔고, 연동을 늘리고, 복합 워크플로우로
13. 케이스 스터디 — 신규 고객 한 명이 오면 세 시스템이 움직인다
신규 고객 계약이 성사됐습니다. 예전 같으면 영업이 CRM에 입력하고, 메일로 재무팀에 거래처 등록을 요청하고, 또 메일로 킥오프 미팅을 잡는 — 사람이 릴레이하는 사흘짜리 일이죠. 적용 전략 문서의 신규 고객 온보딩 워크플로우는 이것을 에이전트 릴레이로 바꿉니다.
1
고객 등록
CRM Agent
고객 정보 등록 · 신용 등급 조회 · 담당 영업 배정
→
2
계정 설정
ERP Agent
거래처 마스터 생성 · 결제 조건(NET30) · 가격 정책
→
3
협업 설정
GW Agent
담당자 업무 할당 · 킥오프 미팅 · 환영 메일
→
4
완료 통보
Notification Agent
관련자 알림 · 대시보드 갱신 · 진행 리포트
Phase 1의 출력(고객 ID·신용 한도)이 Phase 2의 입력이 되는 A2A 핸드오프 체인 — 실패 시 보상 트랜잭션으로 되감기
그림 15. 신규 고객 온보딩 4단계 — CRM → ERP → Groupware → Notification
PYTHON · 온보딩 워크플로우 — 세 에이전트를 잇고, 실패하면 보상 트랜잭션
async def customer_onboarding_workflow(customer_data: dict): workflow_id = str(uuid.uuid4()) results = {"workflow_id": workflow_id, "phases": {}} try: # Phase 1: CRM — 고객 등록 + 신용 확인 async with A2AClient(CRM_AGENT_URL) as crm: customer = await crm.send_task(f"customer_creation: {json.dumps(customer_data)}") credit = await crm.send_task(f"credit_check: {customer['customer_id']}") # Phase 2: ERP — 앞 단계 결과(신용 한도)가 입력으로 흐른다 async with A2AClient(ERP_AGENT_URL) as erp: vendor = await erp.send_task( f"vendor_master_create: customer_id={customer['customer_id']} " f"credit_limit={credit['approved_limit']} payment_terms=NET30") # Phase 3: GW — 업무 할당 + 킥오프 미팅 + 환영 메일 async with A2AClient(GW_AGENT_URL) as gw: task = await gw.send_task("task_management: create title='신규 고객 온보딩...' ...") meeting = await gw.send_task("calendar_management: schedule_meeting ...") notification = await gw.send_task("notification_service: send_email ...") results["status"] = "completed" except Exception as e: results["status"] = "failed" # 롤백 또는 보상 트랜잭션 — 12편(예외 처리)의 Saga 패턴이 여기 붙는다 await handle_onboarding_failure(workflow_id, results) return results
이 코드가 이번 편 전체의 축소판입니다. 각 시스템 에이전트는 자기 Agent Card로 능력을 공개하고(3절), 클라이언트는 send_task로 일을 맡기고(8절), 앞 결과가 뒤 입력으로 흐르고(9절), 전 과정이 감사 로그에 남습니다(11절). 그리고 중간에 무너지면? except 블록의 보상 트랜잭션이 앞 단계를 되감습니다 — CRM에는 등록됐는데 ERP에는 없는 "반쪽 고객"을 막는 12편(예외 처리)의 Saga 패턴이, 분산 에이전트 시대에 더 절실해진 모습입니다.
14. 베스트 프랙티스 — 쓸지 말지, 그리고 체크리스트
모든 표준이 그렇듯 A2A도 만능이 아닙니다. 교재의 적합성 판단 기준부터 — 핵심은 "경계를 넘는가"입니다.
✅ A2A를 사용해야 할 때
✅ 2개 이상의 AI 에이전트 협업이 필요할 때
✅ 서로 다른 프레임워크(ADK·LangGraph·CrewAI) 혼용
✅ 전문 에이전트가 워크플로우의 특정 부분 담당
✅ 에이전트 능력의 동적 발견이 필요할 때
❌ A2A가 불필요한 경우
❌ 단일 에이전트로 충분한 작업 — 프로토콜은 비용이다
❌ 모든 에이전트가 같은 프레임워크 내에서 동작 — 7편의 Supervisor/Crew로 충분
❌ 외부 도구만 연결하면 되는 경우 — MCP를 쓰라(10편)
그림 16. A2A 사용 적합성 판단 — 프레임워크·조직 경계를 넘는 협업일 때만 꺼내는 카드
도입을 결정했다면, 위험 요소를 미리 짚어둡니다. 적용 전략 문서의 위험 관리 표가 좋은 출발점입니다 — Agent 장애(영향 높음 → 이중화·Circuit Breaker), 네트워크 지연(중간 → 캐싱·비동기 처리), 데이터 불일치(높음 → Saga 패턴·보상 트랜잭션), 보안 취약점(높음 → 정기 보안 감사·Penetration Test). 운영 측면에서는 사용자 저항(교육·점진 도입), 복잡성 증가(문서화·모니터링 강화), 벤더 종속(표준 프로토콜 준수가 그 자체로 방어)이 따라옵니다.
✅ A2A 구현 체크리스트 (교재 10항목)
☐ 1. Agent Card 정의 (name · url · skills · authentication)
☐ 2. 발견 메커니즘 선택 (Well-Known URI / Registry / Direct)
☐ 3. 통신 방식 결정 (Sync / Polling / SSE / Webhook — 그림 6)
☐ 4. 보안 설정 (mTLS · OAuth · API Key)
☐ 5. A2A 서버 구현 (HTTP 엔드포인트)
☐ 6. A2A 클라이언트 구현 (요청 전송)
☐ 7. 오류 처리·재시도 로직 (12편의 백오프·Circuit Breaker)
☐ 8. 로깅·모니터링 설정 (감사 로그 포함)
☐ 9. 테스트 및 검증
☐ 10. 프로덕션 배포
★ 마치며
순서대로 쪼개고(체이닝), 갈래를 나누고(라우팅), 병렬로 돌리고(병렬화), 스스로 비평하고(리플렉션), 도구를 쥐고(도구 사용), 계획하고(플래닝), 팀을 이루고(멀티 에이전트), 기억하고(메모리), 배우고(학습), 표준으로 도구에 연결되고(MCP), 스스로 채점하고(목표 설정), 넘어져도 일어서고(예외 처리), 멈춰야 할 순간 사람을 부르고(휴먼 인 더 루프), 환각 대신 근거로 말하던(RAG) 에이전트 — 이번엔 명함을 들고 다른 에이전트와 표준으로 협업하는 법(A2A)을 배웠습니다. 다음 편은 자원 인식 최적화(Resource-Aware Optimization): 모든 요청에 가장 비싼 모델을 쓰지 않도록, 작업 난이도에 맞춰 모델을 바꿔 타고 토큰·예산을 배분하는 법입니다. 에이전트가 똑똑해질수록, 비용은 설계의 문제가 됩니다.
#A2A #Agent2Agent #에이전트간통신 #AgentCard #JSONRPC #SSE #FastAPI #MCP #멀티에이전트 #AI에이전트
📚 참고 자료 · 공식 문서
본문 ROI·효과 수치는 학습 자료 기반 도입 기대효과(목표치)이며 특정 기업의 실측이 아닙니다. A2A 스펙·메서드 명칭·Well-Known 경로는 버전에 따라 변동될 수 있습니다.