에이전트

Google ADK 및 MCP를 활용한 효율적인 MCP 서버 및 클라이언트 구축 가이드

aiagentx 2025. 5. 20. 11:31
728x90
SMALL

Google ADK 및 MCP 간략 소개 (Brief overview of Google ADK and MCP)

Google의 Agent Development Kit (ADK)는 정교한 AI 에이전트 및 멀티 에이전트 시스템 구축 과정을 단순화하면서도 에이전트의 행동에 대한 정밀한 제어를 유지하도록 설계된 오픈소스 프레임워크이다. ADK는 개발자가 직관적인 코드로 AI 에이전트를 만들 수 있도록 지원하며, Google의 Gemini 모델 및 Vertex AI 플랫폼에 최적화되어 있지만, 개발자가 선호하는 다른 언어 모델이나 배포 환경과도 유연하게 연동될 수 있는 개방성을 지향한다.  

 

Model Context Protocol (MCP)은 거대 언어 모델(LLM)이 외부 애플리케이션, 데이터 소스, 그리고 다양한 도구들과 통신하는 방식을 표준화하기 위해 등장한 개방형 표준 프로토콜이다. 기존에는 각 시스템과의 연동을 위해 개별적이고 단편적인 맞춤형 통합(custom integration)이 필요했지만, MCP는 이러한 방식을 보편적이고 표준화된 프레임워크로 대체하여 AI 시스템 간의 상호 운용성을 높이는 것을 목표로 한다.  

 

Google ADK는 이러한 MCP를 적극적으로 지원하며, 이를 통해 ADK로 구축된 AI 에이전트가 MCP 표준을 따르는 다양한 외부 데이터 소스 및 도구들과 안전하고 효율적으로 양방향 통신을 수행할 수 있도록 한다. 이는 ADK 에이전트의 능력을 크게 확장하고, 더 복잡하고 실용적인 작업을 수행할 수 있는 기반을 마련한다. Google이 ADK를 통해 MCP를 지원하는 것은 단순한 기술 통합을 넘어, 개발자들이 MCP 생태계에 더 쉽게 참여하도록 유도하려는 전략적 움직임으로 볼 수 있다. 잘 갖춰진 클라이언트 프레임워크인 ADK를 제공함으로써, MCP 기반 도구 및 서비스의 채택을 가속화하고 Google Cloud 중심의 에이전트 개발 생태계를 강화하려는 의도가 담겨 있을 수 있다. 이는 결과적으로 더 많은 개발자가 MCP를 사용하게 만들고, Google Cloud 플랫폼에서의 에이전트 활용 증대로 이어져 플랫폼의 매력도를 높이는 데 기여할 것이다.  

 

MCP 서버 및 클라이언트 구축의 중요성 및 사용 사례 (Importance and use cases of building MCP servers and clients)

MCP 서버는 특정 기능(예: 데이터베이스 조회, 외부 API 호출, 특정 연산 수행 등)을 표준화된 "도구(tool)" 형태로 에이전트에게 제공하는 역할을 한다. 이를 통해 AI 에이전트는 다양한 종류의 외부 시스템과 데이터에 일관된 방식으로 접근하고 상호작용할 수 있게 된다. 예를 들어, 기업 내부의 데이터베이스에 저장된 정보를 조회하거나, 실시간 날씨 정보를 제공하는 외부 API를 호출하는 기능을 MCP 서버를 통해 도구화할 수 있다.  

 

MCP 클라이언트는 이러한 MCP 서버에 연결하여 제공되는 도구들을 활용하는 주체이다. Google ADK를 사용하여 구축된 AI 에이전트가 대표적인 MCP 클라이언트가 될 수 있다. 에이전트는 사용자의 요청을 이해하거나 내부적인 목표를 달성하기 위해 MCP 서버가 제공하는 도구를 호출하고, 그 결과를 바탕으로 다음 행동을 결정하거나 사용자에게 응답을 생성한다.  

 

