a2a-sdk 1.2.0 기반 — 코드의 구조부터 실행 흐름까지

특히 이번 코드에서 중요한 점은 다음과 같습니다.

WeatherCoreAgent: 사용자의 질문을 분석하고 날씨 정보를 반환하는 핵심 에이전트
WeatherAgentExecutor: 핵심 에이전트를 A2A 서버의 실행 구조에 연결하는 실행기
EventQueue: 에이전트가 생성한 응답을 A2A 서버에 전달하는 이벤트 큐
DefaultRequestHandler: 외부에서 들어온 A2A 요청을 관리하는 요청 처리기
AgentCard: 외부 에이전트가 우리 서버의 기능과 연결 정보를 확인할 수 있도록 제공하는 명세
FastAPI와 A2A Routes: 실제 HTTP 요청을 받아 처리하는 웹 서버의 구조

이번에는 단순히 각 코드가 무엇을 하는지 설명하는 데 그치지 않고, 왜 이런 구조로 작성했는지, 각 클래스와 함수가 서로 어떻게 연결되는지, 실제로 실행하면 어떤 순서로 작동하는지까지 살펴보겠습니다.

1. 먼저 전체 구조를 이해하자

코드를 자세히 살펴보기 전에 전체 구조를 먼저 이해하는 것이 좋습니다.

이번 서버는 크게 다음과 같이 구성되어 있습니다.

                  사용자 또는 외부 AI Agent
                           |
                           | A2A 요청
                           v
                +----------------------+
                |      FastAPI         |
                |                      |
                |   HTTP 요청 수신     |
                +----------------------+
                           |
                           v
                +----------------------+
                |    A2A Routes        |
                |                      |
                | Agent Card           |
                | JSON-RPC             |
                | REST                 |
                +----------------------+
                           |
                           v
                +----------------------+
                | DefaultRequestHandler|
                |                      |
                | A2A 요청 관리        |
                +----------------------+
                           |
                           v
                +----------------------+
                | WeatherAgentExecutor |
                |                      |
                | execute()            |
                +----------------------+
                           |
                           v
                +----------------------+
                |  WeatherCoreAgent    |
                |                      |
                | 사용자 질문 분석     |
                | 도시명 검색           |
                | 날씨 데이터 조회     |
                +----------------------+
                           |
                           v
                +----------------------+
                |   날씨 응답 생성     |
                +----------------------+
                           |
                           v
                +----------------------+
                |     EventQueue       |
                |                      |
                | 응답 이벤트 전달     |
                +----------------------+
                           |
                           v
                +----------------------+
                | A2A 서버 응답 처리   |
                +----------------------+
                           |
                           v
                    사용자에게 응답

이 구조를 이해할 때 가장 중요한 것은 핵심 에이전트와 A2A 서버가 서로 다른 역할을 한다는 점입니다.

구성 요소역할쉽게 비유하면

FastAPI	HTTP 요청을 수신	건물의 출입구
A2A Routes	요청을 적절한 경로로 전달	안내 데스크
DefaultRequestHandler	A2A 요청과 작업 관리	업무 관리자
WeatherAgentExecutor	실제 에이전트 실행	담당 직원
WeatherCoreAgent	날씨 질문 처리	날씨 전문가
EventQueue	실행 결과를 전달	결과 전달 통로
AgentCard	에이전트의 기능과 접속 정보 제공	명함 또는 서비스 안내서
InMemoryTaskStore	작업 정보 임시 저장	임시 업무 기록장

예를 들어 외부 에이전트가 다음과 같은 질문을 보낸다고 가정하겠습니다.

서울의 날씨를 알려줘.

이 질문은 FastAPI에 도착합니다.

FastAPI는 A2A 라우트를 통해 요청을 전달하고, 요청 처리기는 WeatherAgentExecutor를 실행합니다.

실행기는 WeatherCoreAgent에 질문을 전달합니다.

핵심 에이전트는 질문에서 '서울'이라는 도시명을 찾아 목업 데이터에서 날씨 정보를 가져옵니다.

마지막으로 실행기는 결과를 A2A 메시지로 변환하고 이벤트 큐에 등록합니다.

이것이 이번 코드의 전체적인 작동 방식입니다.

2. 라이브러리 Import

먼저 프로그램에서 사용할 라이브러리를 불러오는 부분입니다.

import uvicorn

from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore
from a2a.types import (AgentCapabilities, AgentCard, AgentSkill)
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue

from fastapi import FastAPI
from a2a.utils.constants import TransportProtocol

from a2a.server.routes import (
    add_a2a_routes_to_fastapi,
    create_agent_card_routes,
    create_jsonrpc_routes,
    create_rest_routes,
)

from a2a.types import (
    AgentCard, 
    AgentCapabilities, 
    AgentInterface, 
    AgentSkill,
    Message,
    Part,
    Role,
)

2-1. uvicorn

import uvicorn

uvicorn은 Python으로 작성한 ASGI 애플리케이션을 실행하는 서버입니다.

여기서 ASGI는 비동기 Python 웹 애플리케이션을 위한 표준 인터페이스입니다.

FastAPI는 ASGI를 지원하는 웹 프레임워크입니다.

그런데 FastAPI 애플리케이션을 작성했다고 해서 그 자체로 외부 요청을 받을 수 있는 것은 아닙니다.

실제로 HTTP 요청을 수신하고 응답을 보내는 서버가 필요합니다.

그 역할을 하는 것이 Uvicorn입니다.

이번 코드의 마지막 부분에서 다음과 같이 사용합니다.

uvicorn.run(app, host="0.0.0.0", port=9999)

이 명령은 app이라는 FastAPI 애플리케이션을 9999번 포트에서 실행합니다.

각 매개변수의 의미는 다음과 같습니다.

매개변수의미

app	실행할 FastAPI 애플리케이션
host="0.0.0.0"	모든 네트워크 인터페이스에서 요청 수신
port=9999	서버가 사용할 포트 번호

참고로 0.0.0.0은 외부에서 접속할 수 있는 모든 네트워크 인터페이스에 바인딩한다는 의미입니다.

다만 실제로 외부에서 접속할 수 있는지는 방화벽이나 네트워크 설정에 따라 달라집니다.

2-2. DefaultRequestHandler

from a2a.server.request_handlers import DefaultRequestHandler

DefaultRequestHandler는 A2A 요청을 처리하는 기본 요청 처리기입니다.

외부 에이전트가 A2A 요청을 보내면 요청 처리기는 해당 요청을 해석하고, 필요한 작업을 실행하도록 연결합니다.

이번 코드에서는 다음과 같이 생성합니다.

request_handler = DefaultRequestHandler(
    agent_executor=weather_agent_executor,
    task_store=InMemoryTaskStore(),
    agent_card=agent_card,
)

여기서 세 가지 요소가 중요합니다.

agent_executor: 실제 작업을 수행할 실행기
task_store: 작업 상태와 관련 정보를 저장할 저장소
agent_card: 이 에이전트의 기능과 정보를 설명하는 명세

쉽게 말하면 DefaultRequestHandler는 외부 요청을 받아서 어떤 에이전트를 실행할지 연결하고, A2A 작업을 관리하는 역할을 합니다.

여기서 주의할 점은 DefaultRequestHandler가 직접 날씨를 조회하는 것은 아니라는 사실입니다.

날씨를 조회하는 실제 작업은 WeatherCoreAgent가 담당합니다.

2-3. InMemoryTaskStore

from a2a.server.tasks import InMemoryTaskStore

InMemoryTaskStore는 작업 정보를 메모리에 저장하는 저장소입니다.

여기서 말하는 작업(Task)은 A2A에서 에이전트가 요청받아 수행하는 하나의 작업 단위입니다.

예를 들어 다음과 같은 요청이 있다고 가정하겠습니다.

서울의 날씨를 알려줘.

A2A 서버는 이 요청을 하나의 작업으로 관리할 수 있습니다.

작업에는 식별자와 상태 등의 정보가 연결될 수 있습니다.

InMemoryTaskStore는 이러한 작업 정보를 프로그램이 실행되는 동안 메모리에 저장합니다.

InMemoryTaskStore의 특징

항목설명

저장 위치	서버의 메모리
속도	일반적으로 빠름
서버 재시작	저장된 작업 정보가 사라짐
여러 서버 간 공유	기본적으로 불가능
사용 목적	학습, 테스트, 간단한 서버

이번처럼 간단한 날씨 에이전트를 구현하고 테스트할 때는 충분합니다.

하지만 실제 서비스에서 장시간 실행되는 작업이나 여러 서버가 함께 처리하는 작업을 관리하려면 데이터베이스 등 별도의 저장소를 고려해야 합니다.

여기서 한 가지 구분할 점이 있습니다.

InMemoryTaskStore는 작업 정보를 저장하는 것이지, 날씨 데이터를 저장하는 것은 아닙니다.

날씨 데이터는 별도로 정의한 MOCK_WEATHER_DATA에 들어 있습니다.

2-4. AgentExecutor와 RequestContext

from a2a.server.agent_execution import AgentExecutor, RequestContext

이 두 클래스는 이번 코드에서 매우 중요합니다.

AgentExecutor

AgentExecutor는 A2A 서버에서 실제 에이전트를 실행하기 위한 인터페이스입니다.

이번 코드에서는 다음과 같이 상속합니다.

class WeatherAgentExecutor(AgentExecutor):

즉, WeatherAgentExecutor는 A2A에서 요구하는 실행기 인터페이스를 구현한 클래스입니다.

이 클래스에는 핵심적으로 두 가지 메서드가 있습니다.

async def execute(...)

async def cancel(...)

execute()는 요청받은 작업을 실행합니다.
cancel()은 실행 중인 작업의 취소 요청을 처리합니다.

이번 코드에서는 execute()를 구현하여 날씨 에이전트를 실행하고, cancel()은 지원하지 않는다고 명시했습니다.

RequestContext

RequestContext는 현재 처리 중인 A2A 요청과 관련된 정보를 제공하는 객체입니다.

이번 코드에서는 다음과 같이 사용합니다.

user_query = context.get_user_input()

이 코드는 현재 요청에 포함된 사용자의 입력을 가져옵니다.

예를 들어 외부에서 다음과 같은 질문이 들어왔다면,

서울의 날씨를 알려줘.

