Agentic AI/AI_AGENT

A2A. Weather Client

아톨 2026. 10. 3. 22:17

이 코드는 크게 보면 다음 한 문장입니다.

"Weather Server의 Agent Card를 먼저 확인하고 → A2A Client를 만들고 → 사용자의 질문을 Message로 포장하고 → Request로 한 번 더 포장해서 → 서버에 보내고 → 서버가 보내는 응답 이벤트를 받아서 → Client를 종료한다."

전체 구조부터 잡고 하나씩 보겠습니다.

1. 먼저 전체 그림을 이해하자

현재 우리가 만든 프로그램은 A2A Client입니다.

상대방에는 앞에서 만든 Weather Agent Server가 실행되고 있습니다.

┌─────────────────────────────┐
│      A2A Weather Client     │
│                             │
│  우리가 지금 공부하는 코드   │
└──────────────┬──────────────┘
               │
               │ HTTP
               │ A2A Protocol
               ▼
┌─────────────────────────────┐
│     A2A Weather Server      │
│     localhost:9999          │
│                             │
│  WeatherAgentExecutor       │
│          ↓                  │
│  WeatherCoreAgent           │
│          ↓                  │
│      서울 날씨 검색          │
└─────────────────────────────┘

그리고 Client 내부에서는 다음 순서로 움직입니다.

① 서버 주소 확인
      ↓
② Agent Card 조회
      ↓
③ A2A Client 생성
      ↓
④ 사용자 질문을 Message로 생성
      ↓
⑤ SendMessageRequest 생성
      ↓
⑥ 서버에 전송
      ↓
⑦ 서버 응답 이벤트 수신
      ↓
⑧ Client 종료

이 8단계를 기억하시면 됩니다.

2. import 부분

먼저 처음입니다.

import asyncio
import logging
from uuid import uuid4

import httpx

from a2a.client import A2ACardResolver, create_client
from a2a.types import (
    Message,
    Part,
    Role,
    SendMessageRequest,
)

하나씩 보겠습니다.

2-1. asyncio

import asyncio

Python의 비동기 프로그래밍을 위한 기본 라이브러리입니다.

우리 프로그램의 마지막 부분에 나옵니다.

asyncio.run(main())

이것은 쉽게 말하면:

"async def main()으로 만들어진 비동기 프로그램을 실제로 실행해라."

라는 의미입니다.

3. async def main()

이 부분이 프로그램의 중심입니다.

async def main() -> None:

일반적인 함수는:

def main():

인데 우리는:

async def main():

을 사용했습니다.

왜냐하면 A2A 통신 과정에서 네트워크 작업이 발생하기 때문입니다.

예를 들어:

agent_card = await resolver.get_agent_card()

서버에 HTTP 요청을 보내고 응답을 기다립니다.

또:

client = await create_client(agent_card)

그리고:

async for response in client.send_message(request):

모두 비동기 방식으로 동작합니다.

따라서 main()도 비동기 함수가 되어야 합니다.

4. -> None은 무엇인가?

async def main() -> None:

여기서:

-> None

은 이 함수가 특별한 반환값을 돌려주지 않는다는 의미입니다.

즉,

result = main()

처럼 어떤 결과를 받아서 사용하는 함수가 아니라 프로그램 실행을 담당하는 함수입니다.

5. logging 설정

다음입니다.

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

logging.basicConfig

logging.basicConfig(level=logging.INFO)

로그의 기본 출력 수준을 설정합니다.

그래서 우리가:

logger.info("Agent Card 조회 시작...")

이라고 하면 콘솔에:

INFO:__main__:Agent Card 조회 시작...

같은 메시지가 나타납니다.

logger

logger = logging.getLogger(__name__)

현재 Python 파일을 위한 Logger를 하나 만든 것입니다.

그래서 이후에는:

logger.info(...)
logger.error(...)

등을 사용할 수 있습니다.

이것은 단순히:

print(...)