MCP 서버와 클라이언트 구축의 중요성은 다양한 사용 사례에서 명확히 드러난다.

  • 기업 데이터 접근성 향상: MCP Toolbox for Databases와 같은 MCP 서버를 활용하면, AI 에이전트가 기업의 다양한 데이터베이스(예: AlloyDB, Spanner, Cloud SQL 등)에 안전하고 표준화된 방식으로 접근하여 필요한 정보를 조회하거나 분석할 수 있다.  
     
  • 외부 정보 및 서비스 연동: Wikipedia 기사 내용을 가져오거나 , Google Maps와 같은 지리 정보 서비스의 기능을 활용하는 등, 외부 세계의 방대한 정보와 유용한 서비스들을 에이전트가 쉽게 이용할 수 있게 된다.  
     
  • 복잡한 멀티 에이전트 시스템 구축: 각기 다른 전문성을 가진 여러 에이전트들이 MCP를 통해 서로의 기능을 "도구"로서 호출하고 협업하는 정교한 멀티 에이전트 시스템을 구축할 수 있다. 예를 들어, 사용자 프로필 분석 에이전트, 활동 계획 에이전트, 플랫폼 상호작용 에이전트가 MCP를 통해 데이터를 주고받으며 사용자 맞춤형 서비스를 제공하는 시나리오를 생각해 볼 수 있다.  
     

    다음 표는 MCP의 주요 구성 요소와 ADK와의 연관성을 요약한 것이다.

    Table 1: MCP 주요 구성 요소 및 ADK 연관성 (Key MCP Components and ADK Relevance)

    MCP 구성 요소 (Component) 설명 (Description - Korean) MCP 사양에서의 역할 (Role in MCP Spec) ADK에서의 주요 활용 (Primary Use in ADK) 관련 자료 (Relevant Sources)
    도구 (Tools) 에이전트가 호출할 수 있는 실행 가능한 함수 또는 API. 외부 시스템과의 상호작용을 담당. 에이전트의 행동 및 외부 시스템 제어 MCPToolset을 통해 서버로부터 도구 목록을 가져오고, 특정 도구를 선택하여 실행. ADK의 MCP 연동 핵심.  
    리소스 (Resources) 에이전트가 접근할 수 있는 정적 또는 동적 데이터. LLM에 컨텍스트 제공. (예: 파일, 데이터베이스 레코드) 에이전트의 이해 및 추론을 위한 배경 정보 제공 명시적인 MCPToolset 기능보다는, 도구 실행의 일부로 간접적으로 접근될 가능성. (예: MCP Toolbox for Databases는 DB를 리소스처럼 제공)  
    프롬프트 (Prompts) 서버에 사전 정의된 재사용 가능한 프롬프트 템플릿. LLM과의 표준화된 상호작용 유도. 일관된 사용자 경험 및 LLM 응답 유도 현재 ADK 공식 문서에서 MCPToolset을 통한 직접적인 MCP 프롬프트 활용 방법은 명확히 제시되지 않음.  
     
     
    • 도구 (Tools): 모델(LLM)이 직접 호출하여 특정 작업을 수행하거나 외부 시스템과 상호작용할 수 있도록 하는 실행 가능한 함수이다. 예를 들어, 특정 URL의 웹 페이지 내용을 추출하는 extract_wikipedia_article(url: str) -> str 함수 나, 두 숫자를 더하는 add(a: int, b: int) -> int 함수 등이 MCP 도구가 될 수 있다. ADK의 MCPToolset은 이러한 도구들을 MCP 서버로부터 가져와 에이전트가 사용할 수 있도록 하는 데 핵심적인 역할을 한다.  
    •  
    • 리소스 (Resources): 모델이 요청하여 컨텍스트 정보를 얻을 수 있는 정적 또는 동적 데이터 파일이나 데이터 덩어리를 의미한다. 예를 들어, 로그 파일, 데이터베이스 스키마, 특정 파일의 내용, Git 히스토리 등이 리소스가 될 수 있다. 이러한 리소스는 LLM이 상황을 더 잘 이해하고 정확한 추론을 하는 데 필요한 배경 정보를 제공한다. ADK가 MCP를 통해 데이터 소스에 연결될 수 있다는 점 은 리소스 접근의 한 형태를 시사하지만, ADK의 MCPToolset이 MCP 사양에 정의된 "리소스" 프리미티브를 직접적으로 어떻게 다루는지에 대한 명확한 문서는 현재 부족한 편이다. MCP Toolbox for Databases 와 같은 MCP 서버는 데이터베이스 자체를 일종의 리소스로 간주하고, 이를 조회하는 도구를 제공하는 방식으로 리소스 접근을 간접적으로 지원한다.  
    •  
    • 프롬프트 (Prompts): MCP 서버가 미리 정의해 놓은 재사용 가능한 프롬프트 템플릿이나 워크플로를 지칭한다. 클라이언트는 이러한 프롬프트를 사용자에게 제시하거나 LLM과의 상호작용을 표준화하는 데 활용할 수 있다. 예를 들어, "버그 리포트 작성 프롬프트"나 "코드 분석 요청 프롬프트" 등이 있을 수 있다. MCP 서버가 프롬프트를 관리하고 제공할 수 있다는 자료 는 있지만, ADK 클라이언트가 이러한 MCP 서버의 프롬프트를 MCPToolset을 통해 직접적으로 어떻게 활용하는지에 대한 구체적인 가이드라인은 아직 명확하게 제시되지 않고 있다.  
    •  

    Google ADK가 MCP의 세 가지 주요 구성 요소 중 현재 "도구"에 가장 집중하는 것은 의도된 전략일 가능성이 있다. 이는 ADK를 통해 에이전트가 즉각적으로 외부 시스템과 "상호작용"하고 "행동"할 수 있는 능력, 즉 도구 사용 능력을 우선적으로 강화하려는 전략을 반영하는 것일 수 있다. "리소스" 접근이나 "프롬프트" 활용은 도구 호출의 일부로 간접적으로 통합되거나, 향후 ADK 업데이트를 통해 더욱 명시적으로 지원될 가능성을 열어두고 있다. 이러한 접근 방식은 개발자들이 에이전트의 실행 능력을 빠르게 확장하는 데 집중하도록 유도하며, 이는 실용적인 애플리케이션 구축에 더 직접적인 영향을 미칠 수 있다.

    MCP가 개방형 표준이라는 점은 특정 벤더나 플랫폼에 종속되지 않고 다양한 시스템과 에이전트가 상호 운용될 수 있는 유연하고 확장 가능한 미래를 시사한다. MCP 서버는 데이터베이스 , 웹 콘텐츠 , 파일 시스템 , 지도 서비스 , 심지어 다른 에이전트의 기능까지도 표준화된 방식으로 제공할 수 있는 엄청난 잠재력을 가지고 있다. ADK 에이전트는 MCPToolset을 통해 이러한 다양한 서버에 연결하여 기능을 활용할 수 있으므로, 개발자는 필요한 기능을 매번 직접 구현하는 대신, 잘 정의된 MCP 인터페이스를 통해 기존 서비스나 도구를 "플러그 앤 플레이" 방식으로 통합할 수 있다. 이는 개발 효율성과 시스템의 모듈성을 크게 향상시키는 중요한 이점이다.  

     

    Google ADK를 이용한 MCP 서버 구축 (Building an MCP Server with Google ADK)

    MCP 서버는 외부 시스템의 기능을 표준화된 도구 형태로 제공하여 AI 에이전트(MCP 클라이언트)가 이를 활용할 수 있도록 하는 핵심 구성 요소이다. Python 환경에서는 FastMCP와 Starlette 같은 라이브러리를 사용하여 비교적 쉽게 MCP 서버를 구축할 수 있다.

    A. MCP 서버 기본 구조 설계 (Designing the Basic MCP Server Structure)

    FastMCP 및 Starlette 활용 방안 (Utilizing FastMCP and Starlette)

    • FastMCP: mcp 라이브러리에서 제공하는 FastMCP 클래스는 MCP 서버 인스턴스를 간편하게 생성하고, 서버가 제공할 도구들을 등록하는 데 사용된다. 일반적으로 mcp = FastMCP("my_server_name")과 같이 서버의 식별 이름을 부여하여 초기화한다.  
       
    • Starlette: Starlette은 비동기 처리를 지원하는 고성능 ASGI(Asynchronous Server Gateway Interface) 웹 프레임워크이다. MCP 통신은 주로 HTTP 기반으로 이루어지므로, Starlette을 사용하여 HTTP 요청을 처리하고, 클라이언트와의 통신을 위한 엔드포인트(예: Server-Sent Events (SSE) 엔드포인트, 도구 호출 메시지를 받는 POST 엔드포인트)를 설정하는 데 매우 유용하다.  
       
    • uvicorn: uvicorn은 Starlette과 같은 ASGI 애플리케이션을 실행하는 경량의 ASGI 서버이다. 개발된 Starlette MCP 서버 애플리케이션은 uvicorn을 통해 실행되어 클라이언트의 요청을 받을 준비를 한다. 예를 들어, uvicorn.run(app, host="localhost", port=8001)과 같이 실행할 수 있다.  
       

    MCP 도구 정의 (@mcp.tool() 데코레이터 사용) (Defining MCP tools using @mcp.tool() decorator)

    MCP 서버에서 제공하고자 하는 각 기능은 Python 함수로 작성된 후, @mcp.tool() 데코레이터를 사용하여 MCP 도구로 정식 등록된다. 이 데코레이터는 해당 함수를 MCP 클라이언트가 발견하고 호출할 수 있도록 만들어준다.  

     

    도구를 정의할 때 다음 사항에 유의해야 한다:

    • 타입 어노테이션 (Type Annotations): 함수의 매개변수와 반환 값에 대한 타입 어노테이션을 명시하는 것이 좋다. 이는 MCP 클라이언트(특히 LLM)가 도구의 시그니처를 정확히 이해하는 데 도움을 준다.
    • Docstring: 함수의 docstring은 해당 도구에 대한 설명으로 활용된다. LLM은 이 설명을 읽고 도구의 기능, 사용 방법, 적절한 사용 시점 등을 판단하므로, 명확하고 상세하며 이해하기 쉬운 설명을 제공하는 것이 매우 중요하다. docstring에는 도구가 무엇을 하는지, 각 매개변수는 무엇을 의미하는지, 어떤 값을 반환하는지, 가능하다면 간단한 사용 예시까지 포함하는 것이 좋다. 이는 에이전트의 전반적인 성능 향상에 직접적으로 기여한다.  
       

    다음은 간단한 덧셈 도구를 정의하는 예시이다 :  

     
    Python
     
    from mcp.server.fastmcp import FastMCP