context.get_user_input()은 이 질문을 가져옵니다.

또한 다음과 같은 정보도 접근할 수 있습니다.

context.context_id
context.task_id

각각의 의미는 다음과 같습니다.

속성의미

context_id	대화 또는 요청 문맥의 식별자
task_id	현재 작업의 식별자
get_user_input()	사용자 입력을 가져오는 메서드

이 식별자들은 A2A 메시지와 작업을 연결하는 데 사용됩니다.

특히 하나의 대화에서 여러 작업이 발생할 수 있기 때문에 문맥과 작업을 구분하는 것이 중요합니다.

2-5. EventQueue

from a2a.server.events import EventQueue

EventQueue는 에이전트가 생성한 이벤트를 A2A 요청 처리 흐름에 전달하는 역할을 합니다.

이번 코드에서는 다음과 같이 사용합니다.

await event_queue.enqueue_event(response_message)

여기서 response_message는 에이전트가 생성한 응답 메시지입니다.

이 메시지를 이벤트 큐에 등록하면 A2A 서버가 해당 이벤트를 요청 처리 과정에서 사용할 수 있습니다.

이벤트 큐가 필요한 이유는 A2A가 단순히 함수의 반환값만으로 모든 응답을 처리하는 구조가 아니기 때문입니다.

작업의 상태 변화나 메시지 등의 이벤트를 전달하는 구조를 지원합니다.

예를 들어 다음과 같은 이벤트를 생각할 수 있습니다.

에이전트가 작업을 시작했다는 이벤트
에이전트가 중간 결과를 생성했다는 이벤트
에이전트가 최종 응답을 생성했다는 이벤트

이번 코드는 그중 최종 응답 메시지를 이벤트 큐에 등록하는 간단한 형태입니다.

다만 이번 코드에서는 AgentCapabilities에서 스트리밍을 비활성화했습니다.

따라서 이벤트 큐를 사용한다고 해서 반드시 클라이언트에 실시간 스트리밍이 제공되는 것은 아닙니다.

이벤트 큐를 사용한다는 것과 스트리밍 기능을 활성화한다는 것은 서로 다른 개념입니다.

2-6. FastAPI

from fastapi import FastAPI

FastAPI는 Python으로 웹 API 서버를 구축할 수 있도록 도와주는 프레임워크입니다.

이번 코드에서는 A2A 서버의 HTTP 요청을 받는 웹 애플리케이션을 생성하는 데 사용합니다.

app = FastAPI(
    title="A2A Weather Server",
    description="A simple A2A weather agent server",
    version="1.0.0",
)

여기서 app은 FastAPI 애플리케이션 객체입니다.

이 객체에 A2A 라우트를 등록하고, 마지막에 Uvicorn으로 실행합니다.

2-7. TransportProtocol

from a2a.utils.constants import TransportProtocol

TransportProtocol은 A2A에서 사용하는 전송 프로토콜을 지정할 때 사용합니다.

이번 코드에서는 다음과 같이 사용합니다.

protocol_binding=TransportProtocol.JSONRPC.value

즉, 이 에이전트는 JSON-RPC 방식으로 통신한다는 의미입니다.

JSON-RPC는 JSON 형식으로 원격 프로시저 호출을 수행하는 프로토콜입니다.

예를 들어 외부 에이전트가 서버에 다음과 같은 요청을 보낼 수 있습니다.

{
  "jsonrpc": "2.0",
  "method": "message/send",
  "params": {
    "message": {
      "messageId": "123",
      "role": "user",
      "parts": [
        {
          "text": "서울의 날씨를 알려줘."
        }
      ]
    }
  },
  "id": "request-1"
}

위 JSON은 A2A JSON-RPC 요청의 개념을 보여주는 예시입니다. 실제 요청 형식은 SDK 버전과 메서드에 맞춰야 합니다.

여기서 핵심은 외부 에이전트가 JSON 형식으로 요청을 보내고, 서버가 이를 처리한 뒤 응답을 반환한다는 것입니다.

2-8. A2A Routes

from a2a.server.routes import (
    add_a2a_routes_to_fastapi,
    create_agent_card_routes,
    create_jsonrpc_routes,
    create_rest_routes,
)

이번 코드에서 이전 구현과 비교해 중요한 부분입니다.

A2A SDK 버전이 바뀌면서 기존에 사용하던 클래스나 함수가 제공되지 않는 경우가 있습니다.

이번 코드는 FastAPI를 직접 생성하고, A2A SDK에서 제공하는 라우트 생성 함수를 사용하여 서버를 구성합니다.

각 함수의 역할은 다음과 같습니다.

함수역할

create_agent_card_routes()	Agent Card를 조회할 수 있는 경로 생성
create_jsonrpc_routes()	JSON-RPC 요청을 처리하는 경로 생성
create_rest_routes()	REST 방식의 A2A 요청 경로 생성
add_a2a_routes_to_fastapi()	생성한 라우트를 FastAPI에 등록

여기서 중요한 개념은 라우트(Route) 입니다.

라우트는 특정 URL로 들어온 요청을 어떤 코드가 처리할지 연결하는 규칙입니다.

예를 들어 다음과 같은 URL이 있다고 가정하겠습니다.

http://localhost:9999/

이 주소로 들어온 JSON-RPC 요청을 처리하도록 라우트를 등록할 수 있습니다.

또한 다음 주소로 Agent Card를 조회할 수 있습니다.

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

이처럼 서로 다른 URL에 서로 다른 기능을 연결하는 것이 라우트의 역할입니다.

3. 목업 날씨 데이터

이제 실제 날씨 데이터를 정의한 부분을 살펴보겠습니다.

MOCK_WEATHER_DATA = {
    "서울": {
        "condition": "맑음",
        "temperature_c": 22,
        "min_temp_c": 12,
        "max_temp_c": 22,
        "humidity": 45,
        "summary": "서울은 맑고 쾌청합니다.",
    },
    "대전": {
        "condition": "흐림",
        "temperature_c": 20,
        "min_temp_c": 10,
        "max_temp_c": 18,
        "humidity": 60,
        "summary": "대전은 흐리고 바람이 약간 붑니다.",
    },
    "부산": {
        "condition": "비",
        "temperature_c": 21,
        "min_temp_c": 17,
        "max_temp_c": 24,
        "humidity": 70,
        "summary": "부산은 비가 내리고 습도가 높습니다.",
    },    
}

3-1. 목업(Mock)이란?

목업은 실제 데이터를 대신하여 테스트나 개발에 사용하는 가상의 데이터를 의미합니다.

예를 들어 실제 날씨 정보를 제공하려면 기상청 API 등의 외부 서비스를 연결해야 합니다.

그러나 이번 코드는 외부 API를 사용하지 않습니다.

대신 미리 정의한 날씨 데이터를 사용합니다.

따라서 사용자가 서울의 날씨를 물어보면 실제 서울의 현재 날씨가 아니라 코드에 저장된 데이터를 반환합니다.

이것이 목업 데이터입니다.

3-2. 딕셔너리 구조

이 데이터는 Python의 딕셔너리로 구성되어 있습니다.

딕셔너리는 키(Key)와 값(Value)을 쌍으로 저장하는 자료형입니다.

예를 들어 다음과 같습니다.

weather = {
    "condition": "맑음",
    "temperature_c": 22,
    "humidity": 45,
}

여기서 각 항목은 다음과 같습니다.

키값의미

condition	맑음	날씨 상태
temperature_c	22	현재 기온
humidity	45	습도

따라서 다음과 같이 데이터를 가져올 수 있습니다.

weather["condition"]

결과:

맑음

또한 다음과 같이 사용할 수 있습니다.

weather["temperature_c"]

결과:

이번 코드에서는 도시명을 가장 바깥쪽 딕셔너리의 키로 사용합니다.

MOCK_WEATHER_DATA["서울"]

그러면 서울의 날씨 정보 전체를 가져올 수 있습니다.

그 안에서 특정 항목만 가져오려면 다음과 같이 작성합니다.

MOCK_WEATHER_DATA["서울"]["temperature_c"]

결과는 다음과 같습니다.

이처럼 딕셔너리 안에 또 다른 딕셔너리가 들어 있는 구조를 중첩 딕셔너리라고 합니다.

3-3. 데이터에서 주의할 점

현재 코드의 대전 데이터는 다음과 같습니다.

"temperature_c": 20,
"min_temp_c": 10,
"max_temp_c": 18,

현재 기온이 20도인데 최고 기온은 18도로 설정되어 있습니다.

실제 날씨 데이터라면 일반적으로 최고 기온이 현재 기온보다 낮게 나오는 것은 어색할 수 있습니다.

다만 이번 코드는 목업이기 때문에 테스트 데이터의 일관성을 위해 수정할 수 있습니다.

또한 min_temp_c, max_temp_c, summary는 현재 핵심 에이전트에서 사용하지 않습니다.

이 항목들은 데이터에 존재하지만, 현재 응답을 생성하는 코드에서는 condition, temperature_c, humidity만 사용하고 있습니다.

이러한 차이를 이해하는 것이 중요합니다.

데이터에 정의되어 있다고 해서 프로그램에서 반드시 사용하는 것은 아닙니다.

4. WeatherCoreAgent — 핵심 에이전트

이제 가장 중요한 부분 중 하나인 WeatherCoreAgent를 살펴보겠습니다.

class WeatherCoreAgent:
    """ 사용자의 질문에서 도시명을 찾아, 목업 날씨 데이터를 반환하는 핵심 에이전트 """
    async def invoke(self, user_query:str)->str:    
        
        # 질문에서 도시명 찾기    
        for city, weather in MOCK_WEATHER_DATA.items():
            if city in user_query:
                condition = weather["condition"]
                temperature = weather["temperature_c"]
                humidity = weather["humidity"]
                
                return (
                    f"{city}의 현재 날씨는 {condition}입니다."
                    f"기온은 {temperature}°C 이며,"
                    f"습도는 {humidity}% 입니다."
                )

        # 도시명을 찾지 못한 경우        
        return "질의하신 지역의 날씨 정보를 찾을 수 없습니다...(예: 서울특별시, 부산광역시, 대전광역시)"

이 클래스는 실제 날씨 질문을 처리하는 핵심 로직을 담당합니다.

