2025년 4월 12일 토요일

Gemini 기반 MCP 서버 및 클라이언트 개발해 보기

이 글은 Gemini 기반 MCP 서버 및 클라이언트를 개발하는 방법을 간략히 보여준다.

MCP 기반 멀티 AI 에이전트 아키텍처 개념도

MCP의 개념과 상세한 동작 방식은 다음 글을 참고한다.

개요
MCP는 클라이언트-서버 구조를 따른다. 클라이언트는 서버의 MCP 도구를 사용하는 AI 앱이나 LLM을 의미한다. 서버는 MCP 도구를 공급하고, API, 데이터소스 인터페이스를 제공한다. 

MCP를 통해 LLM이 해결하지 못하는 작업은 외부 시스템과 연결해 서비스 받을 수 있다.

MCP서버는 파일 시스템 조작, 웹 검색, 데이터베이스 조작, 버전 관리 등 다양한 도구를 제공할 수 있다. 

제미니 LLM 기반 MCP 구조
다음은 제미니 LLM 기반 MCP 구조 예시를 보여준다. 이 예는 비행기 예약 유스케이스를 구현한다.
구조의 각 번호는 시퀀스 시나리오를 보여준다. 이 내용은 다음과 같다.
  1. MCP 호스트가 사용자 명령 입력. 예) 내일 인천에서 애틀란타 가는 비행편 찾기
    1. 클라이언트 스크립트가 입력을 처리(CLIENT.PY)
  2. 클라이언트가 MCP 서버 프로세스 시작(MCP-FLIGHT-SEARCH). STDIO 통신 채널 연결 및 관련 도구 검색
  3. 클라이언트가 사용자 명령에 대한 함수 호출 방법을 수신함
  4. 클라리언트가 함수 호출 방법에 대한 정확한 함수 호출 형식을 GEMINI에서 획득. 함수 호출 형식에 부합하는 적절한 MCP 도구를 서버에 호출. 서버의 도구 함수 호출 결과를 리턴
  5. MCP 서버가 구글 항공편 검색을 위한 SerpAPI를 호출. 구글 항공편 데이터 질의.
  6. 구글 항공편 정보 리턴
  7. 서버에서 클라이언트로 해당 정보 리턴
  8. 클라이언트가 호스로 해당 정보 전달
개발 환경
개발을 위한 최소한의 환경은 파이썬 3.8+이다. 이외 다음을 준비한다.

다음 종속성을 터미널에서 설치한다. google-genai는 google 생성AI 라이브러리이며, mcp는 MCP 서버 통신을 위한 파이썬 SDK이다. 
pip install google-genai mcp

환경변수를 설정한다. 
export GEMINI_API_KEY="your-google-api-key"
export SERP_API_KEY="your-serpapi-key"

항공편 검색 MCP 서버 설치
MCP 프로토콜 공개 이후로 많은 MCP 서버가 개발되었다. 우리는 항공편 검색 MCP 서버 오픈소스인 mcp-flgiht-search 를 사용한다. 다음을 설치한다.
pip install mcp-flight-search

코딩해보기
다음과 같이 client.py를 코딩한다. 
import os, sys, time, asyncio
from google import genai
from google.genai import types
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from dotenv import load_dotenv

load_dotenv()

gemini_api_key = os.getenv("GEMINI_API_KEY")
serp_api_key = os.getenv("SERP_API_KEY")

client = genai.Client(api_key=gemini_api_key)

server_params = StdioServerParameters(
    command="mcp-flight-search",
    args=["--connection_type", "stdio"],
    env={"SERP_API_KEY": serp_api_key},
)

async def run():
    async with stdio_client(server_params) as (read, write): # 항공 예약 검색 도구 등록
        async with ClientSession(read, write) as session:
            prompt = f"Find Flights from Atlanta to Las Vegas 2025-08-15" # 사용자 질의 명령
            await session.initialize()

            mcp_tools = await session.list_tools() # 도구 리스트 획득
            tools = [
                types.Tool(
                    function_declarations=[
                        {
                            "name": tool.name,
                            "description": tool.description,
                            "parameters": {
                                k: v
                                for k, v in tool.inputSchema.items()
                                if k not in ["additionalProperties", "$schema"]
                            },
                        }
                    ] # 해당 도구 함수 선언 생성
                )
                for tool in mcp_tools.tools
            ]

            response = client.models.generate_content(
                model="gemini-2.5-pro-exp-03-25",
                contents=prompt,
                config=types.GenerateContentConfig(
                    temperature=0,
                    tools=tools,
                ), # LLM 모델에 프롬프트 전달.
            )

            if response.candidates[0].content.parts[0].function_call:
                function_call = response.candidates[0].content.parts[0].function_call # 함수호출정보

                result = await session.call_tool(
                    function_call.name, arguments=dict(function_call.args)
                ) # 도구 함수 호출

                print("--- Formatted Result ---") # Add header for clarity
                try:
                    flight_data = json.loads(result.content[0].text)
                    print(json.dumps(flight_data, indent=2))
                except json.JSONDecodeError:
                    print("MCP server returned non-JSON response:")
                    print(result.content[0].text)
                except (IndexError, AttributeError):
                     print("Unexpected result structure from MCP server:")
                     print(result)
            else:
                print("No function call was generated by the model.")
                if response.text:
                     print("Model response:")
                     print(response.text)

asyncio.run(run()) # 클라이언트 실행

실행한다. 그럼 프롬프트에 대해 LLM이 적절한 도구와 파라메터를 확인해 함수 호출 정보를 생성한다. 이를 call_tool로 호출한 결과가 표시된다 

레퍼런스


시간:

댓글 없음:

댓글 쓰기