epicure_client.py

"""Minimal MCP (Model Context Protocol) client for the Epicure server by Kaikaku.

The server (https://github.com/KAIKAKU-AI/epicure-mcp) is public, anonymous and
stateless, exposing 13 flavour-analysis tools (find_pairings, pairing_score,
neighbors, ...) over MCP streamable-HTTP. We speak raw JSON-RPC over httpx
instead of pulling in the full `mcp` SDK — small, tinkerable, easy to debug.
"""

import itertools
import json
import os
import time

import httpx

EPICURE_URL = os.environ.get("EPICURE_MCP_URL", "https://epicure-mcp.kaikaku.ai/mcp")
PROTOCOL_VERSION = "2025-03-26"


class EpicureMCP:
    def __init__(self, url: str = EPICURE_URL, timeout: float = 20.0,
                 cache_path: str | None = None):
        self.url = url
        self._ids = itertools.count(1)
        self._session_id = None
        self._client = httpx.Client(timeout=timeout)
        self._initialized = False
        self._tools = None
        # The server is deterministic and stateless, so tool results are safely
        # memoizable. A disk cache makes the demo independent of the public
        # endpoint's rate limit: warm entries never touch the network.
        self._cache_path = cache_path or os.path.join(
            os.path.dirname(os.path.abspath(__file__)), "epicure_cache.json")
        self._min_interval = 0.4  # throttle only real network calls, not cache hits
        self._last_call = 0.0
        self._cache = {}
        if os.path.exists(self._cache_path):
            try:
                with open(self._cache_path) as fh:
                    self._cache = json.load(fh)
            except (json.JSONDecodeError, OSError):
                self._cache = {}

    # ------------------------------------------------------------------ rpc

    def _headers(self) -> dict:
        headers = {
            "Content-Type": "application/json",
            "Accept": "application/json, text/event-stream",
        }
        if self._session_id:
            headers["Mcp-Session-Id"] = self._session_id
        return headers

    @staticmethod
    def _parse_body(response: httpx.Response) -> dict | None:
        """Handle both plain-JSON and SSE-framed (`data: {...}`) responses."""
        content_type = response.headers.get("content-type", "")
        if "text/event-stream" in content_type:
            for line in response.text.splitlines():
                if line.startswith("data:"):
                    return json.loads(line[len("data:"):].strip())
            return None
        if response.text.strip():
            return json.loads(response.text)
        return None

    def _rpc(self, method: str, params: dict | None = None, *, notification: bool = False,
             max_retries: int = 4):
        payload = {"jsonrpc": "2.0", "method": method}
        if params is not None:
            payload["params"] = params
        if not notification:
            payload["id"] = next(self._ids)

        # The public server is flaky: it rate-limits (429/503) AND randomly 404s
        # individual requests that succeed on retry. Retry all three with backoff
        # so transient failures degrade to slow, not silently-ungrounded.
        for attempt in range(max_retries + 1):
            response = self._client.post(self.url, json=payload, headers=self._headers())
            if (not notification and response.status_code in (429, 503, 404)
                    and attempt < max_retries):
                wait = float(response.headers.get("Retry-After", 0)) or min(2 ** attempt, 8)
                time.sleep(wait)
                continue
            break

        if session_id := response.headers.get("mcp-session-id"):
            self._session_id = session_id
        # Notifications are fire-and-forget: the server may answer 202/404 with no
        # body, and that's fine — never let it abort the handshake.
        if notification:
            return None
        response.raise_for_status()
        body = self._parse_body(response)
        if body and "error" in body:
            raise RuntimeError(f"Epicure MCP error on {method}: {body['error']}")
        return body["result"] if body else None

    # ---------------------------------------------------------------- public

    def ensure_initialized(self) -> None:
        if self._initialized:
            return
        self._rpc(
            "initialize",
            {
                "protocolVersion": PROTOCOL_VERSION,
                "capabilities": {},
                "clientInfo": {"name": "epicurean-simmer", "version": "0.1.0"},
            },
        )
        self._rpc("notifications/initialized", {}, notification=True)
        self._initialized = True

    def list_tools(self) -> list[dict]:
        """Return [{name, description, inputSchema}, ...], cached after first call."""
        if self._tools is None:
            self.ensure_initialized()
            result = self._rpc("tools/list", {})
            self._tools = result.get("tools", [])
        return self._tools

    def call_tool(self, name: str, arguments: dict) -> str:
        """Call a tool and flatten its content blocks to a plain string.

        Results are memoized to disk (the server is deterministic), so warm
        entries are returned without a network call — keeping the demo immune
        to the public endpoint's rate limit.
        """
        key = name + ":" + json.dumps(arguments, sort_keys=True)
        if key in self._cache:
            return self._cache[key]

        gap = self._min_interval - (time.monotonic() - self._last_call)
        if gap > 0:
            time.sleep(gap)
        self._last_call = time.monotonic()

        self.ensure_initialized()
        result = self._rpc("tools/call", {"name": name, "arguments": arguments})
        blocks = result.get("content", []) if result else []
        texts = [block.get("text", "") for block in blocks if block.get("type") == "text"]
        text = "\n".join(texts) if texts else json.dumps(result)

        # Don't cache transient error payloads (resolution failures, etc.).
        if not text.lstrip().startswith('{\n  "error"') and '"error":' not in text[:40]:
            self._cache[key] = text
            self._save_cache()
        return text

    def _save_cache(self) -> None:
        try:
            with open(self._cache_path, "w") as fh:
                json.dump(self._cache, fh, indent=0)
        except OSError:
            pass


if __name__ == "__main__":
    # Smoke test: python epicure_client.py
    mcp = EpicureMCP()
    tools = mcp.list_tools()
    print(f"{len(tools)} tools available:")
    for tool in tools:
        print(f"  - {tool['name']}: {tool.get('description', '')[:80]}")