warehouse_mcp.py

"""Stateless MCP server (Streamable HTTP, JSON mode) — warehouse stok.

Tek POST endpoint uzerinden JSON-RPC 2.0 konusur. Oturum (session) ve SSE
stream yok: her istek bagimsiz. Bu sayede telefon/web/desktop Claude Code dahil
tum remote MCP client'lardan tek bir URL ile erisilebilir — Python kurulumu veya
lokal stdio process gerekmez.

Veri kaynagi: ayni Space'in cache'lenmis warehouse XML'i (get_cached_warehouse_xml).
Arama mantigi mcp-warehouse/server.py ile birebir ayni tutulur.
"""
from __future__ import annotations

import asyncio
import json
import xml.etree.ElementTree as ET
from typing import Any

from stock_service import get_cached_warehouse_xml

PROTOCOL_VERSION = "2025-06-18"
MAX_RESULTS = 100

TOOLS = [
    {
        "name": "search_stock",
        "description": (
            "Urun adi, SKU/urun kodu veya barkod ile stok arar. "
            "Eslesen urunlerin her depodaki adedini ve toplam stogunu doner. "
            'Ornek sorgular: "GOBIK forma", "10-01-027", "8434738174740".'
        ),
        "inputSchema": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "Aranacak urun adi, SKU/urun kodu veya barkod.",
                }
            },
            "required": ["query"],
        },
    },
    {
        "name": "list_warehouse",
        "description": (
            "Belirli bir deponun urunlerini ve adetlerini listeler. "
            'warehouse: depo adi (orn. "CADDEBOSTAN", "ALSANCAK", "BAHCEKOY"; '
            "buyuk/kucuk harf onemsiz, kismi eslesir). "
            "only_in_stock: True ise sadece stogu 0'dan buyuk urunler."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {
                "warehouse": {
                    "type": "string",
                    "description": "Depo adi (kismi eslesme, buyuk/kucuk harf onemsiz).",
                },
                "only_in_stock": {
                    "type": "boolean",
                    "description": "True ise sadece stogu 0'dan buyuk urunler.",
                    "default": False,
                },
            },
            "required": ["warehouse"],
        },
    },
]


# ---------- XML parse + arama (mcp-warehouse/server.py ile ayni) ----------
def _parse_products(xml: str) -> list[dict]:
    if not xml:
        return []
    root = ET.fromstring(xml)
    products = []
    for p in root.findall("Product"):
        warehouses = {}
        for w in p.findall("Warehouse"):
            name = (w.findtext("Name") or "").strip()
            try:
                stock = int(w.findtext("Stock") or 0)
            except ValueError:
                stock = 0
            if name:
                warehouses[name] = stock
        products.append({
            "code": (p.findtext("ProductCode") or "").strip(),
            "name": (p.findtext("ProductName") or "").strip(),
            "variant": (p.findtext("ProductVariant") or "").strip(),
            "barcode": (p.findtext("Barcode") or "").strip(),
            "total_stock": int(p.findtext("TotalStock") or 0),
            "warehouses": warehouses,
        })
    return products


def _search_stock(query: str, products: list[dict]) -> dict:
    q = query.strip().lower()
    if not q:
        return {"error": "Bos sorgu", "results": []}
    matches = []
    for p in products:
        haystack = f"{p['name']} {p['code']} {p['barcode']} {p['variant']}".lower()
        if q in haystack:
            matches.append(p)
    total = len(matches)
    return {
        "query": query,
        "match_count": total,
        "truncated": total > MAX_RESULTS,
        "results": matches[:MAX_RESULTS],
    }


def _list_warehouse(warehouse: str, only_in_stock: bool, products: list[dict]) -> dict:
    wq = warehouse.strip().lower()
    if not wq:
        return {"error": "Bos depo adi", "results": []}
    results = []
    for p in products:
        for wname, stock in p["warehouses"].items():
            if wq in wname.lower():
                if only_in_stock and stock <= 0:
                    continue
                results.append({
                    "code": p["code"],
                    "name": p["name"],
                    "variant": p["variant"],
                    "warehouse": wname,
                    "stock": stock,
                })
    total = len(results)
    return {
        "warehouse": warehouse,
        "match_count": total,
        "truncated": total > MAX_RESULTS,
        "results": results[:MAX_RESULTS],
    }


async def _run_tool(name: str, args: dict) -> dict:
    xml = await asyncio.to_thread(get_cached_warehouse_xml)
    products = _parse_products(xml or "")
    if name == "search_stock":
        return _search_stock(str(args.get("query", "")), products)
    if name == "list_warehouse":
        return _list_warehouse(
            str(args.get("warehouse", "")),
            bool(args.get("only_in_stock", False)),
            products,
        )
    raise ValueError(f"Bilinmeyen arac: {name}")


# ---------- JSON-RPC 2.0 dispatch ----------
def _result(id_: Any, result: dict) -> dict:
    return {"jsonrpc": "2.0", "id": id_, "result": result}


def _error(id_: Any, code: int, message: str) -> dict:
    return {"jsonrpc": "2.0", "id": id_, "error": {"code": code, "message": message}}


async def handle_message(msg: dict) -> dict | None:
    """Tek bir JSON-RPC mesajini isler. Notification ise None doner (HTTP 202)."""
    if not isinstance(msg, dict):
        return _error(None, -32600, "Invalid Request")

    method = msg.get("method")
    id_ = msg.get("id")

    if method == "initialize":
        client_ver = (msg.get("params") or {}).get("protocolVersion") or PROTOCOL_VERSION
        return _result(id_, {
            "protocolVersion": client_ver,
            "capabilities": {"tools": {"listChanged": False}},
            "serverInfo": {"name": "warehouse", "version": "1.0.0"},
        })

    if method == "ping":
        return _result(id_, {})

    if method == "tools/list":
        return _result(id_, {"tools": TOOLS})

    if method == "tools/call":
        params = msg.get("params") or {}
        name = params.get("name")
        args = params.get("arguments") or {}
        try:
            data = await _run_tool(name, args)
        except Exception as e:  # noqa: BLE001
            return _result(id_, {
                "content": [{"type": "text", "text": f"Hata: {e}"}],
                "isError": True,
            })
        return _result(id_, {
            "content": [{"type": "text", "text": json.dumps(data, ensure_ascii=False)}],
            "isError": False,
        })

    # notifications/* ve diger bildirimlerin id'si yoktur -> cevap verme
    if id_ is None:
        return None

    return _error(id_, -32601, f"Method not found: {method}")