mcp = FastMCP("calculator_server") # MCP 서버 인스턴스 생성

@mcp.tool() # 도구 등록 데코레이터
def add(a: int, b: int) -> int:
    """
    두 개의 정수를 입력받아 그 합을 반환합니다.
    예시: add(a=5, b=3)은 8을 반환합니다.

    Args:
        a (int): 첫 번째 정수.
        b (int): 두 번째 정수.

    Returns:
        int: 두 정수의 합.
    """
    return a + b
     

    전송 프로토콜 설정 (Setting up Transport Protocol - 예: SSE)

    MCP 서버는 클라이언트와 통신하기 위한 전송 프로토콜을 설정해야 한다. 여러 전송 방식이 있지만, Server-Sent Events (SSE)는 서버에서 클라이언트로의 단방향 스트리밍 통신에 적합하며, MCP에서 자주 사용되는 방법 중 하나이다.  

     

    mcp 라이브러리의 SseServerTransport (일반적으로 mcp.server.sse 모듈에 위치)를 사용하여 SSE 통신을 설정할 수 있다. Starlette 애플리케이션에는 SSE 연결을 수립하는 엔드포인트(예: /sse)와 클라이언트로부터 도구 실행 요청과 같은 메시지를 받는 엔드포인트(예: /messages/)를 라우팅 규칙에 추가해야 한다.  

     

    예제: 간단한 Wikipedia 아티클 추출 MCP 서버 구현 (Example: Implementing a simple Wikipedia article extraction MCP server)

    다음은 및 에서 제시된 Wikipedia 기사 내용을 추출하여 마크다운 형식으로 반환하는 MCP 서버의 구체적인 구현 예제이다.  

     

    server.py (Wikipedia 아티클 추출 MCP 서버)

    Python
     
    import requests
