1. 소개
에이전트 간 (A2A) 프로토콜은 특히 외부 시스템에 배포된 AI 에이전트 간의 통신을 표준화하도록 설계되었습니다. 이전에는 이러한 프로토콜이 LLM을 데이터 및 리소스와 연결하는 신흥 표준인 Model Context Protocol (MCP)이라는 도구에 대해 설정되었습니다. A2A는 MCP를 보완하려고 합니다. A2A는 다른 문제에 중점을 두고 MCP는 에이전트를 도구 및 데이터와 연결하는 복잡성을 줄이는 데 중점을 두는 반면 A2A는 에이전트가 자연스러운 방식으로 공동작업할 수 있는 방법에 중점을 둡니다. 이를 통해 상담사는 도구가 아닌 상담사 (또는 사용자)로서 커뮤니케이션할 수 있습니다. 예를 들어 물건을 주문할 때 상담사와의 양방향 커뮤니케이션을 사용 설정할 수 있습니다.
A2A는 MCP를 보완하기 위한 것입니다. 공식 문서에 따르면 애플리케이션은 A2A 에이전트를 AgentCard로 표시되는 MCP 리소스로 모델링하는 것이 좋습니다 ( 이 내용은 나중에 설명). 그러면 프레임워크는 A2A를 사용하여 사용자, 원격 에이전트, 기타 에이전트와 통신할 수 있습니다.
이 데모에서는 A2A만 처음부터 구현하는 것으로 시작합니다. 이러한 샘플 저장소 에서 파생된 개인 구매 컨시어지를 통해 버거 및 피자 판매자 상담사와 소통하여 주문을 처리하는 사용 사례를 살펴보겠습니다.
A2A는 클라이언트-서버 원리를 활용합니다. 다음은 이 데모에서 예상되는 일반적인 A2A 흐름입니다.
- A2A 클라이언트는 먼저 액세스할 수 있는 모든 A2A 서버 상담사 카드를 검색하고 이 정보를 사용하여 연결 클라이언트를 빌드합니다.
- 필요한 경우 A2A 클라이언트는 A2A 서버로 태스크를 전송합니다. 푸시 알림 수신기 URL이 A2A 클라이언트에 구성된 경우 A2A 서버는 수신 엔드포인트에 작업 진행 상태를 게시할 수도 있습니다.
- 작업이 완료되면 A2A 서버가 응답 아티팩트를 A2A 클라이언트로 전송합니다.
이 Codelab에서는 다음과 같이 단계별로 접근합니다.
- Google Cloud 프로젝트를 준비하고 필요한 모든 API를 사용 설정합니다.
- 코딩 환경의 워크스페이스 설정
- 버거 및 피자 상담사 서비스의 env 변수 준비
- Cloud Run에 버거 및 피자 에이전트 배포
- A2A 서버가 설정된 방식에 관한 세부정보 검사
- 구매 컨시어지를 위한 환경 변수 준비
- Cloud Run에 구매 컨시어지를 배포합니다.
- A2A 클라이언트 설정 방법 및 데이터 모델링에 관한 세부정보 검사
- A2A 클라이언트와 서버 간의 페이로드 및 상호작용 검사
아키텍처 개요
다음과 같은 서비스 아키텍처를 배포합니다.
A2A 서버 역할을 하는 두 가지 서비스, Burger 에이전트 ( CrewAI 에이전트 프레임워크 지원)와 Pizza 에이전트 ( Langgraph 에이전트 프레임워크 지원)를 배포합니다. 사용자는 A2A 클라이언트 역할을 하는 Agent Development Kit (ADK) 프레임워크를 사용하여 실행되는 Purchasing 컨시어지와만 직접 상호작용합니다.
이러한 각 에이전트는 자체 환경과 배포를 갖습니다.
기본 요건
- 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 Code 프로젝트가 아래 이미지에서 강조 표시된 것처럼 Cloud Shell 편집기의 왼쪽 하단 (상태 표시줄)에 설정되어 있고 결제가 사용 설정된 활성 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 편집기 상단 섹션으로 이동하여 파일 -> 폴더 열기를 클릭하고 사용자 이름 디렉터리를 찾아 purchasing-concierge-a2a 디렉터리를 찾은 다음 확인 버튼을 클릭합니다. 이렇게 하면 선택한 디렉터리가 기본 작업 디렉터리가 됩니다. 이 예시에서 사용자 이름은 alvinprayuda이므로 디렉터리 경로는 아래와 같습니다.
이제 Cloud Shell 편집기가 다음과 같이 표시됩니다.
환경 설정
다음 단계는 개발 환경을 준비하는 것입니다. 현재 활성 터미널이 purchasing-concierge-a2a 작업 디렉터리 내에 있어야 합니다. 이 Codelab에서는 Python 3.12를 활용하고 uv python project manager를 사용하여 Python 버전과 가상 환경을 만들고 관리하는 작업을 간소화합니다.
- 아직 터미널을 열지 않았다면 터미널 -> 새 터미널을 클릭하거나 Ctrl + Shift + C를 사용하여 터미널을 엽니다. 그러면 브라우저 하단에 터미널 창이 열립니다.
uv를 다운로드하고 다음 명령어를 사용하여 Python 3.12를 설치합니다.
curl -LsSf https://astral.sh/uv/0.7.2/install.sh | sh && \
source $HOME/.local/bin/env && \
uv python install 3.12
- 이제
uv를 사용하여 구매 컨시어즈의 가상 환경을 초기화해 보겠습니다. 이 명령어를 실행합니다.
uv sync --frozen
이렇게 하면 .venv 디렉터리가 생성되고 종속 항목이 설치됩니다. pyproject.toml을 살짝 살펴보면 다음과 같이 종속 항목에 관한 정보를 확인할 수 있습니다.
dependencies = [
"google-adk>=0.3.0",
"gradio>=5.28.0",
"httpx>=0.28.1",
"jwcrypto>=1.5.6",
"pydantic>=2.10.6",
"pyjwt>=2.10.1",
"sse-starlette>=2.3.3",
"starlette>=0.46.2",
"typing-extensions>=4.13.2",
"uvicorn>=0.34.0",
]
- 가상 환경을 테스트하려면 새 파일 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. Cloud Run에 원격 판매자 에이전트 - A2A 서버 배포
이 단계에서는 빨간색 상자로 표시된 두 개의 원격 판매자 에이전트를 배포합니다. 버거 상담사는 CrewAI 상담사 프레임워크를 기반으로 하고 피자 상담사는 Langgraph 상담사를 기반으로 합니다. 두 상담사 모두 Gemini Flash 2.0 모델을 기반으로 합니다.
원격 버거 에이전트 배포
버거 에이전트 소스 코드는 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.0-flash", # 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,
output_pydantic=ResponseFormat,
agent=burger_agent,
expected_output=(
"A JSON object with 'status' and 'message' fields."
"Set response status to input_required if asking for user order confirmation."
"Set response status to error if there is an error while processing the request."
"Set response status to completed if the request is complete."
),
)
crew = Crew(
tasks=[agent_task],
agents=[burger_agent],
verbose=False,
process=Process.sequential,
)
inputs = {"user_prompt": query, "session_id": sessionId}
response = crew.kickoff(inputs)
...
이제 먼저 .env 변수를 준비해야 합니다. .env.example을 .env 파일에 복사합니다.
cp remote_seller_agents/burger_agent/.env.example remote_seller_agents/burger_agent/.env
이제 remote_seller_agents/burger_agent/.env 파일을 열면 다음 콘텐츠가 표시됩니다.
AUTH_USERNAME=burgeruser123
AUTH_PASSWORD=burgerpass123
GCLOUD_LOCATION=us-central1
GCLOUD_PROJECT_ID={your-project-id}
버거 에이전트 A2A 서버는 기본 HTTP 인증 ( base64로 인코딩된 사용자 이름 및 비밀번호 사용)을 사용하여 실행되며 이 데모에서는 .env 파일에서 허용된 사용자 이름과 비밀번호를 구성합니다. 이제 GCLOUD_PROJECT_ID 변수를 현재 활성 프로젝트 ID로 업데이트합니다.
변경사항을 저장하는 것을 잊지 마세요. 다음으로 서비스를 직접 배포할 수 있습니다. 코드 콘텐츠는 나중에 검사합니다. 다음 명령어를 실행하여 배포합니다.
gcloud run deploy burger-agent \
--source remote_seller_agents/burger_agent \
--port=8080 \
--allow-unauthenticated \
--min 1 \
--region us-central1
소스에서 배포하기 위해 컨테이너 저장소가 생성된다는 메시지가 표시되면 Y를 입력합니다. 배포가 완료되면 다음과 같은 로그가 표시됩니다.
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 부분은 서비스를 배포할 때 고유 식별자가 됩니다.
이제 브라우저를 통해 배포된 버거 에이전트 서비스의 /.well-known/agent.json 경로를 살펴보겠습니다. 다음과 같은 출력이 표시됩니다.
탐색 목적으로 액세스할 수 있어야 하는 버거 상담사 카드 정보입니다. 이 부분에 대해서는 나중에 살펴보겠습니다. 지금은 나중에 사용할 버거 상담사 서비스의 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.0-flash",
location=os.getenv("GCLOUD_LOCATION"),
project=os.getenv("GCLOUD_PROJECT_ID"),
)
self.tools = [create_pizza_order]
self.graph = create_react_agent(
self.model,
tools=self.tools,
checkpointer=memory,
prompt=self.SYSTEM_INSTRUCTION,
response_format=ResponseFormat,
)
...
이제 먼저 .env 변수를 준비해야 합니다. .env.example을 .env 파일에 복사합니다.
cp remote_seller_agents/pizza_agent/.env.example remote_seller_agents/pizza_agent/.env
이제 remote_seller_agents/pizza_agent/.env 파일을 열면 다음 콘텐츠가 표시됩니다.
API_KEY=pizza123
GCLOUD_LOCATION=us-central1
GCLOUD_PROJECT_ID={your-project-id}
피자 상담사 A2A 서버는 Bearer HTTP 인증 (API 키 사용)을 사용하여 실행되며, 이 데모에서는 .env 파일에서 허용된 API 키를 구성합니다. 이제 GCLOUD_PROJECT_ID 변수를 현재 활성 프로젝트 ID로 업데이트합니다.
변경사항을 저장하는 것을 잊지 마세요. 다음으로 서비스를 직접 배포할 수 있습니다. 코드 콘텐츠는 나중에 검사합니다. 다음 명령어를 실행하여 배포합니다.
gcloud run deploy pizza-agent \
--source remote_seller_agents/pizza_agent \
--port=8080 \
--allow-unauthenticated \
--min 1 \
--region us-central1
배포가 완료되면 다음과 같은 로그가 표시됩니다.
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 부분은 서비스를 배포할 때 고유 식별자가 됩니다.
이제 브라우저를 통해 배포된 피자 상담사 서비스의 /.well-known/agent.json 경로를 살펴보겠습니다. 다음과 같은 출력이 표시됩니다.
이는 검색 목적으로 액세스할 수 있어야 하는 피자 상담사 카드 정보입니다. 이 부분에 대해서는 나중에 살펴보겠습니다. 지금은 피자 상담사 서비스의 URL만 기억해 두세요.
이제 햄버거 서비스와 피자 서비스를 모두 Cloud Run에 배포했습니다. 이제 A2A 서버의 핵심 구성요소를 살펴보겠습니다.
4. A2A 서버의 핵심 구성요소
이제 A2A 서버의 핵심 개념과 구성요소를 살펴보겠습니다.
상담사 카드
각 A2A 서버에는 /.well-known/agent.json 리소스에서 액세스할 수 있는 에이전트 카드가 있어야 합니다. 이는 A2A 클라이언트의 검색 단계를 지원하기 위한 것으로, 상담사에게 액세스하고 모든 기능을 파악하는 방법에 관한 완전한 정보와 컨텍스트를 제공해야 합니다. Swagger 또는 Postman을 사용하여 잘 문서화된 API 문서와 비슷합니다.
배포된 버거 상담사 카드의 콘텐츠입니다.
{
"name": "burger_seller_agent",
"description": "Helps with creating burger orders",
"url": "http://0.0.0.0:8080/",
"version": "1.0.0",
"capabilities": {
"streaming": false,
"pushNotifications": true,
"stateTransitionHistory": false
},
"authentication": {
"schemes": [
"Basic"
]
},
"defaultInputModes": [
"text",
"text/plain"
],
"defaultOutputModes": [
"text",
"text/plain"
],
"skills": [
{
"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"
]
}
]
}
상담사 카드에는 상담사 기술, 스트리밍 기능, 지원되는 모달리티, 인증과 같은 여러 중요한 구성요소가 강조 표시됩니다.
이 모든 정보를 활용하여 A2A 클라이언트가 올바르게 통신할 수 있는 적절한 통신 메커니즘을 개발할 수 있습니다. 지원되는 모달리티와 인증 메커니즘을 통해 통신을 적절하게 설정할 수 있으며, 에이전트 skills 정보를 A2A 클라이언트 시스템 프롬프트에 삽입하여 호출할 원격 에이전트 기능과 스킬에 관한 클라이언트의 에이전트 컨텍스트를 제공할 수 있습니다. 이 상담사 카드의 자세한 필드는 이 문서를 참고하세요.
이 코드에서는 a2a_types.py ( remote_seller_agents/burger_agent 또는 remote_seller_agents/pizza_agent)에서 Pydantic을 사용하여 상담사 카드의 구현이 설정됩니다.
...
class AgentProvider(BaseModel):
organization: str
url: str | None = None
class AgentCapabilities(BaseModel):
streaming: bool = False
pushNotifications: bool = False
stateTransitionHistory: bool = False
class AgentAuthentication(BaseModel):
schemes: List[str]
credentials: str | None = None
class AgentSkill(BaseModel):
id: str
name: str
description: str | None = None
tags: List[str] | None = None
examples: List[str] | None = None
inputModes: List[str] | None = None
outputModes: List[str] | None = None
class AgentCard(BaseModel):
name: str
description: str | None = None
url: str
provider: AgentProvider | None = None
version: str
documentationUrl: str | None = None
capabilities: AgentCapabilities
authentication: AgentAuthentication | None = None
defaultInputModes: List[str] = ["text"]
defaultOutputModes: List[str] = ["text"]
skills: List[AgentSkill]
...
객체 생성은 아래에 반영된 remote_seller_agents/burger_agent/__main__.py에 있습니다.
...
def main(host, port):
"""Starts the Burger Seller Agent server."""
try:
capabilities = AgentCapabilities(pushNotifications=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_card = AgentCard(
name="burger_seller_agent",
description="Helps with creating burger orders",
# The URL provided here is for the sake of demo,
# in production you should use a proper domain name
url=f"http://{host}:{port}/",
version="1.0.0",
authentication=AgentAuthentication(schemes=["Basic"]),
defaultInputModes=BurgerSellerAgent.SUPPORTED_CONTENT_TYPES,
defaultOutputModes=BurgerSellerAgent.SUPPORTED_CONTENT_TYPES,
capabilities=capabilities,
skills=[skill],
)
notification_sender_auth = PushNotificationSenderAuth()
notification_sender_auth.generate_jwk()
server = A2AServer(
agent_card=agent_card,
task_manager=AgentTaskManager(
agent=BurgerSellerAgent(),
notification_sender_auth=notification_sender_auth,
),
host=host,
port=port,
auth_username=os.environ.get("AUTH_USERNAME"),
auth_password=os.environ.get("AUTH_PASSWORD"),
)
...
작업 정의 및 작업 관리자
A2A의 핵심 구성요소 중 하나는 태스크 정의입니다. JSON-RPC 표준을 기반으로 페이로드 형식을 조정합니다. 이 데모에서는 이 섹션의 a2a_types.py ( remote_seller_agents/burger_agent 또는 remote_seller_agents/pizza_agent)에서도 Pydantic을 사용하여 구현됩니다.
...
## RPC Messages
class JSONRPCMessage(BaseModel):
jsonrpc: Literal["2.0"] = "2.0"
id: int | str | None = Field(default_factory=lambda: uuid4().hex)
class JSONRPCRequest(JSONRPCMessage):
method: str
params: dict[str, Any] | None = None
...
class SendTaskRequest(JSONRPCRequest):
method: Literal["tasks/send"] = "tasks/send"
params: TaskSendParams
class SendTaskStreamingRequest(JSONRPCRequest):
method: Literal["tasks/sendSubscribe"] = "tasks/sendSubscribe"
params: TaskSendParams
class GetTaskRequest(JSONRPCRequest):
method: Literal["tasks/get"] = "tasks/get"
params: TaskQueryParams
class CancelTaskRequest(JSONRPCRequest):
method: Literal["tasks/cancel",] = "tasks/cancel"
params: TaskIdParams
class SetTaskPushNotificationRequest(JSONRPCRequest):
method: Literal["tasks/pushNotification/set",] = "tasks/pushNotification/set"
params: TaskPushNotificationConfig
class GetTaskPushNotificationRequest(JSONRPCRequest):
method: Literal["tasks/pushNotification/get",] = "tasks/pushNotification/get"
params: TaskIdParams
class TaskResubscriptionRequest(JSONRPCRequest):
method: Literal["tasks/resubscribe",] = "tasks/resubscribe"
params: TaskIdParams
...
다양한 유형의 통신 (예: 동기화, 스트리밍, 비동기)을 지원하고 작업 상태에 대한 알림을 구성하는 데 사용할 수 있는 다양한 작업 메서드가 있습니다. A2A 서버는 이러한 태스크 정의 표준을 처리하도록 유연하게 구성할 수 있습니다.
A2A 서버는 여러 상담사 또는 사용자의 요청을 처리하고 각 태스크를 완벽하게 격리할 수 있습니다. 이러한 컨텍스트를 더 잘 시각화하려면 아래 이미지를 살펴보세요.
따라서 각 A2A 서버는 수신되는 작업을 추적하고 이에 관한 적절한 정보를 저장할 수 있어야 합니다. 일반적으로 수신되는 각 요청에는 작업 ID 및 세션 ID가 있습니다. 이 코드에서 이 작업 관리자의 구현은 remote_seller_agents/burger_agent/task_manager.py에 있습니다 (피자 상담사도 유사한 작업 관리자를 공유함).
...
class AgentTaskManager(InMemoryTaskManager):
def __init__(
self,
agent: BurgerSellerAgent,
notification_sender_auth: PushNotificationSenderAuth,
):
super().__init__()
self.agent = agent
self.notification_sender_auth = notification_sender_auth
...
async def on_send_task(self, request: SendTaskRequest) -> SendTaskResponse:
"""Handles the 'send task' request."""
validation_error = self._validate_request(request)
if validation_error:
return SendTaskResponse(id=request.id, error=validation_error.error)
await self.upsert_task(request.params)
if request.params.pushNotification:
if not await self.set_push_notification_info(
request.params.id, request.params.pushNotification
):
return SendTaskResponse(
id=request.id,
error=InvalidParamsError(
message="Push notification URL is invalid"
),
)
task = await self.update_store(
request.params.id, TaskStatus(state=TaskState.WORKING), None
)
await self.send_task_notification(task)
task_send_params: TaskSendParams = request.params
query = self._get_user_query(task_send_params)
try:
agent_response = self.agent.invoke(query, task_send_params.sessionId)
except Exception as e:
logger.error(f"Error invoking agent: {e}")
raise ValueError(f"Error invoking agent: {e}")
return await self._process_agent_response(request, agent_response)
...
async def _process_agent_response(
self, request: SendTaskRequest, agent_response: dict
) -> SendTaskResponse:
"""Processes the agent's response and updates the task store."""
task_send_params: TaskSendParams = request.params
task_id = task_send_params.id
history_length = task_send_params.historyLength
task_status = None
parts = [{"type": "text", "text": agent_response["content"]}]
artifact = None
if agent_response["require_user_input"]:
task_status = TaskStatus(
state=TaskState.INPUT_REQUIRED,
message=Message(role="agent", parts=parts),
)
else:
task_status = TaskStatus(state=TaskState.COMPLETED)
artifact = Artifact(parts=parts)
task = await self.update_store(
task_id, task_status, None if artifact is None else [artifact]
)
task_result = self.append_task_history(task, history_length)
await self.send_task_notification(task)
return SendTaskResponse(id=request.id, result=task_result)
...
위 코드에서 수신 작업을 처리할 때 ( 수신 요청 메서드가 tasks/send인 경우 on_send_task 메서드가 실행됨) 작업 저장소 업데이트 ( self.update_store 메서드 호출) 및 알림 전송 ( self.send_task_notification 메서드 호출)에 관한 여러 작업을 실행한다는 것을 확인할 수 있습니다. 이는 A2A 서버가 동기식 전송 작업 요청 전반에서 작업 상태 업데이트 및 알림을 관리하는 방법의 예 중 하나입니다.
요약
간단히 말해 지금까지 배포된 A2A 서버는 다음 두 가지 기능을 지원할 수 있습니다.
/.well-known/agent.json경로에 상담사 카드 게시tasks/send메서드로 JSON-RPC 요청 처리
이러한 기능을 시작하는 진입점은 main.py 스크립트 ( remote_seller_agents/burger_agent 또는 remote_seller_agents/pizza_agent)에서 검사할 수 있습니다. 아래 코드 스니펫에서 먼저 상담사 카드를 구성한 다음 서버를 시작해야 합니다.
...
capabilities = AgentCapabilities(pushNotifications=True)
skill = AgentSkill(
id="create_pizza_order",
name="Pizza Order Creation Tool",
description="Helps with creating pizza orders",
tags=["pizza order creation"],
examples=["I want to order 2 pepperoni pizzas"],
)
agent_card = AgentCard(
name="pizza_seller_agent",
description="Helps with creating pizza orders",
# The URL provided here is for the sake of demo,
# in production you should use a proper domain name
url=f"http://{host}:{port}/",
version="1.0.0",
authentication=AgentAuthentication(schemes=["Bearer"]),
defaultInputModes=PizzaSellerAgent.SUPPORTED_CONTENT_TYPES,
defaultOutputModes=PizzaSellerAgent.SUPPORTED_CONTENT_TYPES,
capabilities=capabilities,
skills=[skill],
)
...
server = A2AServer(
agent_card=agent_card,
task_manager=AgentTaskManager(
agent=PizzaSellerAgent(),
notification_sender_auth=notification_sender_auth,
),
host=host,
port=port,
api_key=os.environ.get("API_KEY"),
)
...
logger.info(f"Starting server on {host}:{port}")
server.start()
...
5. Cloud Run에 구매 컨시어지 - A2A 클라이언트 배포
이 단계에서는 구매 컨시어지 에이전트를 배포합니다. 이 에이전트는 우리가 상호작용할 에이전트입니다.
구매 컨시어지 에이전트의 소스 코드는 purchasing_concierge 디렉터리에 있습니다. 에이전트 초기화는 purchasing_agent.py 스크립트에서 검사할 수 있습니다. 다음은 초기화된 에이전트의 코드 스니펫입니다.
from google.adk import Agent
...
def create_agent(self) -> Agent:
return Agent(
model="gemini-2.0-flash-001",
name="purchasing_agent",
instruction=self.root_instruction,
before_model_callback=self.before_model_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.list_remote_agents,
self.send_task,
],
)
...
이제 먼저 .env 변수를 준비해야 합니다. .env.example을 .env 파일에 복사합니다.
cp purchasing_concierge/.env.example purchasing_concierge/.env
이제 purchasing_concierge/.env 파일을 열면 다음 콘텐츠가 표시됩니다.
PIZZA_SELLER_AGENT_AUTH=pizza123
PIZZA_SELLER_AGENT_URL=http://localhost:10000
BURGER_SELLER_AGENT_AUTH=burgeruser123:burgerpass123
BURGER_SELLER_AGENT_URL=http://localhost:10001
GOOGLE_GENAI_USE_VERTEXAI=TRUE
GOOGLE_CLOUD_PROJECT={your-project-id}
GOOGLE_CLOUD_LOCATION=us-central1
이 상담사는 버거 상담사와 피자 상담사와 통신하므로 두 상담사 모두에게 적절한 사용자 인증 정보를 제공해야 합니다. 이제 GCLOUD_PROJECT_ID 변수를 현재 활성 프로젝트 ID로 업데이트합니다.
이제 PIZZA_SELLER_AGENT_URL 및 BURGER_SELLER_AGENT_URL도 이전 단계의 Cloud Run URL로 업데이트해야 합니다. 잊어버린 경우 Cloud Run 콘솔을 방문하세요. 콘솔 상단의 검색창에 'Cloud Run'을 입력하고 Cloud Run 아이콘을 마우스 오른쪽 버튼으로 클릭하여 새 탭에서 엽니다.
아래와 같이 이전에 배포된 원격 판매자 상담사 서비스가 표시됩니다.
이제 이러한 서비스의 공개 URL을 보려면 서비스를 클릭하면 서비스 세부정보 페이지로 리디렉션됩니다. URL은 지역 정보 바로 옆의 상단 영역에서 확인할 수 있습니다.
이 URL의 값을 복사하여 PIZZA_SELLER_AGENT_URL 및 BURGER_SELLER_AGENT_URL에 각각 붙여넣습니다.
최종 환경 변수는 다음과 유사해야 합니다.
PIZZA_SELLER_AGENT_AUTH=pizza123
PIZZA_SELLER_AGENT_URL=https://pizza-agent-xxxxx.us-central1.run.app
BURGER_SELLER_AGENT_AUTH=burgeruser123:burgerpass123
BURGER_SELLER_AGENT_URL=https://burger-agent-xxxxx.us-central1.run.app
GOOGLE_GENAI_USE_VERTEXAI=TRUE
GOOGLE_CLOUD_PROJECT={your-project-id}
GOOGLE_CLOUD_LOCATION=us-central1
이제 구매 컨시어지 에이전트를 배포할 준비가 되었습니다. 이 에이전트를 배포하려면 아래 명령어를 실행합니다.
gcloud run deploy purchasing-concierge \
--source . \
--port=8080 \
--allow-unauthenticated \
--min 1 \
--region us-central1 \
--memory 1024Mi
배포가 완료되면 다음과 같은 로그가 표시됩니다.
Service [purchasing-concierge] revision [purchasing-concierge-xxxxx-xxx] has been deployed and is serving 100 percent of traffic. Service URL: https://purchasing-concierge-xxxxxx.us-central1.run.app
여기서 xxxx 부분은 서비스를 배포할 때 고유 식별자가 됩니다.
이제 UI를 통해 구매 컨시어지 상담사와 상호작용할 수 있습니다. 서비스 URL에 액세스하면 아래와 같이 Gradio 웹 인터페이스가 표시됩니다.
이제 A2A 클라이언트의 핵심 구성요소와 일반적인 흐름을 살펴보겠습니다.
6. A2A 클라이언트의 핵심 구성요소
위의 이미지는 A2A 상호작용의 일반적인 흐름을 보여줍니다.
- 클라이언트는 경로
/.well-known/agent.json의 제공된 원격 상담사 URL에서 게시된 상담사 카드를 찾으려고 시도합니다. - 그런 다음 필요한 경우 메시지와 필요한 메타데이터 매개변수 ( 예: 세션 ID, 이전 컨텍스트 등)가 포함된 태스크를 상담사에게 전송합니다.
- A2A 서버는 요청을 인증하고 처리합니다. 서버에서 푸시 알림을 지원하는 경우 작업 처리 중에 일부 알림을 게시하려고 시도합니다.
- 완료되면 A2A 서버가 응답 아티팩트를 클라이언트로 다시 전송합니다.
위 상호작용의 핵심 객체는 다음과 같습니다 (자세한 내용은 여기를 참고하세요).
- 작업: 클라이언트와 원격 상담사가 특정 결과를 달성하고 결과를 생성할 수 있는 상태 개체입니다.
- 아티팩트: 작업의 최종 결과
- 메시지: 아티팩트가 아닌 모든 콘텐츠입니다. 예: 상담사 의견, 사용자 컨텍스트, 안내, 오류, 상태 또는 메타데이터
- 항목: 클라이언트와 원격 에이전트 간에 메시지 또는 아티팩트의 일부로 교환되는 완전히 형성된 콘텐츠입니다. 부분은 텍스트, 이미지, 동영상, 파일 등이 될 수 있습니다.
- 푸시 알림 (선택사항): 상담사가 연결된 세션 외부에서 클라이언트에게 업데이트를 알릴 수 있는 보안 알림 메커니즘입니다.
카드 검색
A2A 클라이언트 서비스를 시작할 때 일반적인 프로세스는 상담사 카드 정보를 가져와 저장하여 필요할 때 쉽게 액세스하는 것입니다. 여기의 purchasing_concierge/purchasing_agent.py 스크립트에서 확인할 수 있습니다.
...
class PurchasingAgent:
"""The purchasing agent.
This is the agent responsible for choosing which remote seller agents to send
tasks to and coordinate their work.
"""
def __init__(
self,
remote_agent_addresses: List[str],
task_callback: TaskUpdateCallback | None = None,
):
self.task_callback = task_callback
self.remote_agent_connections: dict[str, RemoteAgentConnections] = {}
self.cards: dict[str, AgentCard] = {}
for address in remote_agent_addresses:
card_resolver = A2ACardResolver(address)
try:
card = card_resolver.get_agent_card()
# The URL accessed here should be the same as the one provided in the agent card
# However, in this demo we are using the URL provided in the key arguments
remote_connection = RemoteAgentConnections(
agent_card=card, agent_url=address
)
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)
...
프롬프트 및 전송 태스크 도구
그러면 구매 컨시어지 시스템 프롬프트에 원격 상담사 컨텍스트를 제공하고 상담사에게 작업을 전송하는 도구도 제공합니다. 다음은 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}")
if "task_id" in state:
taskId = state["task_id"]
else:
taskId = str(uuid.uuid4())
sessionId = state["session_id"]
task: Task
messageId = ""
metadata = {}
if "input_message_metadata" in state:
metadata.update(**state["input_message_metadata"])
if "message_id" in state["input_message_metadata"]:
messageId = state["input_message_metadata"]["message_id"]
if not messageId:
messageId = str(uuid.uuid4())
metadata.update(**{"conversation_id": sessionId, "message_id": messageId})
request: TaskSendParams = TaskSendParams(
id=taskId,
sessionId=sessionId,
message=Message(
role="user",
parts=[TextPart(text=task)],
metadata=metadata,
),
acceptedOutputModes=["text", "text/plain"],
# pushNotification=None,
metadata={"conversation_id": sessionId},
)
task = await client.send_task(request, self.task_callback)
# Assume completion unless a state returns that isn't complete
state["session_active"] = task.status.state not in [
TaskState.COMPLETED,
TaskState.CANCELED,
TaskState.FAILED,
TaskState.UNKNOWN,
]
if task.status.state == TaskState.INPUT_REQUIRED:
# Force user input back
tool_context.actions.escalate = True
elif task.status.state == TaskState.COMPLETED:
# Reset active agent is task is completed
state["active_agent"] = "None"
response = []
if task.status.message:
# Assume the information is in the task message.
response.extend(convert_parts(task.status.message.parts, tool_context))
if task.artifacts:
for artifact in task.artifacts:
response.extend(convert_parts(artifact.parts, tool_context))
return response
...
프롬프트에서 구매 컨시어지 상담사에게 사용 가능한 모든 원격 상담사 이름과 설명을 제공하고 self.send_task 도구에서 상담사에게 연결하고 TaskSendParams 객체를 사용하여 필요한 메타데이터를 전송하는 적절한 클라이언트를 검색하는 메커니즘을 제공합니다.
도구에서 태스크가 완료되지 않는 경우 에이전트가 어떻게 동작할지 지정할 수도 있습니다. 마지막으로 작업이 완료될 때 반환된 응답 아티팩트를 처리해야 합니다.
7. 통합 테스트 및 페이로드 검사
이제 다음 대화를 시도하고 구매 컨시어지 UI 및 서비스 로그를 살펴보겠습니다. 다음과 같은 대화를 시도해 보세요.
- 버거와 피자 메뉴 보여 줘
- 바비큐 치킨 피자 1개와 매운 케이준 버거 1개를 주문하고 싶습니다.
주문을 완료할 때까지 대화를 계속합니다. 상호작용 진행 상황과 도구 호출 및 응답을 검사합니다. 다음 이미지는 상호작용 결과의 예입니다.
두 명의 서로 다른 상담사와 통신하면 서로 다른 두 가지 동작이 발생하며 A2A는 이를 잘 처리할 수 있습니다. 버거 판매자 상담사는 Google의 구매 상담사 요청을 직접 수락하지만 피자 상담사는 Google의 요청을 진행하기 전에 Google의 확인을 받아야 하며 Google에서 확인한 후에는 상담사가 피자 상담사에게 확인을 제공할 수 있습니다.
이제 purchasing-agent 서비스 로그에서 교환된 데이터를 확인해 보겠습니다. 먼저 Cloud Run 콘솔로 이동하여 콘솔 상단의 검색창에 'Cloud Run'을 입력하고 Cloud Run 아이콘을 우클릭하여 새 브라우저 탭에서 열어 봅니다.
이제 아래와 같이 이전에 배포된 서비스가 표시됩니다. purchasing-concierge를 클릭합니다.
서비스 세부정보 페이지로 이동한 다음 로그 탭을 클릭합니다.
이제 배포된 purchasing-concierge 서비스의 로그가 표시됩니다. 아래로 스크롤하여 상호작용에 관한 최근 로그를 찾으세요.
A2A 클라이언트와 서버 간의 요청과 응답이 JSON-RPC 형식이며 A2A 표준을 준수하는 것을 확인할 수 있습니다.
이제 A2A의 기본 개념을 마쳤으며 클라이언트 및 서버 아키텍처로 구현된 방식을 살펴보았습니다.
8. 삭제
이 Codelab에서 사용한 리소스의 비용이 Google Cloud 계정에 청구되지 않도록 하려면 다음 단계를 따르세요.