프로젝트 1 — 사내 재고 시스템 연결

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"))
RESERVED = []                         # 이 서버가 떠 있는 동안의 출고 예약
FAILS_LEFT = {"search_items": 1}      # 세션의 첫 찾기는 일시적 오류를 낸다(불안정한 시스템 흉내)
mcp = MCPServer("솔내 재고")

@mcp.tool()
def search_items(keyword: str, warehouse: str = "") -> dict:
    """품목 이름이나 코드에 keyword 가 든 재고 행을 돌려준다. warehouse(본사 창고, 물류 센터)를 주면 그 창고만, 비우면 모든 창고. 수량, 최소수량, 단가를 함께 준다."""
    if FAILS_LEFT["search_items"] > 0:
        FAILS_LEFT["search_items"] -= 1
        raise ConnectionError("일시적인 연결 오류")          # 일시적 오류는 예외로
    rows = STOCK[STOCK["품목"].str.contains(keyword, regex=False) | STOCK["코드"].str.contains(keyword, regex=False)]
    if warehouse:
        rows = rows[rows["창고"] == warehouse]
    return {"찾은 행": rows.to_dict("records")}

@mcp.tool()
def low_stock(warehouse: str) -> dict:
    """창고(본사 창고 또는 물류 센터)에서 수량이 최소수량보다 적은 품목을 돌려준다."""
    rows = STOCK[(STOCK["창고"] == warehouse) & (STOCK["수량"] < STOCK["최소수량"])]
    return {"부족한 품목": rows[["코드", "품목", "수량", "최소수량"]].to_dict("records")}

@mcp.tool()
def reserve(item: str, warehouse: str, quantity: int) -> dict:
    """창고의 품목을 출고 예약한다. item 은 품목 코드(예: E-401) 또는 품목 이름의 일부. 재고가 모자라거나 그 창고에 없으면 오류를 돌려준다."""
    rows = STOCK[((STOCK["코드"] == item) | STOCK["품목"].str.contains(item, regex=False)) & (STOCK["창고"] == warehouse)]
    if rows.empty:
        return {"오류": f"{warehouse} 에는 '{item}' 품목이 없습니다."}            # 업무 오류는 결과로
    have = int(rows["수량"].iloc[0])
    if quantity > have:
        return {"오류": f"{warehouse} 의 '{item}' 재고는 {have}개라 {quantity}개를 예약할 수 없습니다."}
    RESERVED.append({"코드": rows["코드"].iloc[0], "창고": warehouse, "수량": quantity})
    return {"예약됨": RESERVED[-1]}

@mcp.resource("stock://reservations")
def reservations() -> str:
    """지금까지의 출고 예약(JSON)"""
    import json
    return json.dumps(RESERVED, ensure_ascii=False)

if __name__ == "__main__":
    mcp.run()