보다 규모가 큰 프로그램에서 관리하기 좋습니다.

6. 서버 주소

base_url = "http://localhost:9999"

아주 중요합니다.

우리 Weather Server가:

localhost:9999

에서 실행되고 있기 때문입니다.

즉:

http://localhost:9999

는 A2A Weather Server의 기본 주소입니다.

7. httpx.AsyncClient

이 부분도 매우 중요합니다.

async with httpx.AsyncClient() as httpx_client:

httpx는 Python에서 HTTP 통신을 하기 위한 라이브러리입니다.

쉽게 말하면:

"웹 서버와 HTTP로 대화할 수 있게 해주는 도구"

입니다.

우리는 비동기 통신을 하기 때문에:

httpx.AsyncClient()

를 사용합니다.

왜 async with인가?

async with httpx.AsyncClient() as httpx_client:

는 HTTP Client를 만들고 사용한 뒤 자동으로 정리하는 구조입니다.

개념적으로:

HTTP Client 생성
     ↓
HTTP 통신
     ↓
사용 완료
     ↓
자동 정리

라고 이해하면 됩니다.

그래서 직접:

httpx_client.close()

같은 정리 코드를 작성할 필요가 없습니다.

8. Agent Card 조회

이제 A2A에서 매우 중요한 부분입니다.

resolver = A2ACardResolver(
    httpx_client=httpx_client,
    base_url=base_url,
)

여기서:

A2ACardResolver

는 무엇일까요?

이름 그대로입니다.

Agent Card를 찾아오는 역할을 하는 객체

입니다.

9. Agent Card란?

A2A를 공부하면서 이것은 꼭 기억해 두시는 것이 좋습니다.

Agent Card = "나는 어떤 Agent인가?"를 설명하는 명세서

라고 생각하면 됩니다.

예를 들어 우리의 Agent Card에는:

name:
Weather Agent

description:
An A2A agent that provides weather information

supportedInterfaces:
JSONRPC

version:
1.0.0

capabilities:
streaming = false
pushNotifications = false

skills:
weather

등이 들어 있습니다.

즉 Client가 무작정 서버에 질문하는 것이 아닙니다.

먼저:

"당신은 누구고, 무엇을 할 수 있습니까?"

라고 확인합니다.

그것이 Agent Card입니다.

10. 실제 Agent Card 조회

agent_card = await resolver.get_agent_card()

이 한 줄에서 실제 HTTP 요청이 발생합니다.

우리가 실행했을 때:

GET http://localhost:9999/.well-known/agent-card.json

이 요청이 발생했습니다.

그리고:

HTTP/1.1 200 OK

가 나왔습니다.

즉,

Client
  │
  │ GET /.well-known/agent-card.json
  ▼
Weather Server
  │
  │ Agent Card
  ▼
Client

입니다.

11. 왜 await인가?

agent_card = await resolver.get_agent_card()

Agent Card를 가져오는 데 시간이 걸립니다.

네트워크 통신이기 때문입니다.

그래서 Python에게:

"서버에서 응답이 올 때까지 이 작업을 기다려라."

라고 하는 것이:

await

입니다.

12. 예외 처리

다음입니다.

try:
    agent_card = await resolver.get_agent_card()
except Exception as e:
    logger.error(
        f'에이전트 카드 조회 실패: {e}',
        exc_info=True
    )
    raise RuntimeError(
        'Weather Server의 Agent Card를 가져오지 못했습니다.'
    ) from e

이 부분은 실무적으로 상당히 중요합니다.

정상적인 경우

Agent Card 요청
      ↓
200 OK
      ↓
Agent Card 받음
      ↓
계속 진행

문제가 생기는 경우

예를 들어 서버가 꺼져 있다면:

Agent Card 요청
      ↓
연결 실패
      ↓
Exception

이때 프로그램이 그냥 죽는 것이 아니라:

except Exception as e:

가 받아줍니다.