from requests.exceptions import RequestException
from bs4 import BeautifulSoup
from html2text import html2text
import uvicorn
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.routing import Route, Mount
from mcp.server.fastmcp import FastMCP
from mcp.shared.exceptions import McpError
from mcp.types import ErrorData, INTERNAL_ERROR, INVALID_PARAMS
from mcp.server.sse import SseServerTransport

# 1. FastMCP 인스턴스 생성
mcp = FastMCP("wikipedia_extractor")

# 2. MCP 도구 정의: extract_wikipedia_article
@mcp.tool()
def extract_wikipedia_article(url: str) -> str:
    """
    지정된 Wikipedia URL에서 주요 기사 내용을 추출하여 Markdown 형식으로 반환합니다.
    URL은 'http' 또는 'https://'로 시작해야 합니다.

    Args:
        url (str): 내용을 추출할 Wikipedia 기사의 전체 URL.

    Returns:
        str: 추출된 기사 내용 (Markdown 형식).

    Raises:
        McpError: URL이 유효하지 않거나, 페이지 접근에 실패하거나,
                  기사 내용을 찾을 수 없는 경우 발생합니다.
    """
    try:
        if not url.startswith("http"):
            raise ValueError("URL은 http 또는 https 프로토콜로 시작해야 합니다.")

        response = requests.get(url, timeout=10) # 타임아웃 설정
        response.raise_for_status() # 200 OK가 아니면 HTTPError 발생

        soup = BeautifulSoup(response.text, "html.parser")
        # Wikipedia 페이지의 주요 콘텐츠 영역 ID는 'mw-content-text'일 수 있음
        # 실제로는 더 견고한 선택자 전략이 필요할 수 있음
        content_div = soup.find("div", {"id": "mw-content-text"})

        if not content_div:
            # 주요 콘텐츠 영역을 찾지 못한 경우
            # mw-body-content 등 다른 선택자도 고려해볼 수 있음
            content_div = soup.find("div", class_="mw-parser-output") # 대체 선택자
            if not content_div:
                 raise McpError(
                    ErrorData(
                        code=INVALID_PARAMS,
                        message="지정된 Wikipedia URL에서 주요 기사 내용 섹션을 찾을 수 없습니다."
                    )
                )

        # 불필요한 요소 제거 (예: 목차, 편집 링크 등)
        for unwanted_tag in content_div.find_all(["div", "table"], class_=["toc", "infobox", "metadata", "navbox"]):
            unwanted_tag.decompose()
        for unwanted_tag in content_div.find_all("span", class_=["mw-editsection"]):
            unwanted_tag.decompose()


        markdown_text = html2text(str(content_div))
        return markdown_text

    except ValueError as ve: # URL 프로토콜 오류
        raise McpError(ErrorData(code=INVALID_PARAMS, message=str(ve)))
    except RequestException as e: # 네트워크 오류 또는 HTTP 오류
        raise McpError(
            ErrorData(
                code=INTERNAL_ERROR,
                message=f"Wikipedia 기사 접근 중 오류 발생: {e}"
            )
        )
    except Exception as e: # 기타 예기치 않은 오류
        raise McpError(ErrorData(code=INTERNAL_ERROR, message=f"예기치 않은 오류 발생: {str(e)}"))