앞서 설명한 WeatherAgentExecutor가 A2A 서버와 핵심 에이전트를 연결하는 역할이라면, WeatherCoreAgent는 사용자의 질문을 직접 처리하는 역할입니다.

4-1. 클래스 정의

class WeatherCoreAgent:

class는 Python에서 클래스를 정의할 때 사용하는 키워드입니다.

클래스는 객체를 만들기 위한 설계도라고 이해하면 됩니다.

예를 들어 자동차를 만든다고 생각해 보겠습니다.

자동차 설계도에는 자동차가 어떻게 움직이고 어떤 기능을 제공하는지 정의되어 있습니다.

마찬가지로 WeatherCoreAgent 클래스에는 날씨 질문을 처리하는 기능이 정의되어 있습니다.

이 클래스를 실제로 사용하려면 객체를 생성해야 합니다.

agent = WeatherCoreAgent()

그러면 agent라는 객체가 생성됩니다.

이번 코드에서는 나중에 WeatherAgentExecutor 안에서 객체를 생성합니다.

self.agent = WeatherCoreAgent()

4-2. async def invoke()

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

이 한 줄에는 Python의 중요한 개념이 여러 가지 들어 있습니다.

하나씩 살펴보겠습니다.

① async

async는 비동기 함수를 정의할 때 사용하는 키워드입니다.

일반적인 함수는 다음과 같이 작성합니다.

def hello():
    return "Hello"

반면 비동기 함수는 다음과 같이 작성합니다.

async def hello():
    return "Hello"

비동기 함수는 호출하면 일반적인 결과 문자열을 즉시 반환하는 대신 코루틴 객체를 반환합니다.

실행 결과를 얻으려면 일반적으로 await를 사용해야 합니다.

result = await hello()

이번 코드에서도 다음과 같이 호출합니다.

result = await self.agent.invoke(user_query)

여기서 await는 비동기 함수의 실행 결과를 기다리는 역할을 합니다.

② self

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

self는 해당 메서드를 호출한 객체 자신을 가리킵니다.

예를 들어 다음과 같이 객체를 생성했다고 가정하겠습니다.

agent = WeatherCoreAgent()

이후 다음과 같이 호출합니다.

await agent.invoke("서울의 날씨를 알려줘.")

Python은 메서드에 해당 객체 자신을 self로 전달합니다.

따라서 self를 통해 객체의 속성이나 다른 메서드에 접근할 수 있습니다.

이번 WeatherCoreAgent에서는 현재 self를 직접 사용하지 않지만, 클래스의 인스턴스 메서드이므로 첫 번째 매개변수로 선언되어 있습니다.

③ user_query: str

user_query: str

사용자의 질문을 문자열로 전달받는다는 의미입니다.

예를 들어 다음과 같은 질문이 전달될 수 있습니다.

"서울의 날씨를 알려줘."

여기서 : str은 타입 힌트입니다.

즉, 이 매개변수에는 문자열을 전달하는 것이 의도되어 있다는 것을 나타냅니다.

Python은 타입 힌트를 반드시 강제하지는 않습니다.

④ -> str

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

-> str은 함수가 문자열을 반환하도록 설계되었다는 것을 나타냅니다.

예를 들어 다음과 같은 문자열을 반환합니다.

서울의 현재 날씨는 맑음입니다.기온은 22°C 이며,습도는 45% 입니다.

여기서 반환 타입이 문자열이라는 것은 이후 WeatherAgentExecutor가 결과를 A2A 메시지로 변환하는 데 중요한 역할을 합니다.

4-3. for문으로 도시명 검색하기

for city, weather in MOCK_WEATHER_DATA.items():

이 부분은 목업 데이터에 저장된 도시명을 하나씩 확인하는 코드입니다.

MOCK_WEATHER_DATA.items()는 딕셔너리의 키와 값을 함께 가져옵니다.

현재 데이터에는 다음 세 도시가 있습니다.

서울
대전
부산

따라서 for문은 세 번 반복됩니다.

각 반복에서 다음과 같은 값을 얻습니다.

반복cityweather

1회	서울	서울 날씨 딕셔너리
2회	대전	대전 날씨 딕셔너리
3회	부산	부산 날씨 딕셔너리

예를 들어 첫 번째 반복에서는 다음과 같은 상태가 됩니다.

city = "서울"

weather = {
    "condition": "맑음",
    "temperature_c": 22,
    "min_temp_c": 12,
    "max_temp_c": 22,
    "humidity": 45,
    "summary": "서울은 맑고 쾌청합니다.",
}

두 번째 반복에서는 city가 "대전"으로 바뀌고, weather에는 대전의 데이터가 들어갑니다.

이런 식으로 모든 도시를 순서대로 확인합니다.

4-4. if city in user_query

if city in user_query:

이 코드는 사용자의 질문에 현재 검색 중인 도시명이 포함되어 있는지 확인합니다.

예를 들어 다음과 같은 질문이 들어왔다고 가정하겠습니다.

user_query = "서울의 날씨를 알려줘."

첫 번째 반복에서 다음 조건을 검사합니다.

if "서울" in "서울의 날씨를 알려줘.":

결과는 True입니다.

따라서 조건문 안에 있는 코드가 실행됩니다.

반면 다음과 같은 질문이 들어왔다고 가정하겠습니다.

user_query = "부산의 날씨를 알려줘."

첫 번째 반복에서는 다음과 같이 검사합니다.

if "서울" in "부산의 날씨를 알려줘.":

결과는 False입니다.

그러면 첫 번째 반복을 마치고 두 번째 도시인 대전을 확인합니다.

대전도 포함되어 있지 않으므로 세 번째 반복으로 넘어갑니다.

세 번째 반복에서 부산을 발견하면 조건문 안의 코드가 실행됩니다.

이것이 현재 코드에서 도시명을 찾는 방식입니다.

다만 이 방식은 단순한 문자열 검색입니다.

따라서 다음과 같은 질문은 처리할 수 있습니다.

서울 날씨 알려줘.
서울의 현재 기온은?
오늘 서울은 맑아?

하지만 도시명이 다르게 표현되거나 오타가 있으면 찾지 못할 수 있습니다.

예를 들어 "서울특별시"라는 표현은 "서울"을 포함하므로 현재 코드에서도 정상적으로 검색됩니다.

반면 "서울 날씨"를 "수도권 날씨"라고 표현하면 현재 코드에서는 서울을 찾지 못합니다.

4-5. 날씨 데이터 가져오기

condition = weather["condition"]
temperature = weather["temperature_c"]
humidity = weather["humidity"]

도시명을 찾으면 해당 도시의 날씨 데이터를 가져옵니다.

예를 들어 서울이 검색되었다면 다음과 같은 값이 만들어집니다.

condition = "맑음"
temperature = 22
humidity = 45

각 변수의 의미는 다음과 같습니다.

변수값의미

condition	맑음	현재 날씨
temperature	22	현재 기온
humidity	45	현재 습도

여기서 weather는 현재 검색된 도시의 날씨 딕셔너리입니다.

따라서 weather["condition"]은 해당 도시의 날씨 상태를 의미합니다.

4-6. f-string으로 응답 만들기

return (
    f"{city}의 현재 날씨는 {condition}입니다."
    f"기온은 {temperature}°C 이며,"
    f"습도는 {humidity}% 입니다."
)

이 부분은 사용자에게 반환할 최종 응답을 만드는 코드입니다.

여기서 f가 붙은 문자열을 f-string이라고 합니다.

f-string은 문자열 안에 변수의 값을 삽입할 수 있게 해줍니다.

예를 들어 다음과 같습니다.

city = "서울"
temperature = 22

message = f"{city}의 기온은 {temperature}도입니다."

결과는 다음과 같습니다.

서울의 기온은 22도입니다.

이번 코드에서는 세 개의 f-string을 연속해서 작성했습니다.

f"{city}의 현재 날씨는 {condition}입니다."
f"기온은 {temperature}°C 이며,"
f"습도는 {humidity}% 입니다."

Python에서는 괄호 안에 인접한 문자열 리터럴을 작성하면 하나의 문자열로 연결합니다.

따라서 위 코드는 다음과 같이 작성한 것과 같은 결과를 냅니다.

return (
    f"{city}의 현재 날씨는 {condition}입니다."
    + f"기온은 {temperature}°C 이며,"
    + f"습도는 {humidity}% 입니다."
)

다만 현재 코드에서는 문자열 사이에 공백이나 줄바꿈을 넣지 않았습니다.

그래서 실제 결과는 다음과 같습니다.

서울의 현재 날씨는 맑음입니다.기온은 22°C 이며,습도는 45% 입니다.

문장을 좀 더 자연스럽게 표현하려면 다음과 같이 수정할 수 있습니다.

return (
    f"{city}의 현재 날씨는 {condition}입니다. "
    f"기온은 {temperature}°C이며, "
    f"습도는 {humidity}%입니다."
)

그러면 다음과 같은 결과가 나옵니다.

서울의 현재 날씨는 맑음입니다. 기온은 22°C이며, 습도는 45%입니다.

이처럼 문자열을 작성할 때는 문장 사이의 공백도 신경 써야 합니다.

4-7. 도시명을 찾지 못한 경우

return "질의하신 지역의 날씨 정보를 찾을 수 없습니다...(예: 서울특별시, 부산광역시, 대전광역시)"

이 코드는 for문에서 도시명을 찾지 못했을 때 실행됩니다.

예를 들어 사용자가 다음과 같이 질문했다고 가정하겠습니다.

인천 날씨 알려줘.

현재 목업 데이터에는 인천이 없습니다.

따라서 서울, 대전, 부산을 모두 검색해도 일치하는 도시명을 찾지 못합니다.

그러면 마지막 return이 실행됩니다.

결과는 다음과 같습니다.

질의하신 지역의 날씨 정보를 찾을 수 없습니다...(예: 서울특별시, 부산광역시, 대전광역시)

여기서 중요한 점은 return이 실행되면 함수가 즉시 종료된다는 것입니다.

따라서 도시명을 발견한 경우에는 해당 도시의 날씨를 반환하고, 발견하지 못한 경우에는 안내 메시지를 반환합니다.

핵심 정리

WeatherCoreAgent의 역할은 다음 네 단계로 정리할 수 있습니다.