13. logger.error

logger.error(
    f'에이전트 카드 조회 실패: {e}',
    exc_info=True
)

오류 내용을 로그로 남깁니다.

특히:

exc_info=True

는 traceback까지 보여주도록 하는 옵션입니다.

개발할 때 매우 유용합니다.

14. raise RuntimeError

raise RuntimeError(
    'Weather Server의 Agent Card를 가져오지 못했습니다.'
) from e

여기서 중요한 것은 단순히 오류를 출력하고 끝내는 것이 아니라 명확한 오류를 다시 발생시킨다는 것입니다.

즉:

원래 오류
   ↓
로그 기록
   ↓
"Weather Server의 Agent Card를 가져오지 못했습니다."
라는 의미가 명확한 오류로 변환

됩니다.

from e는 원래 발생했던 오류와 새 오류의 연결관계를 유지합니다.

15. Agent Card 출력

print(agent_card)

여기서 우리가 앞에서 한 번 오류를 만났습니다.

처음에는:

agent_card.model_dump()

를 사용했었죠.

그런데 현재 a2a-sdk 1.2.0의 AgentCard는 우리가 기대했던 Pydantic 객체가 아니라 Protocol Buffers 기반 객체이기 때문에 model_dump()가 없습니다.

그래서:

print(agent_card)

를 사용합니다.

실행하면:

name: "Weather Agent"
description: "An A2A agent that provides weather information"
...

처럼 나옵니다.

16. 이제 중요한 변화: create_client

다음입니다.

client = await create_client(agent_card)

이 부분이 이번 디버깅에서 핵심이었습니다.

create_client의 역할

쉽게 말하면:

"이 Agent Card를 기준으로 실제 A2A 통신을 담당할 Client를 만들어라."

입니다.

구조적으로 보면:

Agent Card
    ↓
create_client()
    ↓
A2A Client

입니다.

17. 왜 await가 필요한가?

우리는 처음에:

client = create_client(agent_card)

라고 했습니다.

그러자:

'coroutine' object has no attribute 'send_message'

가 발생했습니다.

왜냐하면 create_client()가 실제 Client를 즉시 반환하는 것이 아니라 coroutine을 반환했기 때문입니다.

따라서:

client = await create_client(agent_card)

라고 해야 합니다.

즉:

create_client()
      ↓
coroutine
      ↓ await
실제 Client

입니다.

이 차이는 앞으로 Python 비동기 프로그래밍을 공부하면서 굉장히 중요합니다.

18. 사용자 질문 만들기

이제 실제 질문을 만듭니다.

message = Message(
    message_id=uuid4().hex,
    role=Role.ROLE_USER,
    parts=[Part(text="이번 주 서울 날씨 알려줘.")],
)

여기에는 A2A에서 중요한 개념이 여러 개 들어 있습니다.

19. Message

Message(...)

는 말 그대로 하나의 메시지입니다.

우리가 사람에게:

이번 주 서울 날씨 알려줘.

라고 말하는 것을 A2A가 이해할 수 있는 구조로 만든 것입니다.

20. message_id

message_id=uuid4().hex

메시지마다 고유한 ID를 부여합니다.

uuid4()는 랜덤 UUID를 생성합니다.

예:

51fe2f85bf404bb39afcc528d74dac72

왜 필요할까요?

서버 입장에서는 여러 사용자가 동시에 메시지를 보낼 수 있기 때문입니다.

예를 들어:

사용자 A → 메시지 1
사용자 B → 메시지 2
사용자 C → 메시지 3

이것들을 구분할 수 있어야 합니다.

그래서 Message마다 ID를 붙입니다.

21. Role

role=Role.ROLE_USER

이 메시지를 누가 작성했는지 알려줍니다.

현재는:

ROLE_USER

입니다.

즉:

"이 메시지는 사용자 측에서 보낸 메시지다."

라는 의미입니다.