# 3. SSE 전송 프로토콜 설정
# 클라이언트가 메시지를 POST할 경로를 지정 (예: /messages/)
sse_transport = SseServerTransport("/messages/")

# 4. Starlette 애플리케이션 및 라우트 정의
async def handle_sse_connection(request: Request):
    """SSE 연결을 처리하는 핸들러"""
    # mcp._mcp_server는 FastMCP 내부의 실제 Server 객체를 가리킬 수 있음
    # 또는 mcp.server() 와 같은 메서드를 통해 Server 객체를 가져와야 할 수 있음
    # FastMCP의 내부 구현에 따라 달라질 수 있으므로, 공식 문서나 소스코드 확인 필요
    # 여기서는 mcp 객체가 run 메서드를 직접 제공한다고 가정하거나,
    # SseServerTransport가 이를 처리한다고 가정함.
    # [6]의 예제에서는 SseServerTransport의 connect_sse를 사용함.
    _server = mcp._mcp_server # [6] 참조
    async with sse_transport.connect_sse(
        request.scope,
        request.receive,
        request._send, # Starlette의 내부 send 사용
    ) as (reader, writer):
        await _server.run(reader, writer, _server.create_initialization_options())


app = Starlette(
    debug=True, # 개발 중에는 True, 프로덕션에서는 False
    routes=,
)

