mcp_server/server.py

"""AutoSource warehouse MCP server (plan §7.1).

The ONLY component that touches the SQLite DB. Exposes exactly the five
contracted tools. `cost_floor` and internal thresholds never leave this process
via supplier-facing tools (a separate _floor lookup exists for the in-process
vendor validator, which is part of the simulation's ground truth, not an agent).

Run standalone (stdio transport):  python -m mcp_server.server
"""
from __future__ import annotations

import sqlite3
import sys
import uuid
from datetime import datetime, timezone
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from mcp.server.fastmcp import FastMCP  # noqa: E402

from core.config import db_path  # noqa: E402

mcp = FastMCP("autosource-warehouse")


def _conn() -> sqlite3.Connection:
    conn = sqlite3.connect(db_path())
    conn.row_factory = sqlite3.Row
    return conn


@mcp.tool()
def get_stock(sku: str) -> dict:
    """Current stock position for one SKU."""
    with _conn() as c:
        row = c.execute(
            "SELECT sku, qty_on_hand, reorder_point, status FROM inventory WHERE sku=?",
            (sku,),
        ).fetchone()
    if row is None:
        return {"error": f"unknown sku {sku!r}"}
    return dict(row)


@mcp.tool()
def list_low_stock() -> list[dict]:
    """All items below their reorder point (status='low')."""
    with _conn() as c:
        rows = c.execute(
            """SELECT sku, name, qty_on_hand, reorder_point, reorder_qty, list_price
               FROM inventory WHERE status='low'"""
        ).fetchall()
    return [dict(r) for r in rows]


@mcp.tool()
def get_suppliers_for(sku: str) -> list[dict]:
    """Vendors stocking a SKU, with their vendor memory. cost_floor is NEVER returned."""
    with _conn() as c:
        rows = c.execute(
            """SELECT s.vendor_id, s.name, s.persona, ss.base_price,
                      s.lead_time_days, s.reliability,
                      vm.last_price AS mem_last_price, vm.rounds AS mem_rounds,
                      vm.settled AS mem_settled, vm.reliability_obs AS mem_reliability_obs
               FROM supplier_skus ss
               JOIN suppliers s ON s.vendor_id = ss.vendor_id
               LEFT JOIN vendor_memory vm ON vm.vendor_id = ss.vendor_id AND vm.sku = ss.sku
               WHERE ss.sku = ? AND ss.in_stock = 1
               ORDER BY ss.base_price ASC""",
            (sku,),
        ).fetchall()
    return [dict(r) for r in rows]


@mcp.tool()
def log_purchase_order(sku: str, vendor_id: str, qty: int, unit_price: float,
                       approved_by: str) -> dict:
    """Write an approved PO and mark the item on-order. Only called after human approval."""
    po_id = f"PO-{datetime.now(timezone.utc).strftime('%Y%m%d')}-{uuid.uuid4().hex[:8].upper()}"
    total = round(qty * unit_price, 2)
    with _conn() as c:
        c.execute(
            "INSERT INTO purchase_orders VALUES (?,?,?,?,?,?,?,?,?)",
            (po_id, sku, vendor_id, qty, unit_price, total, "created",
             datetime.now(timezone.utc).isoformat(), approved_by),
        )
        c.execute("UPDATE inventory SET status='on_order' WHERE sku=?", (sku,))
    return {"po_id": po_id, "status": "created", "total": total}


@mcp.tool()
def update_vendor_memory(vendor_id: str, sku: str, last_price: float, rounds: int,
                         settled: bool, reliability_obs: float) -> dict:
    """Persist negotiation outcome so future runs learn vendor behavior."""
    with _conn() as c:
        c.execute(
            """INSERT INTO vendor_memory (vendor_id, sku, last_price, rounds, settled,
                                          reliability_obs, updated_at)
               VALUES (?,?,?,?,?,?,?)
               ON CONFLICT(vendor_id, sku) DO UPDATE SET
                 last_price=excluded.last_price, rounds=excluded.rounds,
                 settled=excluded.settled, reliability_obs=excluded.reliability_obs,
                 updated_at=excluded.updated_at""",
            (vendor_id, sku, last_price, rounds, int(settled), reliability_obs,
             datetime.now(timezone.utc).isoformat()),
        )
    return {"ok": True}


def get_cost_floor(vendor_id: str) -> float:
    """Ground-truth floor for the deterministic vendor validator (plan §9).

    NOT an MCP tool — never exposed to any agent. Imported only by the
    negotiation engine's validator, which plays the vendor's own 'books'.
    """
    with _conn() as c:
        row = c.execute(
            "SELECT cost_floor FROM suppliers WHERE vendor_id=?", (vendor_id,)
        ).fetchone()
    if row is None:
        raise KeyError(f"unknown vendor {vendor_id!r}")
    return float(row["cost_floor"])


if __name__ == "__main__":
    mcp.run()  # stdio transport