사용자 질문을 입력받습니다.
질문에서 도시명을 검색합니다.
도시명이 발견되면 해당 도시의 날씨 데이터를 조회합니다.
날씨 정보를 문자열로 반환하거나, 도시를 찾지 못하면 안내 메시지를 반환합니다.

여기까지는 A2A 프로토콜과 관계없이 독립적으로 동작할 수 있는 일반적인 Python 에이전트입니다.

이제 이 핵심 에이전트를 A2A 서버에 연결하는 부분을 살펴보겠습니다.

5. WeatherAgentExecutor — 핵심 에이전트와 A2A를 연결하는 실행기

class WeatherAgentExecutor(AgentExecutor):
    """A2A 요청을 실제 WeatherCoreAgent에 전달하고, 결과를 이벤트 큐에 등록하는 실행기"""

    def __init__(self):
        self.agent = WeatherCoreAgent()

이 클래스는 A2A 서버와 실제 날씨 에이전트를 연결하는 역할을 합니다.

앞에서 WeatherCoreAgent는 사용자의 질문을 처리하는 핵심 로직이라고 설명했습니다.

그런데 외부에서 A2A 요청이 들어왔을 때 이 핵심 에이전트를 직접 호출하려면 A2A 서버의 요청 처리 구조와 연결해야 합니다.

그 연결을 담당하는 것이 WeatherAgentExecutor입니다.

5-1. AgentExecutor 상속

class WeatherAgentExecutor(AgentExecutor):

여기서 AgentExecutor를 상속하고 있습니다.

상속은 기존 클래스의 기능이나 인터페이스를 바탕으로 새로운 클래스를 만드는 방법입니다.

이번 코드에서는 A2A SDK가 정의한 AgentExecutor를 상속합니다.

따라서 WeatherAgentExecutor는 A2A 서버가 요구하는 실행기 인터페이스를 구현해야 합니다.

앞에서 살펴본 것처럼 대표적인 메서드는 다음 두 가지입니다.

execute()
cancel()

실제로 이번 코드에서도 두 메서드를 구현했습니다.

5-2. init() 생성자

def __init__(self):
    self.agent = WeatherCoreAgent()

__init__()은 객체가 생성될 때 자동으로 호출되는 초기화 메서드입니다.

예를 들어 다음과 같이 실행하면,

weather_agent_executor = WeatherAgentExecutor()

Python은 WeatherAgentExecutor 객체를 생성하면서 __init__()을 호출합니다.

그 안에서 다음 코드가 실행됩니다.

self.agent = WeatherCoreAgent()

즉, WeatherAgentExecutor 객체 안에 WeatherCoreAgent 객체를 생성하여 저장합니다.

여기서 self.agent는 WeatherAgentExecutor 객체의 속성입니다.

따라서 나중에 다음과 같이 사용할 수 있습니다.

self.agent.invoke(user_query)

이 코드는 실행기 내부에 저장된 핵심 에이전트의 invoke() 메서드를 호출하는 것입니다.

이 구조를 통해 A2A 요청 처리 코드와 실제 날씨 처리 코드를 분리할 수 있습니다.

이러한 분리는 코드의 유지보수에 매우 유용합니다.

예를 들어 나중에 날씨 데이터를 목업에서 실제 API로 바꾸더라도 WeatherCoreAgent를 수정하면 됩니다.

반면 A2A 요청을 받아 처리하는 WeatherAgentExecutor의 기본 구조는 그대로 유지할 수 있습니다.

6. execute() — 실제 요청을 처리하는 핵심 메서드

이번 코드에서 가장 중요한 부분입니다.

async def execute(
    self,
    context: RequestContext,
    event_queue: EventQueue,
) -> None:

이 메서드는 A2A 요청이 들어왔을 때 실제 에이전트를 실행하는 역할을 합니다.

먼저 함수의 매개변수를 살펴보겠습니다.

매개변수의미

self	현재 실행기 객체
context	현재 A2A 요청의 문맥 정보
event_queue	에이전트가 생성한 이벤트를 등록할 큐
-> None	명시적으로 반환값을 지정하지 않는다는 의미

여기서 context와 event_queue는 A2A 서버가 실행기에 전달해 주는 객체입니다.

실행기는 이 객체들을 활용하여 사용자의 질문을 가져오고, 실행 결과를 A2A 서버에 전달합니다.

이제 내부 코드를 하나씩 살펴보겠습니다.

6-1. 사용자 질문 가져오기

user_query = context.get_user_input()

이 코드는 현재 A2A 요청에서 사용자의 질문을 가져옵니다.

예를 들어 외부 에이전트가 다음과 같은 질문을 전송했다고 가정하겠습니다.

서울의 날씨를 알려줘.

그러면 다음과 같은 값이 만들어집니다.

user_query = "서울의 날씨를 알려줘."

이렇게 가져온 사용자 질문은 이후 핵심 에이전트에 전달됩니다.

여기서 기억해야 할 점은 context가 단순히 사용자 질문만 담고 있는 객체는 아니라는 것입니다.

앞서 설명한 것처럼 작업 ID나 문맥 ID 등 요청 처리에 필요한 정보도 제공합니다.

6-2. 핵심 에이전트 실행

result = await self.agent.invoke(user_query)

이 코드는 WeatherCoreAgent의 invoke() 메서드를 실행합니다.

앞에서 생성자에서 다음과 같이 핵심 에이전트 객체를 만들었습니다.

self.agent = WeatherCoreAgent()

따라서 self.agent.invoke()를 호출하면 해당 객체의 invoke() 메서드가 실행됩니다.

여기서 user_query를 인자로 전달합니다.

예를 들어 다음과 같습니다.

user_query = "서울의 날씨를 알려줘."

result = await self.agent.invoke(user_query)

그러면 핵심 에이전트는 다음 과정을 수행합니다.

사용자 질문
    |
    v
"서울의 날씨를 알려줘."
    |
    v
도시명 검색
    |
    v
서울 발견
    |
    v
목업 날씨 데이터 조회
    |
    v
응답 문자열 생성
    |
    v
result에 저장

최종적으로 result에는 다음과 같은 문자열이 저장됩니다.

서울의 현재 날씨는 맑음입니다.기온은 22°C 이며,습도는 45% 입니다.

이제 실행기는 이 문자열을 A2A 메시지로 변환해야 합니다.

여기서 await를 사용한 이유는 invoke()가 비동기 함수이기 때문입니다.

비동기 함수의 실행 결과를 얻으려면 일반적으로 await를 사용해야 합니다.

6-3. 응답을 A2A Message로 변환

response_message = Message(
    message_id="weather_response",
    context_id=context.context_id,
    task_id=context.task_id,
    role=Role.ROLE_AGENT,
    parts=[Part(text=result)],
)

이 부분은 이번 코드에서 특히 중요합니다.

앞에서 WeatherCoreAgent는 일반적인 문자열을 반환했습니다.

하지만 A2A 서버에서는 단순한 문자열만으로 응답을 전달하는 것이 아니라 A2A에서 정의한 메시지 구조를 사용합니다.

따라서 문자열을 A2A Message 객체로 변환해야 합니다.

이를 위해 다음과 같은 클래스를 사용합니다.

Message
Part
Role

하나씩 살펴보겠습니다.

① Message

response_message = Message(...)

Message는 A2A 메시지를 표현하는 객체입니다.

A2A 메시지에는 메시지 ID, 역할, 메시지 내용 등 여러 정보가 포함될 수 있습니다.

이번 코드에서는 다음과 같이 구성합니다.

Message(
    message_id="weather_response",
    context_id=context.context_id,
    task_id=context.task_id,
    role=Role.ROLE_AGENT,
    parts=[Part(text=result)],
)

각 매개변수의 의미는 다음과 같습니다.

매개변수설명

message_id	메시지의 고유 식별자
context_id	해당 메시지가 속한 대화 문맥
task_id	해당 메시지가 속한 작업
role	메시지를 작성한 주체
parts	메시지의 실제 내용

② message_id

message_id="weather_response"

이 값은 메시지를 식별하기 위한 ID입니다.

현재 코드는 이해하기 쉽도록 고정 문자열을 사용했습니다.

하지만 실제 서버에서 여러 요청을 처리한다면 서로 다른 메시지가 동일한 ID를 갖게 될 수 있습니다.

따라서 실제 운영 환경에서는 UUID 등을 이용해 고유한 메시지 ID를 생성하는 것이 좋습니다.

예를 들어 다음과 같이 작성할 수 있습니다.

import uuid

message_id = str(uuid.uuid4())

그러면 실행할 때마다 서로 다른 ID가 생성됩니다.

예시:

a3f0e9d2-4a61-4f15-8b76-3b8a2d9c6e10

실제 운영 환경에서는 고유 ID를 사용하는 편이 안전합니다.

③ context_id

context_id=context.context_id

이 값은 현재 메시지가 어떤 대화 문맥에 속하는지를 나타냅니다.

예를 들어 하나의 대화에서 사용자가 다음과 같이 질문할 수 있습니다.

사용자: 서울의 날씨를 알려줘.
에이전트: 서울의 날씨는 맑습니다.

사용자: 그럼 부산은?
에이전트: 부산은 비가 내립니다.

두 번째 질문은 첫 번째 질문과 별개의 문장처럼 보이지만, 앞선 대화 문맥을 고려하면 의미가 명확해집니다.

A2A에서는 이러한 대화 문맥을 식별하는 데 context_id를 활용할 수 있습니다.

이번 코드는 현재 요청의 context_id를 그대로 응답 메시지에 전달합니다.

④ task_id

task_id=context.task_id

task_id는 현재 작업을 식별하는 ID입니다.

예를 들어 사용자가 서울의 날씨를 요청하면 서버는 해당 요청을 하나의 작업으로 관리할 수 있습니다.

이때 작업을 식별하는 데 task_id가 사용됩니다.

응답 메시지에 동일한 task_id를 포함하면 해당 메시지가 어떤 작업에 대한 응답인지 연결할 수 있습니다.

즉, context_id가 대화 문맥을 식별한다면 task_id는 개별 작업을 식별하는 데 초점을 둡니다.

⑤ role

role=Role.ROLE_AGENT

role은 메시지를 작성한 주체를 나타냅니다.

이번 코드에서는 에이전트가 생성한 응답이므로 다음 값을 사용합니다.

Role.ROLE_AGENT

