ADK (Agent Development Kit) - MCP를 활용한 멀티 에이전트 시스템 구축

Agent Development Kit

https://google.github.io/adk-docs

What is Agent Development Kit?

Agent Development Kit (ADK) is a flexible and modular framework for developing and deploying AI agents. While optimized for Gemini and the Google ecosystem, ADK is model-agnostic, deployment-agnostic, and is built for compatibility with other frameworks. ADK was designed to make agent development feel more like software development, to make it easier for developers to create, deploy, and orchestrate agentic architectures that range from simple tasks to complex workflows.

ADK란 무엇인가요?

Agent Development Kit(ADK)는 AI 에이전트를 개발하고 배포하기 위한 유연하고 모듈식 프레임워크입니다. Gemini와 Google 생태계에 최적화되어 있지만, ADK는 모델에 구애받지 않고, 배포 방식에 구애받지 않으며, 다른 프레임워크와의 호환성을 위해 설계되었습니다. ADK는 에이전트 개발이 소프트웨어 개발처럼 느껴지도록 설계되었으며, 개발자가 간단한 작업부터 복잡한 워크플로우까지 다양한 에이전트 아키텍처를 쉽게 생성, 배포 및 조율할 수 있도록 만들어졌습니다.

Let’s try using ADK examples with Python.

파이썬을 이용해서 ADK 예제를 사용해보겠습니다.

Requirements:

Visual Studio Code
Python 3.10 or higher (Note: Official documentation states 3.9+, but 3.10+ is recommended due to potential errors)
pip
Google Gemini API key

준비물

Visual Studio Code

Python 3.10 이상 버전(※ 공식 문서상은 3.9 이상이지만 오류 때문에 3.10 이상 진행)

pip

google gemini api key

QuickStart

https://google.github.io/adk-docs/get-started/quickstart/

1. Create a Virtual Environment

1. 가상환경 만들기

python -m venv venv
source venv/bin/activate  # (Windows venvScriptsactivate)

2. Install Required Packages

2. 필요 패키지 설치
pip install google-adk mcp

3. Create Project Folder Structure

3. 프로젝트 폴더 구조 만들기
multi_tool_agent/
    ├── __init__.py
    ├── agent.py
    ├── .env

4. init.py

Set up the module to import agent.py during initialization.

모듈을 초기화할 때 agent.py를 불러오도록 설정합니다.
from . import agent

5. agent.py

# Function to get weather information
# Function to get current time
# Create agent instance

# 날씨 정보를 가져오는 함수
# 현재 시간을 가져오는 함수
# 에이전트 인스턴스 생성

import datetime
from zoneinfo import ZoneInfo
from google.adk.agents import Agent

# Function to get weather information
def get_weather(city: str) -> dict:
    if city.lower() == "new york":
        return {
            "status": "success",
            "report": "The weather in New York is sunny with a temperature of 25°C (77°F)."
        }
    else:
        return {
            "status": "error",
            "error_message": f"Weather information for '{city}' is not available."
        }

# Function to get current time
def get_current_time(city: str) -> dict:
    if city.lower() == "new york":
        tz_identifier = "America/New_York"
    else:
        return {
            "status": "error",
            "error_message": f"Sorry, I don't have timezone information for {city}."
        }

    tz = ZoneInfo(tz_identifier)
    now = datetime.datetime.now(tz)
    report = f'The current time in {city} is {now.strftime("%Y-%m-%d %H:%M:%S %Z%z")}'
    return {"status": "success", "report": report}

# Create agent instance
root_agent = Agent(
    name="weather_time_agent",
    model="gemini-2.0-flash",
    description="Agent to answer questions about the time and weather in a city.",
    instruction="You are a helpful agent who can answer user questions about the time and weather in a city.",
    tools=[get_weather, get_current_time],
)

6. .env

Configure ADK to use the Gemini API.

ADK가 Gemini API를 쓸 수 있도록 설정합니다.
GOOGLE_GENAI_USE_VERTEXAI=FALSE
GOOGLE_API_KEY=PASTE_YOUR_ACTUAL_API_KEY_HERE

7. Run

Now run it using adk web, and the agent will be able to provide appropriate answers to questions about the weather or time in New York.

7. 실행

이제 adk web을 이용해서 실행하면, 뉴욕의 날씨나 시간을 물어보는 질문에 대해서 에이전트가 적절한 답변을 줄 수 있습니다.
adk web

MCP(Model Context Protocol) Tools