A2A의 메시지 구조에서는 Agent 측 메시지도 존재할 수 있습니다.

개념적으로:

USER
 ↓
Message
 ↓
AGENT

라고 생각하시면 됩니다.

22. Part

이 부분도 매우 중요합니다.

parts=[
    Part(text="이번 주 서울 날씨 알려줘.")
]

왜 그냥:

text="이번 주 서울 날씨 알려줘."

라고 하지 않고 Part를 사용했을까요?

A2A의 메시지는 여러 종류의 콘텐츠를 포함할 수 있도록 설계되어 있기 때문입니다.

현재는:

Part
 └── text

입니다.

즉:

Message
 └── Part
      └── Text

구조입니다.

23. Message 구조를 그림으로 보면

현재 우리가 만든 Message는 사실 이렇게 생겼습니다.

Message
│
├── message_id
│      └── "51fe..."
│
├── role
│      └── ROLE_USER
│
└── parts
       │
       └── Part
            │
            └── text
                 └── "이번 주 서울 날씨 알려줘."

이 구조를 기억해 두시면 좋습니다.

24. SendMessageRequest

그런데 여기서 한 단계가 더 필요합니다.

request = SendMessageRequest(
    message=message,
)

왜 Message를 또 SendMessageRequest 안에 넣을까요?

이 둘은 역할이 다르기 때문입니다.

Message

Message

는 무슨 말을 하는가입니다.

예:

"이번 주 서울 날씨 알려줘."

SendMessageRequest

SendMessageRequest

는 그 Message를 서버에 보내기 위한 요청입니다.

쉽게 비유하면:

Message
= 편지 내용

SendMessageRequest
= 편지를 우체국에 보내기 위한 발송 요청

정도로 생각하시면 이해하기 쉽습니다.

25. 전체 구조

따라서:

message = Message(...)

를 만들고

request = SendMessageRequest(message=message)

로 감쌉니다.

구조는:

SendMessageRequest
│
└── Message
     │
     ├── message_id
     ├── role
     └── parts
          │
          └── Part
               │
               └── text

입니다.

이 구조가 현재 A2A Client 코드에서 가장 중요한 부분 중 하나입니다.

26. 실제 서버로 보내기

이제 드디어 핵심입니다.

async for response in client.send_message(request):

여기서 실제로 A2A 통신이 발생합니다.

27. 그런데 왜 await가 아니고 async for인가?

이것도 이번 과정에서 아주 중요한 부분이었습니다.

처음에는 우리가:

response = await client.send_message(request)

라고 생각하기 쉬웠습니다.

그런데 현재 SDK의 send_message()는:

AsyncIterator[StreamResponse]

를 반환합니다.

쉽게 말하면:

응답 하나를 반환하는 것이 아니라 응답 이벤트들을 순차적으로 제공할 수 있는 구조

입니다.

그래서:

async for

를 사용합니다.

28. await와 async for 차이

아주 중요한 차이입니다.

await

result = await some_async_function()

개념적으로:

비동기 작업
     ↓
기다림
     ↓
결과 하나

입니다.

async for

async for item in some_async_iterator():
    ...

개념적으로:

비동기 데이터
     ↓
이벤트 1
     ↓
이벤트 2
     ↓
이벤트 3
     ↓
...

를 순차적으로 받는 것입니다.

29. 현재 Weather Agent는 streaming=False인데?

여기서 재미있는 부분이 있습니다.

Agent Card에는:

streaming: false

라고 되어 있습니다.

즉 우리 Weather Agent는 스트리밍을 지원하지 않습니다.

그런데 Client API는:

async for response in client.send_message(request):

를 사용합니다.

이것은 모순이 아닙니다.

Client API 자체가 응답 이벤트를 AsyncIterator 형태로 제공하도록 설계되어 있기 때문입니다.

즉:

현재 Weather Agent
→ 실제 응답 이벤트는 사실상 한 번

하지만

Client API
→ 여러 이벤트를 받을 수 있는 공통 구조