일반적으로 A2A 메시지에는 사용자 메시지와 에이전트 메시지를 구분하는 역할 정보가 포함됩니다.

예를 들어 다음과 같이 생각할 수 있습니다.

역할의미

ROLE_USER	사용자가 작성한 메시지
ROLE_AGENT	에이전트가 작성한 메시지

따라서 이번 코드는 에이전트가 생성한 응답임을 명시하고 있습니다.

⑥ parts

parts=[Part(text=result)]

parts는 메시지의 실제 내용을 담는 부분입니다.

A2A에서는 메시지 내용을 하나 이상의 Part로 구성할 수 있습니다.

이번 코드에서는 텍스트만 사용하므로 다음과 같이 작성합니다.

Part(text=result)

예를 들어 result에 다음 문자열이 저장되어 있다고 가정하겠습니다.

서울의 현재 날씨는 맑음입니다. 기온은 22°C이며, 습도는 45%입니다.

그러면 Part 객체에 해당 텍스트가 저장됩니다.

그리고 이를 리스트로 감싸서 parts에 전달합니다.

parts=[Part(text=result)]

여기서 리스트를 사용하는 이유는 하나의 메시지가 여러 개의 Part로 구성될 수 있기 때문입니다.

예를 들어 메시지에 텍스트와 다른 유형의 콘텐츠를 함께 포함하는 구조를 생각할 수 있습니다.

이번 코드는 간단한 텍스트 응답이므로 Part 하나만 사용합니다.

6-4. EventQueue에 응답 등록

await event_queue.enqueue_event(response_message)

이제 생성한 A2A 메시지를 이벤트 큐에 등록합니다.

앞에서 response_message는 다음과 같은 객체였습니다.

response_message = Message(
    message_id="weather_response",
    context_id=context.context_id,
    task_id=context.task_id,
    role=Role.ROLE_AGENT,
    parts=[Part(text=result)],
)

이 객체를 다음 코드로 이벤트 큐에 전달합니다.

await event_queue.enqueue_event(response_message)

여기서 enqueue_event()는 이벤트 큐에 이벤트를 등록하는 메서드입니다.

await를 사용하므로 비동기 작업으로 실행됩니다.

이벤트 큐를 사용하는 이유

A2A 서버에서는 에이전트가 작업을 수행하고 생성한 메시지나 상태 변화를 이벤트로 전달할 수 있습니다.

이벤트 큐는 이러한 이벤트를 요청 처리 흐름에 전달하는 통로 역할을 합니다.

이번 코드는 에이전트가 생성한 응답 메시지를 이벤트 큐에 등록하는 방식으로 구현되어 있습니다.

여기서 중요한 점은 다음과 같습니다.

return result

와

await event_queue.enqueue_event(response_message)

는 서로 다른 역할을 합니다.

첫 번째는 일반적인 Python 함수에서 호출자에게 값을 반환하는 방식입니다.

두 번째는 A2A 서버의 이벤트 전달 구조를 이용해 결과를 전달하는 방식입니다.

이번 코드에서는 execute()의 반환 타입이 None이므로 일반적인 문자열을 반환하지 않습니다.

대신 응답 메시지를 이벤트 큐에 등록합니다.

따라서 실행기에서 return result를 사용하는 대신 enqueue_event()를 사용한 것입니다.

7. cancel() — 작업 취소 기능

async def cancel(
    self, 
    context: RequestContext,
    event_queue: EventQueue
) -> None:
    raise Exception('cancel not supported')

이 메서드는 작업 취소 요청을 처리하는 기능입니다.

A2A 서버에서는 작업을 실행하는 도중 취소 요청이 발생할 수 있습니다.

예를 들어 사용자가 에이전트에 복잡한 분석을 요청했다고 가정하겠습니다.

그런데 분석이 완료되기 전에 사용자가 작업을 취소할 수 있습니다.

이런 경우 실행기는 취소 요청을 처리해야 합니다.

이번 코드에서는 다음과 같이 작성했습니다.

raise Exception('cancel not supported')

즉, 현재 에이전트는 취소 기능을 지원하지 않는다고 예외를 발생시키는 것입니다.

여기서 raise는 예외를 발생시키는 Python 키워드입니다.

예를 들어 다음과 같이 작성할 수 있습니다.

raise Exception("오류가 발생했습니다.")

그러면 해당 위치에서 예외가 발생합니다.

이번 코드에서는 취소 기능을 구현하지 않았기 때문에 호출되면 예외가 발생합니다.

다만 실제 운영 환경에서는 단순히 예외를 발생시키는 것보다 SDK에서 요구하는 취소 처리 방식을 확인하고, 작업이 취소될 수 있는 상태인지 판단하는 등의 로직을 구현하는 것이 좋습니다.

특히 현재 날씨 에이전트는 실행 시간이 매우 짧기 때문에 취소 기능이 크게 필요하지 않을 수 있습니다.

반면 장시간 실행되는 AI 에이전트라면 취소 기능이 중요해집니다.

8. AgentSkill — 에이전트의 기능 정의

weather_skill = AgentSkill(
    id="weather",
    name="Weather Information",
    description="Provide weather information for Seoul, Daejeon, and Busan.",
    tags=["weather", "temperature", "humidity", "forecast",],
    examples=["서울 날시 알려줘", "대전의 날씨는 어때?", "부산 기온과 습도를 알려줘",],
)

AgentSkill은 에이전트가 어떤 기능을 제공하는지 설명하는 객체입니다.

쉽게 말하면 에이전트의 기능 목록이라고 생각하면 됩니다.

예를 들어 하나의 AI 에이전트가 다음과 같은 기능을 제공한다고 가정하겠습니다.

날씨 조회
계산
문서 검색
이메일 작성

이러한 기능을 외부 에이전트가 이해할 수 있도록 설명하는 것이 AgentSkill의 역할입니다.

이번 코드에서는 날씨 조회 기능 하나만 정의했습니다.

8-1. id

id="weather"

기능을 식별하기 위한 ID입니다.

이번 코드에서는 날씨 조회 기능을 의미하는 weather를 사용했습니다.

8-2. name

name="Weather Information"

기능의 이름입니다.

외부에서 이 에이전트가 어떤 기능을 제공하는지 이해할 수 있도록 이름을 지정합니다.

8-3. description

description="Provide weather information for Seoul, Daejeon, and Busan."

이 기능에 대한 설명입니다.

현재 에이전트는 서울, 대전, 부산의 날씨 정보를 제공한다고 설명하고 있습니다.

이 설명은 외부 에이전트가 어떤 기능을 요청할 수 있는지 이해하는 데 도움을 줍니다.

8-4. tags

tags=["weather", "temperature", "humidity", "forecast",]

기능을 설명하는 태그 목록입니다.

현재는 다음과 같은 태그가 정의되어 있습니다.

태그의미

weather	날씨
temperature	기온
humidity	습도
forecast	예보

이 태그들은 에이전트의 기능을 분류하거나 검색하는 데 활용할 수 있습니다.

다만 태그가 있다고 해서 자동으로 검색 기능이 구현되는 것은 아닙니다.

태그를 활용하는 것은 이를 사용하는 클라이언트나 에이전트의 역할입니다.

8-5. examples

examples=[
    "서울 날시 알려줘",
    "대전의 날씨는 어때?",
    "부산 기온과 습도를 알려줘",
]

이 항목은 해당 기능을 사용할 수 있는 예시 질문을 제공합니다.

예를 들어 다음과 같은 질문입니다.

서울 날씨 알려줘.
대전의 날씨는 어때?
부산 기온과 습도를 알려줘.

여기서 첫 번째 예시의 "날시"는 "날씨"의 오타입니다.

실제 기능 설명에 넣을 예시라면 "서울 날씨 알려줘"로 수정하는 것이 좋겠습니다.

또한 예시를 정의했다고 해서 에이전트가 자동으로 해당 질문들을 학습하는 것은 아닙니다.

이 예시는 에이전트의 기능을 설명하는 메타데이터입니다.

AgentSkill의 핵심

AgentSkill은 에이전트가 어떤 기능을 제공하는지 외부에 알리는 역할을 합니다.

실제 기능을 수행하는 코드는 WeatherCoreAgent에 있습니다.

따라서 다음 두 가지를 구분해야 합니다.

구분역할

AgentSkill	에이전트가 제공하는 기능 설명
WeatherCoreAgent	실제 날씨 조회 수행

9. AgentCard — 에이전트의 명함

agent_card = AgentCard(
    name="Weather Agent",
    description="An A2A agent that provides weather information",
    version="1.0.0",
    supported_interfaces=[
        AgentInterface(
            url="http://localhost:9999/",
            protocol_binding=TransportProtocol.JSONRPC.value,
            protocol_version="1.0",
        )
    ],
    capabilities=AgentCapabilities(
        streaming=False,
        push_notifications=False,
    ),
    default_input_modes=["text",],
    default_output_modes=["text",],
    skills=[weather_skill,],
)

AgentCard는 A2A에서 매우 중요한 개념입니다.

외부 에이전트가 다른 에이전트와 통신하려면 먼저 해당 에이전트가 어떤 기능을 제공하는지, 어디에 접속해야 하는지 알아야 합니다.

이를 위해 사용하는 것이 AgentCard입니다.

쉽게 말하면 에이전트의 명함이나 서비스 안내서라고 생각하면 됩니다.

AgentCard에는 다음과 같은 정보가 포함됩니다.

에이전트의 이름
에이전트의 설명
버전
통신 인터페이스
지원하는 기능
입력 및 출력 데이터 유형
제공하는 스킬

이번 코드의 각 항목을 살펴보겠습니다.

9-1. name

name="Weather Agent"

에이전트의 이름입니다.

외부 에이전트가 이 에이전트를 식별할 때 사용할 수 있습니다.

9-2. description

description="An A2A agent that provides weather information"

에이전트가 어떤 역할을 수행하는지 설명합니다.

현재는 날씨 정보를 제공하는 A2A 에이전트라고 설명하고 있습니다.

9-3. version

version="1.0.0"

에이전트의 버전을 나타냅니다.

예를 들어 이후 기능을 추가하거나 인터페이스를 변경하면 버전을 변경할 수 있습니다.

일반적으로 소프트웨어 버전은 다음과 같은 형식을 사용합니다.

