실습 1 — MCP 서버 만들기

from pathlib import Path

SERVER = r'''
import os
import pandas as pd
from mcp.server.mcpserver import MCPServer

HERE = os.path.dirname(os.path.abspath(__file__))
STOCK = pd.read_csv(os.path.join(HERE, "stock.csv"))       # 서버 파일 옆의 재고 표
mcp = MCPServer("솔내 재고")

@mcp.tool()
def get_stock(code: str) -> dict:
    """품목 코드(예: P-101)의 창고별 재고 수량과 최소수량을 돌려준다."""
    rows = STOCK[STOCK["코드"] == code]
    if rows.empty:
        return {"오류": f"{code} 라는 품목 코드는 없습니다."}
    return {"품목": rows["품목"].iloc[0], "창고별": rows[["창고", "수량", "최소수량"]].to_dict("records")}

@mcp.tool()
def list_items(category: str = "") -> dict:
    """분류(용지, 토너, 필기구, 전자, 탕비)의 품목 코드와 이름 목록을 돌려준다. 비우면 모든 품목."""
    rows = STOCK if not category else STOCK[STOCK["분류"] == category]
    return {"품목": rows[["코드", "품목"]].drop_duplicates().to_dict("records")}

@mcp.resource("stock://warehouses")
def warehouses() -> str:
    """창고 목록"""
    return ", ".join(STOCK["창고"].unique())

if __name__ == "__main__":
    mcp.run()          # stdio 로 클라이언트와 대화한다
'''
Path("out_stock_server.py").write_text(SERVER, encoding="utf-8")
print("서버 파일:", Path("out_stock_server.py").stat().st_size, "바이트")

import asyncio, json, os, sys
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

# sys.executable: 지금 쓰는 파이썬(.venv)으로 서버를 띄운다
PARAMS = StdioServerParameters(command=sys.executable, args=["out_stock_server.py"])

async def show_tools():
    # errlog: 서버가 표준 오류에 남기는 기록을 버린다(출력을 깔끔하게)
    async with stdio_client(PARAMS, errlog=open(os.devnull, "w")) as (read, write):
        async with ClientSession(read, write) as session:
            init = await session.initialize()                     # 인사: 서버 이름과 판을 받는다
            print("서버:", init.server_info.name)
            for t in (await session.list_tools()).tools:          # tools/list
                print("도구:", t.name, "|", t.description)
                print("  스키마:", json.dumps(t.input_schema["properties"], ensure_ascii=False))

asyncio.run(show_tools())