Google의 Agent Development Kit과 Agent2Agent 사용하기



목차

Agent2Agent 프로토콜이란 무엇인가?

워크플로 생산성 향상 에이전트 간의 작업 위임, 컨텍스트 공유, 정보 교환 방식을 표준화하여 보다 매끄러운 자동화를 가능하게 합니다.
통합 오버헤드 감소 공통된 통신 백본과 인터페이스 정의를 제공함으로써, 서로 다른 에이전트 시스템을 연결하는 데 필요한 엔지니어링 노력과 비용을 크게 낮춥니다.
복합 혁신 가속화 이러한 표준화된 인터페이스를 통해 다양한 에이전트의 특화된 기능을 구성하거나 연쇄하여 결합하는 새로운 애플리케이션을 개발자가 기술적으로 더 쉽게 구축할 수 있도록 함으로써.

Agent2Agent 프로토콜의 핵심 설계 원칙

에이전트형 능력 수용 이는 단순한 요청-응답 상호작용만을 다루는 것이 아닙니다. 이 프로토콜은 근본적으로 AI 에이전트가 갖춘 능력을 전제로 설계되었으며, 에이전시 – 목표 달성을 위해 추론하고, 계획을 세우며, 자율적으로 작동하고, 의사결정을 내리는 능력. A2A는 이러한 정교한 능력을 지원하여, 에이전트들이 단순한 데이터 교환을 넘어 복잡하고 다단계의 협업에 참여할 수 있도록 하는 것을 목표로 합니다.
기존 표준 위에 구축하기 Agent2Agent은 가능한 모든 곳에서 HTTP, RESTful 원칙, OAuth와 같은 보안 패턴 등 이미 확립되어 널리 채택된 웹·인터넷 표준을 활용합니다. 인용되는 개별 표준은 시간이 지나며 발전할 수 있지만, 핵심 원칙은 검증된 기존 기술 위에 구축하는 것입니다. 이러한 접근은 개발 속도를 높이고, 기존 인프라와의 상호 운용성을 개선하며, 해당 표준에 익숙한 개발자가 Agent2Agent 프로토콜을 더 쉽게 도입하고 구현할 수 있게 합니다.
기본값으로 보안을 보장하기 가지고 있을 때 자율 에이전트 사용자나 시스템을 대신해 행동할 가능성이 있는 만큼, 보안을 사후에 덧붙이는 요소로 취급할 수는 없습니다. Agent2Agent는 보안을 핵심 설계 단계에서부터 통합합니다. 이는 처음부터 인증, 인가, 안전한 데이터 교환을 위한 메커니즘을 정의하여, 에이전트 간 상호작용이 신뢰할 수 있고 오남용으로부터 보호되도록 보장한다는 뜻입니다.
장기 실행 작업 지원하기: 현실 세계의 업무는 즉시 끝나지 않는 경우가 많습니다. 에이전트는 전체 여행 일정을 계획하거나, 상세 보고서를 생성하거나, 대규모 데이터셋을 처리하는 것처럼 상당한 시간이 걸리는 복잡한 작업을 맡을 수 있습니다. Agent2Agent 프로토콜은 이러한 비동기적이고 장기 실행 작업을 처리하도록 명시적으로 설계되었으며, 수분, 수시간, 혹은 그보다 더 오래 걸릴 수 있는 작업에 필요한 통신과 상태 추적을 관리합니다.
모달리티에 구애받지 않기: 협업은 다음에 의해 제한되어서는 안 됩니다 유형 처리되는 정보의 유형과 관계없이, 에이전트가 텍스트, 이미지, 구조화된 데이터, 음성 명령 등 다양한 형태의 정보(모달리티)를 교환해야 할 때에도 프로토콜은 유연하게 동작하도록 설계되었습니다. Agent2Agent는 서로 다른 데이터 유형이 관여된 상호작용을 원활히 지원하도록 고안되어, 단일한 통신 방식에 묶이지 않고 폭넓은 애플리케이션과 에이전트 역량을 뒷받침할 수 있습니다.

Agent2Agent 프로토콜의 핵심 구성 요소



에이전트 카드: 본질적으로 에이전트의 디지털 신원과 역량 선언서에 해당합니다. 표준화된 명함 또는 AI 에이전트를 위한 매니페스트 파일을 떠올리면 됩니다. 에이전트 카드는 다음과 같은 핵심 메타데이터를 포함할 가능성이 큽니다:
식별: 에이전트가 누구인지.
기능: 에이전트가 수행할 수 있는 작업이나 제공하는 서비스.
엔드포인트: 다른 에이전트가 어떻게 통신할 수 있는지(예: API 주소).
요구 사항: 상호작용에 필요한 모든 정보로, 요구하는 보안 프로토콜이나 데이터 형식 등이 포함될 수 있습니다. 에이전트 카드의 기본 기능은 검색입니다. 잠재적 클라이언트(다른 에이전트나 시스템)가 에이전트를 찾고, 제공하는 기본 기능과 연결 방법을 이해할 수 있도록 합니다.