라고 이해하시면 됩니다.

앞으로 더 복잡한 Agent에서는 이 구조가 훨씬 중요해집니다.

30. 서버에서는 무슨 일이 일어나는가?

Client가:

client.send_message(request)

를 실행하면 서버에서는 우리가 앞에서 만든 코드가 작동합니다.

대략:

Client
  │
  │ SendMessageRequest
  ▼
Weather Server
  │
  ▼
DefaultRequestHandler
  │
  ▼
WeatherAgentExecutor.execute()
  │
  ▼
WeatherCoreAgent.invoke()
  │
  ▼
"서울" 발견
  │
  ▼
서울 날씨 생성
  │
  ▼
EventQueue
  │
  ▼
Client

입니다.

이것이 바로 지금 우리가 만든 A2A의 핵심 흐름입니다.

31. 서버의 WeatherCoreAgent

앞에서 만든 서버를 기억하면:

class WeatherCoreAgent:
    async def invoke(self, user_query: str) -> str:

여기에서 사용자 질문을 받습니다.

현재 질문:

이번 주 서울 날씨 알려줘.

이 들어갑니다.

그리고:

for city, weather in MOCK_WEATHER_DATA.items():

를 돌면서:

서울
대전
부산

중에서 질문에 포함된 도시를 찾습니다.

서울을 발견하면:

맑음
22°C
습도 45%

등의 정보를 이용해 응답을 만듭니다.

32. 서버가 다시 Message를 만든다

서버에서는:

response_message = Message(
    ...
    role=Role.ROLE_AGENT,
    parts=[Part(text=result)],
)

처럼 Agent의 응답 Message를 만듭니다.

Client에서는:

ROLE_USER

였지만 서버에서는:

ROLE_AGENT

가 됩니다.

즉:

Client → ROLE_USER
Server → ROLE_AGENT

라는 관계가 만들어집니다.

33. Client가 응답을 받는다

다시 Client로 돌아옵니다.

async for response in client.send_message(request):

    print("\n" + "="*60)
    print("서버 응답 이벤트")
    print("="*60)
    print(response)

여기서 response에는 서버에서 전달한 응답 이벤트가 들어옵니다.

즉:

서버
  ↓
response
  ↓
print(response)

입니다.

34. 마지막으로 Client 종료

await client.close()

이제 A2A Client 사용이 끝났습니다.

따라서 Client가 사용한 리소스를 정리합니다.

구조적으로:

Client 생성
   ↓
통신
   ↓
통신 완료
   ↓
Client 종료

입니다.

35. 프로그램 실행 부분

마지막입니다.

if __name__=="__main__":
    asyncio.run(main())

이것도 Python에서 아주 중요한 패턴입니다.

name

Python 파일은 실행될 때 특별한 변수:

__name__

을 가지고 있습니다.

직접 실행하면:

__name__ == "__main__"

이 됩니다.

따라서:

if __name__ == "__main__":

은:

"이 파일을 직접 실행했을 때만 아래 코드를 실행하라."

라는 의미입니다.

36. asyncio.run(main())

그런데 main()은:

async def main():

입니다.

따라서 일반적으로:

main()

이라고 실행할 수 없습니다.

대신:

asyncio.run(main())

을 사용합니다.

즉:

asyncio.run()
      ↓
이벤트 루프 생성
      ↓
main() 실행
      ↓
비동기 작업 처리
      ↓
main() 종료
      ↓
이벤트 루프 정리

라고 이해하면 됩니다.

37. 이 코드 전체를 한 장으로 정리하면

제가 보기에는 지금 단계에서 이 그림을 기억하시는 것이 가장 중요합니다.

                 [A2A Client]
                      │
                      │
             ① Server 주소 설정
                      │
                      ▼
             A2ACardResolver
                      │
             ② Agent Card 조회
                      │
                      ▼
              [Agent Card]
                      │
                      ▼
              ③ create_client()
                      │
                      ▼
                 [Client]
                      │
                      │
             ④ Message 생성
                      │
                      ▼
                [Message]
                      │
                      ▼
          ⑤ SendMessageRequest
                      │
                      ▼
             [A2A Request]
                      │
                      │
             ⑥ send_message()
                      │
                      ▼
