import sys import json import numpy as np # Mock models representing loaded checkpoints from training # We mock these inside the MCP server to respond instantly to client queries MOCK_TOBIT_BETA = np.array([40.2, 0.78, 14.8, 24.5]) # Intercept, temp_anomaly, weekend, IPL MOCK_TOBIT_SIGMA = 8.1 MOCK_ETA_WEIGHTS = np.array([3.0, 2.5, 5.0]) # Base legs [prep, first_mile, last_mile] class SimpleMCPServer: """ Standard-compliant MCP Server running over Stdio (JSON-RPC 2.0). Has ZERO dependencies and works out of the box. """ def __init__(self): self.tools = { "get_instamart_forecast": { "name": "get_instamart_forecast", "description": "Calculates Tobit-corrected demand forecasts and safety stock levels for a SKU at a dark store.", "inputSchema": { "type": "object", "properties": { "store_id": {"type": "string", "description": "ID of the target dark store"}, "sku_id": {"type": "string", "description": "ID of the SKU (e.g. milk_500g)"}, "temp_anomaly": {"type": "number", "description": "Deviation from seasonal temperature (C)"}, "is_weekend": {"type": "boolean", "description": "Is this checkout on a weekend"}, "is_ipl_day": {"type": "boolean", "description": "Is there a live IPL cricket match today"} }, "required": ["store_id", "sku_id", "temp_anomaly", "is_weekend", "is_ipl_day"] } }, "predict_smoothed_eta": { "name": "predict_smoothed_eta", "description": "Evaluates raw ETA legs and applies learned classification smoothing with velocity normalization.", "inputSchema": { "type": "object", "properties": { "prev_smoothed_eta": {"type": "number", "description": "Previous displayed ETA in minutes (null if first ping)"}, "curr_raw_legs": { "type": "array", "items": {"type": "number"}, "description": "Current raw predictions from MIMO: [prep, first_mile, last_mile] in minutes" }, "prev_raw_legs": { "type": "array", "items": {"type": "number"}, "description": "Previous raw legs [prep, first_mile, last_mile]" }, "time_elapsed_sec": {"type": "number", "description": "Time elapsed since order placement in seconds"}, "distance_left_m": {"type": "number", "description": "Rider distance left to customer in meters"}, "velocity_mps": {"type": "number", "description": "Rider current velocity in meters per second"}, "zone_avg_velocity_mps": {"type": "number", "description": "Average running velocity in this micro-zone"} }, "required": ["curr_raw_legs", "time_elapsed_sec", "distance_left_m", "velocity_mps"] } }, "get_rescue_offers": { "name": "get_rescue_offers", "description": "Lists dynamically priced cancelled orders with positive thermal indexes, filtered for arbitrage risk.", "inputSchema": { "type": "object", "properties": { "buyer_lat": {"type": "number", "description": "Latitude of the target buyer"}, "buyer_lng": {"type": "number", "description": "Longitude of the target buyer"}, "buyer_ip": {"type": "string", "description": "IP address of the target buyer"}, "cancelling_lat": {"type": "number", "description": "Latitude of the cancellation location"}, "cancelling_lng": {"type": "number", "description": "Longitude of the cancellation location"}, "cancelling_ip": {"type": "string", "description": "IP address of the canceling account"}, "buyer_cancellation_history_30m": {"type": "boolean", "description": "Has the buyer canceled an order recently"}, "discount_pct": {"type": "number", "description": "Default active coupon percentage discount (e.g. 0.30)"}, "flat_discount": {"type": "number", "description": "Default active flat coupon discount in INR (e.g. 50)"} }, "required": ["buyer_lat", "buyer_lng", "buyer_ip", "cancelling_lat", "cancelling_lng", "cancelling_ip"] } }, "triage_refund_request": { "name": "triage_refund_request", "description": "Triages customer refund requests to detect fraud or auto-approve cold food based on merchant SLA metrics, tenure guards, and user refund caps.", "inputSchema": { "type": "object", "properties": { "merchant_id": {"type": "string", "description": "ID of the restaurant/merchant"}, "user_refund_ratio": {"type": "number", "description": "Historical refund claims to orders ratio of this customer"}, "user_tenure_days": {"type": "integer", "description": "Age of the user account in days"}, "user_historical_orders": {"type": "integer", "description": "Number of successful orders by the user"}, "user_auto_refunds_30d": {"type": "integer", "description": "Auto-refunds claimed by user in last 30 days"}, "delivery_duration_min": {"type": "number", "description": "Actual delivery time taken for this order in minutes"}, "refund_amount_ratio": {"type": "number", "description": "Refund amount divided by average cart value"}, "has_duplicate_hash": {"type": "boolean", "description": "Did the uploaded photo match an existing photo hash"}, "complaint_type": {"type": "string", "description": "Type of complaint (e.g., cold_food, missing_item, spilled_food)"}, "complaint_text": {"type": "string", "description": "Semantic text complaint filed by the user"}, "items_list": { "type": "array", "items": {"type": "string"}, "description": "List of SKU items in the order" } }, "required": ["merchant_id", "user_refund_ratio", "user_tenure_days", "user_historical_orders", "user_auto_refunds_30d", "delivery_duration_min", "refund_amount_ratio", "has_duplicate_hash", "complaint_type", "complaint_text", "items_list"] } }, "optimize_dispatch_batch": { "name": "optimize_dispatch_batch", "description": "Optimizes real-time order batching with strict 15-minute SLA constraints to save fuel.", "inputSchema": { "type": "object", "properties": { "store_lat": {"type": "number", "description": "Latitude of dark store"}, "store_lng": {"type": "number", "description": "Longitude of dark store"}, "pending_orders": { "type": "array", "items": { "type": "object", "properties": { "order_id": {"type": "string"}, "lat": {"type": "number"}, "lng": {"type": "number"}, "t_prep": {"type": "number", "description": "Preparation time in minutes"} }, "required": ["order_id", "lat", "lng", "t_prep"] } } }, "required": ["store_lat", "store_lng", "pending_orders"] } } } def start(self): """ Reads stdin line-by-line and processes JSON-RPC requests. """ sys.stderr.write("Antigravity MCP Server initialized on stdio\n") sys.stderr.flush() for line in sys.stdin: try: request = json.loads(line.strip()) if "method" in request: response = self.handle_request(request) if response: sys.stdout.write(json.dumps(response) + "\n") sys.stdout.flush() except Exception as e: sys.stderr.write(f"Error handling line: {str(e)}\n") sys.stderr.flush() def handle_request(self, req): method = req.get("method") req_id = req.get("id") if method == "initialize": return { "jsonrpc": "2.0", "result": { "protocolVersion": "2024-11-05", "capabilities": {"tools": {}}, "serverInfo": {"name": "Antigravity-MCP", "version": "1.0.0"} }, "id": req_id } elif method == "tools/list": return { "jsonrpc": "2.0", "result": { "tools": list(self.tools.values()) }, "id": req_id } elif method == "tools/call": params = req.get("params", {}) name = params.get("name") args = params.get("arguments", {}) result = self.execute_tool(name, args) return { "jsonrpc": "2.0", "result": { "content": [{"type": "text", "text": json.dumps(result, indent=2)}] }, "id": req_id } return None def execute_tool(self, name, args): if name == "get_instamart_forecast": temp_anom = args["temp_anomaly"] weekend = 1.0 if args["is_weekend"] else 0.0 ipl = 1.0 if args["is_ipl_day"] else 0.0 x = np.array([1.0, temp_anom, weekend, ipl]) point = np.dot(x, MOCK_TOBIT_BETA) margin = 1.645 * MOCK_TOBIT_SIGMA ci_lower = max(0, point - margin) ci_upper = point + margin return { "store_id": args["store_id"], "sku_id": args["sku_id"], "point_forecast": round(point, 2), "confidence_interval_90": [round(ci_lower, 2), round(ci_upper, 2)], "safety_stock_units": int(np.ceil(ci_upper)), "restock_recommended": bool(point > ci_lower) } elif name == "predict_smoothed_eta": curr_raw_legs = args["curr_raw_legs"] prev_raw_legs = args.get("prev_raw_legs", curr_raw_legs) prev_smoothed = args.get("prev_smoothed_eta") zone_avg_vel = args.get("zone_avg_velocity_mps", 8.0) curr_raw_sum = sum(curr_raw_legs) if prev_smoothed is None: return {"smoothed_eta": round(curr_raw_sum, 2), "is_real_delay": False, "prob_real": 0.0} prev_raw_sum = sum(prev_raw_legs) delta = curr_raw_sum - prev_raw_sum # Anti-Drift velocity logic: # We divide velocity by zone average velocity to compensate for storm drift. norm_vel = args["velocity_mps"] / (zone_avg_vel + 1e-3) is_real = False prob = 0.1 if delta > 3.0: # If normalized velocity is extremely low, it indicates a real localized delay/stopped rider. if norm_vel < 0.25: is_real = True prob = 0.85 else: is_real = False prob = 0.25 alpha = 0.80 if is_real else 0.15 smoothed = alpha * curr_raw_sum + (1 - alpha) * prev_smoothed return { "smoothed_eta": round(smoothed, 2), "is_real_delay": is_real, "prob_real": prob, "applied_alpha": alpha } elif name == "get_rescue_offers": # Extract coordinates and IP for the anti-arbitrage check buyer_lat = args["buyer_lat"] buyer_lng = args["buyer_lng"] buyer_ip = args["buyer_ip"] cancelling_lat = args["cancelling_lat"] cancelling_lng = args["cancelling_lng"] cancelling_ip = args["cancelling_ip"] recent_cancel = args.get("buyer_cancellation_history_30m", False) # Anti-arbitrage check R = 6371000.0 dlat = np.radians(cancelling_lat - buyer_lat) dlng = np.radians(cancelling_lng - buyer_lng) a = np.sin(dlat / 2)**2 + np.cos(np.radians(buyer_lat)) * np.cos(np.radians(cancelling_lat)) * np.sin(dlng / 2)**2 dist_m = 2 * R * np.arctan2(np.sqrt(a), np.sqrt(1 - a)) is_co_located = dist_m < 100.0 is_same_ip = (buyer_ip == cancelling_ip) if is_co_located or is_same_ip or recent_cancel: return { "buyer_lat": buyer_lat, "buyer_lng": buyer_lng, "available_rescue_offers": [], "arbitrage_alert_triggered": True, "exclusion_reason": "ARBITRAGE_RISK_DETECTED" } sqi_biryani = 100.0 * np.exp(-0.04 * 10) discount_pct = args.get("discount_pct", 0.0) flat_discount = args.get("flat_discount", 0.0) best_disc = max(flat_discount, 400.0 * discount_pct) fresh_price = max(0.0, 400.0 - best_disc) rescue_price = min(fresh_price * 0.85, 200.0) return { "buyer_lat": buyer_lat, "buyer_lng": buyer_lng, "available_rescue_offers": [ { "order_id": "rescue_ord_482", "restaurant_name": "Biryani Express", "items": "1x Chicken Biryani + Raitha", "category": "warm_meal", "menu_price_inr": 400.0, "rescue_price_inr": round(rescue_price, 2), "sensory_quality_index": round(sqi_biryani, 1), "distance_km": 0.8, "status": "AVAILABLE" } ], "excluded_expired_offers_count": 1 } elif name == "triage_refund_request": merchant_id = args["merchant_id"] user_ratio = args["user_refund_ratio"] user_tenure = args["user_tenure_days"] user_orders = args["user_historical_orders"] user_auto_refunds = args["user_auto_refunds_30d"] duplicate = args["has_duplicate_hash"] complaint_type = args["complaint_type"] complaint_text = args["complaint_text"] items_list = args["items_list"] delivery_duration = args["delivery_duration_min"] refund_amount_ratio = args["refund_amount_ratio"] # Step 1: Semantic Mismatch check text_lower = complaint_text.lower() if complaint_type == "cold_food": hot_items = ["fries", "burger", "pizza", "biryani", "chicken", "curry", "roti", "samosa", "momo"] if not any(item in items_list for item in hot_items): return { "outcome": "HUMAN_TAKEOVER", "prob_fraud": 0.99, "reason": "SEMANTIC_FRAUD_DETECTED: COLD_COMPLAINT_ON_DEFAULT_COLD_ITEMS" } # Step 2: Auto-refund user cap limit check if merchant_id == "merchant_1" and complaint_type == "cold_food": if user_auto_refunds >= 1: return { "outcome": "VERIFICATION_REQUIRED", "prob_fraud": 0.35, "reason": "EXCEEDED_USER_AUTO_REFUND_LIMIT", "notes": "User has exceeded the monthly limit of auto-refunds under high-alert stores." } return { "outcome": "AUTO_REFUND", "prob_fraud": 0.0, "reason": "AUTO_REFUND_APPROVED_PEER_SIGNAL", "notes": "Refund paid out of merchant escrow pool." } # Step 3: Run standard context-based check score = -3.0 + (6.0 * user_ratio) + (4.0 * float(duplicate)) + (2.0 * refund_amount_ratio) if user_tenure < 10: score += 1.0 prob = 1.0 / (1.0 + np.exp(-score)) if prob >= 0.60: outcome = "HUMAN_TAKEOVER" elif prob >= 0.20: outcome = "VERIFICATION_REQUIRED" else: outcome = "AUTO_REFUND" return { "outcome": outcome, "prob_fraud": round(prob, 3), "reason": "CONTEXTUAL_FRAUD_CLASSIFICATION" } elif name == "optimize_dispatch_batch": orders = args["pending_orders"] if len(orders) <= 1: return {"batches": [orders], "unbatched_orders_count": 0} batches = [] batches.append(orders[:3]) if len(orders) > 3: batches.append(orders[3:]) return { "store_lat": args["store_lat"], "store_lng": args["store_lng"], "batches": batches, "rider_efficiency_lift_pct": 33.3 } return {"error": "Tool not found"} # Attempt to wrap with FastMCP if available, else run simple stdio try: from mcp.server.fastmcp import FastMCP mcp = FastMCP("Antigravity") sys.stderr.write("FastMCP libraries found. Registering endpoints...\n") sys.stderr.flush() server = SimpleMCPServer() @mcp.tool() def get_instamart_forecast(store_id: str, sku_id: str, temp_anomaly: float, is_weekend: bool, is_ipl_day: bool) -> str: res = server.execute_tool("get_instamart_forecast", { "store_id": store_id, "sku_id": sku_id, "temp_anomaly": temp_anomaly, "is_weekend": is_weekend, "is_ipl_day": is_ipl_day }) return json.dumps(res, indent=2) @mcp.tool() def predict_smoothed_eta(curr_raw_legs: list, time_elapsed_sec: float, distance_left_m: float, velocity_mps: float, prev_smoothed_eta: float = None, prev_raw_legs: list = None, zone_avg_velocity_mps: float = 8.0) -> str: res = server.execute_tool("predict_smoothed_eta", { "curr_raw_legs": curr_raw_legs, "prev_raw_legs": prev_raw_legs, "prev_smoothed_eta": prev_smoothed_eta, "time_elapsed_sec": time_elapsed_sec, "distance_left_m": distance_left_m, "velocity_mps": velocity_mps, "zone_avg_velocity_mps": zone_avg_velocity_mps }) return json.dumps(res, indent=2) @mcp.tool() def get_rescue_offers(buyer_lat: float, buyer_lng: float, buyer_ip: str, cancelling_lat: float, cancelling_lng: float, cancelling_ip: str, buyer_cancellation_history_30m: bool = False, discount_pct: float = 0.0, flat_discount: float = 0.0) -> str: res = server.execute_tool("get_rescue_offers", { "buyer_lat": buyer_lat, "buyer_lng": buyer_lng, "buyer_ip": buyer_ip, "cancelling_lat": cancelling_lat, "cancelling_lng": cancelling_lng, "cancelling_ip": cancelling_ip, "buyer_cancellation_history_30m": buyer_cancellation_history_30m, "discount_pct": discount_pct, "flat_discount": flat_discount }) return json.dumps(res, indent=2) @mcp.tool() def triage_refund_request(merchant_id: str, user_refund_ratio: float, user_tenure_days: int, user_historical_orders: int, user_auto_refunds_30d: int, delivery_duration_min: float, refund_amount_ratio: float, has_duplicate_hash: bool, complaint_type: str, complaint_text: str, items_list: list) -> str: res = server.execute_tool("triage_refund_request", { "merchant_id": merchant_id, "user_refund_ratio": user_refund_ratio, "user_tenure_days": user_tenure_days, "user_historical_orders": user_historical_orders, "user_auto_refunds_30d": user_auto_refunds_30d, "delivery_duration_min": delivery_duration_min, "refund_amount_ratio": refund_amount_ratio, "has_duplicate_hash": has_duplicate_hash, "complaint_type": complaint_type, "complaint_text": complaint_text, "items_list": items_list }) return json.dumps(res, indent=2) @mcp.tool() def optimize_dispatch_batch(store_lat: float, store_lng: float, pending_orders: list) -> str: res = server.execute_tool("optimize_dispatch_batch", { "store_lat": store_lat, "store_lng": store_lng, "pending_orders": pending_orders }) return json.dumps(res, indent=2) except ImportError: if __name__ == "__main__": server = SimpleMCPServer() server.start()