A2A 서버: 이 구성 요소는 일반적으로 서비스나 기능을 제공하는 에이전트(상호작용에서 흔히 ‘원격 에이전트’라고 함)와 함께 실행됩니다. 주요 책임은 다음과 같습니다:
수신: Agent2Agent 프로토콜 명세에 따라 형식화된 수신 요청을 수신하기 위해 리슨하는 엔드포인트를 노출합니다.
처리 중 유효한 Agent2Agent 요청을 수신해 해석하고, 실제 기반 AI 에이전트 로직과 연동하여 요청된 작업을 실행합니다.
관리: 이 구성 요소는 작업의 라이프사이클을 관리하며, 설계 원칙에서 언급한 장기 실행 작업의 경우 특히 중요합니다.
응답: 표준화된 Agent2Agent 형식을 사용해 요청 클라이언트에게 응답, 상태 업데이트, 최종 결과를 다시 전송합니다. 본질적으로 A2A 서버는 A2A 프레임워크 내에서 에이전트의 기능을 호스트하고 실행을 처리하는 핸들러 역할을 합니다.
A2A 클라이언트: 이 구성 요소는 시작한다 상호작용을 시작하는 주체입니다. 다른 AI 에이전트, 애플리케이션, 또는 원격 에이전트의 기능을 활용해야 하는 사용자용 인터페이스일 수 있습니다. Agent2Agent 클라이언트의 역할은 다음을 포함합니다:
탐색(Agent Card 사용) 적절한 에이전트를 찾고, 그와 어떻게 상호작용할지 이해하기.
요청 구성: Agent2Agent 프로토콜 표준을 준수하도록 요청 메시지를 구성하고, 수행할 작업을 명시하며 필요한 입력 데이터를 제공하기.
통신: 대상 에이전트의 A2A 서버 엔드포인트로 요청을 전송하기.
응답 처리: A2A 서버로부터 응답(또는 업데이트)을 수신하고 처리하기. A2A 클라이언트는 agent2agent 상호작용에서 서비스를 시작하고 소비하는 주체입니다.

통신과 작업 관리 촉진

작업 시작: 일반적으로 A2A 클라이언트에서 시작됩니다. 예를 들어, 한 에이전트가(클라이언트 역할) 다른 에이전트(원격 에이전트)의 특화된 능력이 필요할 수 있습니다. 예를 들면 문서 요약이나 이미지 분석 같은 작업입니다. 클라이언트는 에이전트 카드 등을 통해 원격 에이전트와 그 능력을 발견한 뒤, 작업 요청을 구성합니다. 이는 가벼운 메시지가 아니라 A2A 프로토콜의 명세를 준수해 정교하게 포맷된 요청입니다. 표준화된 이 요청은 원하는 동작을 명확히 기술하고, 요약할 문서와 같은 필요한 입력 데이터도 함께 포함합니다.
전송 및 처리: 그다음 클라이언트는 프로토콜을 준수한 이 요청을 원격 에이전트의 A2A 서버에서 지정한 엔드포인트로 전송합니다. 서버가 요청을 수신하면 가장 먼저 하는 일은 보통 유효성 검증으로, Agent2Agent 규칙을 따르는지 확인하는 것입니다. 검증이 끝나면 서버는 요청을 해석하고, 관련된 AI 에이전트의 내부 로직을 호출하면서 작업 상세와 입력 데이터를 함께 전달합니다. 이로써 실제 작업이 시작됩니다.
상호작용 관리(특히 장기 작업의 경우): 바로 여기서 장기 실행 작업을 처리하기 위한 설계 원칙이 중요해집니다. 요청된 작업에 몇 초 이상이 소요될 경우, A2A 서버는 묵묵히 대기만 하지 않을 수 있습니다. 프로토콜은 서버가 작업 수신을 즉시 확인(ack)한 뒤, 필요에 따라 A2A 클라이언트에게 주기적인 상태 업데이트(“처리 시작”, “50% 완료” 등)를 제공할 수 있는 메커니즘을 허용합니다. 이를 통해 클라이언트가 암중모색하며 기다리는 상황을 방지하고, 진행 상황을 더욱 역동적이고 실시간에 가깝게 모니터링할 수 있습니다.
완료 및 응답 전송 원격 에이전트가 할당된 작업을 완료하면, A2A 서버는 결과(또는 처리 중 발생한 오류)를 Agent2Agent 프로토콜에서 정의한 표준화된 응답 형식으로 포장하여 원래 요청을 보낸 A2A 클라이언트로 반환합니다.
결과 수신 및 활용: A2A 클라이언트는 이 구조화된 응답을 수신합니다. 응답이 Agent2Agent 표준에 따라 포맷되어 있기 때문에, 클라이언트는 이를 정확히 파싱하고 요약본, 분석 결과 등 관련 정보를 추출한 뒤, 사용자에게 표시하거나 더 큰 워크플로의 다음 단계로 전달하는 등 자체 용도에 맞게 활용할 수 있습니다.

