Gemini 3.5 Flash + Interactions API로 MCP 에이전트 만들기 — 완전 실전 가이드

지난 편에서 Interactions API의 구조와 브레이킹 체인지를 다뤘습니다. 이번 편은 코드만 남습니다. MCP 서버를 Flash에 연결하고, 멀티툴 체인을 구성하고, 에이전트 루프를 프로덕션에서 돌리는 전 과정입니다.

---

핵심 요약 → 이전 편 복습: Interactions API = 서버사이드 히스토리, previous_interaction_id 패턴 → 이번 편: MCP 서버 연결 → 멀티툴 체인 → 에이전트 루프 → 비용 최적화 → 에러 핸들링 → Gemini SDK의 MCP 통합: ClientSession을 tools= 파라미터에 전달하면 자동 루프 실행 → Flash MCP Atlas 83.6% = 6번에 1번은 도구 호출 실패 — 재시도 로직 필수 → thinking_level: Low가 에이전트 루프 최적 (빠르고 싸면서 SWE 성능 Medium 수준) → 멀티툴 동시 사용: Google Search + 커스텀 MCP + File Search 조합 실전 코드 포함

---

사전 준비

# SDK 설치 (2.0.0 이상 필수)
pip install -U google-genai

# MCP SDK 설치
pip install mcp fastmcp

# 환경 변수
export GEMINI_API_KEY="your_api_key"

---

1. 가장 단순한 MCP 연결 — 원격 MCP 서버

import google.generativeai as genai
import datetime

client = genai.Client()
MODEL = "gemini-3.5-flash"

# ── 원격 MCP 서버 연결 (가장 간단한 형태) ──

def simple_mcp_agent(user_query: str) -> str:
    """
    원격 MCP 서버를 도구로 사용하는 기본 에이전트
    """
    mcp_server = {
        "type": "mcp_server",
        "name": "weather_service",
        "url": "https://your-mcp-server.example.com/mcp"
        # stdio, SSE, HTTP 모두 지원
    }

    today = datetime.date.today().strftime("%Y년 %m월 %d일")

    interaction = client.interactions.create(
        model=MODEL,
        input=user_query,
        tools=[mcp_server],
        config={
            "thinking_level": "low",       # 에이전트 루프 최적값
            "system_instruction": f"오늘 날짜는 {today}입니다."
        }
        # store=True 기본값 → 서버가 55일 보관
    )

    # steps 배열에서 최종 응답 추출 (신규 스키마)
    for step in interaction.steps:
        if step.type == "model_output":
            return step.content[0].text

    return ""

# 실행
result = simple_mcp_agent("서울 지금 날씨 어때?")
print(result)

---

2. 로컬 MCP 서버 — FastMCP로 직접 만들기

# ── 1단계: FastMCP로 커스텀 MCP 서버 정의 ──

from fastmcp import FastMCP
import httpx

mcp = FastMCP("내 서비스 에이전트")

@mcp.tool()
async def search_internal_docs(query: str) -> str:
    """사내 문서 데이터베이스에서 관련 내용을 검색합니다."""
    # 실제 구현: 벡터 DB 조회, Elasticsearch 등
    response = await httpx.AsyncClient().post(
        "http://internal-search.company.com/search",
        json={"q": query, "limit": 5}
    )
    results = response.json()["results"]
    return "\n\n".join([r["content"] for r in results])

@mcp.tool()
async def get_customer_info(customer_id: str) -> dict:
    """고객 ID로 고객 정보를 조회합니다."""
    # 실제 구현: 내부 DB 조회
    response = await httpx.AsyncClient().get(
        f"http://internal-api.company.com/customers/{customer_id}"
    )
    return response.json()

