Google ADK를 이용한 MCP 클라이언트 구축 (Building an MCP Client with Google ADK)

Google ADK를 사용하여 AI 에이전트를 구축하고, 이 에이전트가 MCP 서버에 연결하여 외부 도구를 활용하도록 하는 것은 ADK의 강력한 기능 중 하나이다. 이를 통해 에이전트는 자체적으로 구현되지 않은 다양한 기능을 수행할 수 있게 된다.

A. ADK 에이전트 기본 설정 (Basic ADK Agent Setup)

LlmAgent 또는 AdkApp 초기화 (Initializing LlmAgent or AdkApp)

ADK 에이전트를 만들기 위한 기본 클래스로 LlmAgent 또는 AdkApp을 사용할 수 있다.

• LlmAgent: 거대 언어 모델(LLM)을 핵심 추론 엔진으로 사용하는 단일 에이전트를 정의하는 데 적합하다. 일반적으로 모델, 에이전트 이름, 기본 지침(instruction), 그리고 사용할 도구 목록을 인자로 받아 초기화한다.  
   
• AdkApp: 보다 포괄적인 ADK 애플리케이션 컨테이너 역할을 하며, 세션 관리, 여러 에이전트 통합 등의 기능을 포함할 수 있다. 간단한 MCP 클라이언트 에이전트의 경우 LlmAgent로 시작하는 것이 일반적이다.  
   

예시:

Python

 

from google.adk.agents import LlmAgent

# 모델 이름은 사용 가능한 최신 모델 또는 특정 요구에 맞는 모델로 지정
# 예: "gemini-1.5-flash-preview-0514" 또는 Vertex AI에서 제공하는 다른 모델
agent = LlmAgent(
    model="gemini-1.5-flash-preview-0514", # Vertex AI 모델 사용 시 GCP 프로젝트 및 위치 설정 필요
    name="my_mcp_client_agent",
    instruction="당신은 사용자의 요청을 이해하고, 필요한 경우 외부 도구를 사용하여 답변하는 지능형 비서입니다."
)

 

 

 

 

 

 

 

 

모델 및 기본 지침 설정 (Configuring model and basic instructions)

• 모델 설정: 에이전트가 추론 및 의사결정에 사용할 LLM 모델을 지정한다. Google Cloud 환경에서는 Vertex AI를 통해 Gemini와 같은 최신 모델을 사용할 수 있다. LlmAgent의 model 매개변수에 모델 이름을 문자열로 전달한다.  
   
• 기본 지침 (Instruction) 설정: instruction 매개변수는 LLM에게 에이전트의 역할, 목표, 행동 방식, 그리고 특히 사용 가능한 도구들을 어떻게 활용해야 하는지에 대한 명확하고 구체적인 지침을 제공한다. 이 지침은 LLM이 MCP 서버로부터 가져온 도구들을 언제, 어떻게, 왜 사용해야 하는지 결정하는 데 결정적인 역할을 한다. MCP를 통해 외부 도구를 사용하는 에이전트의 경우, 이 instruction의 중요성은 더욱 커진다. 단순히 에이전트의 페르소나를 정의하는 것을 넘어, 사용 가능한 MCP 도구들과 그 도구들을 활용한 문제 해결 전략을 명시하는 "메타-프롬프트"로서 기능해야 한다. 예를 들어, "사용자가 특정 주제에 대한 Wikipedia 기사를 요청하면, 'extract_wikipedia_article' 도구를 사용하여 해당 URL에서 내용을 가져온 후, 그 내용을 바탕으로 사용자에게 몇 문장으로 요약해 주세요."와 같이 구체적인 도구 사용법을 포함하는 것이 좋다. 이는 LLM이 모호한 상황에서 올바른 도구를 선택하고 효과적으로 사용하는 데 큰 도움을 준다.  
   

B. MCP 서버 연결 및 도구 연동 (Connecting to MCP Server and Integrating Tools)

MCPToolset 소개 및 사용법 (Introduction and usage of MCPToolset)

MCPToolset은 ADK 에이전트가 외부 MCP 서버에 연결하고, 해당 서버가 제공하는 도구들을 가져와 에이전트의 기능으로 통합할 수 있도록 지원하는 핵심 클래스이다. MCPToolset은 MCP 서버와의 통신 복잡성(연결 설정, 도구 목록화, 도구 호출 등)을 상당 부분 추상화하여 ADK 개발자가 MCP 도구를 마치 로컬 ADK 도구처럼 비교적 쉽게 사용할 수 있도록 한다. 이는 개발 효율성을 크게 높이는 장점이 있지만, 추상화 수준이 높을수록 세밀한 통신 제어나 MCPToolset이 직접 지원하지 않는 MCP 기능(예: MCP 사양의 '리소스'나 '프롬프트' 프리미티브에 대한 직접 접근)을 사용하기에는 한계가 있을 수 있다.  

 