════════════════════════════════════
              HTTP / A2A
════════════════════════════════════
                      │
                      ▼
              [Weather Server]
                      │
                      ▼
            DefaultRequestHandler
                      │
                      ▼
          WeatherAgentExecutor
                      │
                      ▼
            WeatherCoreAgent
                      │
                      ▼
                서울 날씨
                      │
                      ▼
                 EventQueue
                      │
════════════════════════════════════
                      │
                      ▼
             ⑦ Client 응답 수신
                      │
                      ▼
                response
                      │
                      ▼
             ⑧ client.close()

38. 이 코드에서 꼭 기억해야 할 핵심 8개

지금은 모든 세부사항을 외우려고 하지 않으셔도 됩니다.

다음 8개 개념을 이해하는 것이 중요합니다.

단계코드핵심 의미

①	base_url	Agent Server 주소
②	A2ACardResolver	Agent Card를 가져오는 도구
③	get_agent_card()	Agent의 능력/정보 확인
④	create_client()	실제 A2A Client 생성
⑤	Message	사용자의 메시지 표현
⑥	SendMessageRequest	메시지를 서버로 보내기 위한 요청
⑦	send_message()	실제 A2A 통신
⑧	async for response	서버 응답 이벤트 수신

그리고 비동기 측면에서는 세 가지를 기억하면 됩니다.

async def
   ↓
비동기 함수

await
   ↓
비동기 작업의 완료를 기다림

async for
   ↓
비동기적으로 여러 이벤트/응답을 순차적으로 받음

39. 특히 이번에 배운 가장 중요한 것

이번 A2A Client를 통해 사실 A2A만 배운 것이 아닙니다.

Python 비동기 프로그래밍의 중요한 개념도 같이 배웠습니다.

첫 번째

agent_card = await resolver.get_agent_card()

→ 비동기 작업의 결과를 기다린다.

두 번째

client = await create_client(agent_card)

→ coroutine을 실제 결과로 받아온다.

세 번째

async for response in client.send_message(request):

→ 비동기 iterator에서 이벤트를 순차적으로 받는다.

이 세 가지는 앞으로 AI Agent, FastAPI, A2A, MCP 등을 공부할 때 계속 등장합니다.

40. 그리고 A2A의 본질을 한 문장으로

지금 만든 프로그램을 아주 쉽게 표현하면:

"상대 Agent가 어떤 능력을 가지고 있는지 Agent Card로 확인한 후, 정해진 A2A 형식의 Message/Request를 만들어 상대 Agent에게 보내고, 그 Agent가 처리한 결과를 다시 이벤트 형태로 받는 것"

입니다.

그래서 A2A의 핵심 구조를 아주 간단히 줄이면:

Agent Card
    ↓
"너 누구야? 뭘 할 수 있어?"
    ↓
Message
    ↓
"이 일을 해줘."
    ↓
Request
    ↓
"이 요청을 공식적으로 전달할게."
    ↓
Agent
    ↓
"처리했어."
    ↓
Response Event

입니다.

그리고 AgentCard → Client → Message → Request → AsyncIterator → Response라는 현재 A2A SDK의 실제 구조를 직접 확인하였습니다.

다음 단계에서는 이 Client 코드와 앞서 만든 Server 코드를 좌우로 놓고, Client의 한 줄이 Server의 어느 코드로 들어가서 어떻게 다시 Client로 돌아오는지를 1:1로 연결해서 보는 것이 가장 좋습니다. 그러면 지금까지 배운 A2A가 훨씬 선명해집니다.

저작자표시 비영리 변경금지 (새창열림)