1. 소개
에이전트 간 (A2A) 프로토콜은 특히 외부 시스템에 배포된 AI 에이전트 간의 통신을 표준화하도록 설계되었습니다. 이전에는 LLM을 데이터 및 리소스에 연결하는 새로운 표준인 Model Context Protocol (MCP)이라는 도구에 대해 이러한 프로토콜이 설정되었습니다. A2A는 MCP가 에이전트를 도구 및 데이터에 연결하는 복잡성을 낮추는 데 중점을 두는 반면, A2A는 에이전트가 자연스러운 방식으로 공동작업할 수 있도록 하는 방법에 중점을 두는 등 서로 다른 문제에 초점을 맞추므로 MCP를 보완하려고 합니다. 이를 통해 에이전트는 도구가 아닌 에이전트 (또는 사용자)로 소통할 수 있습니다. 예를 들어 무언가를 주문할 때 양방향 커뮤니케이션을 사용 설정할 수 있습니다.
A2A는 MCP를 보완하도록 설계되었으며, 공식 문서에서는 애플리케이션이 도구에는 MCP를 사용하고 에이전트에는 A2A를 사용하도록 권장합니다 ( AgentCard로 표시됨. 이 내용은 나중에 자세히 설명함). 그러면 프레임워크가 A2A를 사용하여 사용자, 원격 에이전트, 기타 에이전트와 통신할 수 있습니다.
이 데모에서는 Python SDK를 사용하여 A2A를 구현하는 것으로 시작합니다. 버거 및 피자 판매자 상담사와 소통하여 주문을 처리하는 데 도움이 되는 개인 구매 컨시어지가 있는 사용 사례를 살펴보겠습니다.
A2A는 클라이언트-서버 원칙을 활용합니다. 이 데모에서 예상되는 일반적인 A2A 흐름은 다음과 같습니다.
- A2A 클라이언트는 먼저 액세스 가능한 모든 A2A 서버 에이전트 카드에서 검색을 실행하고 해당 정보를 활용하여 연결 클라이언트를 빌드합니다.
- 필요한 경우 A2A 클라이언트는 A2A 서버에 메시지를 전송하고 서버는 이를 완료해야 하는 작업으로 평가합니다. 푸시 알림 수신자 URL이 A2A 클라이언트에 구성되어 있고 A2A 서버에서 지원되는 경우 서버는 클라이언트의 수신 엔드포인트에 작업 진행 상태를 게시할 수도 있습니다.
- 작업이 완료되면 A2A 서버가 응답 아티팩트를 A2A 클라이언트에 전송합니다.
Codelab을 통해 다음과 같이 단계별 접근 방식을 사용합니다.
- Google Cloud 프로젝트를 준비하고 필요한 API를 모두 사용 설정합니다.
- 코딩 환경의 작업공간 설정
- 버거 및 피자 에이전트 서비스의 환경 변수를 준비하고 로컬에서 시도
- Cloud Run에 버거 및 피자 에이전트 배포
- A2A 서버가 설정된 방식에 관한 세부정보 검사
- 구매 컨시어지용 환경 변수 준비 및 로컬에서 시도
- 구매 컨시어지를 Agent Engine에 배포
- 로컬 인터페이스를 통해 에이전트 엔진에 연결
- A2A 클라이언트가 설정된 방식과 데이터 모델링에 관한 세부정보를 검사합니다.
- A2A 클라이언트와 서버 간의 페이로드 및 상호작용 검사
아키텍처 개요
다음 서비스 아키텍처를 배포합니다.
A2A 서버 역할을 하는 두 가지 서비스, 즉 Burger 에이전트 ( CrewAI 에이전트 프레임워크 지원)와 Pizza 에이전트 ( Langgraph 에이전트 프레임워크 지원)를 배포합니다. 사용자는 A2A 클라이언트 역할을 하는 에이전트 개발 키트 (ADK) 프레임워크를 사용하여 실행되는 구매 컨시어지와만 직접 상호작용합니다.
이러한 각 에이전트는 자체 환경과 배포를 갖습니다.
기본 요건
- Python 사용에 능숙함
- HTTP 서비스를 사용하는 기본 풀 스택 아키텍처에 대한 이해
학습할 내용
- A2A 서버의 핵심 구조
- A2A 클라이언트의 핵심 구조
- Cloud Run에 에이전트 서비스 배포
- 에이전트 엔진에 에이전트 서비스 배포
- A2A 클라이언트가 A2A 서버에 연결하는 방법
- 스트리밍되지 않는 연결의 요청 및 응답 구조
필요한 항목
- Chrome 웹브라우저
- Gmail 계정
- 결제가 사용 설정된 Cloud 프로젝트
이 Codelab은 초보자를 포함한 모든 수준의 개발자를 대상으로 하며 샘플 애플리케이션에서 Python을 사용합니다. 하지만 제시된 개념을 이해하는 데 Python 지식이 필요하지는 않습니다.
2. 시작하기 전에
Cloud 콘솔에서 활성 프로젝트 선택
이 Codelab에서는 결제가 사용 설정된 Google Cloud 프로젝트가 이미 있다고 가정합니다. 아직 설치하지 않았다면 아래 안내에 따라 시작하세요.
- Google Cloud 콘솔의 프로젝트 선택기 페이지에서 Google Cloud 프로젝트를 선택하거나 만듭니다.
- Cloud 프로젝트에 결제가 사용 설정되어 있어야 하므로 프로젝트에 결제가 사용 설정되어 있는지 확인하는 방법을 알아보세요.
Cloud Shell 터미널에서 Cloud 프로젝트 설정
- bq가 미리 로드되어 제공되는 Google Cloud에서 실행되는 명령줄 환경인 Cloud Shell을 사용합니다. Google Cloud 콘솔 상단에서 Cloud Shell 활성화를 클릭합니다. 승인을 요청하는 메시지가 표시되면 승인을 클릭합니다.
- Cloud Shell에 연결되면 다음 명령어를 사용하여 인증이 완료되었고 프로젝트가 해당 프로젝트 ID로 설정되었는지 확인합니다.
gcloud auth list
- Cloud Shell에서 다음 명령어를 실행하여 gcloud 명령어가 프로젝트를 알고 있는지 확인합니다.
gcloud config list project
- 프로젝트가 설정되지 않은 경우 다음 명령어를 사용하여 설정합니다.
gcloud config set project <YOUR_PROJECT_ID>
또는 콘솔에서 PROJECT_ID ID를 확인할 수도 있습니다.
클릭하면 오른쪽에 모든 프로젝트와 프로젝트 ID가 표시됩니다.
- 아래에 표시된 명령어를 통해 필수 API를 사용 설정합니다. 몇 분 정도 걸릴 수 있으니 잠시만 기다려 주세요.
gcloud services enable aiplatform.googleapis.com \
run.googleapis.com \
cloudbuild.googleapis.com \
cloudresourcemanager.googleapis.com
명령어가 성공적으로 실행되면 아래와 비슷한 메시지가 표시됩니다.
Operation "operations/..." finished successfully.
gcloud 명령의 대안은 콘솔을 통해 각 제품을 검색하거나 이 링크를 사용하는 것입니다.
API가 누락된 경우 구현 과정에서 언제든지 사용 설정할 수 있습니다.
gcloud 명령어 및 사용법은 문서를 참조하세요.
Cloud Shell 편집기로 이동하여 애플리케이션 작업 디렉터리 설정
이제 코드 편집기를 설정하여 코딩 작업을 할 수 있습니다. 이를 위해 Cloud Shell 편집기를 사용합니다.
- 편집기 열기 버튼을 클릭하면 Cloud Shell 편집기가 열립니다. 여기에 코드를 작성할 수 있습니다.
- 아래 이미지에 강조 표시된 것처럼 Cloud Shell 편집기의 왼쪽 하단 (상태 표시줄)에 Cloud Code 프로젝트가 설정되어 있고 결제가 사용 설정된 활성 Google Cloud 프로젝트로 설정되어 있는지 확인합니다. 메시지가 표시되면 승인을 클릭합니다. 이전 명령어를 이미 따른 경우 버튼이 로그인 버튼 대신 활성화된 프로젝트를 직접 가리킬 수도 있습니다.
- 다음으로 GitHub에서 이 Codelab의 템플릿 작업 디렉터리를 클론합니다. 다음 명령어를 실행하세요. purchasing-concierge-a2a 디렉터리에 작업 디렉터리가 생성됩니다.
git clone https://github.com/alphinside/purchasing-concierge-intro-a2a-codelab-starter.git purchasing-concierge-a2a
- 그런 다음 Cloud Shell 편집기의 상단 섹션으로 이동하여 File->Open Folder를 클릭하고 username 디렉터리를 찾아 purchasing-concierge-a2a 디렉터리를 찾은 다음 OK 버튼을 클릭합니다. 이렇게 하면 선택한 디렉터리가 기본 작업 디렉터리가 됩니다. 이 예시에서 사용자 이름은 alvinprayuda이므로 디렉터리 경로는 아래와 같습니다.
이제 Cloud Shell 편집기가 다음과 같이 표시됩니다.
환경 설정
다음 단계는 개발 환경을 준비하는 것입니다. 현재 활성 터미널은 purchasing-concierge-a2a 작업 디렉터리 내에 있어야 합니다. 이 Codelab에서는 Python 3.12를 사용하고 uv python 프로젝트 관리자를 사용하여 Python 버전과 가상 환경을 만들고 관리할 필요성을 간소화합니다.
- 터미널을 아직 열지 않은 경우 터미널 -> 새 터미널을 클릭하여 열거나 Ctrl + Shift + C를 사용합니다. 그러면 브라우저 하단에 터미널 창이 열립니다.
- 이제
uv(클라우드 터미널에 이미 사전 설치됨)를 사용하여 구매 컨시어지의 가상 환경을 초기화합니다. 다음 명령어를 실행합니다.
uv sync --frozen
그러면 .venv 디렉터리가 생성되고 종속 항목이 설치됩니다. pyproject.toml을 간단히 살펴보면 다음과 같이 표시된 종속 항목에 관한 정보를 확인할 수 있습니다.
dependencies = [
"a2a-sdk>=0.2.16",
"google-adk>=1.8.0",
"gradio>=5.38.2",
]
- 가상 환경을 테스트하려면 main.py라는 새 파일을 만들고 다음 코드를 복사합니다.
def main():
print("Hello from purchasing-concierge-a2a!")
if __name__ == "__main__":
main()
- 그런 다음 다음 명령어를 실행합니다.
uv run main.py
아래와 같은 출력이 표시됩니다.
Using CPython 3.12 Creating virtual environment at: .venv Hello from purchasing-concierge-a2a!
이는 Python 프로젝트가 올바르게 설정되고 있음을 보여줍니다.
이제 다음 단계로 이동하여 원격 판매자 에이전트를 구성하고 배포할 수 있습니다.
3. 원격 판매자 에이전트(A2A 서버)를 Cloud Run에 배포
이 단계에서는 빨간색 상자로 표시된 두 개의 원격 판매자 에이전트를 배포합니다. 햄버거 에이전트는 CrewAI 에이전트 프레임워크를 기반으로 하고, 피자 에이전트는 Langgraph 에이전트를 기반으로 합니다. 두 에이전트 모두 Gemini Flash 2.0 모델을 기반으로 합니다.
원격 Burger Agent 배포
햄버거 에이전트 소스 코드는 remote_seller_agents/burger_agent 디렉터리에 있습니다. agent.py 스크립트에서 에이전트 초기화를 검사할 수 있습니다. 다음은 초기화된 에이전트의 코드 스니펫입니다.
from crewai import Agent, Crew, LLM, Task, Process
from crewai.tools import tool
...
model = LLM(
model="vertex_ai/gemini-2.5-flash-lite", # Use base model name without provider prefix
)
burger_agent = Agent(
role="Burger Seller Agent",
goal=(
"Help user to understand what is available on burger menu and price also handle order creation."
),
backstory=("You are an expert and helpful burger seller agent."),
verbose=False,
allow_delegation=False,
tools=[create_burger_order],
llm=model,
)
agent_task = Task(
description=self.TaskInstruction,
agent=burger_agent,
expected_output="Response to the user in friendly and helpful manner",
)
crew = Crew(
tasks=[agent_task],
agents=[burger_agent],
verbose=False,
process=Process.sequential,
)
inputs = {"user_prompt": query, "session_id": sessionId}
response = crew.kickoff(inputs)
return response
...
remote_seller_agents/burger_agent 디렉터리에 있는 모든 파일은 서비스로 액세스할 수 있도록 Cloud Run에 에이전트를 배포하는 데 충분합니다. 이 부분은 나중에 설명하겠습니다. 다음 명령어를 실행하여 배포합니다.
gcloud run deploy burger-agent \
--source remote_seller_agents/burger_agent \
--port=8080 \
--allow-unauthenticated \
--min 1 \
--region us-central1 \
--update-env-vars GOOGLE_CLOUD_LOCATION=us-central1 \
--update-env-vars GOOGLE_CLOUD_PROJECT={your-project-id}
소스에서 배포하기 위해 컨테이너 저장소가 생성된다는 메시지가 표시되면 Y라고 대답합니다. 이 문제는 이전에 소스에서 Cloud Run에 배포한 적이 없는 경우에만 발생했습니다. 배포가 완료되면 다음과 같은 로그가 표시됩니다.
Service [burger-agent] revision [burger-agent-xxxxx-xxx] has been deployed and is serving 100 percent of traffic. Service URL: https://burger-agent-xxxxxxxxx.us-central1.run.app
여기서 xxxx 부분은 서비스를 배포할 때 고유 식별자가 됩니다. 이제 브라우저를 통해 배포된 버거 에이전트 서비스의 https://burger-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json 경로로 이동해 보겠습니다. 배포된 A2A 서버 에이전트 카드에 액세스하는 URL입니다.
배포에 성공하면 https://burger-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json에 액세스할 때 브라우저에 아래와 같은 응답이 표시됩니다.
이 정보는 검색 목적으로 액세스할 수 있어야 하는 햄버거 에이전트 카드 정보입니다. 이 부분은 나중에 설명하겠습니다. 여기서는 url 값이 여전히 http://0.0.0.0:8080/로 설정되어 있습니다. 이 url 값은 A2A 클라이언트가 외부에서 메시지를 전송하는 데 필요한 기본 정보이며 올바르게 구성되지 않았습니다. 이 데모에서는 추가 환경 변수 HOST_OVERRIDE를 추가하여 이 값을 버거 에이전트 서비스의 URL로 업데이트해야 합니다.
환경 변수를 통해 에이전트 카드에서 Burger Agent URL 값 업데이트
버거 에이전트 서비스에 HOST_OVERRIDE을 추가하려면 다음 단계를 따르세요.
- Cloud 콘솔 상단의 검색창에서 Cloud Run을 검색합니다.
- 이전에 배포된 burger-agent Cloud Run 서비스를 클릭합니다.
- 버거 서비스 URL을 복사한 다음 새 버전 수정 및 배포를 클릭합니다.
- 그런 다음 변수 및 보안 비밀 섹션을 클릭합니다.
- 그런 다음 변수 추가를 클릭하고
HOST_OVERRIDE값을 서비스 URL (https://burger-agent-xxxxxxxxx.us-central1.run.app패턴이 있는 URL)로 설정합니다.
- 마지막으로 배포 버튼을 클릭하여 서비스를 재배포합니다.
이제 브라우저 https://burger-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json에서 burger-agent 에이전트 카드를 다시 액세스하면 url 값이 이미 올바르게 구성되어 있습니다.
원격 피자 에이전트 배포
마찬가지로 피자 에이전트 소스 코드는 remote_seller_agents/pizza_agent 디렉터리에 있습니다. agent.py 스크립트에서 에이전트 초기화를 검사할 수 있습니다. 다음은 초기화된 에이전트의 코드 스니펫입니다.
from langchain_google_vertexai import ChatVertexAI
from langgraph.prebuilt import create_react_agent
...
self.model = ChatVertexAI(
model="gemini-2.5-flash-lite",
location=os.getenv("GOOGLE_CLOUD_LOCATION"),
project=os.getenv("GOOGLE_CLOUD_PROJECT"),
)
self.tools = [create_pizza_order]
self.graph = create_react_agent(
self.model,
tools=self.tools,
checkpointer=memory,
prompt=self.SYSTEM_INSTRUCTION,
)
...
이전 burger-agent 배포 단계와 마찬가지로 remote_seller_agents/pizza_agent 디렉터리에 있는 모든 파일은 서비스로 액세스할 수 있도록 Cloud Run에 에이전트를 배포하는 데 충분합니다. 다음 명령어를 실행하여 배포합니다.
gcloud run deploy pizza-agent \
--source remote_seller_agents/pizza_agent \
--port=8080 \
--allow-unauthenticated \
--min 1 \
--region us-central1 \
--update-env-vars GOOGLE_CLOUD_LOCATION=us-central1 \
--update-env-vars GOOGLE_CLOUD_PROJECT={your-project-id}
배포가 완료되면 다음과 같은 로그가 표시됩니다.
Service [pizza-agent] revision [pizza-agent-xxxxx-xxx] has been deployed and is serving 100 percent of traffic. Service URL: https://pizza-agent-xxxxxxxxx.us-central1.run.app
여기서 xxxx 부분은 서비스를 배포할 때 고유 식별자가 됩니다. 브라우저를 통해 배포된 피자 에이전트 서비스의 https://pizza-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json 경로로 이동하여 A2A 서버 에이전트 카드에 액세스하려고 할 때 햄버거 에이전트의 경우도 마찬가지입니다. 에이전트 카드의 피자 에이전트 url 값이 아직 제대로 구성되지 않았습니다. 또한 환경 변수에 HOST_OVERRIDE를 추가해야 합니다.
환경 변수를 통해 에이전트 카드에서 Pizza Agent URL 값 업데이트
피자 에이전트 서비스에 HOST_OVERRIDE을 추가하려면 다음 단계를 따르세요.
- Cloud 콘솔 상단의 검색창에서 Cloud Run을 검색합니다.
- 이전에 배포된 pizza-agent Cloud Run 서비스를 클릭합니다.
- 새 버전 수정 및 배포를 클릭합니다.
- pizza-service URL을 복사한 다음 변수 및 보안 비밀 섹션을 클릭합니다.
- 그런 다음 변수 추가를 클릭하고
HOST_OVERRIDE값을 서비스 URL (https://pizza-agent-xxxxxxxxx.us-central1.run.app패턴이 있는 URL)로 설정합니다.
- 마지막으로 배포 버튼을 클릭하여 서비스를 재배포합니다.
이제 브라우저 https://pizza-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json에서 pizza-agent 에이전트 카드를 다시 액세스하면 url 값이 이미 올바르게 구성되어 있습니다.
이제 햄버거 서비스와 피자 서비스를 모두 Cloud Run에 성공적으로 배포했습니다. 이제 A2A 서버의 핵심 구성요소를 살펴보겠습니다.
4. A2A 서버의 핵심 구성요소
이제 A2A 서버의 핵심 개념과 구성요소를 살펴보겠습니다.
에이전트 카드
각 A2A 서버에는 /.well-known/agent.json 리소스에서 액세스할 수 있는 에이전트 카드가 있어야 합니다. 이는 A2A 클라이언트의 검색 단계를 지원하기 위한 것으로, 에이전트에 액세스하는 방법과 모든 기능을 파악하는 방법에 관한 완전한 정보와 컨텍스트를 제공해야 합니다. Swagger 또는 Postman을 사용하여 문서화가 잘 된 API 문서와 비슷합니다.
배포된 버거 에이전트 에이전트 카드의 콘텐츠입니다.
{
"capabilities": {
"streaming": true
},
"defaultInputModes": [
"text",
"text/plain"
],
"defaultOutputModes": [
"text",
"text/plain"
],
"description": "Helps with creating burger orders",
"name": "burger_seller_agent",
"protocolVersion": "0.2.6",
"skills": [
{
"description": "Helps with creating burger orders",
"examples": [
"I want to order 2 classic cheeseburgers"
],
"id": "create_burger_order",
"name": "Burger Order Creation Tool",
"tags": [
"burger order creation"
]
}
],
"url": "https://burger-agent-109790610330.us-central1.run.app",
"version": "1.0.0"
}
이러한 에이전트 카드에는 에이전트 기술, 스트리밍 기능, 지원되는 모달리티, 프로토콜 버전 등 다양한 중요한 구성요소가 강조 표시됩니다.
이 모든 정보를 활용하여 A2A 클라이언트가 올바르게 통신할 수 있도록 적절한 통신 메커니즘을 개발할 수 있습니다. 지원되는 모달리티와 인증 메커니즘을 통해 커뮤니케이션을 적절하게 설정할 수 있으며, 에이전트 skills 정보를 A2A 클라이언트 시스템 프롬프트에 삽입하여 호출할 원격 에이전트 기능과 기술에 관한 컨텍스트를 클라이언트의 에이전트에게 제공할 수 있습니다. 이 상담사 카드에 대한 자세한 필드는 이 문서에서 확인할 수 있습니다.
코드에서 에이전트 카드 구현은 A2A Python SDK를 사용하여 설정됩니다. 구현은 아래 remote_seller_agents/burger_agent/main.py 스니펫을 확인하세요.
...
capabilities = AgentCapabilities(streaming=True)
skill = AgentSkill(
id="create_burger_order",
name="Burger Order Creation Tool",
description="Helps with creating burger orders",
tags=["burger order creation"],
examples=["I want to order 2 classic cheeseburgers"],
)
agent_host_url = (
os.getenv("HOST_OVERRIDE")
if os.getenv("HOST_OVERRIDE")
else f"http://{host}:{port}/"
)
agent_card = AgentCard(
name="burger_seller_agent",
description="Helps with creating burger orders",
url=agent_host_url,
version="1.0.0",
defaultInputModes=BurgerSellerAgent.SUPPORTED_CONTENT_TYPES,
defaultOutputModes=BurgerSellerAgent.SUPPORTED_CONTENT_TYPES,
capabilities=capabilities,
skills=[skill],
)
...
다음과 같은 여러 필드가 표시됩니다.
AgentCapabilities: 스트리밍 기능 또는 푸시 알림 지원과 같이 에이전트 서비스에서 지원하는 추가 선택적 기능 선언AgentSkill: 에이전트에서 지원하는 도구 또는 기능Input/OutputModes: 지원되는 입력/출력 유형 모달리티Url: 에이전트와 통신할 주소
이 구성에서는 동적 에이전트 호스트 URL 생성을 제공하므로 로컬 테스트와 클라우드 배포 간에 더 쉽게 전환할 수 있습니다. 따라서 이전 단계에서 HOST_OVERRIDE 변수를 추가해야 합니다.
태스크 큐 및 에이전트 실행기
A2A 서버는 다양한 상담사 또는 사용자의 요청을 처리하고 각 작업을 완벽하게 격리할 수 있습니다. 이러한 상황을 더 잘 시각화하려면 아래 이미지를 검사하세요.
따라서 각 A2A 서버는 수신 작업을 추적하고 이에 관한 적절한 정보를 저장할 수 있어야 합니다. A2A SDK는 A2A 서버에서 이 문제를 해결하는 모듈을 제공합니다. 먼저 수신 요청을 처리하는 방법을 나타내는 로직을 인스턴스화할 수 있습니다. AgentExecutor 추상 클래스를 상속하여 작업 실행 및 취소를 관리하는 방법을 제어할 수 있습니다. 이 예시 구현은 remote_seller_agents/burger_agent/agent_executor.py 모듈에서 검사할 수 있습니다 ( 피자 판매자 사례의 유사한 경로).
...
class BurgerSellerAgentExecutor(AgentExecutor):
"""Burger Seller AgentExecutor."""
def __init__(self):
self.agent = BurgerSellerAgent()
async def execute(
self,
context: RequestContext,
event_queue: EventQueue,
) -> None:
query = context.get_user_input()
try:
result = self.agent.invoke(query, context.context_id)
print(f"Final Result ===> {result}")
parts = [Part(root=TextPart(text=str(result)))]
await event_queue.enqueue_event(
completed_task(
context.task_id,
context.context_id,
[new_artifact(parts, f"burger_{context.task_id}")],
[context.message],
)
)
except Exception as e:
print("Error invoking agent: %s", e)
raise ServerError(error=ValueError(f"Error invoking agent: {e}")) from e
async def cancel(
self, request: RequestContext, event_queue: EventQueue
) -> Task | None:
raise ServerError(error=UnsupportedOperationError())
...
위 코드에서는 요청이 수신될 때 에이전트가 직접 호출되고 호출이 완료된 후 완료된 작업 이벤트를 전송하는 기본 처리 스킴을 구현합니다. 하지만 짧은 실행 작업으로 간주되므로 여기서는 취소 메서드를 구현하지 않았습니다.
실행기를 빌드한 후 빌드된 DefaultRequestHandler, InMemoryTaskStore, A2AStarletteApplication을 직접 활용하여 HTTP 서버를 시작할 수 있습니다. 이 구현은 remote_seller_agents/burger_agent/__main__.py에서 검사할 수 있습니다.
...
request_handler = DefaultRequestHandler(
agent_executor=BurgerSellerAgentExecutor(),
task_store=InMemoryTaskStore(),
)
server = A2AStarletteApplication(
agent_card=agent_card, http_handler=request_handler
)
uvicorn.run(server.build(), host=host, port=port)
...
이 모듈은 에이전트 카드에 액세스하는 /.well-known/agent.json 경로와 A2A 프로토콜을 지원하는 POST 엔드포인트를 구현합니다.
요약
간단히 말해 지금까지는 아래의 두 기능을 지원할 수 있는 Python SDK를 사용하여 A2A 서버를 배포했습니다.
/.well-known/agent.json경로에 상담사 카드 게시- 메모리 내 작업 대기열을 사용하여 JSON-RPC 요청 처리
이러한 기능을 시작할 때의 진입점은 __main__.py 스크립트 ( remote_seller_agents/burger_agent 또는 remote_seller_agents/pizza_agent)에서 검사할 수 있습니다.
5. 구매 컨시어지 - A2A 클라이언트를 에이전트 엔진에 배포
이 단계에서는 구매 컨시어지 에이전트를 배포합니다. 이 에이전트와 상호작용하게 됩니다.
구매 컨시어지 에이전트의 소스 코드는 purchasing_concierge 디렉터리에 있습니다. purchasing_agent.py 스크립트에서 에이전트 초기화를 검사할 수 있습니다. 다음은 초기화된 에이전트의 코드 스니펫입니다.
from google.adk import Agent
...
def create_agent(self) -> Agent:
return Agent(
model="gemini-2.5-flash-lite",
name="purchasing_agent",
instruction=self.root_instruction,
before_model_callback=self.before_model_callback,
before_agent_callback=self.before_agent_callback,
description=(
"This purchasing agent orchestrates the decomposition of the user purchase request into"
" tasks that can be performed by the seller agents."
),
tools=[
self.send_task,
],
)
...
이 에이전트를 에이전트 엔진에 배포합니다. Vertex AI Agent Engine은 개발자가 프로덕션에서 AI 에이전트를 배포, 관리, 확장할 수 있도록 지원하는 서비스 집합입니다. 프로덕션에서 에이전트를 확장하기 위한 인프라를 처리하므로 애플리케이션을 만드는 데 집중할 수 있습니다. 자세한 내용은 이 문서 를 참고하세요. 이전에는 에이전트 서비스를 배포하는 데 필요한 파일 (예: main 서버 스크립트 및 Dockerfile)을 준비해야 했지만, 이 경우에는 ADK와 Agent Engine을 조합하여 자체 백엔드 서비스를 개발하지 않고도 Python 스크립트에서 직접 에이전트를 배포할 수 있습니다. 다음 단계에 따라 배포하세요.
- 먼저 Cloud Storage에 스테이징 스토리지를 만들어야 합니다.
gcloud storage buckets create gs://purchasing-concierge-{your-project-id} --location=us-central1
- 이제 먼저 .env 변수를 준비해야 합니다. .env.example을 .env 파일에 복사해 보겠습니다.
cp .env.example .env
- 이제 .env 파일을 열면 다음과 같은 콘텐츠가 표시됩니다.
GOOGLE_GENAI_USE_VERTEXAI=TRUE
GOOGLE_CLOUD_PROJECT={your-project-id}
GOOGLE_CLOUD_LOCATION=us-central1
STAGING_BUCKET=gs://purchasing-concierge-{your-project-id}
PIZZA_SELLER_AGENT_URL={your-pizza-agent-url}
BURGER_SELLER_AGENT_URL={your-burger-agent-url}
AGENT_ENGINE_RESOURCE_NAME={your-agent-engine-resource-name}
이 상담사는 햄버거 및 피자 상담사와 통신하므로 두 상담사 모두에 적절한 사용자 인증 정보를 제공해야 합니다. 이전 단계의 Cloud Run URL로 PIZZA_SELLER_AGENT_URL 및 BURGER_SELLER_AGENT_URL을 업데이트해야 합니다.
이 사실을 잊은 경우 Cloud Run 콘솔을 방문해 보겠습니다. 콘솔 상단의 검색창에 'Cloud Run'을 입력하고 Cloud Run 아이콘을 마우스 오른쪽 버튼으로 클릭하여 새 탭에서 엽니다.
아래와 같이 이전에 배포된 원격 판매자 에이전트 서비스가 표시됩니다.
이제 서비스의 공개 URL을 확인하려면 서비스 중 하나를 클릭하면 서비스 세부정보 페이지로 리디렉션됩니다. 리전 정보 바로 옆의 상단 영역에서 URL을 확인할 수 있습니다.
최종 환경 변수는 다음과 유사해야 합니다.
GOOGLE_GENAI_USE_VERTEXAI=TRUE
GOOGLE_CLOUD_PROJECT={your-project-id}
GOOGLE_CLOUD_LOCATION=us-central1
STAGING_BUCKET=gs://purchasing-concierge-{your-project-id}
PIZZA_SELLER_AGENT_URL=https://pizza-agent-xxxxx.us-central1.run.app
BURGER_SELLER_AGENT_URL=https://burger-agent-xxxxx.us-central1.run.app
AGENT_ENGINE_RESOURCE_NAME={your-agent-engine-resource-name}
- 이제 구매 컨시어지 에이전트를 배포할 준비가 되었습니다. 이 데모에서는 콘텐츠가 아래에 표시된 스크립트
deploy_to_agent_engine.py를 사용하여 배포합니다.
"""
Copyright 2025 Google LLC
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
https://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
import vertexai
from vertexai.preview import reasoning_engines
from vertexai import agent_engines
from dotenv import load_dotenv
import os
from purchasing_concierge.agent import root_agent
load_dotenv()
PROJECT_ID = os.getenv("GOOGLE_CLOUD_PROJECT")
LOCATION = os.getenv("GOOGLE_CLOUD_LOCATION")
STAGING_BUCKET = os.getenv("STAGING_BUCKET")
vertexai.init(
project=PROJECT_ID,
location=LOCATION,
staging_bucket=STAGING_BUCKET,
)
adk_app = reasoning_engines.AdkApp(
agent=root_agent,
)
remote_app = agent_engines.create(
agent_engine=adk_app,
display_name="purchasing-concierge",
requirements=[
"google-cloud-aiplatform[adk,agent_engines]",
"a2a-sdk==0.2.16",
],
extra_packages=[
"./purchasing_concierge",
],
env_vars={
"GOOGLE_GENAI_USE_VERTEXAI": os.environ["GOOGLE_GENAI_USE_VERTEXAI"],
"PIZZA_SELLER_AGENT_URL": os.environ["PIZZA_SELLER_AGENT_URL"],
"BURGER_SELLER_AGENT_URL": os.environ["BURGER_SELLER_AGENT_URL"],
},
)
print(f"Deployed remote app resource: {remote_app.resource_name}")
ADK 에이전트를 에이전트 엔진에 배포하는 데 필요한 단계입니다. 먼저 ADK root_agent에서 AdkApp 객체를 만들어야 합니다. 그런 다음 adk_app 객체를 제공하고, requirements 필드에 요구사항을 지정하고, extra_packages에 에이전트 디렉터리 경로를 지정하고, 필요한 env 변수를 제공하여 agent_engines.create 메서드를 실행할 수 있습니다.
스크립트를 실행하여 배포할 수 있습니다.
uv run deploy_to_agent_engine.py
배포가 완료되면 다음과 같은 로그가 표시됩니다. xxxx는 프로젝트 ID이고 yyyy는 에이전트 엔진 리소스 ID입니다.
AgentEngine created. Resource name: projects/xxxx/locations/us-central1/reasoningEngines/yyyy
To use this AgentEngine in another session:
agent_engine = vertexai.agent_engines.get('projects/xxxx/locations/us-central1/reasoningEngines/yyyy)
Deployed remote app resource: projects/xxxx/locations/us-central1/reasoningEngines/xxxx
상담사 엔진 대시보드에서 검사하면(검색창에서 '상담사 엔진' 검색) 이전 배포가 표시됩니다.
에이전트 엔진에서 배포된 에이전트 테스트
curl 명령 및 SDK를 통해 에이전트 엔진과 상호작용할 수 있습니다. 예를 들어 다음 명령어를 실행하여 배포된 에이전트와 상호작용해 보세요.
이 쿼리를 전송하여 에이전트가 성공적으로 배포되었는지 확인할 수 있습니다.
curl \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
https://us-central1-aiplatform.googleapis.com/v1/projects/{YOUR_PROJECT_ID}/locations/us-central1/reasoningEngines/{YOUR_AGENT_ENGINE_RESOURCE_ID}:streamQuery?alt=sse -d '{
"class_method": "stream_query",
"input": {
"user_id": "user_123",
"message": "List available burger menu please",
}
}'
성공하면 다음과 같이 콘솔에 스트리밍된 여러 응답 이벤트가 표시됩니다.
{
"content": {
"parts": [
{
"text": "Here is our burger menu:\n- Classic Cheeseburger: IDR 85K\n- Double Cheeseburger: IDR 110K\n- Spicy Chicken Burger: IDR 80K\n- Spicy Cajun Burger: IDR 85K"
}
],
"role": "model"
},
"usage_metadata": {
"candidates_token_count": 51,
"candidates_tokens_details": [
{
"modality": "TEXT",
"token_count": 51
}
],
"prompt_token_count": 907,
"prompt_tokens_details": [
{
"modality": "TEXT",
"token_count": 907
}
],
"total_token_count": 958,
"traffic_type": "ON_DEMAND"
},
"invocation_id": "e-14679918-af68-45f1-b942-cf014368a733",
"author": "purchasing_agent",
"actions": {
"state_delta": {},
"artifact_delta": {},
"requested_auth_configs": {}
},
"id": "dbe7fc43-b82a-4f3e-82aa-dd97afa8f15b",
"timestamp": 1754287348.941454
}
다음 단계에서는 UI를 사용해 보겠습니다. 먼저 A2A 클라이언트의 핵심 구성요소와 일반적인 흐름을 살펴보겠습니다.
6. A2A 클라이언트의 핵심 구성요소
위 이미지에 표시된 내용은 A2A 상호작용의 일반적인 흐름입니다.
- 클라이언트는
/.well-known/agent.json경로에서 제공된 원격 상담사 URL에서 게시된 상담사 카드를 찾으려고 시도합니다. - 그런 다음 필요한 경우 메시지와 필요한 메타데이터 매개변수 ( 예: 세션 ID, 이전 컨텍스트 등)를 사용하여 해당 에이전트에게 메시지를 전송합니다. 서버는 이 메시지를 완료해야 하는 작업으로 인식합니다.
- A2A 서버는 요청을 처리합니다. 서버가 푸시 알림을 지원하는 경우 작업 처리 전반에 걸쳐 일부 알림을 게시할 수도 있습니다. 이 기능은 이 Codelab의 범위를 벗어납니다.
- 완료되면 A2A 서버가 응답 아티팩트를 클라이언트에 다시 전송합니다.
위 상호작용의 핵심 객체는 다음과 같습니다 (자세한 내용은 여기에서 확인).
- 메시지: 클라이언트와 원격 상담사 간의 커뮤니케이션 턴
- 작업: A2A에서 관리하는 기본 작업 단위로, 고유 ID로 식별됩니다.
- 아티팩트: 태스크의 결과로 에이전트가 생성한 출력 (예: 문서, 이미지, 구조화된 데이터)으로, 파트로 구성됩니다.
- Part: 메시지 또는 아티팩트 내의 가장 작은 콘텐츠 단위입니다. 부분은 텍스트, 이미지, 동영상, 파일 등이 될 수 있습니다.
카드 검색
A2A 클라이언트 서비스가 시작될 때 일반적인 프로세스는 상담사 카드 정보를 가져와 필요할 때 쉽게 액세스할 수 있도록 저장하는 것입니다. 이 Codelab에서는 before_agent_callback에서 구현합니다. purchasing_concierge/purchasing_agent.py에서 구현을 확인할 수 있습니다. 아래 코드 스니펫을 참고하세요.
...
async def before_agent_callback(self, callback_context: CallbackContext):
if not self.a2a_client_init_status:
httpx_client = httpx.AsyncClient(timeout=httpx.Timeout(timeout=30))
for address in self.remote_agent_addresses:
card_resolver = A2ACardResolver(
base_url=address, httpx_client=httpx_client
)
try:
card = await card_resolver.get_agent_card()
remote_connection = RemoteAgentConnections(
agent_card=card, agent_url=card.url
)
self.remote_agent_connections[card.name] = remote_connection
self.cards[card.name] = card
except httpx.ConnectError:
print(f"ERROR: Failed to get agent card from : {address}")
agent_info = []
for ra in self.list_remote_agents():
agent_info.append(json.dumps(ra))
self.agents = "\n".join(agent_info)
...
여기에서 내장된 A2A 클라이언트 A2ACardResolver 모듈을 사용하여 사용 가능한 모든 에이전트 카드에 액세스하려고 시도한 다음 에이전트에 메시지를 보내는 데 필요한 연결을 수집합니다. 그런 다음 에이전트가 이러한 에이전트와 통신할 수 있음을 알 수 있도록 사용 가능한 모든 에이전트와 사양을 프롬프트에 나열해야 합니다.
프롬프트 및 작업 전송 도구
다음은 Google에서 ADK 에이전트에게 제공하는 프롬프트와 도구입니다.
...
def root_instruction(self, context: ReadonlyContext) -> str:
current_agent = self.check_active_agent(context)
return f"""You are an expert purchasing delegator that can delegate the user product inquiry and purchase request to the
appropriate seller remote agents.
Execution:
- For actionable tasks, you can use `send_task` to assign tasks to remote agents to perform.
- When the remote agent is repeatedly asking for user confirmation, assume that the remote agent doesn't have access to user's conversation context.
So improve the task description to include all the necessary information related to that agent
- Never ask user permission when you want to connect with remote agents. If you need to make connection with multiple remote agents, directly
connect with them without asking user permission or asking user preference
- Always show the detailed response information from the seller agent and propagate it properly to the user.
- If the remote seller is asking for confirmation, rely the confirmation question to the user if the user haven't do so.
- If the user already confirmed the related order in the past conversation history, you can confirm on behalf of the user
- Do not give irrelevant context to remote seller agent. For example, ordered pizza item is not relevant for the burger seller agent
- Never ask order confirmation to the remote seller agent
Please rely on tools to address the request, and don't make up the response. If you are not sure, please ask the user for more details.
Focus on the most recent parts of the conversation primarily.
If there is an active agent, send the request to that agent with the update task tool.
Agents:
{self.agents}
Current active seller agent: {current_agent["active_agent"]}
"""
...
async def send_task(self, agent_name: str, task: str, tool_context: ToolContext):
"""Sends a task to remote seller agent
This will send a message to the remote agent named agent_name.
Args:
agent_name: The name of the agent to send the task to.
task: The comprehensive conversation context summary
and goal to be achieved regarding user inquiry and purchase request.
tool_context: The tool context this method runs in.
Yields:
A dictionary of JSON data.
"""
if agent_name not in self.remote_agent_connections:
raise ValueError(f"Agent {agent_name} not found")
state = tool_context.state
state["active_agent"] = agent_name
client = self.remote_agent_connections[agent_name]
if not client:
raise ValueError(f"Client not available for {agent_name}")
session_id = state["session_id"]
task: Task
message_id = ""
metadata = {}
if "input_message_metadata" in state:
metadata.update(**state["input_message_metadata"])
if "message_id" in state["input_message_metadata"]:
message_id = state["input_message_metadata"]["message_id"]
if not message_id:
message_id = str(uuid.uuid4())
payload = {
"message": {
"role": "user",
"parts": [
{"type": "text", "text": task}
], # Use the 'task' argument here
"messageId": message_id,
"contextId": session_id,
},
}
message_request = SendMessageRequest(
id=message_id, params=MessageSendParams.model_validate(payload)
)
send_response: SendMessageResponse = await client.send_message(
message_request=message_request
)
print(
"send_response",
send_response.model_dump_json(exclude_none=True, indent=2),
)
if not isinstance(send_response.root, SendMessageSuccessResponse):
print("received non-success response. Aborting get task ")
return None
if not isinstance(send_response.root.result, Task):
print("received non-task response. Aborting get task ")
return None
return send_response.root.result
...
프롬프트에서 구매 컨시어지 에이전트에게 사용 가능한 모든 원격 에이전트 이름과 설명을 제공하고, 도구 self.send_task에서 에이전트에 연결할 적절한 클라이언트를 가져오고 SendMessageRequest 객체를 사용하여 필요한 메타데이터를 전송하는 메커니즘을 제공합니다.
통신 프로토콜
작업 정의는 A2A 서버가 소유한 도메인입니다. 하지만 A2A 클라이언트의 관점에서 보면 서버로 전송되는 메시지로 표시됩니다. 클라이언트에서 수신되는 메시지를 어떤 작업으로 정의할지, 작업을 완료하는 데 클라이언트의 상호작용이 필요한지는 서버에 달려 있습니다. 작업 수명 주기에 관한 자세한 내용은 이 문서를 참고하세요. 이러한 개념을 시각화하면 아래와 같습니다.
메시지->작업의 이러한 교환은 아래 message/send 프로토콜 예시와 같이 JSON-RPC 표준 위에 페이로드 형식을 사용하여 구현됩니다.
{
# identifier for this request
"id": "abc123",
# version of JSON-RPC protocol
"jsonrpc": "2.0",
# method name
"method": "message/send",
# parameters/arguments of the method
"params": {
"message": "hi, what can you help me with?"
}
}
다양한 유형의 통신 (예: 동기화, 스트리밍, 비동기)을 지원하거나 작업 상태에 대한 알림을 구성하는 등 다양한 방법을 사용할 수 있습니다. A2A 서버는 이러한 작업 정의 표준을 처리하도록 유연하게 구성할 수 있습니다. 이러한 메서드의 세부정보는 이 문서에서 확인할 수 있습니다.
7. 통합 테스트 및 페이로드 검사
이제 웹 UI를 사용하여 원격 상담사 상호작용으로 구매 컨시어지를 검사해 보겠습니다.
먼저 .AGENT_ENGINE_RESOURCE_NAMEenv 파일 올바른 에이전트 엔진 리소스 이름을 제공해야 합니다. .env 파일은 다음과 같이 표시됩니다.
GOOGLE_GENAI_USE_VERTEXAI=TRUE
GOOGLE_CLOUD_PROJECT={your-project-id}
GOOGLE_CLOUD_LOCATION=us-central1
STAGING_BUCKET=gs://purchasing-concierge-{your-project-id}
PIZZA_SELLER_AGENT_URL=https://pizza-agent-xxxxx.us-central1.run.app
BURGER_SELLER_AGENT_URL=https://burger-agent-xxxxx.us-central1.run.app
AGENT_ENGINE_RESOURCE_NAME=projects/xxxx/locations/us-central1/reasoningEngines/yyyy
그런 다음 다음 명령어를 실행하여 Gradio 앱을 배포합니다.
uv run purchasing_concierge_ui.py
성공하면 다음과 같은 출력이 표시됩니다.
* Running on local URL: http://0.0.0.0:8080 * To create a public link, set `share=True` in `launch()`.
그런 다음 터미널에서 http://0.0.0.0:8080 URL을 Ctrl + 클릭하거나 웹 미리보기 버튼을 클릭하여 웹 UI를 엽니다.
다음과 같은 대화를 시도해 보세요.
- 햄버거와 피자 메뉴 보여 줘
- 바비큐 치킨 피자 1개와 스파이시 케이준 버거 1개를 주문하고 싶어.
주문이 완료될 때까지 대화를 계속합니다. 상호작용이 어떻게 진행되고 있는지, 도구 호출과 응답은 무엇인지 확인하세요. 다음 이미지는 상호작용 결과의 예입니다.
서로 다른 두 에이전트와 통신하면 서로 다른 두 가지 동작이 발생하며 A2A는 이를 잘 처리할 수 있습니다. 피자 판매자 상담사는 Google의 구매 상담사 요청을 직접 수락하는 반면, 햄버거 상담사는 Google의 요청을 진행하기 전에 Google의 확인이 필요하며, Google에서 확인한 후 상담사는 햄버거 상담사에게 확인을 전달할 수 있습니다.
이제 A2A의 기본 개념을 마치고 클라이언트 및 서버 아키텍처로 구현되는 방식을 살펴보겠습니다.
8. 도전과제
이제 필요한 파일을 준비하고 Gradio 앱을 Cloud Run에 직접 배포할 수 있나요? 이제 챌린지에 참여할 시간입니다.
9. 삭제
이 Codelab에서 사용한 리소스의 비용이 Google Cloud 계정에 청구되지 않도록 하려면 다음 단계를 따르세요.