infrapuzzle/.agents/skills/marktguru/scripts/marktguru_mcp.py

#!/usr/bin/env python3
"""
Marktguru MCP (Model Context Protocol) Server (Production Grade)
Exposes German supermarket flyers, discounts, and price comparison tools over standard JSON-RPC 2.0 stdio.

Key Design Principles:
- Zero external dependencies: Uses standard library only.
- Strict Protocol Compliance: Conforms to MCP (2024-11-05) and JSON-RPC 2.0 specifications.
- Defensive Argument Coercion: Tolerates string/number variations commonly output by LLMs.
- Graceful Signal Handling: Clean exit on SIGINT, SIGTERM, and EOF.
- Stream Isolation: All logs and diagnostics route to stderr, ensuring stdout is 100% clean JSON-RPC.
"""

import sys
import json
import signal
import logging
from typing import Dict, Any, List, Optional

# Import business logic from hardened client script
from marktguru_client import (
    search_offers,
    compare_prices,
    list_supported_supermarkets,
    DEFAULT_ZIP_CODE,
    MAJOR_RETAILERS,
    RETAILER_ALIASES,
)

# Configure logging strictly to stderr
logging.basicConfig(
    stream=sys.stderr,
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s"
)
logger = logging.getLogger("marktguru-mcp")

PROTOCOL_VERSION = "2024-11-05"
SERVER_NAME = "marktguru-mcp-server"
SERVER_VERSION = "1.1.0"

TOOLS = [
    {
        "name": "search_supermarket_offers",
        "description": (
            "Search current weekly supermarket offers and promotional discounts in Germany via the Marktguru API. "
            "Supports filtering by postal code (PLZ) and specific supermarket chains (e.g. Aldi Süd, Aldi Nord, Lidl, Edeka, Rewe, Kaufland, Penny, Netto)."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "The product or keyword to search for (e.g. 'Butter', 'Kaffee', 'Bier', 'Milch', 'Nutella', 'Schokolade')."
                },
                "zip_code": {
                    "type": "string",
                    "description": f"5-digit German postal code (PLZ). Defaults to '{DEFAULT_ZIP_CODE}' (Regensburg)."
                },
                "supermarkets": {
                    "type": "array",
                    "items": {"type": "string"},
                    "description": "Optional list of supermarket names to filter by (e.g. ['Aldi', 'Lidl', 'Edeka', 'Rewe', 'Kaufland', 'Penny', 'Netto', 'dm', 'Rossmann']). If omitted, all available retailers are included."
                },
                "limit": {
                    "type": "integer",
                    "description": "Maximum number of offers to return (default: 20, max: 50)."
                }
            },
            "required": ["query"]
        }
    },
    {
        "name": "compare_supermarket_prices",
        "description": (
            "Compare current prices for a product across German supermarkets in a given postal code. "
            "Returns results sorted from lowest to highest price, highlighting the cheapest deal, regular prices (UVP), and required loyalty app memberships."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "The product or grocery item to compare (e.g. 'Kaffee', 'Butter', 'Bier', 'Olivenöl')."
                },
                "zip_code": {
                    "type": "string",
                    "description": f"5-digit German postal code (PLZ). Defaults to '{DEFAULT_ZIP_CODE}' (Regensburg)."
                },
                "supermarkets": {
                    "type": "array",
                    "items": {"type": "string"},
                    "description": "Optional list of supermarket names to compare (e.g. ['Aldi', 'Lidl', 'Edeka', 'Rewe'])."
                }
            },
            "required": ["query"]
        }
    },
    {
        "name": "list_supported_supermarkets",
        "description": (
            "List all supermarket, discounter, hypermarket, and drugstore chains supported by the Marktguru service in Germany, "
            "including their categories (Discounter, Supermarkt, Großfläche, Drogerie) and accepted name aliases."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {}
        }
    }
]


def _sanitize_supermarkets(sm_input: Any) -> Optional[List[str]]:
    """Coerces various LLM input formats (list, comma-string, single string) into a clean list."""
    if not sm_input:
        return None
    if isinstance(sm_input, list):
        return [str(s).strip() for s in sm_input if str(s).strip()]
    if isinstance(sm_input, str):
        return [s.strip() for s in sm_input.split(",") if s.strip()]
    return None


def handle_initialize(req_id: Any, params: Dict[str, Any]) -> Dict[str, Any]:
    return {
        "jsonrpc": "2.0",
        "id": req_id,
        "result": {
            "protocolVersion": PROTOCOL_VERSION,
            "capabilities": {
                "tools": {
                    "listChanged": False
                }
            },
            "serverInfo": {
                "name": SERVER_NAME,
                "version": SERVER_VERSION
            }
        }
    }


def handle_tools_list(req_id: Any) -> Dict[str, Any]:
    return {
        "jsonrpc": "2.0",
        "id": req_id,
        "result": {
            "tools": TOOLS
        }
    }