# 5. Uvicorn으로 서버 실행
if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8001)

이 아이디어를 구체적으로 구현한 예제를 제공하는데, 핵심은 기존 ADK 도구(AgentTool 인스턴스)를 가져와서, MCP 서버의 list_tools 및 call_tool 요청 핸들러 내에서 해당 ADK 도구의 스키마를 MCP가 이해할 수 있는 형식으로 변환하고(adk_to_mcp_tool_type 유틸리티 사용 가능성), 실제 ADK 도구를 실행(adk_tool.run_async)하는 것이다.  

 

이 과정에는 ADK 도구가 사용하는 입력 및 출력 데이터 형식을 MCP 표준에 부합하도록 변환하는 작업이 포함될 수 있다. 예를 들어, ADK 도구가 복잡한 Python 객체를 반환한다면, 이를 JSON 직렬화 가능한 형태로 변환하여 MCP 클라이언트에게 전달해야 할 수 있다.

예제 코드 (adk_mcp_server.py) 및 설명 (Example code and explanation)

다음은 의 코드를 기반으로, ADK의 AgentTool (여기서는 특정 ADK 에이전트에 포함된 도구를 가정)을 MCP 서버를 통해 노출하는 adk_mcp_server.py 스크립트의 예제이다. 이 예제에서는 mcp.server.Server를 직접 사용하여 MCP 서버를 구성한다.  

 
Python
 
# adk_mcp_server.py
import asyncio
import json
import uvicorn
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.routing import Route, Mount

# MCP 관련 import
from mcp import types as mcp_types
from mcp.server.server import Server as McpServer # FastMCP 대신 기본 Server 사용
from mcp.server.sse import SseServerTransport

# ADK 관련 import (가상의 ADK 도구 및 변환 유틸리티)
# 실제 ADK 프로젝트에서는 해당 프로젝트의 에이전트 및 도구를 import 해야 함
# from my_adk_project.agents import my_root_agent
# from my_adk_project.tools import some_adk_tool_instance
# from google.adk.tools.utils import adk_tool_to_mcp_schema # 가상의 유틸리티