실제 활용 사례와 Agent2Agent의 미래

표준화된 요청을 해당 기준과 함께 전문 항공편 검색 에이전트(자체 A2A 서버에서 실행 중)에 전송합니다.
동시에 또는 순차적으로 다른 A2A 서버인 호텔 예약 에이전트에 가용 여부와 가격을 조회하고, 필요하다면 첫 번째 에이전트의 잠재적 항공편 시간 정보를 입력으로 활용합니다.
해당 날짜에 맞춰 관심사에 부합하는 콘서트, 투어, 또는 레스토랑 추천을 찾기 위해 지역 이벤트·추천 에이전트(또 다른 A2A 서버)에 문의합니다.

멀티 에이전트 시스템 핵심 내용 Agent Development Kit은 여러 전문화된 에이전트로 구성된 애플리케이션을 구축하고, 이들 간의 복잡한 조정과 위임을 원활히 수행할 수 있도록 명시적으로 설계되었습니다.
모델과 도구의 유연성: 개발자가 선호하는 것을 선택할 수 있도록 합니다 대규모 언어 모델 (제미니, 모델을 통해 Vertex AI Model Garden 또는 LiteLLM 통합을 통한 다양한 모델을 활용할 수 있으며, 에이전트에 검색과 같은 사전 구축 기능부터 서드파티 라이브러리에 이르는 폭넓은 도구를 갖추게 할 수 있습니다 (LangChain, LlamaIndex) 그리고 다른 에이전트를 도구로 활용하는 것까지 가능합니다.
풍부한 상호작용: Agent Development Kit은 더욱 자연스럽고 사람과 유사한 대화를 위해, 고유한 양방향 오디오 및 비디오 스트리밍 기능을 포함한 고급 상호작용 모델을 지원합니다.
전체 수명주기 지원: 구축을 넘어, ADK는 유연한 오케스트레이션 옵션, 통합된 로컬 개발 환경(CLI와 Web UI), 에이전트 성능을 평가하기 위한 내장 평가 프레임워크, 그리고 배포를 위한 간소화된 경로(컨테이너화 또는 Google Cloud의 Vertex AI에 최적화)를 제공합니다.

자습서: 멀티 에이전트 시스템 구축

도구 키트 설정하기

google-adk: 쇼의 주인공은 바로 Google의 Agent Development Kit입니다. 이 툴킷은 에이전트와 그 도구를 정의하고, 상호작용을 관리하며(앞서 언급한 에이전트 간 협업을 가능하게 하고), 전체 시스템을 오케스트레이션하는 핵심 프레임워크를 제공합니다.
litellm: ADK의 유연성을 보여 주기 위해 우리는 litellm을 사용할 것입니다. 이 라이브러리는 브리지 역할을 하여, Google의 모델뿐 아니라 OpenAI, Anthropic 등 다양한 대규모 언어 모델을 우리 ADK 에이전트가 폭넓게 사용할 수 있도록 해 줍니다.
weave 에이전트 시스템이 복잡해질수록 내부에서 무슨 일이 일어나는지 파악하는 것이 중요합니다. 우리는 이를 통합할 것입니다. W&B Weave 가시성을 추가하여 에이전트의 동작과 의사결정 과정을 추적하고, 시각화하며, 디버그할 수 있도록 돕습니다.
pyowm: 우리 에이전트에 실용적인 능력을 하나 추가해 봅시다! 이 라이브러리는 OpenWeatherMap API에 손쉽게 연결하고 실시간 날씨 데이터를 가져오는 편리한 방법을 제공합니다.
newsapi-python: 또 다른 차원을 더하기 위해, 이 라이브러리를 사용해 NewsAPI를 통해 특정 주제의 최신 뉴스 헤드라인을 가져오겠습니다.

우리가 만들 것

전문화된 인사 및 작별 하위 에이전트를 통한 대화상 예의 관리.
pyowm 라이브러리와 외부 API를 사용해 실시간 날씨 상태를 가져오고 보고하기.
사용자가 입력한 주제에 대해 newsapi-python을 사용해 최신 뉴스 헤드라인 가져오기.
보안 점검 구현하기
특정 키워드를 포함한 요청을 차단하는 입력 가드레일.
특정 도시(예: “Paris”)에 대해 날씨 도구 사용을 금지하는 도구 사용 가드레일.

import os
import warnings
import logging
import asyncio
from typing import Optional, Dict, Any
import weave

# External API Clients
import pyowm
from newsapi import NewsApiClient

# ADK Components
from google.adk.agents import Agent
from google.adk.models.lite_llm import LiteLlm  # For potential multi-model use
from google.adk.sessions import InMemorySessionService
from google.adk.runners import Runner
from google.adk.tools.tool_context import ToolContext
from google.adk.tools.base_tool import BaseTool
from google.adk.agents.callback_context import CallbackContext
from google.adk.models.llm_request import LlmRequest
from google.adk.models.llm_response import LlmResponse

# Google Generative AI types
from google.genai import types as google_types
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

# --- Configuration ---
warnings.filterwarnings("ignore")
logging.basicConfig(level=logging.INFO)


weave.init("Google-Agent2Agent")

# --- Define Model Constants ---
MODEL_GEMINI_2_0_FLASH = "gemini-2.0-flash"

도구 정의하기

ADK에서 Tools는 단순한 텍스트 생성 이상으로 에이전트에 구체적인 능력을 부여하는 구성 요소입니다. 보통 API 호출, 데이터베이스 조회, 계산 수행처럼 특정 작업을 수행하는 일반적인 Python 함수로 구현됩니다.

# Greeting and Farewell Tools
@weave.op()  # Add weave decorator
def say_hello(name: str = "there") -> str:
    """Provides a simple greeting."""
    logging.info(f"Tool 'say_hello' executed with name: {name}")
    return f"Hello, {name}!"


@weave.op()  # Add weave decorator
def say_goodbye() -> str:
    """Provides a simple farewell message."""
    logging.info("Tool 'say_goodbye' executed.")
    return "Goodbye! Have a great day."

현실 세계 기능 추가하기: 날씨 및 뉴스 도구

실시간 날씨: 우리는 …을(를) 정의할 것입니다 get_real_weather 함수입니다. 이 도구는 pyowm 라이브러리를 사용해 OpenWeatherMap API에 연결하고, 사용자가 지정한 도시의 현재 날씨 정보를 가져옵니다. 특히 이 도구는 에이전트의 메모리(세션 상태를 통해)와도 상호작용합니다. ToolContext) 사용자 선호도(예: 선호하는 온도 단위인 섭씨 또는 화씨)를 활용하거나 기억하도록 합니다.
최신 뉴스 헤드라인: 다음으로, 우리는 …을(를) 추가하겠습니다 get_latest_news function이 도구는 …을(를) 활용하여 newsapi-python 라이브러리를 사용해 사용자가 제공한 주제를 바탕으로 NewsAPI에서 최신 헤드라인을 조회합니다.