'''
Path("out_inventory_server.py").write_text(SERVER, encoding="utf-8")
print("서버 파일을 썼습니다:", len(SERVER), "글자")

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

client = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")
MODEL = "qwen2.5:3b"
PARAMS = StdioServerParameters(command=sys.executable, args=["out_inventory_server.py"])
tasks = [json.loads(l) for l in open("inventory_tasks.jsonl", encoding="utf-8")]
SYSTEM = ("당신은 솔내기술(가상 회사)의 사무용품 재고 도우미입니다. 창고는 '본사 창고'와 '물류 센터' 두 곳입니다. "
          "재고는 반드시 도구로 확인하고, 도구 결과에 있는 숫자만 말하세요. 출고 예약은 사용자가 예약해 달라고 할 때만 reserve 로 합니다. "
          "도구가 오류를 돌려주면 그 내용을 읽고 다시 시도하거나 사용자에게 알리세요.")

def unload(model):
    body = json.dumps({"model": model, "keep_alive": 0}).encode()
    urllib.request.urlopen(urllib.request.Request("http://localhost:11434/api/generate", data=body,
                                                  headers={"Content-Type": "application/json"})).read()

def chat(**kw):
    """LLM 호출도 실패할 수 있다: 모델을 내렸다 다시 보낸다(7차시)"""
    for _ in range(3):
        try:
            return client.chat.completions.create(model=MODEL, temperature=0, seed=42, **kw).choices[0].message
        except APIError:
            unload(MODEL)
            time.sleep(2)
    raise RuntimeError("Ollama 가 계속 오류를 냅니다")

async def call_tool(session, name, args, retries):
    """retries 번까지 다시 부른다. 기다림은 0.5초, 1초로 늘린다(백오프). (글, 오류였나, 다시 부른 수)"""
    for k in range(retries + 1):
        r = await session.call_tool(name, args)
        text = "\n".join(c.text for c in r.content)
        if r.is_error and k < retries:
            await asyncio.sleep(0.5 * 2 ** k)
            continue
        return text, r.is_error, k

async def run_task(request, retries):
    async with stdio_client(PARAMS, errlog=open(os.devnull, "w")) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            tools = [{"type": "function", "function": {"name": t.name, "description": t.description, "parameters": t.input_schema}}
                     for t in (await session.list_tools()).tools]
            messages = [{"role": "system", "content": SYSTEM}, {"role": "user", "content": request}]
            calls, answer = [], ""
            for _ in range(6):
                msg = chat(messages=messages, tools=tools)
                if not msg.tool_calls:
                    answer = msg.content or ""
                    break
                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:
                    args = json.loads(tc.function.arguments)
                    text, is_error, again = await call_tool(session, tc.function.name, args, retries)
                    calls.append((tc.function.name, is_error, again))
                    messages.append({"role": "tool", "tool_call_id": tc.id, "content": text})
            booked = json.loads((await session.read_resource("stock://reservations")).contents[0].text)   # 실제 상태
            return answer, calls, booked

CLAIM = ["예약했", "예약이 완료", "예약되었", "예약 완료"]

def judge(task, answer, booked):
    key_ok = any(k in answer for k in task["정답열쇠"].split("|"))
    want = task["예약"]
    state_ok = (booked == []) if want is None else (len(booked) == 1 and [booked[0]["코드"], booked[0]["창고"], booked[0]["수량"]] == want)
    false_claim = any(c in answer for c in CLAIM) and not booked       # 말로는 예약했는데 목록이 비었다
    return key_ok and state_ok and not false_claim, false_claim

answer, calls, booked = asyncio.run(run_task(tasks[0]["요청"], retries=2))
print("첫 요청:", tasks[0]["요청"], "→", answer[:80], "| 도구:", calls)

rows, logs = [], {}
for retries in [0, 2]:
    start, ok, lies, errors_seen = time.time(), 0, 0, 0
    logs[retries] = []
    for t in tasks:
        answer, calls, booked = asyncio.run(run_task(t["요청"], retries))
        good, false_claim = judge(t, answer, booked)
        ok, lies = ok + good, lies + false_claim
        errors_seen += sum(e for _, e, _ in calls)                     # 모델이 받은 오류 결과 수
        logs[retries].append((t["요청"], good, false_claim, [c[0] for c in calls], booked, answer))
    rows.append({"설정": f"재시도 {retries}번", "성공(/8)": ok, "모델이 본 도구 오류": errors_seen,
                 "거짓 보고": lies, "초": round(time.time() - start)})
print(rows)

for retries, items in logs.items():
    print(f"■ 재시도 {retries}번")
    for request, good, false_claim, tools_used, booked, answer in items:
        mark = "O" if good else ("거짓" if false_claim else "X")
        print(f"  {mark:2s} {request[:24]} | {tools_used} | 예약 {booked} | {answer.replace(chr(10), ' ')[:70]}")

def safe_reply(answer, booked):
    """거짓 보고면 사용자에게 보이기 전에 바로잡는다"""
    if any(c in answer for c in CLAIM) and not booked:
        return "확인 결과 실제로 예약된 것은 없습니다. 다시 요청해 주세요. (원래 답: " + answer[:40] + "…)"
    return answer

for request, good, false_claim, tools_used, booked, answer in logs[2]:
    if false_claim:
        print("바로잡은 답:", safe_reply(answer, booked))