@mcp.tool()
async def create_support_ticket(
    customer_id: str,
    issue: str,
    priority: str = "normal"
) -> dict:
    """고객 지원 티켓을 생성합니다."""
    response = await httpx.AsyncClient().post(
        "http://internal-api.company.com/tickets",
        json={
            "customer_id": customer_id,
            "issue": issue,
            "priority": priority
        }
    )
    return {"ticket_id": response.json()["id"], "status": "created"}


# ── 2단계: FastMCP 클라이언트로 Gemini에 연결 ──

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
import google.generativeai as genai

client = genai.Client()

async def agent_with_local_mcp(user_query: str) -> str:
    """로컬 MCP 서버를 사용하는 에이전트"""

    # stdio transport로 로컬 MCP 서버 연결
    server_params = StdioServerParameters(
        command="python",
        args=["my_mcp_server.py"]
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # ClientSession을 tools에 직접 전달
            # → SDK가 자동으로 도구 목록 가져오고 루프 실행
            interaction = client.interactions.create(
                model="gemini-3.5-flash",
                input=user_query,
                tools=session,           # ← ClientSession 직접 전달
                config={"thinking_level": "low"}
            )

            for step in interaction.steps:
                if step.type == "model_output":
                    return step.content[0].text

---

3. 멀티툴 체인 — Google Search + 커스텀 MCP + File Search

# ── 세 가지 도구를 동시에 사용하는 에이전트 ──

async def multi_tool_agent(user_query: str, session: ClientSession) -> str:
    """
    Google Search + 커스텀 MCP + File Search를 동시에 사용
    Flash MCP Atlas 83.6% 수치의 실제 활용 패턴
    """

    interaction = client.interactions.create(
        model="gemini-3.5-flash",
        input=user_query,
        tools=[
            # 도구 1: Google Search (실시간 웹 검색)
            {"google_search": {}},

            # 도구 2: 커스텀 MCP 서버 (내부 시스템)
            session,  # ClientSession

            # 도구 3: File Search (사내 문서 벡터 검색)
            {
                "file_search": {
                    "corpus_id": "company-docs-2026"
                }
            }
        ],
        config={
            "thinking_level": "medium",  # 멀티툴은 medium 권장
            "system_instruction": """
                당신은 고객 지원 에이전트입니다.
                1. 사내 문서에서 관련 정책을 먼저 확인하세요
                2. 필요시 내부 시스템에서 고객 정보를 조회하세요
                3. 최신 정보가 필요하면 웹 검색을 활용하세요
                항상 근거를 명시하고 한국어로 답변하세요.
            """
        }
    )

    # steps 전체 출력 (디버깅용)
    for step in interaction.steps:
        print(f"[{step.type}]", end=" ")
        if step.type == "google_search_call":
            print(f"검색어: {step.query}")
        elif step.type == "function_call":
            print(f"함수: {step.name}({step.args})")
        elif step.type == "model_output":
            print("→ 최종 응답")
            return step.content[0].text
        else:
            print()

    return ""

---

4. 멀티턴 에이전트 루프 — previous_interaction_id 활용

# ── 상태를 유지하는 멀티턴 에이전트 ──

class GeminiMCPAgent:
    """
    previous_interaction_id로 컨텍스트를 유지하는 에이전트
    히스토리 관리 코드 없음 — 서버가 전담
    """

    def __init__(self, mcp_session: ClientSession):
        self.client = genai.Client()
        self.session = mcp_session
        self.previous_id: str | None = None
        self.turn_count = 0

    def chat(self, user_message: str) -> str:
        self.turn_count += 1

        interaction = self.client.interactions.create(
            model="gemini-3.5-flash",
            input=user_message,
            previous_interaction_id=self.previous_id,  # 컨텍스트 연결
            tools=[
                self.session,
                {"google_search": {}}
            ],
            config={
                # 에이전트 루프는 Low가 최적
                # → Medium과 SWE 성능 비슷하면서 토큰 45% 절감
                "thinking_level": "low",
            }
        )

        # 다음 턴을 위해 ID 저장
        self.previous_id = interaction.id

        # 최종 응답 추출
        for step in interaction.steps:
            if step.type == "model_output":
                return step.content[0].text

        return ""

    def reset(self):
        """대화 컨텍스트 초기화"""
        self.previous_id = None
        self.turn_count = 0


# 사용 예시
async def main():
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            agent = GeminiMCPAgent(mcp_session=session)

            # 대화 1: 고객 조회
            r1 = agent.chat("고객 ID C-12345 정보 조회해줘")
            print(f"에이전트: {r1}")

            # 대화 2: 이전 컨텍스트 기반 (히스토리 자동 유지)
            r2 = agent.chat("이 고객 최근 문의 내역도 보여줘")
            print(f"에이전트: {r2}")

            # 대화 3: 도구 호출 연계
            r3 = agent.chat("이 고객 이름으로 긴급 티켓 하나 만들어줘")
            print(f"에이전트: {r3}")

---

5. 에러 핸들링 — MCP Atlas 83.6%의 현실적 대응

import asyncio
import logging
from typing import Optional

logger = logging.getLogger(__name__)

# MCP Atlas 83.6% = 약 16%의 도구 호출 실패 가능성
# 프로덕션에서는 재시도 + fallback 필수

async def robust_mcp_agent(
    user_query: str,
    session: ClientSession,
    max_retries: int = 3
) -> Optional[str]:
    """
    재시도 로직이 포함된 프로덕션 수준 에이전트
    """
    last_error = None

    for attempt in range(max_retries):
        try:
            interaction = client.interactions.create(
                model="gemini-3.5-flash",
                input=user_query,
                tools=[session, {"google_search": {}}],
                config={"thinking_level": "low"}
            )

            # 실패 스텝 감지
            failed_steps = [
                s for s in interaction.steps
                if s.type == "function_call" and
                   getattr(s, "status", None) == "error"
            ]

            if failed_steps:
                logger.warning(
                    f"시도 {attempt+1}: 도구 호출 실패 "
                    f"({len(failed_steps)}개) — 재시도"
                )
                # 지수 백오프
                await asyncio.sleep(2 ** attempt)
                continue

            # 성공
            for step in interaction.steps:
                if step.type == "model_output":
                    return step.content[0].text

        except Exception as e:
            last_error = e
            logger.error(f"시도 {attempt+1} 오류: {e}")
            await asyncio.sleep(2 ** attempt)

    # 모든 재시도 실패 → fallback: 도구 없이 응답
    logger.warning("모든 재시도 실패 — 도구 없이 응답")
    try:
        fallback = client.interactions.create(
            model="gemini-3.5-flash",
            input=f"{user_query}\n(주의: 실시간 데이터 조회 불가, 일반 지식으로만 답변)",
            config={"thinking_level": "low"}
        )
        for step in fallback.steps:
            if step.type == "model_output":
                return f"[도구 조회 실패, 일반 응답]\n{step.content[0].text}"
    except Exception as fe:
        logger.error(f"Fallback도 실패: {fe}")
        return None

    return None

---

6. 비용 최적화 — thinking_level 라우팅

# ── 태스크 복잡도에 따른 thinking_level 자동 선택 ──

def get_thinking_level(task_type: str) -> str:
    """
    태스크 유형별 최적 thinking_level 선택
    토큰 비용 비교:
      Low    → 기준치
      Medium → Low 대비 ~2배
      High   → Low 대비 ~4배
    """
    routing = {
        # Low: 단순 조회, 빠른 응답 필요
        "simple_lookup":    "low",   # 고객 정보 조회
        "classification":   "low",   # 문의 유형 분류
        "data_extraction":  "low",   # 구조화 데이터 추출
        "routing":          "low",   # 다음 단계 결정

        # Medium: 일반 에이전트 루프 (기본값)
        "multi_tool":       "medium", # 멀티툴 조합
        "summarization":    "medium", # 문서 요약
        "code_review":      "medium", # 코드 리뷰

        # High: 복잡한 추론이 필요한 경우만
        "complex_analysis": "high",  # 심층 분석
        "hard_coding":      "high",  # 어려운 알고리즘
        "math_reasoning":   "high",  # 수학적 추론
    }
    return routing.get(task_type, "medium")  # 기본값 medium


async def cost_optimized_agent(
    user_query: str,
    task_type: str,
    session: ClientSession,
    previous_id: str | None = None
) -> tuple[str, str]:  # (응답, interaction_id)
    """
    비용 최적화된 에이전트
    Low 설정 시 Medium 대비 토큰 45% 절감
    """
    level = get_thinking_level(task_type)

    interaction = client.interactions.create(
        model="gemini-3.5-flash",
        input=user_query,
        previous_interaction_id=previous_id,
        tools=[session],
        config={"thinking_level": level}
    )

    # 토큰 사용량 모니터링
    if hasattr(interaction, "usage"):
        logger.info(
            f"thinking={level} | "
            f"input={interaction.usage.input_tokens} | "
            f"output={interaction.usage.output_tokens} | "
            f"thoughts={getattr(interaction.usage, 'thought_tokens', 0)}"
        )

    for step in interaction.steps:
        if step.type == "model_output":
            return step.content[0].text, interaction.id

    return "", interaction.id

---

7. 스트리밍 + 백그라운드 — 프로덕션 패턴

# ── 스트리밍: 실시간 응답 표시 ──

def streaming_mcp_agent(user_query: str, session: ClientSession):
    """
    토큰이 생성되는 즉시 출력
    인터랙티브 UI에 적합
    """
    for event in client.interactions.create(
        model="gemini-3.5-flash",
        input=user_query,
        tools=[session, {"google_search": {}}],
        config={"thinking_level": "low"},
        stream=True
    ):
        if event.type == "step.delta":
            if event.delta.type == "model_output":
                # 토큰 단위로 즉시 출력
                print(event.delta.text, end="", flush=True)
        elif event.type == "step.complete":
            if event.step.type == "function_call":
                print(f"\n[도구 호출: {event.step.name}]")
    print()  # 줄바꿈


# ── 백그라운드: 장시간 실행 에이전트 ──

import time

async def background_analysis_agent(
    large_dataset_query: str,
    webhook_url: str | None = None
) -> str:
    """
    수 분 이상 걸리는 태스크를 비동기로 실행
    Webhook URL 설정 시 완료 후 콜백
    """
    # 시작 (즉시 반환)
    interaction = client.interactions.create(
        model="gemini-3.5-flash",
        input=large_dataset_query,
        config={
            "thinking_level": "high",
            "background": True,   # 비동기 실행
        }
        # webhook_url=webhook_url  # 완료 시 POST 콜백
    )

    print(f"분석 시작됨 (ID: {interaction.id})")
    print("백그라운드에서 실행 중...")

    # 폴링 (Webhook 없는 경우)
    while True:
        updated = client.interactions.get(interaction.id)

        if updated.status == "completed":
            for step in updated.steps:
                if step.type == "model_output":
                    return step.content[0].text
            break

        elif updated.status in ["failed", "cancelled"]:
            raise Exception(f"에이전트 실패: {updated.status}")

        print(f"  진행 중... ({updated.status})")
        await asyncio.sleep(10)  # 10초 간격 폴링

    return ""

---

8. 실전 완성 예제 — 고객 지원 에이전트

# ── 모든 패턴을 통합한 실전 고객 지원 에이전트 ──

from fastmcp import FastMCP
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
import google.generativeai as genai
import asyncio

# MCP 서버 정의
support_mcp = FastMCP("고객지원 서비스")

@support_mcp.tool()
async def get_customer(customer_id: str) -> dict:
    """고객 정보 조회"""
    # 실제 DB 조회
    return {"id": customer_id, "name": "홍길동", "plan": "Premium"}

@support_mcp.tool()
async def get_order_history(customer_id: str, limit: int = 5) -> list:
    """주문 이력 조회"""
    return [{"order_id": f"ORD-{i}", "status": "완료"} for i in range(limit)]

@support_mcp.tool()
async def create_ticket(customer_id: str, issue: str, priority: str) -> dict:
    """지원 티켓 생성"""
    return {"ticket_id": "TKT-99999", "status": "접수", "priority": priority}


# 에이전트 메인
async def customer_support_agent():
    server_params = StdioServerParameters(
        command="python", args=["support_mcp_server.py"]
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            gemini = genai.Client()
            previous_id = None

            print("고객 지원 에이전트 시작 (종료: 'q')")
            print("-" * 50)

            while True:
                user_input = input("고객: ").strip()
                if user_input.lower() == 'q':
                    break

                # 태스크 분류 기반 thinking_level 선택
                level = "low" if len(user_input) < 50 else "medium"

                try:
                    interaction = gemini.interactions.create(
                        model="gemini-3.5-flash",
                        input=user_input,
                        previous_interaction_id=previous_id,
                        tools=[
                            session,               # 내부 MCP 도구
                            {"google_search": {}}, # 외부 검색
                        ],
                        config={
                            "thinking_level": level,
                            "system_instruction": """
                                당신은 친절한 고객 지원 담당자입니다.
                                - 고객 정보 조회 시 항상 ID를 먼저 확인하세요
                                - 문제 해결 후 만족도를 확인하세요
                                - 긴급한 경우 우선순위 '긴급'으로 티켓을 생성하세요
                            """
                        }
                    )

                    previous_id = interaction.id

                    # 도구 호출 과정 출력
                    for step in interaction.steps:
                        if step.type == "function_call":
                            print(f"  [내부 조회: {step.name}]")
                        elif step.type == "google_search_call":
                            print(f"  [웹 검색: {step.query}]")
                        elif step.type == "model_output":
                            print(f"상담원: {step.content[0].text}")

                except Exception as e:
                    print(f"오류 발생: {e} — 재시도하세요")

if __name__ == "__main__":
    asyncio.run(customer_support_agent())

---

핵심 패턴 요약

# 언제 어떤 패턴을 쓰나

단순 단건 도구 호출
  → simple_mcp_agent() — MCP 서버 URL 하나

멀티툴 복합 조회
  → multi_tool_agent() — Search + MCP + File Search

대화 유지가 필요한 에이전트
  → GeminiMCPAgent 클래스 — previous_interaction_id 관리

프로덕션 안정성 필요
  → robust_mcp_agent() — 재시도 + fallback

실시간 응답 필요 (UI)
  → streaming_mcp_agent() — stream=True

수 분 이상 걸리는 분석
  → background_analysis_agent() — background=True

비용 최적화
  → cost_optimized_agent() — 태스크별 thinking_level 라우팅

---

결론

✅ Flash + Interactions API + MCP 조합이 강한 이유

• previous_interaction_id → 히스토리 코드 제거, 멀티턴 비용 절감
• MCP 자동 루프 → ClientSession만 넘기면 SDK가 도구 탐색·실행 자동 처리
• thinking_level Low → 에이전트 루프에서 Medium과 SWE 성능 비슷하면서 45% 저렴
• 289 tokens/sec → 루프 빠르게 돌면서 UX 체감

❌ 잊으면 안 되는 것

• MCP Atlas 83.6% = 16% 실패 가능성 → 재시도 로직 필수
• store=True 기본값 → 민감 데이터는 store=False 명시
• Remote MCP 아직 미지원 → stdio/SSE 로컬 서버 방식 사용
• 6월 8일 outputs → steps 스키마 변경 완료 필요

---

관련 글

• Gemini Interactions API 완전가이드 — 1편: 구조·브레이킹 체인지