https://google.github.io/adk-docs/tools/mcp-tools/

1. Create mcp_agent Folder

1. mcp_agent 폴더 만들기

adk_mcp_agent/
    ├── __init__.py
    ├── agent.py
    ├── .env
    ├── adk_mcp_server.py

2. adk_mcp_server.py

Expose a web page loading tool called load_web_page through the MCP server.

2. adk_mcp_server.py

load_web_page라는 웹 페이지 로드 툴을 MCP 서버로 노출합니다.

# Initialize ADK web loading tool
# Create MCP server instance
# Return list of tools to register with MCP server
# Call tool from MCP server
# Run MCP server

# ADK 웹 로딩 도구 초기화
# MCP 서버 인스턴스 생성
# MCP 서버에 등록할 도구 목록 반환
# MCP 서버에서 도구 호출
# MCP 서버 실행

import asyncio
import json
from dotenv import load_dotenv

from mcp import types as mcp_types
from mcp.server.lowlevel import Server, NotificationOptions
from mcp.server.models import InitializationOptions
import mcp.server.stdio

from google.adk.tools.function_tool import FunctionTool
from google.adk.tools.load_web_page import load_web_page
from google.adk.tools.mcp_tool.conversion_utils import adk_to_mcp_tool_type

load_dotenv()

# Initialize ADK web loading tool
print("Initializing ADK load_web_page tool...")
adk_web_tool = FunctionTool(load_web_page)

# Create MCP server instance
print("Creating MCP Server instance...")
app = Server("adk-web-tool-mcp-server")

# Return list of tools to register with MCP server
@app.list_tools()
def list_tools() -> list[mcp_types.Tool]:
    print("MCP Server: Received list_tools request.")
    mcp_tool_schema = adk_to_mcp_tool_type(adk_web_tool)
    print(f"MCP Server: Advertising tool: {mcp_tool_schema.name}")
    return [mcp_tool_schema]

# Call tool from MCP server
@app.call_tool()
async def call_tool(name: str, arguments: dict):
    print(f"MCP Server: Received call_tool request for '{name}' with args: {arguments}")

    if name == adk_web_tool.name:
        try:
            adk_response = await adk_web_tool.run_async(args=arguments, tool_context=None)
            response_text = json.dumps(adk_response, indent=2)
            return [mcp_types.TextContent(type="text", text=response_text)]
        except Exception as e:
            error_text = json.dumps({"error": f"Failed to execute tool '{name}': {str(e)}"})
            return [mcp_types.TextContent(type="text", text=error_text)]
    else:
        error_text = json.dumps({"error": f"Tool '{name}' not implemented."})
        return [mcp_types.TextContent(type="text", text=error_text)]

# Run MCP server
async def run_server():
    with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        print("MCP Server starting handshake...")
        await app.run(
            read_stream,
            write_stream,
            InitializationOptions(
                server_name=app.name,
                server_version="0.1.0",
                capabilities=app.get_capabilities(
                    notification_options=NotificationOptions(),
                ),
            ),
        )

if __name__ == "__main__":
    try:
        asyncio.run(run_server())
    except KeyboardInterrupt:
        print("\nMCP Server stopped by user.")
    except Exception as e:
        print(f"MCP Server encountered an error: {e}")
    finally:
        print("MCP Server process exiting.")

3. agent.py

Run the actual MCP server and connect the agent to use the tools from that MCP server.

3. agent.py

실제 MCP 서버를 실행하고, 에이전트가 해당 MCP 서버의 도구를 사용할 수 있도록 연결합니다.

# Get tool list from MCP server
# Create agent connected to MCP tools
# Create a placeholder agent synchronously
# Load MCP tools asynchronously and set them to the agent
# Clean up MCP connection on exit

# MCP 서버로부터 툴 목록 가져오기
# MCP 툴을 연결한 에이전트 생성
# 동기적으로 에이전트 생성
# 비동기적으로 MCP 툴을 로드하고 에이전트에 세팅
# MCP 연결 정리

from asyncio.log import logger
from google.adk.agents import Agent
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset, StdioServerParameters

# Get tool list from MCP server
async def get_tools_async():
    print("Attempting to connect to MCP Filesystem server...")
    tools, exit_stack = await MCPToolset.from_server(
        connection_params=StdioServerParameters(
            command='python',
            args=["d:/GitHub_NJY/ADK/adk_agent_samples/mcp_agent/adk_mcp_server.py"],
        )
    )
    return tools, exit_stack

