실습 3 — MCP 도구를 LLM 에 잇기

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

client = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")
MODEL = "qwen2.5:3b"
PARAMS = StdioServerParameters(command=sys.executable, args=["out_stock_server.py"])

def to_openai(mcp_tool):
    """MCP 도구(name, description, input_schema)를 9장의 tools 꼴로 바꾼다"""
    return {"type": "function", "function": {"name": mcp_tool.name, "description": mcp_tool.description,
                                             "parameters": mcp_tool.input_schema}}

async def peek():
    async with stdio_client(PARAMS, errlog=open(os.devnull, "w")) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            return [to_openai(t) for t in (await session.list_tools()).tools]

print(json.dumps(asyncio.run(peek())[0], ensure_ascii=False)[:200])

SYSTEM = ("당신은 솔내기술(가상 회사)의 사무용품 재고 도우미입니다. 재고는 반드시 도구로 확인하고, "
          "도구 결과에 있는 숫자만 말하세요. 품목 코드를 모르면 list_items 로 먼저 찾으세요.")

async def ask(question, max_steps=5):
    async with stdio_client(PARAMS, errlog=open(os.devnull, "w")) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            tools = [to_openai(t) for t in (await session.list_tools()).tools]
            messages = [{"role": "system", "content": SYSTEM}, {"role": "user", "content": question}]
            for step in range(1, max_steps + 1):
                msg = client.chat.completions.create(model=MODEL, messages=messages, tools=tools,
                                                     temperature=0, seed=42).choices[0].message
                if not msg.tool_calls:
                    return msg.content
                messages.append({"role": "assistant", "content": msg.content or "",
                                 "tool_calls": [tc.model_dump() for tc in msg.tool_calls]})
                for tc in msg.tool_calls:
                    r = await session.call_tool(tc.function.name, json.loads(tc.function.arguments))   # MCP 로 실행
                    text = "\n".join(c.text for c in r.content)                    # 내용 조각을 모두 잇는다
                    print(f"  걸음 {step}: {tc.function.name}({tc.function.arguments}) → {text[:60]!r}")
                    messages.append({"role": "tool", "tool_call_id": tc.id, "content": text})
            return "[걸음 한도에 닿았습니다]"

print("답:", asyncio.run(ask("검정 토너(T-201)는 창고마다 몇 개 있어요?")))

for q in ["본사 창고에 볼펜이 몇 개 있어요?", "탕비 품목에는 무엇이 있어요?"]:
    print("질문:", q)
    print("답:", (asyncio.run(ask(q)) or "").replace("\n", " ")[:120])