1.0.0

각 숫자는 보통 다음과 같은 의미로 사용됩니다.

구분일반적인 의미

첫 번째 숫자	호환성이 깨질 수 있는 주요 변경
두 번째 숫자	기존 기능과 호환되는 기능 추가
세 번째 숫자	버그 수정

단, 이번 코드에서는 단순히 1.0.0이라는 버전 문자열을 지정한 것입니다.

9-4. supported_interfaces

supported_interfaces=[
    AgentInterface(
        url="http://localhost:9999/",
        protocol_binding=TransportProtocol.JSONRPC.value,
        protocol_version="1.0",
    )
]

이 부분은 외부 에이전트가 실제로 어디에 접속해야 하는지 알려주는 정보입니다.

AgentInterface는 에이전트가 제공하는 통신 인터페이스를 나타냅니다.

url

url="http://localhost:9999/"

에이전트의 접속 주소입니다.

현재 코드는 로컬에서 9999번 포트로 서버를 실행하도록 설정했습니다.

따라서 같은 컴퓨터에서 테스트하는 경우 다음 주소로 접근할 수 있습니다.

http://localhost:9999/

실제 서비스에서는 localhost 대신 외부에서 접근할 수 있는 도메인이나 IP 주소를 사용해야 합니다.

protocol_binding

protocol_binding=TransportProtocol.JSONRPC.value

이 에이전트가 JSON-RPC 방식으로 통신한다는 것을 나타냅니다.

여기서 .value는 열거형(Enum)의 실제 값을 가져오는 속성입니다.

예를 들어 다음과 같은 구조를 생각할 수 있습니다.

TransportProtocol.JSONRPC

이는 JSON-RPC라는 프로토콜 항목 자체를 의미합니다.

반면 다음과 같이 작성하면,

TransportProtocol.JSONRPC.value

해당 항목에 지정된 실제 문자열 값을 가져옵니다.

AgentCard에는 이 실제 값을 전달하고 있습니다.

protocol_version

protocol_version="1.0"

사용하는 A2A 프로토콜의 버전을 나타냅니다.

여기서 주의할 점은 SDK의 버전과 프로토콜 버전은 서로 다르다는 것입니다.

이번 코드에서는 다음과 같습니다.

구분버전

A2A SDK	1.2.0
AgentCard의 프로토콜 버전	1.0

SDK가 1.2.0이라고 해서 프로토콜 버전도 반드시 1.2.0인 것은 아닙니다.

9-5. capabilities

capabilities=AgentCapabilities(
    streaming=False,
    push_notifications=False,
)

AgentCapabilities는 에이전트가 지원하는 기능을 나타냅니다.

이번 코드에서는 다음 두 가지를 설정했습니다.

streaming

streaming=False

스트리밍 기능을 지원하지 않는다는 의미입니다.

스트리밍은 에이전트가 응답을 완성할 때까지 기다리는 대신 생성되는 결과를 순차적으로 전달하는 방식입니다.

예를 들어 LLM이 긴 답변을 생성할 때 다음과 같이 전달할 수 있습니다.

안녕하세요.
오늘은
서울의
날씨를
알려드리겠습니다.

실제 스트리밍에서는 이러한 결과가 작은 단위로 전달될 수 있습니다.

이번 날씨 에이전트는 간단한 응답을 한 번에 반환하기 때문에 스트리밍을 비활성화했습니다.

다만 앞서 설명했듯이 이벤트 큐를 사용한다고 해서 스트리밍을 지원한다는 의미는 아닙니다.

push_notifications

push_notifications=False

푸시 알림 기능을 지원하지 않는다는 의미입니다.

푸시 알림은 클라이언트가 계속 요청하지 않아도 서버가 특정 상황에서 알림을 전달하는 기능입니다.

예를 들어 에이전트가 장시간 작업을 수행하고 있다고 가정하겠습니다.

작업이 완료되었을 때 클라이언트가 다시 요청하지 않아도 서버가 알림을 보낼 수 있습니다.

이러한 기능이 푸시 알림입니다.

이번 날씨 에이전트는 요청을 받으면 바로 응답하는 간단한 구조이므로 푸시 알림을 사용하지 않습니다.

9-6. default_input_modes와 default_output_modes

default_input_modes=["text",]
default_output_modes=["text",]

이 두 항목은 에이전트가 기본적으로 어떤 형식의 입력과 출력을 사용하는지 나타냅니다.

이번 에이전트는 텍스트로 질문을 받고 텍스트로 응답합니다.

따라서 두 항목 모두 "text"로 설정했습니다.

예를 들어 다음과 같습니다.

입력: 서울의 날씨를 알려줘.
출력: 서울의 현재 날씨는 맑음입니다.

현재 에이전트는 이미지나 오디오 등의 다른 입력 유형을 처리하는 기능을 구현하지 않았습니다.

9-7. skills

skills=[weather_skill,]

앞에서 정의한 weather_skill을 AgentCard에 등록하는 부분입니다.

즉, AgentCard가 에이전트의 기능을 설명할 때 날씨 조회 기능도 함께 포함하도록 합니다.

여기서 weather_skill은 다음과 같이 정의했습니다.

weather_skill = AgentSkill(
    id="weather",
    name="Weather Information",
    description="Provide weather information for Seoul, Daejeon, and Busan.",
    tags=["weather", "temperature", "humidity", "forecast"],
    examples=["서울 날씨 알려줘", "대전의 날씨는 어때?", "부산 기온과 습도를 알려줘"],
)

이렇게 정의한 스킬을 AgentCard에 연결하면 외부 에이전트가 이 서버의 기능을 확인할 수 있습니다.

AgentCard와 AgentSkill의 관계

이 두 객체의 관계를 간단히 정리하면 다음과 같습니다.

AgentCard
   |
   |-- name
   |-- description
   |-- version
   |-- supported_interfaces
   |-- capabilities
   |-- default_input_modes
   |-- default_output_modes
   |
   |-- skills
          |
          |-- AgentSkill
                 |
                 |-- id
                 |-- name
                 |-- description
                 |-- tags
                 |-- examples

즉, AgentCard는 에이전트 전체에 대한 정보를 담고, AgentSkill은 에이전트가 제공하는 개별 기능을 설명합니다.

10. AgentExecutor 및 RequestHandler 생성

이제 앞에서 정의한 객체들을 실제로 연결하는 부분입니다.

weather_agent_executor = WeatherAgentExecutor()

request_handler = DefaultRequestHandler(
    agent_executor=weather_agent_executor,
    task_store=InMemoryTaskStore(),
    agent_card=agent_card,
)

이 부분은 A2A 서버의 실행 구조를 완성하는 단계입니다.

10-1. WeatherAgentExecutor 객체 생성

weather_agent_executor = WeatherAgentExecutor()

앞에서 정의한 WeatherAgentExecutor 클래스를 실제 객체로 생성합니다.

이때 생성자 내부의 코드가 실행됩니다.

def __init__(self):
    self.agent = WeatherCoreAgent()

따라서 실행기 객체 안에는 WeatherCoreAgent 객체가 생성되어 저장됩니다.

이제 실행기는 사용자 요청이 들어왔을 때 핵심 에이전트를 호출할 수 있습니다.

10-2. DefaultRequestHandler 객체 생성

request_handler = DefaultRequestHandler(
    agent_executor=weather_agent_executor,
    task_store=InMemoryTaskStore(),
    agent_card=agent_card,
)

이제 A2A 요청을 관리할 요청 처리기를 생성합니다.

여기서는 세 가지 객체를 전달합니다.

agent_executor

agent_executor=weather_agent_executor

요청 처리기가 실제 작업을 실행할 수 있도록 실행기를 전달합니다.

즉, A2A 요청이 들어오면 weather_agent_executor가 실행될 수 있도록 연결합니다.

task_store

task_store=InMemoryTaskStore()

작업 정보를 저장할 저장소를 전달합니다.

이번 코드에서는 메모리 기반 저장소를 사용합니다.

agent_card

agent_card=agent_card

에이전트의 정보를 전달합니다.

이렇게 하면 요청 처리기가 에이전트의 기능과 관련된 정보를 사용할 수 있습니다.

전체 연결 구조

AgentCard
    |
    |
    v
DefaultRequestHandler <------ InMemoryTaskStore
    |
    |
    v
WeatherAgentExecutor
    |
    |
    v
WeatherCoreAgent

여기서 중요한 것은 각각의 객체가 독립적으로 존재하는 것이 아니라 서로 연결되어 있다는 점입니다.

이러한 구조를 통해 A2A 서버는 요청을 받아 실제 에이전트를 실행하고 결과를 전달할 수 있습니다.

11. FastAPI 애플리케이션 생성

app = FastAPI(
    title="A2A Weather Server",
    description="A simple A2A weather agent server",
    version="1.0.0",
)

이 부분에서는 FastAPI 애플리케이션 객체를 생성합니다.

앞서 설명한 것처럼 FastAPI는 HTTP 요청을 처리하는 웹 프레임워크입니다.

여기서 생성한 app 객체는 이후 A2A 라우트를 등록하고 Uvicorn으로 실행할 때 사용합니다.

각 매개변수는 다음과 같습니다.

매개변수값의미

title	A2A Weather Server	애플리케이션 이름
description	A simple A2A weather agent server	애플리케이션 설명
version	1.0.0	애플리케이션 버전

이 정보는 FastAPI의 자동 문서화 기능 등에서 활용할 수 있습니다.

다만 이 정보와 AgentCard의 정보는 서로 다른 용도로 사용됩니다.

구분FastAPIAgentCard

목적	웹 애플리케이션 설명	A2A 에이전트 설명
이름	A2A Weather Server	Weather Agent
버전	1.0.0	1.0.0
주요 사용자	개발자 및 API 문서 이용자	A2A 클라이언트 및 에이전트

12. A2A 라우트 등록

이제 FastAPI 애플리케이션에 A2A 라우트를 등록합니다.

add_a2a_routes_to_fastapi(
    app,

    agent_card_routes=create_agent_card_routes(agent_card),

    jsonrpc_routes=create_jsonrpc_routes(
        request_handler,
        rpc_url="/",
    ),

    rest_routes=create_rest_routes(request_handler),
)

이 부분은 A2A 서버가 외부 요청을 실제로 받을 수 있도록 만드는 중요한 단계입니다.