# Create agent connected to MCP tools
async def get_agent_async():
    tools, exit_stack = await get_tools_async()
    agent = Agent(
        model='gemini-2.0-flash',
        name='filesystem_assistant',
        instruction='Help user interact with the local filesystem using available tools.',
        tools=tools,
    )
    return agent

# Create a placeholder agent synchronously
root_agent = Agent(
    model='gemini-2.0-flash',
    name='filesystem_assistant_placeholder',
    instruction='Help user interact with the local filesystem. Tools will be loaded asynchronously.',
    tools=[],
)

_mcp_exit_stack = None

# Load MCP tools asynchronously and set them to the agent
async def load_and_set_mcp_tools(agent_instance):
    global _mcp_exit_stack
    tools, exit_stack = await get_tools_async()
    agent_instance.tools = tools
    _mcp_exit_stack = exit_stack

# Clean up MCP connection on exit
def cleanup_mcp_connection():
    global _mcp_exit_stack
    if _mcp_exit_stack:
        _mcp_exit_stack.close()
        _mcp_exit_stack = None

4. Run

4. 실행
adk web

Other Topics

LLM Agents

https://google.github.io/adk-docs/agents/llm-agents

In ADK, LLM agents serve as the core components responsible for the “thinking” of the application. These agents leverage large language models to perform natural language understanding, reasoning, decision-making, response generation, and interaction with tools. Unlike workflow agents, which follow a fixed execution path, LLM agents are non-deterministic and act dynamically based on given instructions and context. To build effective LLM agents, it’s crucial to define the agent’s identity and purpose, provide clear instructions, and ensure the necessary tools and functionalities are available.

ADK에서 LLM 에이전트는 애플리케이션의 ‘사고’를 담당하는 핵심 구성 요소입니다. 이 에이전트는 대형 언어 모델을 활용하여 자연어 이해, 추론, 의사 결정, 응답 생성 및 도구와의 상호 작용을 수행합니다. LLM 에이전트는 고정된 실행 경로를 따르는 워크플로우 에이전트와 달리 비결정론적이며, 주어진 지침과 문맥을 기반으로 동적으로 행동합니다. 효과적인 LLM 에이전트를 구축하려면 에이전트의 정체성과 목적을 정의하고, 명확한 지침을 제공하며, 필요한 도구와 기능을 갖추는 것이 중요합니다.

Multi-Agent

https://google.github.io/adk-docs/agents/multi-agents

ADK allows the creation of multi-agent systems (MAS) by combining multiple independent instances of the BaseAgent. In such systems, different agents may cooperate or coordinate within a hierarchical structure to achieve larger goals. This structure enhances modularity, specialization, reusability, and maintainability, and allows for structured control flows using workflow agents. ADK provides essential building blocks to compose complex multi-agent systems by combining LLM agents, workflow agents (SequentialAgent, ParallelAgent, LoopAgent), and custom agents.

ADK는 여러 개의 독립적인 BaseAgent 인스턴스를 조합하여 멀티 에이전트 시스템(MAS)을 구축할 수 있도록 지원합니다. 이러한 시스템에서는 서로 다른 에이전트들이 계층 구조를 이루며 협력하거나 조정하여 더 큰 목표를 달성합니다. 이러한 구조는 모듈화, 전문화, 재사용성 및 유지 관리성을 향상시키며, 워크플로우 에이전트를 사용하여 구조화된 제어 흐름을 정의할 수 있습니다. ADK는 LLM 에이전트, 워크플로우 에이전트(SequentialAgent, ParallelAgent, LoopAgent) 및 사용자 정의 에이전트 등을 조합하여 복잡한 멀티 에이전트 시스템을 구성할 수 있는 핵심 빌딩 블록을 제공합니다.

Agent Development Kit(ADK)

Agent Development Kit

ADK란 무엇인가요?

QuickStart

1. Create a Virtual Environment

1. 가상환경 만들기

2. Install Required Packages

2. 필요 패키지 설치

3. Create Project Folder Structure

3. 프로젝트 폴더 구조 만들기

4. init.py

5. agent.py

6. .env

7. Run

7. 실행

MCP(Model Context Protocol) Tools

1. Create mcp_agent Folder

1. mcp_agent 폴더 만들기

2. adk_mcp_server.py

2. adk_mcp_server.py

3. agent.py

3. agent.py

4. Run

4. 실행

Other Topics

LLM Agents

Multi-Agent

Related Pages