@weave.op()  # Add weave decorator
def get_real_weather(city: str, tool_context: ToolContext) -> dict:
    """Retrieves the current real weather report for a specified city using OpenWeatherMap."""
    logging.info(f"Attempting to get real weather for city: '{city}'")

    owm_api_key = os.environ.get("OWM_API_KEY")

    try:
        # Initialize PyOWM
        owm = pyowm.OWM(owm_api_key)
        mgr = owm.weather_manager()

        # Get Weather Observation
        logging.info(f"Querying OpenWeatherMap API for city: {city}")
        observation = mgr.weather_at_place(city)
        w = observation.weather
        if w is None:
            # This case might be rare if NotFoundError is caught, but good to check
            logging.error(f"OWM observation for '{city}' did not contain weather data.")
            raise ValueError(f"Weather data not available in OWM response for {city}")
        # Extract Temperature (Kelvin)
        temp_data = w.temperature("kelvin")
        if not temp_data or "temp" not in temp_data:
            logging.error(f"Could not find 'temp' in OWM temperature data: {temp_data}")
            raise ValueError("Temperature data missing 'temp' key in OWM response.")
        temp_k = temp_data["temp"]  # Direct access after check
        logging.debug(f"Temperature (Kelvin): {temp_k}")

        # Read unit preference from state
        preferred_unit = tool_context.state.get(
            "user_preference_temperature_unit", "Celsius"
        )
        # Convert temperature
        if preferred_unit == "Fahrenheit":
            temp_c = temp_k - 273.15
            temp_value = (temp_c * 9 / 5) + 32
            temp_unit = "°F"
        else:  # Default to Celsius
            temp_value = temp_k - 273.15
            temp_unit = "°C"
        logging.debug(f"Converted temperature: {temp_value:.1f}{temp_unit}")

        # Extract other details
        status = w.detailed_status
        humidity = w.humidity
        wind_speed = w.wind().get("speed", "N/A")
        # Format Report
        report = (
            f"The current weather in {city.capitalize()} is '{status}' "
            f"with a temperature of {temp_value:.1f}{temp_unit}. "
            f"Humidity is {humidity}%, and wind speed is {wind_speed} m/s."
        )

        result = {"status": "success", "report": report}
        # Update last checked city in state
        tool_context.state["last_city_checked_stateful"] = city.capitalize()
        logging.info(f"Updated state 'last_city_checked_stateful': {city.capitalize()}")
        return result