앞에서 app을 생성했지만 아직 A2A 요청을 처리할 경로를 등록하지 않았습니다.

이제 A2A SDK가 제공하는 라우트 생성 함수를 사용하여 필요한 경로를 만들고, 이를 FastAPI 애플리케이션에 등록합니다.

12-1. add_a2a_routes_to_fastapi()

add_a2a_routes_to_fastapi(
    app,
    ...
)

이 함수는 A2A 라우트들을 FastAPI 애플리케이션에 등록하는 역할을 합니다.

첫 번째 인자로 app을 전달합니다.

즉, 우리가 생성한 FastAPI 애플리케이션에 A2A 관련 경로를 추가하는 것입니다.

여기서 중요한 점은 각 라우트를 생성하는 함수와 실제로 라우트를 등록하는 함수가 분리되어 있다는 것입니다.

예를 들어 다음과 같이 생각할 수 있습니다.

create_agent_card_routes()
          |
          v
Agent Card 경로 생성
          |
          v
add_a2a_routes_to_fastapi()
          |
          v
FastAPI에 등록

JSON-RPC와 REST 라우트도 같은 방식입니다.

12-2. create_agent_card_routes()

agent_card_routes=create_agent_card_routes(agent_card)

이 함수는 AgentCard를 조회할 수 있는 라우트를 생성합니다.

여기서는 앞에서 정의한 agent_card 객체를 전달합니다.

그러면 해당 AgentCard 정보를 외부에서 조회할 수 있도록 경로가 만들어집니다.

이번 코드에서 출력하는 주소는 다음과 같습니다.

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

이 주소를 브라우저에서 열면 AgentCard 정보를 JSON 형태로 확인할 수 있습니다.

AgentCard를 조회하는 목적은 외부 에이전트가 이 서버의 기능과 연결 정보를 확인하도록 하기 위한 것입니다.

예를 들어 외부 에이전트는 다음과 같은 정보를 확인할 수 있습니다.

에이전트 이름: Weather Agent
기능: 날씨 정보 제공
지원 프로토콜: JSON-RPC
접속 주소: http://localhost:9999/
입력 형식: text
출력 형식: text

실제 응답은 SDK가 정의한 JSON 구조로 반환됩니다.

12-3. create_jsonrpc_routes()

jsonrpc_routes=create_jsonrpc_routes(
    request_handler,
    rpc_url="/",
)

이 함수는 A2A JSON-RPC 요청을 처리할 라우트를 생성합니다.

여기서 request_handler를 전달합니다.

즉, JSON-RPC 요청이 들어오면 앞에서 생성한 DefaultRequestHandler가 요청을 처리하도록 연결합니다.

또한 다음과 같이 설정했습니다.

rpc_url="/"

이는 JSON-RPC 요청을 서버의 루트 경로에서 처리하도록 지정한 것입니다.

따라서 JSON-RPC 요청 주소는 다음과 같습니다.

http://localhost:9999/

이 주소로 A2A JSON-RPC 요청을 보내면 서버가 요청을 처리합니다.

여기서 중요한 점은 JSON-RPC 요청의 실제 처리 로직을 직접 구현하지 않았다는 것입니다.

A2A SDK가 제공하는 create_jsonrpc_routes()와 DefaultRequestHandler를 사용하여 요청 처리 구조를 구성했습니다.

따라서 개발자는 실제 날씨 에이전트의 핵심 로직에 집중할 수 있습니다.

12-4. create_rest_routes()

rest_routes=create_rest_routes(request_handler)

이 함수는 REST 방식의 A2A 요청을 처리할 라우트를 생성합니다.

REST는 HTTP 메서드와 URL을 이용하여 자원을 다루는 웹 API 설계 방식입니다.

이번 코드에서는 A2A SDK가 제공하는 REST 라우트 생성 함수를 사용하고 있습니다.

따라서 JSON-RPC와 REST라는 서로 다른 요청 처리 경로를 함께 등록할 수 있습니다.

다만 AgentCard의 supported_interfaces에는 JSON-RPC 인터페이스만 명시되어 있습니다.

즉, REST 라우트를 등록했다고 해서 AgentCard에서 REST 인터페이스까지 외부에 광고하는 것은 아닙니다.

실제 클라이언트가 어떤 인터페이스를 사용할 수 있는지는 서버가 등록한 라우트와 AgentCard의 인터페이스 명세를 함께 고려해야 합니다.

세 가지 라우트의 역할 정리

함수목적이번 코드에서의 역할

create_agent_card_routes()	에이전트 정보 제공	AgentCard 조회
create_jsonrpc_routes()	JSON-RPC 요청 처리	핵심 A2A 통신 경로
create_rest_routes()	REST 요청 처리	REST 방식의 요청 경로
add_a2a_routes_to_fastapi()	라우트 등록	FastAPI에 경로 연결

13. 서버 실행

마지막으로 서버를 실행하는 부분입니다.

if __name__ == "__main__":

    print("="*60)
    print("A2A Weather Server")
    print("="*60)

    print("Server URL: htpp://localhost:9999")
    print("Agent Card: http://localhost:9999/.well-known/agent-card.json")
    print("JSON-RPC: http://localhost:9999")
    print("="*60)

    uvicorn.run(app, host="0.0.0.0", port=9999)

이 부분은 Python 프로그램을 직접 실행했을 때 서버를 시작하도록 구성한 코드입니다.

13-1. if name == "main":

if __name__ == "__main__":

이 구문은 Python 프로그램을 실행할 때 자주 사용합니다.

Python 파일은 직접 실행할 수도 있고, 다른 Python 파일에서 모듈로 불러올 수도 있습니다.

예를 들어 파일 이름이 weather_server.py라고 가정하겠습니다.

직접 실행하는 경우에는 다음과 같습니다.

python weather_server.py

이때 Python은 해당 파일의 __name__ 변수에 "__main__"을 저장합니다.

따라서 다음 조건문이 참이 됩니다.

if __name__ == "__main__":

그러면 조건문 안의 코드가 실행됩니다.

반면 다른 파일에서 다음과 같이 불러오는 경우를 생각해 보겠습니다.

import weather_server

이때는 일반적으로 weather_server.py의 __name__ 값이 "weather_server"가 됩니다.

따라서 다음 조건은 거짓이 됩니다.

if __name__ == "__main__":

즉, 해당 파일을 다른 프로그램에서 불러오더라도 서버가 자동으로 실행되지 않습니다.

이것이 이 구문을 사용하는 중요한 이유입니다.

13-2. print()

print("="*60)
print("A2A Weather Server")
print("="*60)

이 코드는 콘솔에 안내 문구를 출력합니다.

"=" * 60

은 = 문자를 60번 반복한 문자열을 생성합니다.

따라서 다음과 같은 구분선이 출력됩니다.

============================================================

이후 서버 이름과 접속 주소를 출력합니다.

다만 현재 코드에는 다음과 같은 오타가 있습니다.

print("Server URL: htpp://localhost:9999")

htpp가 아니라 http가 맞습니다.

다음과 같이 수정하는 것이 좋습니다.

print("Server URL: http://localhost:9999")

이 부분은 단순히 콘솔에 출력되는 문자열이므로 서버 실행 자체에는 영향을 주지 않습니다.

13-3. uvicorn.run()

uvicorn.run(app, host="0.0.0.0", port=9999)

이 명령이 실행되면 실제 웹 서버가 시작됩니다.

앞서 생성한 FastAPI 애플리케이션인 app을 Uvicorn에 전달합니다.

그리고 서버가 사용할 네트워크 주소와 포트를 지정합니다.

설정의미

app	실행할 FastAPI 애플리케이션
host="0.0.0.0"	모든 네트워크 인터페이스에서 요청 수신
port=9999	9999번 포트 사용

서버가 정상적으로 실행되면 일반적으로 콘솔에 다음과 비슷한 메시지가 출력됩니다.

INFO:     Started server process
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:9999

이제 서버는 외부 요청을 받을 준비가 된 것입니다.

14. 실제로 요청이 들어오면 어떻게 작동할까?

이제 전체 코드를 하나의 실행 흐름으로 연결해 보겠습니다.

사용자가 다음과 같은 질문을 보낸다고 가정하겠습니다.

서울의 날씨를 알려줘.

이 질문이 서버에 도착하면 다음과 같은 순서로 처리됩니다.

1단계: FastAPI가 HTTP 요청을 받습니다.

사용자 또는 외부 에이전트
          |
          | HTTP 요청
          v
       FastAPI

외부 에이전트가 http://localhost:9999/로 요청을 보냅니다.

FastAPI는 해당 URL에 등록된 라우트를 확인합니다.

2단계: JSON-RPC 라우트가 요청을 처리합니다.

create_jsonrpc_routes(
    request_handler,
    rpc_url="/",
)

JSON-RPC 요청이므로 해당 라우트를 통해 요청 처리기로 전달됩니다.

3단계: DefaultRequestHandler가 요청을 관리합니다.

request_handler = DefaultRequestHandler(
    agent_executor=weather_agent_executor,
    task_store=InMemoryTaskStore(),
    agent_card=agent_card,
)

요청 처리기는 A2A 요청을 처리하고 실행기를 통해 실제 에이전트를 실행하도록 연결합니다.

4단계: WeatherAgentExecutor의 execute()가 실행됩니다.

async def execute(
    self,
    context: RequestContext,
    event_queue: EventQueue,
) -> None:

요청 처리기는 실행기에 요청 문맥과 이벤트 큐를 전달합니다.

5단계: 사용자 질문을 가져옵니다.

user_query = context.get_user_input()

결과:

서울의 날씨를 알려줘.

6단계: WeatherCoreAgent를 실행합니다.

result = await self.agent.invoke(user_query)

실행기는 사용자 질문을 핵심 에이전트에 전달합니다.

7단계: 핵심 에이전트가 도시명을 검색합니다.

for city, weather in MOCK_WEATHER_DATA.items():
    if city in user_query:

질문에 포함된 도시명을 확인합니다.

서울이라는 도시명을 발견하면 해당 도시의 날씨 정보를 조회합니다.

8단계: 날씨 응답을 생성합니다.

return (
    f"{city}의 현재 날씨는 {condition}입니다."
    f"기온은 {temperature}°C 이며,"
    f"습도는 {humidity}% 입니다."
)

