실습 4 — 상품 자료 MCP 서버 만들고 붙이기

from pathlib import Path

# 서버 코드를 글로 적어 파일로 저장한다. r''' ''' 는 \n 같은 글자를 그대로 두는 글자열이다
SERVER_CODE = r'''
"""구름다리여행 자료 MCP 서버 (5차시 실습 4가 만든 파일)"""
from pathlib import Path

import pandas as pd
from mcp.server.fastmcp import FastMCP

# 이 파일이 있는 폴더(data)만 도구가 볼 수 있는 곳으로 정한다
BASE = Path(__file__).resolve().parent
# 그 폴더 안에서도 이 목록에 적은 파일만 내준다
SHARED = ["trip_products.csv", "trip_notice.txt", "faq.csv"]

mcp = FastMCP("trip-data")


@mcp.tool()
def list_files() -> str:
    """이 서버가 읽어 줄 수 있는 파일의 이름 목록을 돌려준다."""
    return ", ".join(SHARED)


@mcp.tool()
def read_file(filename: str, max_lines: int = 5) -> str:
    """공유한 파일 하나를 앞에서부터 max_lines 줄까지 읽어 돌려준다. filename 은 list_files 가 알려 준 이름이다."""
    path = (BASE / filename).resolve()
    # 1겹: data 폴더 밖을 가리키는 경로(../.env, C:/Windows/...)는 거절한다
    if not path.is_relative_to(BASE):
        return f"거절: 허락된 폴더 밖의 파일입니다 ({filename})"
    # 2겹: 폴더 안이라도 공유 목록에 없는 파일은 거절한다
    if path.name not in SHARED:
        return f"거절: 공유 목록에 없는 파일입니다 ({filename})"
    lines = path.read_text(encoding="utf-8-sig").splitlines()
    return "\n".join(lines[:max_lines])


@mcp.tool()
def find_products(region: str = "", max_price: int = 0) -> str:
    """구름다리여행 상품을 찾는다. region 은 지역 이름(예: 제주, 부산), max_price 는 어른 1명 가격의 상한(만 원)이다. 0 이면 따지지 않는다."""
    df = pd.read_csv(BASE / "trip_products.csv")
    if region:
        df = df[df["지역"] == region]
    if max_price > 0:
        df = df[df["가격"] <= max_price]
    if df.empty:
        return "조건에 맞는 상품이 없습니다."
    return "\n".join(
        f"{r.상품코드} {r.상품명} | {r.가격}만 원 | 남은 자리 {r.남은자리} | 출발 {r.출발일}" for r in df.itertuples()
    )


if __name__ == "__main__":
    # 기본 전송 방식은 stdio: 표준 입력으로 요청을 받고 표준 출력으로 답한다
    mcp.run()
'''

Path("out_ch05_server.py").write_text(SERVER_CODE.lstrip(), encoding="utf-8")
print("저장: out_ch05_server.py,", len(SERVER_CODE.strip().splitlines()), "줄")

import asyncio
import json
import sys

from langchain_mcp_adapters.client import MultiServerMCPClient

# 서버 이름 "trip": 지금 쓰는 파이썬(sys.executable)으로 out_ch05_server.py 를 켜고 stdio 로 잇는다
client = MultiServerMCPClient({
    "trip": {"command": sys.executable, "args": ["out_ch05_server.py"], "transport": "stdio"},
})


async def main():
    # get_tools: 서버를 켜서 도구 목록(tools/list)을 받아 LangChain 도구로 바꾼다
    tools = await client.get_tools()
    for t in tools:
        print(t.name, "|", type(t).__name__)
        print("  설명:", t.description)
        print("  인자:", {k: v["type"] for k, v in t.args.items()})
    return tools


mcp_tools = asyncio.run(main())
info = [{"name": t.name, "description": t.description, "args": t.args} for t in mcp_tools]
Path("out_ch05_mcp_tools.json").write_text(json.dumps(info, ensure_ascii=False, indent=2), encoding="utf-8")
print("저장: out_ch05_mcp_tools.json")

def text_of(result):
    """MCP 도구의 결과(글 조각 목록)에서 글만 꺼내 잇는다. 조각의 id 는 부를 때마다 바뀌어 버린다"""
    if isinstance(result, str):
        return result
    return "\n".join(block["text"] for block in result if block.get("type") == "text")


async def main():
    tools = {t.name: t for t in await client.get_tools()}
    # MCP 도구는 비동기라 invoke 대신 ainvoke 를 await 로 부른다
    raw = await tools["find_products"].ainvoke({"region": "제주", "max_price": 40})
    print("결과의 모양:", type(raw).__name__, len(raw), "조각,", list(raw[0]))
    print(text_of(raw))
    print("---")
    print(text_of(await tools["list_files"].ainvoke({})))
    print("---")
    print(text_of(await tools["read_file"].ainvoke({"filename": "trip_notice.txt"})))


asyncio.run(main())

async def main():
    tools = {t.name: t for t in await client.get_tools()}
    tries = ["faq.csv", "../.env", "../content/lesson01.md", "C:/Windows/win.ini", "staff_memo.txt"]
    for name in tries:
        answer = text_of(await tools["read_file"].ainvoke({"filename": name, "max_lines": 2}))
        # 허락된 파일은 첫 줄만 보여 준다
        print(f"{name:24} → {answer.splitlines()[0]}")


asyncio.run(main())