MCPToolset.from_server()를 이용한 연결 (Connecting using MCPToolset.from_server())

MCPToolset.from_server()는 비동기 메서드로서, 지정된 MCP 서버에 실제로 연결하고 해당 서버가 제공하는 도구 목록을 가져오는 주된 방법이다.  

 

• connection_params: 이 인자를 통해 연결할 MCP 서버의 주소, 포트, 그리고 사용할 전송 프로토콜에 대한 정보를 전달한다.
• 반환 값: 성공적으로 연결되면 (tools, exit_stack) 형태의 튜플을 반환한다.
  • tools: MCP 서버로부터 가져온 도구 객체들의 리스트이다. 이 도구 객체들은 ADK 에이전트가 직접 사용할 수 있는 형태로 제공된다.
  • exit_stack: 연결과 관련된 리소스(예: 네트워크 연결, 백그라운드 프로세스)를 안전하게 관리하고 해제하기 위한 AsyncExitStack 객체이다. 이는 주로 async with 구문과 함께 사용되어 프로그램 종료 시 리소스가 자동으로 정리되도록 한다.

지원되는 전송 프로토콜: SSE (SseServerParams), Stdio (StdioServerParameters) (Supported transport protocols: SSE, Stdio)

MCPToolset.from_server()의 connection_params에 전달되는 객체에 따라 MCP 서버와의 통신 방식이 결정된다. 현재 Google ADK 공식 문서에서 주로 언급되는 방식은 다음과 같다 :  

 

• SSE (SseServerParams): 원격 MCP 서버와 HTTP 기반의 Server-Sent Events를 통해 통신할 때 사용된다.  
  Python

   

  from google.adk.tools.mcp_tool.mcp_toolset import SseServerParams
  
  sse_params = SseServerParams(url="http://<mcp_server_host>:<port>/sse")
  # 예: sse_params = SseServerParams(url="http://localhost:8001/sse")
  

   

   

   

   

   

   

   

   
  여기서 url은 MCP 서버가 SSE 연결을 위해 열어둔 엔드포인트 주소이다.
•  
• Stdio (StdioServerParameters): 로컬 시스템에서 실행 중인 MCP 서버 프로세스와 표준 입출력(stdin/stdout)을 통해 직접 통신할 때 사용된다. 이는 주로 개발 및 테스트 단계에서 로컬 MCP 서버를 연동할 때 유용하다.  
  Python

   

  from google.adk.tools.mcp_tool.mcp_toolset import StdioServerParameters
  
  # 예: 로컬 파일 시스템 MCP 서버 (Node.js 기반) 실행
  stdio_params = StdioServerParameters(
      command='npx', # 실행할 명령어
      args=[ # 명령어에 전달할 인자들
          "-y",
          "@modelcontextprotocol/server-filesystem", # MCP 파일 시스템 서버 패키지
          "/path/to/accessible/folder" # 서버가 접근할 로컬 폴더 경로 (절대 경로 권장)
      ]
  )
  

   

   

   

   

   

   

   

   
•  
• Streamable HTTP 관련 참고: 에서는 omarbenhamid/google-adk-python이라는 커뮤니티 포크(fork)에서 Streamable HTTP 전송 방식을 지원한다고 언급하고 있다. Streamable HTTP는 양방향 스트리밍에 더 적합할 수 있는 최신 MCP 전송 방식 중 하나이다. 그러나 2024년 중반 현재, Google ADK의 공식 배포판 및 문서에서는 SSE와 Stdio를 주요 지원 방식으로 명시하고 있다. 따라서 이 보고서에서는 공식 지원을 기준으로 설명하며, Streamable HTTP는 향후 공식 지원될 가능성이 있는 커뮤니티 동향으로 간주한다. 프로덕션 환경에서는 공식 지원되는 안정적인 기능을 우선적으로 고려하는 것이 좋다.  
•  

가져온 도구를 ADK 에이전트에 통합 (Integrating fetched tools into ADK Agent)

MCPToolset.from_server()를 통해 성공적으로 가져온 tools 리스트는 LlmAgent를 초기화할 때 tools= 인자에 전달된다. 이렇게 함으로써 ADK 에이전트는 해당 MCP 도구들을 자신의 능력 중 하나로 인식하고, LLM의 판단에 따라 필요시 호출할 수 있게 된다.  

 

Python

 

# (이전 단계에서 mcp_tools, exit_stack = await MCPToolset.from_server(...) 로 도구들을 가져왔다고 가정)

agent = LlmAgent(
    model="gemini-1.5-flash-preview-0514",
    name="my_mcp_client_agent",
    instruction="사용자의 질문에 답하기 위해 Wikipedia 검색 도구를 활용하세요.",
    tools=mcp_tools # MCP 서버로부터 가져온 도구 리스트를 전달
)

 

 

 

 

 

 

 

 