@weave.op()
def get_latest_news(topic: str) -> dict:
    """Fetches the latest 5-10 news headlines for a given topic using NewsAPI."""
    logging.info(f"Tool 'get_latest_news' called for topic: {topic}")

    news_api_key = os.environ.get("NEWS_API_KEY")
    try:
        newsapi = NewsApiClient(api_key=news_api_key)

        # Fetch news articles related to the topic, sorted by published date (latest first)
        # Using 'everything' endpoint for broader search on a topic
        articles_data = newsapi.get_everything(
            q=topic, language="en", sort_by="publishedAt", page_size=10
        )  # Get up to 10 articles

        if articles_data["status"] != "ok":
            raise Exception(
                f"NewsAPI returned status: {articles_data.get('code', 'unknown')} - {articles_data.get('message', 'No message')}"
            )

        articles = articles_data["articles"]

        if not articles:
            summary = f"I couldn't find any recent news articles about '{topic}'."
            logging.warning(f"No news articles found for topic: {topic}")
            return {
                "status": "success",
                "news_summary": summary,
            }  # Success, but no articles

        # Format the top 5-10 headlines
        num_headlines = min(len(articles), 10)  # Show up to 10
        headlines_list = []
        for i, article in enumerate(articles[:num_headlines]):
            title = article.get("title", "No Title")
            source = article.get("source", {}).get("name", "Unknown Source")
            headlines_list.append(f"{i+1}. {title} ({source})")

        summary = (
            f"Here are the latest {num_headlines} headlines I found about '{topic}':\n"
            + "\n".join(headlines_list)
        )
        result = {"status": "success", "news_summary": summary}
        logging.info(f"Fetched {num_headlines} news headlines for topic: {topic}")
        return result

안전성 구현: 입력 가드레일 도입 before_model_callback

검사: LLM으로 전달될 전체 요청 패키지를 면밀히 검토하세요.
수정: 필요하다면 요청의 일부를 신중하게 수정하세요(예: 컨텍스트 추가, 콘텐츠 필터링).
블록: 사전에 정의된 규칙을 위반하는 요청은 LLM에 전달되지 않도록 완전히 차단하고, 대신 사용자에게 특정 응답을 직접 반환합니다.

민감한 데이터 필터링 (개인 식별 정보) 또는 사용자 프롬프트의 부적절한 언어
키워드 기반 차단을 구현해 주제와 무관하거나 정책을 위반하는 요청을 막기.
LLM이 프롬프트를 보기 직전에 오늘 날짜나 세션 상태의 정보처럼 시의적절한 정보를 프롬프트 컨텍스트에 동적으로 추가하기.

요청이 적합하다면, 함수는 다음을 반환합니다. None. Agent Development Kit은 이를 감지하고 평소와 동일하게 LLM을 호출합니다.
요청을 차단해야 하는 경우, 함수가 이를 구성하여 반환합니다. LlmResponse 객체입니다. 이 객체에는 대신 사용자에게 전달할 메시지가 포함되어 있습니다. ADK는 이 응답을 가로채어 즉시 반환하며, 해당 턴의 LLM 호출을 사실상 취소합니다.

Python 함수를 정의해 봅시다. 이름은 이렇게 하겠습니다: block_keyword_guardrail, 다음과 같이 설계된 before_model_callback. 이 함수의 단순한 역할은 사용자의 최신 입력에 대소문자를 구분하지 않고 “BLOCK”이라는 단어가 포함되어 있는지 확인하는 것입니다.
메인 루트 에이전트의 정의를 수정하여 Agent Development Kit에 이것을 실행하도록 지시하세요. block_keyword_guardrail 주요 LLM을 호출하기 전에 실행되는 함수.
ADK 설정 Runner 이 새로 구성한 에이전트를 사용하려면
일반 요청 몇 건과 “BLOCK” 키워드를 포함한 요청 한 건을 보내 테스트 상호작용을 수행하고, 가드레일이 문제 있는 요청을 올바르게 가로채면서 나머지는 통과시키는지 확인하세요.

레이어 추가하기: 도구 사용 가드레일 설정 before_tool_callback

인자 검증 및 정제 LLM이 제공한 인자가 타당한지 확인하세요. 유형이 올바른가요? 허용 범위 안에 있나요? 도구로 전달하기 전에 정제가 필요한가요?
리소스·정책 집행 인자에 따라 특정 도구 실행을 차단합니다. 예를 들어, 비용 관리를 위해 특정 매개변수의 API 호출을 막거나, 상태에 저장된 사용자 권한을 기준으로 접근을 제한하거나, 곧 살펴볼 예시처럼 특정 제한된 엔터티(예: 특정 도시)에 대한 쿼리를 방지하는 것입니다.
동적 인자 수정: 다음의 정보를 활용할 수 있습니다 ToolContext (예: 세션 상태)와 같은 정보를 사용해, 도구가 실행되기 직전에 인자를 조정하거나 추가할 수 있습니다.

