import sys import os import urllib3 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) # Ensure the mcp directory is on the path when spawned by Claude Desktop sys.path.insert(0, os.path.dirname(__file__)) from mcp.server.fastmcp import FastMCP from laravel_client import get from app_client import post as app_post from dotenv import load_dotenv load_dotenv(dotenv_path=os.path.join(os.path.dirname(__file__), "..", ".env")) MCP_EXECUTION_MODE = os.getenv("MCP_EXECUTION_MODE", "local") MCP_REMOTE_HOST = os.getenv("MCP_REMOTE_HOST", "0.0.0.0") MCP_REMOTE_PORT = os.getenv("MCP_REMOTE_PORT", "8200") MCP_REMOTE_TRANSPORT = os.getenv("MCP_REMOTE_TRANSPORT", "sse") mcp = FastMCP( name="IDEAL OMS Analytics", instructions="Real-time analytics and operational tools for the IDEAL textbook OMS.", ) # ── Analytics & Reporting ────────────────────────────────────────────────────── @mcp.tool() def get_season_summary(account_year_id: int = None) -> dict: """ High-level KPIs for the current (or specified) season. Returns: total orders, order value, dispatched count, pending dispatch, collections received, overdue amount, open credit note requests, stock alert count. """ return get("season-summary", {"account_year_id": account_year_id}) @mcp.tool() def get_stock_alerts(days_remaining: int = 30) -> dict: """ Products that will run out of stock within `days_remaining` days at the current order rate. Returns product name, current stock, avg daily orders, days remaining. """ return get("stock-alerts", {"days_remaining": days_remaining}) @mcp.tool() def get_district_performance(zone_id: int = None, account_year_id: int = None) -> dict: """ Order quantity vs. target by district with attainment percentage. Optionally filter by zone_id or account_year_id. """ return get("district-performance", {"zone_id": zone_id, "account_year_id": account_year_id}) @mcp.tool() def get_top_products(limit: int = 10, from_date: str = None, to_date: str = None) -> dict: """ Top products ranked by total quantity ordered. Dates are ISO format YYYY-MM-DD. Default limit is 10. """ return get("top-products", {"limit": limit, "from_date": from_date, "to_date": to_date}) @mcp.tool() def get_product_monthly_sales( product_id: int = None, account_year_id: int = None, from_date: str = None, to_date: str = None, top_n: int = 10, ) -> dict: """ Billed sales (transaction_sales status=DISPATCHED) per product, bucketed by year/month. Source: transaction_sales x transaction_sales_products. Pass product_id to drill into one product, or leave blank to get the top-N products in the window by total sales value. Dates are ISO YYYY-MM-DD; if both dates and account_year_id are omitted, the current accounting year is used. top_n is capped at 50. """ return get("product-monthly-sales", { "product_id": product_id, "account_year_id": account_year_id, "from_date": from_date, "to_date": to_date, "top_n": top_n, }) @mcp.tool() def get_shortfall_products(account_year_id: int = None) -> dict: """ Products where total orders exceed available stock. Critical for purchase order planning. """ return get("shortfall-products", {"account_year_id": account_year_id}) @mcp.tool() def search_orders( client_id: int = None, district_id: int = None, status: str = None, from_date: str = None, to_date: str = None, limit: int = 25, ) -> dict: """ Search orders with flexible filters. Status values: placed, approved, dispatched, invoiced, cancelled. Dates are ISO format YYYY-MM-DD. """ return get("orders", { "client_id": client_id, "district_id": district_id, "status": status, "from_date": from_date, "to_date": to_date, "limit": limit, }) # ── Risk & Collections ───────────────────────────────────────────────────────── @mcp.tool() def get_overdue_collections(min_days_overdue: int = 30, limit: int = 20) -> dict: """ Clients with overdue payments sorted by amount descending. `min_days_overdue` filters to only collections overdue by at least that many days. """ return get("overdue-collections", {"min_days_overdue": min_days_overdue, "limit": limit}) @mcp.tool() def score_collection_risk(client_id: int) -> dict: """ ML-generated payment delay probability (0.0–1.0) for a specific client, plus a risk_level label: LOW, MEDIUM, or HIGH. """ return get(f"collection-risk/{client_id}") @mcp.tool() def get_cn_anomalies(limit: int = 20) -> dict: """ Credit note requests flagged as anomalous. Returns anomaly score, risk level, and the trigger reason for each record. """ return get("cn-anomalies", {"limit": limit}) @mcp.tool() def forecast_demand(product_id: int, district_id: int, next_season_start: str) -> dict: """ Prophet model demand forecast for a product × district combination. `next_season_start` must be an ISO date string e.g. "2026-04-01". Returns predicted_qty with lower and upper confidence bounds. Uses a locally trained Prophet model (model_id = demand_p{product_id}_d{district_id}) when available; otherwise falls back to the Laravel API forecast endpoint. """ model_id = f"demand_p{product_id}_d{district_id}" result = app_post(f"prophet/forecast/{model_id}", { "periods": 12, "freq": "M", "include_history": False, }) if result and result.get("success"): forecast = result.get("data", {}).get("forecast", []) if forecast: predicted = round(sum(f.get("yhat", 0) for f in forecast)) lower = round(sum(f.get("yhat_lower", 0) for f in forecast)) upper = round(sum(f.get("yhat_upper", 0) for f in forecast)) return { "predicted_qty": predicted, "lower_bound": lower, "upper_bound": upper, "model_source": "local", "model_id": model_id, } # Fall back to Laravel API return get("demand-forecast", { "product_id": product_id, "district_id": district_id, "next_season_start": next_season_start, }) # ── Operational ──────────────────────────────────────────────────────────────── @mcp.tool() def get_pending_approvals() -> dict: """ Count and list of items awaiting admin action: credit note requests, purchase orders, and orders on hold. """ return get("pending-approvals") @mcp.tool() def get_client_profile(client_id: int) -> dict: """ Full analytical profile for a client: lifetime orders, lifetime value, current outstanding balance, YTD credit note count, AI segment, collection risk level. """ return get(f"clients/{client_id}/profile") @mcp.tool() def get_sales_performance(account_year_id: int = None, sales_user_id: int = None) -> dict: """ Season attainment percentage per sales user. Shows orders placed, target, attainment %, and rank. """ return get("sales-performance", { "account_year_id": account_year_id, "sales_user_id": sales_user_id, }) @mcp.tool() def get_dispatch_queue(district_id: int = None) -> dict: """ Approved orders ready for dispatch, optionally filtered by district. Grouped by geographic cluster with suggested batching. """ return get("dispatch-queue", {"district_id": district_id}) @mcp.tool() def get_dispatchable_orders() -> dict: """ Returns which child orders (parent_id > 0) in pending statuses (Order Placed, Client Confirm, Challan Date, Approved) can be simultaneously dispatched given current product stock levels. Greedy allocation: orders are prioritised by order_expected_dispatched_date ASC (oldest due date first), with order_id ASC as tiebreaker for same-date orders. Orders with no dispatch date set are treated as lowest priority. Result contains two keys: summary – { total_pending_child_orders, dispatchable_orders, stock_limited_orders } orders – list of orders, each with: order_id, parent_order_id, expected_dispatch_date, order_value, client_name, district, status, status_label, can_dispatch ("Yes"/"No"), is_dispatchable (1/0), products: list of { code, name, qty_needed, stock, cumulative_demand, sufficient (1/0) } Use `sufficient=0` in products to identify which specific item is blocking dispatch. Filter `can_dispatch="No"` to see all stock-limited orders. """ return get("simultaneous-dispatch-orders", { "account_year_id": None, "district_id": None, "limit": 5000, }) # ── Lookups (ID resolution) ──────────────────────────────────────────────────── @mcp.tool() def lookup_account_year(query: str = "current") -> dict: """ Resolve a human-readable year reference to an account year record with its id. Accepted formats: - Short range: '25-26', '2025-26' - Full range: '2025-2026' - Single year: '2025' - Relative: 'current', 'active', 'this year', 'last year', 'past year', 'previous year', 'last 2 years', 'last 3 years', 'past 2 years' Returns id, title, from_period, to_period, status, status_label. Call BEFORE any tool that accepts account_year_id when the user gives a year by name. """ return get("lookup/account-year", {"q": query}) @mcp.tool() def lookup_product(query: str, limit: int = 15) -> dict: """ Search products by name or code. Returns id, name, code, product_head, stock. Call BEFORE forecast_demand or get_shortfall_products when you only have a product name. """ return get("lookup/products", {"q": query, "limit": limit}) @mcp.tool() def lookup_district(query: str, zone_id: int = None, limit: int = 15) -> dict: """ Search districts by name or code. Returns id, district_name, zone_id, zone_name. Call BEFORE forecast_demand, get_district_performance, get_dispatch_queue, or search_orders when you only have a district name. """ return get("lookup/districts", {"q": query, "zone_id": zone_id, "limit": limit}) @mcp.tool() def lookup_zone(query: str, limit: int = 15) -> dict: """ Search zones by name or code. Returns id, zone_code, zone_name. Call BEFORE get_district_performance when filtering by zone name. """ return get("lookup/zones", {"q": query, "limit": limit}) @mcp.tool() def lookup_sales_user(query: str, limit: int = 15) -> dict: """ Search sales users by name or username. Returns id, full_name, username, role. Call BEFORE get_sales_performance when filtering by a sales user's name. """ return get("lookup/sales-users", {"q": query, "limit": limit}) @mcp.tool() def lookup_client(query: str, district_id: int = None, limit: int = 15) -> dict: """ Search clients/dealers by name or client code. Returns id, client_code, client_name, district, zone. Call BEFORE score_collection_risk, get_client_profile, or search_orders when you only have a client name. """ return get("lookup/clients", {"q": query, "district_id": district_id, "limit": limit}) # ── Schema resource ──────────────────────────────────────────────────────────── @mcp.resource("oms://schema/overview") def schema_overview() -> str: """ A plain-English description of the IDEAL OMS database schema, key table relationships, and business terminology. Used by the AI to interpret query results correctly. """ return """ IDEAL OMS Database Overview =========================== The system manages textbook distribution in Gujarat, India. Key Tables: - account_years: Each row is one academic season (April–March). Filter all seasonal data by this. - admin: Sales reps, back office staff, management teams, etc. users - products: Product catalog with stock, price, and category data. - boards: Exam boards (Maharashtra, Gujarat, CBSE, etc.) - mediums: Languages of instruction (Hindi, Marathi, English, Semi-English) - mas_segment: Standard Junior Kg, Senior Kg, 1–12 - mas_product_head: Product head groups product by their category. - mas_series - Series is a collection of books following a common theme. - product_balance - Year wise opening and closing stock of products. - product_stock - Maintains the running stock of products for each accounting year. - clients: Dealers/bookshops who place orders. - sales_structure - The sales structure is a hierarchical structure that defines the sales organization. - sales_users_relationship - The mapping of users in sales hierarchy. - districts / talukas: Geographic hierarchy. Zone → District → Taluka. - orders: Main order table with status, dates, and financial info. - order_products: Line items showing which products are in which order. - purchase_orders: Orders placed to vendors to replenish stock. - credit_note: CN generated from the sales returned by clients. - credit_note_products: Line items showing which products are in which credit note. - transaction_sales: Sales transactions. - transaction_sales_products: Line items showing which products are in which sales transaction. - transaction_cn: Credit note transactions. - transaction_cn_products: Line items showing which products are in which credit note transaction. Key Metrics: - Total Orders: Count of orders in different statuses. - Total Sales: Revenue from sold items. - Stock Balance: Current inventory levels. - Supply Pending: Orders waiting for stock. - Purchase Orders: Pending vendor orders. Business Terms: - Challan: Physical delivery challan. - Stock Balance: Available stock. - Supply Pending: Orders awaiting stock. - Purchase Order: Vendor purchase order. """ # ── Prompts ──────────────────────────────────────────────────────────────────── @mcp.prompt() def daily_ops_briefing() -> str: """Generate a daily operations briefing for the admin.""" return """ You are an analytics assistant for IDEAL, a textbook distribution company. Please provide a concise daily operations briefing by calling these tools in order: 1. get_season_summary() — overall health 2. get_stock_alerts(days_remaining=14) — critical stock issues only 3. get_overdue_collections(min_days_overdue=30) — urgent payment risks 4. get_pending_approvals() — what needs action today 5. get_cn_anomalies(limit=5) — top anomalous credit note requests Format the briefing as: - A 3-sentence executive summary - A bulleted "Action Required" list (items needing immediate attention) - A bulleted "Watch List" (items that may need attention in 3–5 days) Keep the tone factual and direct. Use ₹ for currency. Flag anything with > ₹1L exposure. """ @mcp.prompt() def client_review(client_id: int) -> str: """Deep-dive review of a specific client.""" return f""" You are reviewing client ID {client_id} for a sales manager. Call get_client_profile({client_id}) to retrieve their profile. Then call score_collection_risk({client_id}) for payment risk. Then call search_orders(client_id={client_id}, limit=10) for recent orders. Provide: 1. A 2-sentence summary of who this client is and their value to IDEAL. 2. Their payment behaviour and current risk level. 3. Any red flags (high CN rate, overdue balance, dropping order frequency). 4. A recommended action for the sales team. """ @mcp.prompt() def season_forecast_report(district_id: int) -> str: """Demand forecast and stock planning report for a district.""" return f""" Generate a season stock planning report for district ID {district_id}. Steps: 1. Call get_stock_alerts() and filter for this district. 2. Call get_shortfall_products() to identify gaps. 3. Call get_district_performance() with district_id={district_id} for target context. 4. For the top 3 shortfall products, call forecast_demand(product_id=X, district_id={district_id}, next_season_start="2026-04-01") to get next season estimates. Output a markdown table: Product | Current Stock | Projected Need | Gap | Suggested PO Qty Then write a 2-sentence purchasing recommendation. """ # ── Entry point ──────────────────────────────────────────────────────────────── if __name__ == "__main__": if MCP_EXECUTION_MODE == "local": print("Starting MCP server in LOCAL mode on stdio...") mcp.run() else: print( f"Starting MCP server in REMOTE mode: " f"{MCP_REMOTE_TRANSPORT}://{MCP_REMOTE_HOST}:{MCP_REMOTE_PORT}" ) mcp.run(transport=MCP_REMOTE_TRANSPORT)