예제: Wikipedia 추출 MCP 서버에 연결하는 ADK 에이전트 구현 (Example: Implementing an ADK agent connecting to the Wikipedia extraction MCP server)

앞서 IV 섹션에서 구현한 Wikipedia 아티클 추출 MCP 서버 (server.py가 http://localhost:8001에서 실행 중이라고 가정)에 연결하여 해당 서버의 extract_wikipedia_article 도구를 사용하는 ADK 에이전트의 예제 코드는 다음과 같다.  

 

agent_client.py (Wikipedia 추출 MCP 서버용 ADK 클라이언트)

Python

 

import asyncio
import os
from dotenv import load_dotenv

from google.adk.agents import LlmAgent
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset, SseServerParams
from google.adk.runners import AdkRunner, InProcessAdkRunner # Runner 추가

#.env 파일에서 환경 변수 로드 (예: GOOGLE_CLOUD_PROJECT 등)
load_dotenv()

# MCP 서버의 SSE 엔드포인트 URL
MCP_SERVER_SSE_URL = "http://localhost:8001/sse" # IV 섹션의 서버가 이 주소에서 실행 중이라고 가정

async def get_mcp_tools_async():
    """지정된 MCP 서버에 연결하여 도구들을 가져옵니다."""
    print(f"'{MCP_SERVER_SSE_URL}' MCP 서버에서 도구를 가져오는 중...")
    try:
        tools, exit_stack = await MCPToolset.from_server(
            connection_params=SseServerParams(url=MCP_SERVER_SSE_URL)
        )
        print(f"MCP 서버로부터 {len(tools)}개의 도구를 성공적으로 가져왔습니다.")
        # 가져온 도구 정보 출력 (디버깅용)
        for tool in tools:
            print(f" - 도구 이름: {tool.name}, 설명: {tool.description}")
        return tools, exit_stack
    except Exception as e:
        print(f"MCP 서버에서 도구를 가져오는 데 실패했습니다: {e}")
        return, None # 실패 시 빈 리스트와 None 반환

async def create_wikipedia_agent_async():
    """MCP 도구를 사용하는 Wikipedia 요약 에이전트를 생성합니다."""
    mcp_tools, exit_stack = await get_mcp_tools_async()

    if not mcp_tools or exit_stack is None:
        print("MCP 도구를 가져오지 못해 에이전트를 생성할 수 없습니다.")
        return None, None

    # 에이전트에게 MCP 도구 사용 방법을 명확히 지시
    agent_instruction = (
        "당신은 사용자가 제공한 Wikipedia URL의 기사 내용을 요약하는 AI 비서입니다.\n"
        "사용 가능한 도구 중 'extract_wikipedia_article' 도구를 사용하여 Wikipedia 기사 내용을 가져오세요.\n"
        "가져온 내용이 너무 길 경우, 핵심 내용을 중심으로 3-5 문장으로 요약하여 사용자에게 친절하게 설명해주세요.\n"
        "만약 도구 사용 중 오류가 발생하면, 사용자에게 오류 상황을 알리고 다른 요청을 하도록 안내하세요."
    )

    root_agent = LlmAgent(
        model=os.getenv("GOOGLE_GENAI_MODEL", "gemini-1.5-flash-preview-0514"), # 환경 변수 또는 기본값 사용
        name="wikipedia_summary_agent",
        instruction=agent_instruction,
        tools=mcp_tools, # MCP 서버로부터 가져온 도구 통합
    )
    print("Wikipedia 요약 에이전트가 성공적으로 생성되었습니다.")
    return root_agent, exit_stack

async def main():
    """에이전트를 생성하고 사용자 입력을 받아 처리하는 메인 함수입니다."""
    agent, exit_stack = await create_wikipedia_agent_async()

    if agent is None or exit_stack is None:
        print("에이전트 초기화 실패. 프로그램을 종료합니다.")
        return

    # exit_stack을 사용하여 리소스 자동 정리
    async with exit_stack:
        runner = InProcessAdkRunner() # 간단한 인프로세스 러너 사용
        print("\nWikipedia 요약 에이전트와 대화를 시작합니다. 종료하려면 'exit'을 입력하세요.")
        print("예시 질문: 'https://en.wikipedia.org/wiki/Artificial_intelligence' 기사를 요약해줘.")

        while True:
            user_input = input("사용자: ")
            if user_input.lower() == 'exit':
                print("프로그램을 종료합니다.")
                break

            if not user_input:
                continue

            print("에이전트 처리 중...")
            try:
                # runner.run()은 AdkApp에서 주로 사용. LlmAgent는 직접 query 또는 stream_query 호출 가능
                # 여기서는 LlmAgent의 stream_query를 직접 사용하는 예시를 보여줌
                # 세션 관리가 필요하다면 AdkApp과 함께 사용해야 함
                async for chunk in agent.stream_query(user_input):
                    # chunk는 보통 딕셔너리 형태이며, 'text' 키에 응답 내용이 담겨 있을 수 있음
                    if isinstance(chunk, dict) and "text" in chunk:
                        print(f"에이전트: {chunk['text']}", end="", flush=True)
                    elif isinstance(chunk, str): # 단순 텍스트 응답도 처리
                        print(f"에이전트: {chunk}", end="", flush=True)
                print() # 줄바꿈
            except Exception as e:
                print(f"\n에이전트 처리 중 오류 발생: {e}")

if __name__ == "__main__":
    # GCP 프로젝트 및 위치 설정 (Vertex AI 사용 시 필수)
    # 이 정보는 환경 변수 (GOOGLE_CLOUD_PROJECT, GOOGLE_CLOUD_LOCATION) 또는
    # google.cloud.aiplatform.init() 등을 통해 설정될 수 있음
    # load_dotenv()를 통해.env 파일에서 읽어올 수 있도록 하는 것이 좋음
    # 예:.env 파일 내용
    # GOOGLE_CLOUD_PROJECT="your-gcp-project-id"
    # GOOGLE_CLOUD_LOCATION="us-central1"
    # GOOGLE_GENAI_MODEL="gemini-1.5-flash-preview-0514"
    # GOOGLE_GENAI_USE_VERTEXAI="True" # ADK가 Vertex AI를 사용하도록 지시

    if not os.getenv("GOOGLE_CLOUD_PROJECT") or not os.getenv("GOOGLE_CLOUD_LOCATION"):
        print("경고: GOOGLE_CLOUD_PROJECT 또는 GOOGLE_CLOUD_LOCATION 환경 변수가 설정되지 않았습니다.")
        print("Vertex AI 모델 사용 시 오류가 발생할 수 있습니다.")

    asyncio.run(main())

 

 

 

 

 

 

 

 

코드 설명:

1. 환경 설정 및 Import: 필요한 라이브러리( asyncio, LlmAgent, MCPToolset, SseServerParams 등)를 가져오고, dotenv를 사용하여 .env 파일에 저장된 환경 변수(예: Google Cloud 프로젝트 ID, 위치, 사용할 모델명)를 로드한다.
2. get_mcp_tools_async(): MCPToolset.from_server()를 호출하여 지정된 MCP_SERVER_SSE_URL (여기서는 http://localhost:8001/sse)로 MCP 서버에 연결하고, 서버가 제공하는 도구 목록과 exit_stack을 가져온다. 도구 정보를 간단히 출력하여 확인한다.
3. create_wikipedia_agent_async(): get_mcp_tools_async()를 호출하여 MCP 도구들을 가져온 후, LlmAgent를 생성한다.
   • model: 사용할 LLM 모델을 지정한다 (환경 변수 GOOGLE_GENAI_MODEL 또는 기본값 사용).
   • name: 에이전트의 이름을 부여한다.
   • instruction: 에이전트의 역할과 함께, MCP 서버로부터 가져온 extract_wikipedia_article 도구를 어떻게 사용해야 하는지에 대한 명확한 지침을 제공한다.
   • tools: get_mcp_tools_async()를 통해 가져온 mcp_tools 리스트를 전달하여 에이전트가 해당 도구들을 사용할 수 있도록 한다.
4. main(): 에이전트를 생성하고, async with exit_stack: 구문을 사용하여 MCP 연결 관련 리소스가 프로그램 종료 시 자동으로 정리되도록 한다. InProcessAdkRunner (또는 직접 agent.stream_query)를 사용하여 사용자 입력을 받아 에이전트에게 전달하고, 에이전트의 응답을 스트리밍 형태로 출력한다.
5. 실행: 스크립트 실행 시, 필요한 환경 변수(특히 Vertex AI 사용 시 GOOGLE_CLOUD_PROJECT, GOOGLE_CLOUD_LOCATION)가 설정되어 있는지 확인하는 로직을 추가했다.

이 에이전트를 실행하고 "https://en.wikipedia.org/wiki/Artificial_intelligence 기사를 요약해줘."와 같은 질문을 하면, 에이전트는 LLM의 판단에 따라 extract_wikipedia_article MCP 도구를 호출하여 Wikipedia 서버로부터 해당 URL의 내용을 가져온 후, 그 내용을 바탕으로 사용자에게 요약된 답변을 제공할 것이다.

ADK는 adk web이라는 명령어를 통해 개발자용 웹 UI를 제공하여, 에이전트와의 상호작용, 도구 호출 과정, 내부 상태 등을 시각적으로 확인하며 테스트할 수 있는 편리한 기능을 제공한다. 이를 활용하면 MCP 클라이언트 에이전트의 개발 및 디버깅 효율을 높일 수 있다.