원한다면 허용하다 실행할 도구(당신이 내부에서 직접 조정한 수정된 인자를 포함해) args 딕셔너리)에서, 함수는 반환해야 합니다 None그러면 ADK가 원래의 도구 함수를 실행합니다.
원한다면 블록 도구 호출을 완전히 차단하고 다른 결과로 대체하려면, 함수는 다음을 반환해야 합니다 사전. ADK는 이 사전을 가로채어 다음과 같이 처리합니다 실제 결과 이번 턴의 도구 호출 결과로 간주되어 원래 도구 함수의 실행을 완전히 건너뜁니다. 반환하는 딕셔너리는 이상적으로 해당 도구의 출력이 기대하는 구조와 일치해야 합니다(예: 도구가 사용하는 형식과 동일한 형식으로 사용자 정의 오류 메시지를 제공).

Python 함수를 정의하세요 block_paris_tool_guardrail, 우리의 before_tool_callback. 호출된 도구가 우리의 날씨 도구인지 여부를 구체적으로 확인합니다 (get_real_weather) 그리고 यदि city LLM이 제공한 인자가 “Paris”인지(대소문자 무시) 확인합니다.
이 특정 상황을 감지하면 콜백이 사용자 정의 오류 딕셔너리를 반환하여 실제 OpenWeatherMap API 호출을 효과적으로 차단합니다.
루트 에이전트 정의를 마지막으로 한 번 더 업데이트하여 다음을 사용하도록 구성하세요 둘 다 그 before_model_callback (입력 키워드 차단용) 그리고 이 새로운 before_tool_callback (파리 날씨 확인을 차단하기 위한 용도)
ADK 설정 Runner 최종 이중 보호 에이전트 구성과 연관되어 있습니다.
도구의 가드레일이 제한된 호출은 차단하면서 다른 요청은 허용한다는 점을 보여주기 위해, 허용된 도시들의 날씨를 요청한 뒤 이어서 “파리”의 날씨를 구체적으로 요청하는 테스트 상호작용을 실행하세요.

# Model Guardrail
def block_keyword_guardrail(
    callback_context: CallbackContext, llm_request: LlmRequest
) -> Optional[LlmResponse]:
    """Inspects user input for 'BLOCK', blocks if found."""
    agent_name = callback_context.agent_name
    logging.debug(f"Callback 'block_keyword_guardrail' running for agent: {agent_name}")
    last_user_message_text = ""
    if llm_request.contents:
        for content in reversed(llm_request.contents):
            if content.role == "user" and content.parts:
                if isinstance(content.parts[0], google_types.Part) and hasattr(
                    content.parts[0], "text"
                ):
                    last_user_message_text = content.parts[0].text or ""
                    break
    keyword_to_block = "BLOCK"
    if keyword_to_block in last_user_message_text.upper():
        logging.warning(f"Keyword '{keyword_to_block}' found. Blocking LLM call.")
        callback_context.state["guardrail_block_keyword_triggered"] = True
        logging.info("Set state 'guardrail_block_keyword_triggered': True")
        return LlmResponse(
            content=google_types.Content(
                role="model",
                parts=[
                    google_types.Part(
                        text="I cannot process this request (blocked keyword)."
                    )
                ],
            )
        )
    else:
        logging.debug(f"Keyword not found. Allowing LLM call for {agent_name}.")
        return None


# Tool Guardrail
def block_paris_tool_guardrail(
    tool: BaseTool, args: Dict[str, Any], tool_context: ToolContext
) -> Optional[Dict]:
    """Blocks 'get_real_weather' tool execution for 'Paris'."""
    tool_name = tool.name
    agent_name = tool_context.agent_name
    logging.debug(
        f"Callback 'block_paris_tool_guardrail' running for tool '{tool_name}' in agent '{agent_name}'"
    )
    logging.debug(f"Inspecting tool args: {args}")

    # *** UPDATED target_tool_name ***
    target_tool_name = "get_real_weather"
    blocked_city = "paris"

    if tool_name == target_tool_name:
        city_argument = args.get("city", "")
        if city_argument and city_argument.lower() == blocked_city:
            logging.warning(
                f"Blocked city '{city_argument}' detected for tool '{tool_name}'. Blocking execution."
            )
            tool_context.state["guardrail_tool_block_triggered"] = True
            logging.info("Set state 'guardrail_tool_block_triggered': True")
            return {  # Return error dictionary, skipping the actual tool
                "status": "error",
                "error_message": f"Policy restriction: Weather checks for '{city_argument.capitalize()}' are disabled by a tool guardrail.",
            }
        else:
            logging.debug(f"City '{city_argument}' is allowed for tool '{tool_name}'.")
    else:
        logging.debug(f"Tool '{tool_name}' not targeted by Paris guardrail. Allowing.")

    logging.debug(f"Allowing tool '{tool_name}' to proceed.")
    return None  # Allow tool execution

에이전트 정의하기