# --- 가상의 ADK 도구 준비 ---
# 실제로는 ADK 에이전트를 로드하고, 그 에이전트의 특정 도구를 가져와야 함
# 예시를 위해 간단한 더미 ADK 도구를 정의
class DummyADKTool:
    def __init__(self, name, description, parameters, func_to_run):
        self.name = name
        self.description = description
        self.parameters = parameters # 예: {"param1": {"type": "string", "description": "..."}}
        self._func_to_run = func_to_run

    async def run_async(self, args, tool_context=None):
        # tool_context는 ADK 실행 환경에서 제공됨, 여기서는 None으로 가정
        return await self._func_to_run(**args)

    def to_mcp_schema(self):
        # ADK 도구 정의를 MCP 도구 스키마로 변환하는 로직
        # [22]의 adk_to_mcp_tool_type 와 유사한 기능
        mcp_args =
        for p_name, p_details in self.parameters.items():
            mcp_args.append(
                mcp_types.ToolArgument(
                    name=p_name,
                    description=p_details.get("description", ""),
                    type=p_details.get("type", "string"), # 단순화된 타입 매핑
                    required=p_details.get("required", True)
                )
            )
        return mcp_types.Tool(
            name=self.name,
            description=self.description,
            arguments=mcp_args,
            # MCP는 반환 타입도 명시할 수 있으나, 여기서는 생략
        )

async def dummy_adk_function(message: str) -> str:
    return f"ADK tool processed: {message}"

# 노출할 ADK 도구 인스턴스 생성
adk_tool_to_expose = DummyADKTool(
    name="my_adk_echo_tool",
    description="ADK로 만들어진 에코 기능을 제공하는 도구입니다.",
    parameters={"message": {"type": "string", "description": "에코할 메시지"}},
    func_to_run=dummy_adk_function
)
# --- ADK 도구 준비 완료 ---


# --- MCP 서버 설정 ---
mcp_server_instance = McpServer("adk_wrapped_server") # MCP 서버 인스턴스
sse_transport = SseServerTransport("/messages/")      # SSE 전송 설정

@mcp_server_instance.list_tools()
async def list_mcp_tools() -> list:
    """MCP 클라이언트에게 사용 가능한 도구 목록을 반환합니다."""
    print("MCP Server: list_tools 요청 수신")
    # ADK 도구의 스키마를 MCP 형식으로 변환하여 반환
    mcp_tool_schema = adk_tool_to_expose.to_mcp_schema()
    print(f"MCP Server: 다음 도구를 광고합니다: {mcp_tool_schema.name}")
    return [mcp_tool_schema]

@mcp_server_instance.call_tool()
async def call_mcp_tool(
    name: str,
    arguments: dict[str, any] | None, # MCP 사양에 따라 arguments는 dict 또는 None
    # [22]에서는 list 등을 반환 타입으로 사용
) -> any: # 실제 반환 타입은 MCP Content 타입이어야 함
    """MCP 클라이언트로부터 도구 호출 요청을 받아 처리합니다."""
    print(f"MCP Server: '{name}' 도구 호출 요청 수신, 인자: {arguments}")

    tool_args = arguments if arguments is not None else {}

    if name == adk_tool_to_expose.name:
        try:
            # ADK 도구의 run_async 메서드 호출
            # tool_context는 MCP 서버 환경에서는 일반적으로 None 또는 모의(mock) 객체일 수 있음
            adk_response = await adk_tool_to_expose.run_async(args=tool_args, tool_context=None)
            print(f"MCP Server: ADK 도구 '{name}' 실행 성공.")

            # ADK 도구의 응답을 MCP 형식으로 변환
            # 여기서는 간단히 JSON 문자열로 변환하여 TextContent로 래핑
            # 실제로는 ADK 응답의 구조에 따라 더 정교한 변환 필요
            response_text = json.dumps({"result": adk_response}, ensure_ascii=False, indent=2)
            # MCP는 다양한 콘텐츠 타입(TextContent, ImageContent 등)을 지원
            return
        except Exception as e:
            print(f"MCP Server: ADK 도구 '{name}' 실행 중 오류: {e}")
            # MCP 오류 형식으로 변환하여 반환하는 것이 좋음
            # 예: McpError 사용 또는 표준 오류 구조 사용
            error_response_text = json.dumps({"error": str(e)}, ensure_ascii=False, indent=2)
            # 실제로는 McpError 객체를 발생시키거나, types.ErrorContent 등을 사용해야 할 수 있음
            # 여기서는 간단히 텍스트로 오류를 전달
            return
    else:
        # 요청된 도구를 찾을 수 없는 경우
        unknown_tool_error_text = f"알 수 없는 도구입니다: {name}"
        return