결과:

서울의 현재 날씨는 맑음입니다.기온은 22°C 이며,습도는 45% 입니다.

이 문자열이 result에 저장됩니다.

9단계: A2A 메시지를 생성합니다.

response_message = Message(
    message_id="weather_response",
    context_id=context.context_id,
    task_id=context.task_id,
    role=Role.ROLE_AGENT,
    parts=[Part(text=result)],
)

일반 문자열을 A2A 메시지로 변환합니다.

10단계: 이벤트 큐에 응답을 등록합니다.

await event_queue.enqueue_event(response_message)

생성한 메시지를 이벤트 큐에 등록합니다.

11단계: A2A 서버가 응답을 처리합니다.

A2A 서버의 요청 처리 흐름은 이벤트 큐에 등록된 응답을 받아 클라이언트에 전달할 수 있도록 처리합니다.

여기서 실제 응답의 JSON 구조와 작업 상태 처리는 A2A SDK의 요청 처리기가 담당합니다.

12단계: 클라이언트가 응답을 받습니다.

최종적으로 외부 에이전트는 서울의 날씨에 대한 응답을 받습니다.

전체 흐름을 다시 정리하면 다음과 같습니다.

[외부 에이전트]
      |
      | 서울의 날씨를 알려줘.
      v
[FastAPI]
      |
      v
[A2A JSON-RPC Route]
      |
      v
[DefaultRequestHandler]
      |
      v
[WeatherAgentExecutor.execute()]
      |
      | context.get_user_input()
      v
[WeatherCoreAgent.invoke()]
      |
      | 도시명 검색
      | 날씨 데이터 조회
      v
[날씨 응답 문자열]
      |
      v
[Message 객체 생성]
      |
      v
[EventQueue]
      |
      v
[A2A 요청 처리기]
      |
      v
[외부 에이전트에 응답]

이것이 이번 A2A Weather Agent Server의 전체 실행 흐름입니다.

15. 이번 코드에서 특히 기억해야 할 핵심 개념 7가지

이번 코드를 학습하면서 꼭 기억해야 할 내용을 정리해 보겠습니다.

번호핵심 개념기억해야 할 내용

1	WeatherCoreAgent	사용자 질문을 처리하는 핵심 로직
2	WeatherAgentExecutor	핵심 에이전트를 A2A 실행 구조에 연결
3	RequestContext	사용자 입력과 요청 문맥 등의 정보 제공
4	EventQueue	에이전트의 응답 이벤트를 서버의 처리 흐름에 전달
5	Message / Part	일반 문자열을 A2A 메시지로 구성
6	AgentCard / AgentSkill	에이전트의 접속 정보와 제공 기능을 설명
7	FastAPI Routes	A2A 요청을 받을 HTTP 경로를 생성하고 등록

특히 다음 세 가지는 확실하게 구분하는 것이 좋습니다.

첫째, WeatherCoreAgent는 실제 작업을 수행합니다.

사용자 질문에서 도시명을 찾고 날씨 데이터를 조회한 다음 결과를 문자열로 반환합니다.

둘째, WeatherAgentExecutor는 A2A와 핵심 에이전트를 연결합니다.

요청 문맥에서 사용자 질문을 가져오고, 핵심 에이전트를 실행한 다음 결과를 A2A 메시지로 변환합니다.

셋째, DefaultRequestHandler는 A2A 요청을 관리합니다.

외부에서 들어온 A2A 요청을 처리하고, 실행기와 작업 저장소를 이용하여 요청 처리 흐름을 관리합니다.

이 세 가지의 역할을 명확하게 구분하면 다른 A2A 에이전트를 개발할 때도 구조를 이해하기 쉬워집니다.

16. 현재 코드에서 개선할 수 있는 부분

현재 코드는 학습과 테스트를 위한 간단한 A2A 서버로서 핵심 구조를 잘 보여줍니다.

다만 실제 서비스에 적용하려면 몇 가지 개선할 부분이 있습니다.

구분현재 코드개선 방향

날씨 데이터	목업 데이터	실제 날씨 API 연동
도시명 검색	단순 문자열 포함 여부	별칭, 오타, 자연어 처리
메시지 ID	고정 문자열	UUID 등으로 고유 ID 생성
응답 문장	문자열 연결	공백과 문장 형식 개선
취소 기능	미지원	실제 작업 취소 처리
작업 저장소	메모리	필요에 따라 영구 저장소 사용
스트리밍	비활성화	장시간 작업에 필요하면 구현
오류 처리	기본적인 처리	API 오류와 예외 상황 대응

여기서 모든 항목을 한꺼번에 개선할 필요는 없습니다.

현재 코드는 A2A 서버의 기본적인 구조를 이해하는 데 초점을 맞추고 있으므로, 우선 지금 구현된 코드가 어떻게 작동하는지 이해하는 것이 중요합니다.

이후 실제 날씨 API를 연동하거나 LLM을 추가할 때 각각의 구성 요소가 어떤 역할을 하는지 알고 있으면 훨씬 수월하게 확장할 수 있습니다.

17. 최종 정리: 이번 코드의 핵심은 무엇인가?

이번 코드를 한 문장으로 요약하면 다음과 같습니다.

사용자의 질문을 받아 핵심 에이전트가 처리하고, 그 결과를 A2A 메시지로 변환하여 서버의 요청 처리 흐름을 통해 전달하는 구조입니다.

이를 조금 더 자세히 정리하면 다음과 같습니다.

FastAPI는 외부 HTTP 요청을 받는 웹 애플리케이션입니다.
A2A Routes는 AgentCard 조회와 JSON-RPC 및 REST 요청을 처리할 경로를 제공합니다.
DefaultRequestHandler는 A2A 요청을 관리하고 실행기를 호출합니다.
WeatherAgentExecutor는 사용자 질문을 가져와 핵심 에이전트를 실행합니다.
WeatherCoreAgent는 질문에서 도시명을 찾고 날씨 정보를 반환합니다.
Message와 Part는 반환된 문자열을 A2A 메시지로 구성합니다.
EventQueue는 생성된 응답 이벤트를 A2A 서버의 처리 흐름에 전달합니다.
AgentCard와 AgentSkill은 외부 에이전트가 이 서버의 기능을 이해할 수 있도록 정보를 제공합니다.
Uvicorn은 FastAPI 애플리케이션을 실제 서버로 실행합니다.

마지막으로 이번 코드에서 가장 중요한 설계 원칙을 하나 기억하면 좋겠습니다.

실제 에이전트의 기능과 A2A 통신 기능을 분리했다는 점입니다.

WeatherCoreAgent는 날씨 질문을 처리하는 데 집중하고, WeatherAgentExecutor는 A2A 요청과 에이전트 실행을 연결합니다. 그리고 DefaultRequestHandler와 FastAPI는 외부 요청을 받아 처리하는 역할을 담당합니다.

이렇게 역할을 분리하면 나중에 날씨 에이전트를 다른 종류의 에이전트로 교체하거나, 실제 API를 연결하거나, 여러 개의 도구를 사용하는 에이전트로 확장하기도 쉬워집니다.

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

'Agentic AI > AI_AGENT' 카테고리의 다른 글

A2A. Weather Client (0)	2026.10.03
AI Agent(3/3): MCP(Model Context Protocol) 서버와 MCP 클라이언트(LLM 에이전트) (0)	2026.10.03
AI Agent(2/3): 다단계 도구 호출(Multi-step Tool Calling) 및 자율적 루프(Autonomous Re-act Loop) (0)	2026.10.03
AI Agent(1/3): Function Calling(도구 호출) 에이전트의 정석 (0)	2026.10.03
XVII. AI Agent Engine 기본 완성: Python · Agent Loop · Tool Dispatcher · Function Call · Context Block · History Trim (0)	2026.08.30

A2A. Weather Agent Server

a2a-sdk 1.2.0 기반 — 코드의 구조부터 실행 흐름까지

1. 먼저 전체 구조를 이해하자

2. 라이브러리 Import

2-1. uvicorn

2-2. DefaultRequestHandler

2-3. InMemoryTaskStore

InMemoryTaskStore의 특징

2-4. AgentExecutor와 RequestContext

AgentExecutor

RequestContext

2-5. EventQueue

2-6. FastAPI

2-7. TransportProtocol

2-8. A2A Routes

3. 목업 날씨 데이터

3-1. 목업(Mock)이란?

3-2. 딕셔너리 구조

3-3. 데이터에서 주의할 점

4. WeatherCoreAgent — 핵심 에이전트

4-1. 클래스 정의

4-2. async def invoke()

① async

② self

③ user_query: str

④ -> str

4-3. for문으로 도시명 검색하기

4-4. if city in user_query

4-5. 날씨 데이터 가져오기

4-6. f-string으로 응답 만들기

4-7. 도시명을 찾지 못한 경우

핵심 정리

5. WeatherAgentExecutor — 핵심 에이전트와 A2A를 연결하는 실행기

5-1. AgentExecutor 상속

5-2. init() 생성자

6. execute() — 실제 요청을 처리하는 핵심 메서드

6-1. 사용자 질문 가져오기

6-2. 핵심 에이전트 실행

6-3. 응답을 A2A Message로 변환

① Message

② message_id

③ context_id

④ task_id

⑤ role

⑥ parts

6-4. EventQueue에 응답 등록

이벤트 큐를 사용하는 이유

7. cancel() — 작업 취소 기능

8. AgentSkill — 에이전트의 기능 정의

8-1. id

8-2. name

8-3. description

8-4. tags

8-5. examples

AgentSkill의 핵심

9. AgentCard — 에이전트의 명함

9-1. name

9-2. description

9-3. version

9-4. supported_interfaces

url

protocol_binding

protocol_version

9-5. capabilities

streaming

push_notifications

9-6. default_input_modes와 default_output_modes

9-7. skills

AgentCard와 AgentSkill의 관계

10. AgentExecutor 및 RequestHandler 생성

10-1. WeatherAgentExecutor 객체 생성

10-2. DefaultRequestHandler 객체 생성

agent_executor

task_store

agent_card

전체 연결 구조

11. FastAPI 애플리케이션 생성

12. A2A 라우트 등록

12-1. add_a2a_routes_to_fastapi()

12-2. create_agent_card_routes()