name: 애플리케이션 내에서 이 에이전트를 식별하는 고유 문자열(예: "greeting_agent", "news_fetcher"). 이는 구성 관리, 로깅, 디버깅에 유용합니다.
model: 이 매개변수는 에이전트의 “두뇌”, 즉 추론과 응답을 담당하는 LLM을 지정합니다. 일반적으로 여기에는 모델 식별자 문자열을 제공합니다(예: 우리의 상수 MODEL_GEMINI_2_0_FLASH) 또는 사용합니다 LiteLlm 래퍼(예: OpenAI나 Anthropic과 같은 다른 제공업체의 모델을 활용하려는 경우)
description: 이는 에이전트의 주요 목적이나 기능을 간략하고 상위 수준에서 요약한 설명입니다. 에이전트 팀을 구성하면서 보게 되겠지만, 이는 description 결정적인 역할을 합니다 위임, 다른 에이전트가 이 에이전트의 전문 분야와 어떤 경우에 작업을 넘겨야 하는지 이해하는 데 사용되기 때문입니다.
instruction: 에이전트의 LLM을 위한 상세 운영 매뉴얼로 생각하세요. 여기에는 페르소나, 목표, 대화 스타일, 제약 조건에 대한 구체적인 지침과—무엇보다도—명확한 지시사항을 제공합니다. 언제 그리고 방법 이를 활용해야 합니다 Tools 구성에 나열된 도구입니다. 여기에서의 효과적인 프롬프트 작성은 도구를 신뢰성 있게 사용하고 전체적인 에이전트 동작을 안정화하는 데 매우 중요합니다.
tools: 이것은 실제 도구 함수들(예: …)을 담고 있는 단순한 Python 리스트입니다 say_hello, get_real_weather, get_latest_news) that 이 특정 에이전트 사용 권한이 있으며 이를 사용할 준비가 되어 있습니다

greeting_agent = Agent(
    model=MODEL_GEMINI_2_0_FLASH,
    name="greeting_agent",
    instruction="Greet the user friendly.",
    description="Handles simple greetings and hellos.",
    tools=[say_hello],
)
farewell_agent = Agent(
    model=MODEL_GEMINI_2_0_FLASH,
    name="farewell_agent",
    instruction="Provide a polite goodbye.",
    description="Handles simple farewells and goodbyes.",
    tools=[say_goodbye],
)


news_agent = Agent(
    model=MODEL_GEMINI_2_0_FLASH,  # Can use a different model if desired
    name="news_agent",
    instruction="You are a News Reporter agent. Your goal is to fetch and present the latest news headlines on a specific topic requested by the user. Use the 'get_latest_news' tool. Clearly state the topic and present the headlines returned by the tool. If the tool returns an error or no news, inform the user politely.",
    description="Fetches and presents the latest 5-10 news headlines for a given topic using the 'get_latest_news' tool.",
    tools=[get_latest_news],
)


root_agent = Agent(
    name="weather_news_assistant",
    model=MODEL_GEMINI_2_0_FLASH,  # Orchestration model
    description="Main assistant: Handles real weather requests, delegates news requests, greetings, and farewells. Includes safety guardrails.",
    instruction=(
                "You are the main Assistant coordinating a team. Your primary responsibilities are providing real-time weather and delegating other tasks.\n"
                "1.  **Weather:** If the user asks for weather in a specific city, use the 'get_real_weather' tool yourself. The tool respects temperature unit preferences stored in state.\n"
                "2.  **News:** If the user asks for news on a specific topic (e.g., 'latest news on AI', 'updates on electric vehicles'), delegate the request to the 'news_agent'.\n"
                "3.  **Greetings:** If the user offers a simple greeting ('Hi', 'Hello'), delegate to the 'greeting_agent'.\n"
                "4.  **Farewells:** If the user says goodbye ('Bye', 'Thanks bye'), delegate to the 'farewell_agent'.\n"
                "Analyze the user's query and delegate or handle it appropriately. If unsure, ask for clarification. Only use tools or delegate as described."
            ),
            tools=[get_real_weather],  # Root agent handles weather directly
            sub_agents=[greeting_agent, farewell_agent, news_agent],
            output_key="last_assistant_response",
            before_model_callback=block_keyword_guardrail,
            before_tool_callback=block_paris_tool_guardrail,
        )

adk web

첫 번째 메시지에서는 에이전트가 인사 에이전트에 작업을 위임하고, 두 번째 메시지에서는 실시간 날씨 정보를 가져오는 방식을 확인해 보세요.



또한 에이전트가 실시간 뉴스를 가져오는 방식도 확인해 보세요.

에이전트 모니터링을 위한 Weave

Weave 대시보드에서 서로 다른 에이전트 간의 통신 내역과 사용된 정확한 프롬프트를 확인할 수 있습니다.





세계를 잇다: Agent2Agent로 ADK 에이전트를 연결하기

서로 다른 프레임워크(예: 왼쪽의 ADK, 오른쪽의 다른 프레임워크)로 구축된 에이전트 시스템은 A2A 프로토콜을 사용해 에이전트 간 협업을 위해 상호 통신할 수 있으며, 동시에 자신들의 특정 도구와 API와 상호작용하기 위해 MCP(Model Context Protocol)와 같은 다른 프로토콜을 사용할 수도 있습니다.

A2A와 MCP: 간단한 구분