# Starlette 앱 설정 [6, 22]
async def sse_endpoint(request: Request):
    async with sse_transport.connect_sse(
        request.scope, request.receive, request._send
    ) as (reader, writer):
        await mcp_server_instance.run(reader, writer, mcp_server_instance.create_initialization_options())

starlette_app = Starlette(
    routes=
)
# --- MCP 서버 설정 완료 ---

if __name__ == "__main__":
    print("ADK 도구를 노출하는 MCP 서버를 시작합니다...")
    uvicorn.run(starlette_app, host="0.0.0.0", port=8002) # 다른 포트 사용
 

코드 설명:

  1. ADK 도구 준비: 실제 ADK 프로젝트에서는 google-adk 라이브러리를 사용하여 정의된 AgentTool 인스턴스를 가져와야 한다. 위 예제에서는 설명을 위해 DummyADKTool이라는 간소화된 클래스를 정의하고, 이를 사용하여 노출할 ADK 도구(adk_tool_to_expose)를 생성했다. to_mcp_schema 메서드는 ADK 도구의 정의를 MCP가 이해할 수 있는 mcp_types.Tool 형식으로 변환하는 역할을 한다.
  2. MCP 서버 설정: McpServer("adk_wrapped_server")를 사용하여 MCP 서버 인스턴스를 직접 생성한다. SseServerTransport도 이전 예제와 유사하게 설정한다.
  3. @mcp_server_instance.list_tools() 핸들러: MCP 클라이언트가 사용 가능한 도구 목록을 요청하면 호출된다. 여기서는 준비된 adk_tool_to_expose의 스키마를 MCP 형식으로 변환하여 리스트에 담아 반환한다.
  4. @mcp_server_instance.call_tool() 핸들러: MCP 클라이언트가 특정 도구의 실행을 요청하면 호출된다. 요청된 도구 이름(name)이 우리가 노출하려는 ADK 도구의 이름과 일치하는지 확인한다.
    • 일치하면, adk_tool_to_expose.run_async(args=arguments, tool_context=None)을 호출하여 실제 ADK 도구 로직을 실행한다. tool_context는 ADK 에이전트 실행 컨텍스트에서 제공되므로, MCP 서버 환경에서는 None이거나 필요한 경우 모의(mock) 객체를 전달해야 할 수 있다.
    • ADK 도구의 실행 결과를 MCP 클라이언트가 이해할 수 있는 형식(예: mcp_types.TextContent)으로 변환하여 반환한다. 이 변환 과정은 ADK 도구의 실제 반환 값 구조에 따라 달라진다.
    • 실행 중 오류가 발생하면 적절한 MCP 오류 응답을 반환해야 한다.
  5. Starlette 앱 및 실행: 이전 Wikipedia 예제와 유사하게 Starlette 애플리케이션을 설정하고 uvicorn으로 실행한다.

이러한 방식으로 ADK로 이미 개발된 복잡한 로직이나 여러 도구를 사용하는 에이전트의 특정 기능을 표준 MCP 인터페이스 뒤에 숨겨 재사용성을 높이고, 다른 MCP 호환 시스템과의 통합을 용이하게 할 수 있다.

728x90