def handle_tools_call(req_id: Any, params: Dict[str, Any]) -> Dict[str, Any]:
    tool_name = params.get("name")
    args = params.get("arguments", {})

    logger.info(f"Tool call requested: {tool_name} with args={args}")

    try:
        if tool_name == "search_supermarket_offers":
            query = str(args.get("query") or "").strip()
            if not query:
                return {
                    "jsonrpc": "2.0",
                    "id": req_id,
                    "result": {
                        "content": [{"type": "text", "text": "Error: 'query' argument is required and cannot be empty."}],
                        "isError": True
                    }
                }

            zip_code = str(args.get("zip_code") or DEFAULT_ZIP_CODE).strip()
            supermarkets = _sanitize_supermarkets(args.get("supermarkets"))
            try:
                limit = int(args.get("limit") or 20)
            except (ValueError, TypeError):
                limit = 20

            data = search_offers(query, zip_code=zip_code, supermarkets=supermarkets, limit=limit)
            text_result = json.dumps(data, indent=2, ensure_ascii=False)

        elif tool_name == "compare_supermarket_prices":
            query = str(args.get("query") or "").strip()
            if not query:
                return {
                    "jsonrpc": "2.0",
                    "id": req_id,
                    "result": {
                        "content": [{"type": "text", "text": "Error: 'query' argument is required and cannot be empty."}],
                        "isError": True
                    }
                }

            zip_code = str(args.get("zip_code") or DEFAULT_ZIP_CODE).strip()
            supermarkets = _sanitize_supermarkets(args.get("supermarkets"))

            data = compare_prices(query, zip_code=zip_code, supermarkets=supermarkets)
            text_result = json.dumps(data, indent=2, ensure_ascii=False)

        elif tool_name == "list_supported_supermarkets":
            data = list_supported_supermarkets()
            text_result = json.dumps(data, indent=2, ensure_ascii=False)

        else:
            return {
                "jsonrpc": "2.0",
                "id": req_id,
                "error": {
                    "code": -32601,
                    "message": f"Method or tool '{tool_name}' not found."
                }
            }

        return {
            "jsonrpc": "2.0",
            "id": req_id,
            "result": {
                "content": [
                    {
                        "type": "text",
                        "text": text_result
                    }
                ],
                "isError": False
            }
        }

    except Exception as e:
        logger.exception(f"Exception during tool execution '{tool_name}': {e}")
        return {
            "jsonrpc": "2.0",
            "id": req_id,
            "result": {
                "content": [
                    {
                        "type": "text",
                        "text": f"Tool execution failed: {type(e).__name__}: {str(e)}"
                    }
                ],
                "isError": True
            }
        }


def process_message(line: str) -> Optional[Dict[str, Any]]:
    if not line.strip():
        return None

    try:
        msg = json.loads(line)
    except json.JSONDecodeError as e:
        logger.error(f"Malformed JSON-RPC received: {e}")
        return {
            "jsonrpc": "2.0",
            "id": None,
            "error": {"code": -32700, "message": f"Parse error: {str(e)}"}
        }

    method = msg.get("method")
    req_id = msg.get("id")
    params = msg.get("params", {})

    # Notifications do not return a response
    if req_id is None:
        if method == "notifications/initialized":
            logger.info("Client acknowledged initialized notification.")
        elif method == "notifications/cancelled":
            logger.info(f"Client cancelled operation: {params}")
        return None

    # Standard Requests
    if method == "initialize":
        return handle_initialize(req_id, params)
    elif method == "tools/list":
        return handle_tools_list(req_id)
    elif method == "tools/call":
        return handle_tools_call(req_id, params)
    elif method == "ping":
        return {"jsonrpc": "2.0", "id": req_id, "result": {}}
    else:
        return {
            "jsonrpc": "2.0",
            "id": req_id,
            "error": {
                "code": -32601,
                "message": f"Method '{method}' is not supported."
            }
        }


def _sig_handler(sig, frame):
    logger.info(f"Received termination signal {sig}. Exiting gracefully.")
    sys.exit(0)


def main():
    signal.signal(signal.SIGINT, _sig_handler)
    signal.signal(signal.SIGTERM, _sig_handler)

    logger.info(f"Starting {SERVER_NAME} v{SERVER_VERSION} (stdio mode)...")

    try:
        for raw_line in sys.stdin:
            response = process_message(raw_line)
            if response is not None:
                output = json.dumps(response, ensure_ascii=False) + "\n"
                sys.stdout.write(output)
                sys.stdout.flush()
    except (KeyboardInterrupt, BrokenPipeError):
        pass
    except Exception as e:
        logger.exception(f"Fatal error in stdio loop: {e}")
        sys.exit(1)

    logger.info("Marktguru MCP server shutdown complete.")


if __name__ == "__main__":
    main()