MCP 일반적으로 에이전트를 연결하는 데 중점을 둡니다 도구, API, 리소스 구조화된 방식으로. ADK 자체는 MCP를 지원하므로, 이를 통해 구축된 에이전트는 다양한 도구를 활용할 수 있습니다.
A2A 특히 에이전트 간 협업을 목표로 합니다. 서로 다른 에이전트 시스템이 메모리, 리소스, 특정 도구 구현 같은 내부 세부 정보를 공유하지 않고도, 동적으로—종종 멀티모달 방식으로—소통할 수 있도록 합니다.

불투명 실행: 우리 Weather Bot이 A2A를 통해 서비스를 공개한다면, 호출하는 에이전트는 다음만 알면 됩니다 무엇 요청할 스킬(예: get_weather) 및 Agent Card를 통해 예상 입력/출력 형식. 이는 않는다 ADK, PyOWM을 썼는지, 내부 하위 에이전트를 갖고 있는지는 알 필요가 없습니다. A2A가 내부 복잡성을 숨겨 줍니다.
비동기 우선: 우리 get_latest_news 도구가 시간이 걸릴 수 있습니다. A2A의 비동기 설계(폴링, SSE, 푸시 알림)는 호출자를 블로킹하지 않고도 우리 Weather Bot이 더 오래 걸리는 작업을 처리하고 상태 업데이트를 제공할 수 있음을 의미합니다 (TaskStatusUpdateEvent) 필요한 경우 스트리밍으로.
모달리티 무관: 우리가 텍스트를 사용했지만, A2A의 Part 객체는 다른 데이터 유형에 대한 지원을 정의할 수 있도록 합니다 (application/json, image/png등)을 Agent Card를 통해 정의하여 더 풍부한 상호작용을 가능하게 합니다.
엔터프라이즈 준비 완료 & 간결함: HTTP/JSON-RPC와 표준 인증 방식(헤더에 토큰을 전달하고 Agent Card에서 선언하는 방식 등)을 사용하면, 익숙한 기술을 활용하면서도 A2A를 비즈니스 용도로 적합하게 만들 수 있습니다.

기본 정보: nameADK Weather & News Bot description, url (A2A 엔드포인트)
인증: 필요한 스킴 지정(예: schemes: ["OAuth2"]).
기능: 예를 들어 streaming: true 뉴스 업데이트를 위한 SSE를 지원하는지 여부.
기능: 사용할 수 있는 항목을 설명하는 배열 하다:
기능 1: id: "get_weather", name: "Get Current Weather", description: "Retrieves real-time weather for a city.", tags: ["weather", "real-time"], examples: ["weather in london"], inputModes: ["text/plain"], outputModes: ["text/plain"]
기능 2: id: "get_news", name: "Get Latest News", description: "Fetches recent news headlines for a topic.", tags: ["news", "headlines"], examples: ["news about AI"], inputModes: ["text/plain"], outputModes: ["text/plain"]

요청 (tasks/send) 클라이언트는 HTTP POST를 통해 A2A 서버의 URL을 대상으로 하는 JSON-RPC 요청을 구성합니다. The method 는 "tasks/send". 해당 params 고유한 작업을 포함하십시오 id 그리고 a message 객체 (role: "user") 에는 포함되어 있으며 parts (예: a TextPart 쿼리와 함께)."
처리 중 우리 A2A 서버 래퍼는 이를 수신하고, 요청과 인증을 검증한 뒤, 세부 정보를 추출합니다.
내부 ADK 실행 래퍼가 내부 ADK Weather Bot 로직을 트리거하고(쿼리를 전달), ADK 에이전트는 상태, 가드레일, 도구를 포함한 플로우를 실행합니다.
응답 (Task 객체): ADK 에이전트가 결과를 반환합니다. 우리 A2A 래퍼가 이를 A2A 형식으로 포맷합니다 Task 객체( 원본 id, sessionId, status: "completed", 그리고 an artifacts 날씨 보고서를 포함하는 배열에서 Artifact 와 함께 TextPart). 이는 Task 객체가 JSON-RPC로 다시 전송됩니다 result.

우리 봇을 서버로 사용하기 A2A 서버 레이어로 감싸면, 우리의 ADK Weather Bot은 다음으로부터의 요청을 처리할 수 있습니다 임의의 LangGraph, Crew.AI, Microsoft Autogen 등으로 구축된 A2A 준수 클라이언트
우리 봇을 클라이언트로 사용하기 우리 ADK 에이전트는 호출 다른 에이전트와는 A2A를 통해 상호작용합니다. 항공권 가격을 요청받으면 A2A 클라이언트로서 동작하여, 항공 에이전트의 에이전트 카드(Agent Card)를 검색하고 A2A를 전송할 수 있습니다. tasks/send 요청을 보내고 A2A를 처리합니다 Task 응답 이는 우리 ADK 에이전트와 서로 다른 기술로 구축된 에이전트 간의 직접 협업을 보여줍니다.

결론