"""FastMCP server with product tools exposed over stdio."""
from __future__ import annotations
import json
import logging
from pathlib import Path
from threading import Lock
from typing import Any
try:
from fastmcp import FastMCP
except ImportError: # pragma: no cover - local fallback for offline environments
class FastMCP: # type: ignore[override]
def __init__(self, name: str) -> None:
self.name = name
def tool(self, fn):
return fn
def run(self, transport: str = "stdio") -> None:
raise RuntimeError(
f"fastmcp package is required to run MCP server over {transport}"
)
LOGGER = logging.getLogger(__name__)
DATA_FILE = Path(__file__).resolve().parent / "data" / "products.json"
_FILE_LOCK = Lock()
mcp = FastMCP("product-catalog-server")
def _ensure_data_file() -> None:
DATA_FILE.parent.mkdir(parents=True, exist_ok=True)
if not DATA_FILE.exists():
DATA_FILE.write_text("[]", encoding="utf-8")
def _read_products() -> list[dict[str, Any]]:
_ensure_data_file()
with _FILE_LOCK:
raw = DATA_FILE.read_text(encoding="utf-8")
products = json.loads(raw)
if not isinstance(products, list):
raise ValueError("Product storage is corrupted: expected list")
return products
def _write_products(products: list[dict[str, Any]]) -> None:
payload = json.dumps(products, ensure_ascii=False, indent=2)
with _FILE_LOCK:
DATA_FILE.write_text(payload, encoding="utf-8")
def _next_id(products: list[dict[str, Any]]) -> int:
if not products:
return 1
return max(int(product["id"]) for product in products) + 1
@mcp.tool
def list_products(category: str | None = None) -> list[dict[str, Any]]:
"""Return all products, optionally filtered by category."""
products = _read_products()
if not category:
return products
category_lc = category.lower().strip()
return [p for p in products if str(p.get("category", "")).lower() == category_lc]
@mcp.tool
def get_product(product_id: int) -> dict[str, Any]:
"""Return a product by ID or raise ValueError if product does not exist."""
products = _read_products()
for product in products:
if int(product["id"]) == product_id:
return product
raise ValueError(f"Product with ID={product_id} not found")
@mcp.tool
def add_product(
name: str,
price: float,
category: str,
in_stock: bool = True,
) -> dict[str, Any]:
"""Add a new product and return the created object."""
if not name.strip():
raise ValueError("Product name must not be empty")
if price <= 0:
raise ValueError("Product price must be greater than 0")
if not category.strip():
raise ValueError("Category must not be empty")
products = _read_products()
new_product = {
"id": _next_id(products),
"name": name.strip(),
"price": float(price),
"category": category.strip(),
"in_stock": bool(in_stock),
}
products.append(new_product)
_write_products(products)
LOGGER.info("Product added: id=%s name=%s", new_product["id"], new_product["name"])
return new_product
@mcp.tool
def get_statistics() -> dict[str, float | int]:
"""Return catalog statistics: total count and average price."""
products = _read_products()
if not products:
return {"count": 0, "average_price": 0.0}
total_price = sum(float(product["price"]) for product in products)
average_price = round(total_price / len(products), 2)
return {"count": len(products), "average_price": average_price}
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
mcp.run(transport="stdio")
