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 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, company_id: int = None) -> dict: """ High-level KPIs for the current (or specified) season. Returns: total_orders / total_order_value – regular orders placed in this season, cancelled and deleted excluded. dispatched_orders / dispatched_value – invoices dispatched inside the season's date window, counted from transaction_sales (the same source as the admin dashboard). This can exceed total_orders, because an invoice may belong to an order placed in an earlier season. pending_dispatch – orders from this season still awaiting dispatch. total_collections – collections received, bounced and deleted excluded. uncleared_collections – cheques collected or deposited, not yet credited. overdue_amount – invoice amounts still due past the payment due date. open_cn_requests – pending or approved credit note requests. Optionally scope to one company via company_id (omit for all companies). Returns HTTP 422 when account_year_id does not exist. """ return get("season-summary", {"account_year_id": account_year_id, "company_id": company_id}) @mcp.tool() def get_stock_alerts(days_remaining: int = 30, company_id: int = None) -> dict: """ Regular (non-custom, active) products that will run out of stock within `days_remaining` days at the current order rate, based on the last 30 days of regular orders. Returns product name, current stock, avg daily orders, days remaining. Optionally scope to one company via company_id. """ return get("stock-alerts", {"days_remaining": days_remaining, "company_id": company_id}) @mcp.tool() def get_district_performance(zone_id: int = None, account_year_id: int = None, company_id: int = None) -> dict: """ Per-district order activity plus target attainment. Each district row has: unique_clients, order_count, total_order_value_inr (from regular orders), and target_qty / billed_qty / attainment_pct (dispatched sales qty vs product_head_district_target — same source as the admin dashboard). attainment_pct is null when no target is set. Optionally filter by zone_id, account_year_id, company_id. """ return get("district-performance", { "zone_id": zone_id, "account_year_id": account_year_id, "company_id": company_id, }) @mcp.tool() def get_top_products( limit: int = 10, from_date: str = None, to_date: str = None, account_year_id: int = None, company_id: int = None, district_id: int = None, zone_id: int = None, ) -> dict: """ Top regular products ranked by BILLED REVENUE — the same source and rules as the admin dashboard chart: dispatched invoices only (transaction_sales), custom products excluded, and specimen lines (free copies, amount 0) kept out of the ranking and reported separately as specimen_qty. Each row: product_code, product_name, product_head, total_qty_billed, total_sales_value_inr, specimen_qty, bill_count. This answers "what sells", not "what was asked for" — use search_orders if you need ordered quantities that have not been billed yet. Dates are ISO YYYY-MM-DD and match the invoice bill date. Filter by account_year_id for a season, or district_id / zone_id for geographic popularity (e.g. "best selling products in Surat district"). District and zone come from the client record, the same way the district reports attribute a sale. """ return get("top-products", { "limit": limit, "from_date": from_date, "to_date": to_date, "account_year_id": account_year_id, "company_id": company_id, "district_id": district_id, "zone_id": zone_id, }) @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, company_id: int = None, ) -> dict: """ Billed sales (transaction_sales status=DISPATCHED, regular order mode) 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, "company_id": company_id, }) @mcp.tool() def get_shortfall_products(account_year_id: int = None, company_id: int = None) -> dict: """ Products where total orders exceed available stock. Critical for purchase order planning. Custom (institution-branded) products are excluded. Optionally scope to one company via company_id (omit for all companies). """ return get("shortfall-products", { "account_year_id": account_year_id, "company_id": company_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, account_year_id: int = None, company_id: int = None, order_mode: str = None, limit: int = 25, ) -> dict: """ Search orders with flexible filters. Status values (names, not ids): placed, client confirm, challan, qc, approved, dispatched, cancelled, pending (= all not-yet-dispatched). There is no "invoiced" or "delivered" status in this system. When status is omitted, cancelled and deleted orders are excluded. order_mode is "regular" or "custom"; omit it to search both. Custom orders are the ones placed for institution-branded books. district_id matches the billing district. Dates are ISO format YYYY-MM-DD. Pass account_year_id to scope to a season — without it the search spans every season and the limit may hide older rows. """ return get("orders", { "client_id": client_id, "district_id": district_id, "status": status, "from_date": from_date, "to_date": to_date, "account_year_id": account_year_id, "company_id": company_id, "order_mode": order_mode, "limit": limit, }) # ── Risk & Collections ───────────────────────────────────────────────────────── @mcp.tool() def get_overdue_collections( min_days_overdue: int = 30, limit: int = 20, company_id: int = None, account_year_id: int = None, ) -> dict: """ Clients with overdue unpaid invoices sorted by due amount descending. `min_days_overdue` filters to only invoices overdue by at least that many days. Rows are grouped per client AND company, so a client trading with both companies appears once per company. Pass account_year_id to scope to one season; without it every season is summed. Optionally scope to one company via company_id. """ return get("overdue-collections", { "min_days_overdue": min_days_overdue, "limit": limit, "company_id": company_id, "account_year_id": account_year_id, }) @mcp.tool() def score_collection_risk(client_id: int, company_id: int = None) -> dict: """ Rule-based payment risk score (0.0–1.0) for a specific client, plus a risk_level label: LOW, MEDIUM, or HIGH. Derived from unpaid invoices, max days overdue, and cheque bounce history (not an ML model). Also returns ledger_outstanding — the client's running ledger balance. Optionally scope the invoices and bounces to one company via company_id. """ return get(f"collection-risk/{client_id}", {"company_id": company_id}) @mcp.tool() def get_cn_anomalies(limit: int = 20, account_year_id: int = None, company_id: int = None) -> dict: """ Credit note requests flagged as unusual by business rules (amount far above average, or clients with repeated pending CNs this season). Returns risk level and the trigger reason for each record. The average used as the baseline is taken over the same season and company that the rules are applied to, so it moves with the data being checked. Defaults to the current season and all companies. Results are sorted by amount before the limit is applied. """ return get("cn-anomalies", { "limit": limit, "account_year_id": account_year_id, "company_id": company_id, }) @mcp.tool() def forecast_demand(product_id: int, district_id: int, next_season_start: str, company_id: int = None) -> dict: """ 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 and a `note` field stating the source: "prophet" (stored Prophet model predictions) or a labelled heuristic fallback (3-year order average) when no trained model exists. company_id scopes to one company (models are trained per company; omit to use the latest trained model). """ return get("demand-forecast", { "product_id": product_id, "district_id": district_id, "next_season_start": next_season_start, "company_id": company_id, }) # ── Operational ──────────────────────────────────────────────────────────────── @mcp.tool() def get_pending_approvals(account_year_id: int = None, company_id: int = None) -> dict: """ Counts of items awaiting admin action in the current (or given) season: pending credit note requests, pending purchase orders, and newly placed regular orders awaiting review. (There is no "on hold" order status.) Optionally scope to one company via company_id (omit for all companies). """ return get("pending-approvals", { "account_year_id": account_year_id, "company_id": company_id, }) @mcp.tool() def get_client_profile(client_id: int, company_id: int = None) -> dict: """ Analytical profile for a client: lifetime regular orders and value (cancelled and deleted excluded), current outstanding balance from the customer ledger, and current-season credit note count. Optionally scope orders and credit notes to one company via company_id. Combine with score_collection_risk for the payment risk level. """ return get(f"clients/{client_id}/profile", {"company_id": company_id}) @mcp.tool() def get_sales_performance(account_year_id: int = None, sales_user_id: int = None, company_id: int = None) -> dict: """ Sales users ranked by regular-order value for the season (rank included). Shows unique clients, order count, and total order value per user. NOTE: sales targets in this system are per district, not per sales user — use get_district_performance for target attainment questions. """ return get("sales-performance", { "account_year_id": account_year_id, "sales_user_id": sales_user_id, "company_id": company_id, }) @mcp.tool() def get_dispatch_queue(district_id: int = None, company_id: int = None) -> dict: """ Regular orders ready for dispatch (challan/QC/approved statuses) for the current season, oldest first. Max 100 rows. district_id matches the SHIPPING district, since dispatch follows the shipping address. Optionally scope to one company via company_id. """ return get("dispatch-queue", { "district_id": district_id, "company_id": company_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, company_id: int = None, is_custom_product: int = None, ) -> dict: """ Search products by name or code. Returns id, name, code, product_head, stock, company_id, product_type ("book" or "kit"), and is_custom_product (1 = custom institution-branded book, excluded from demand analytics). Pass is_custom_product=0 for regular catalogue books only, or 1 for custom books. Optionally scope to one company via company_id. Call BEFORE forecast_demand or get_shortfall_products when you only have a product name. """ return get("lookup/products", { "q": query, "limit": limit, "company_id": company_id, "is_custom_product": is_custom_product, }) @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)