shopdb-flask/tools/mcp/shopdb_mcp/client.py

"""HTTP access to a ShopDB instance, and the operation index it publishes.

Everything the server knows about the API is fetched from the instance at
runtime: the operation index, the schemas, the mount path. Nothing is generated
ahead of time and nothing is pinned, so a site running an older or newer build
is described accurately rather than according to whatever was true when this
package was written.
"""

from __future__ import annotations

import json
import os
import re
import threading
import time
from dataclasses import dataclass
from typing import Any
import http.client as http_client
from urllib import error, parse, request

# How long a fetched index or spec stays usable before it is re-read. Long
# enough that a burst of tool calls does not re-download a 700 KB document,
# short enough that enabling a plugin shows up without restarting the server.
CACHE_SECONDS = 300

DEFAULT_TIMEOUT = 30

# Verbs that change or destroy data. Named so the tool descriptions can say so:
# a model told that DELETE is irreversible behaves better than one left to infer
# it from the verb.
WRITE_METHODS = {'POST', 'PUT', 'PATCH', 'DELETE'}
DESTRUCTIVE_METHODS = {'DELETE'}


class ShopdbError(RuntimeError):
    """An error worth showing the caller verbatim."""


@dataclass
class Operation:
    method: str
    path: str
    tag: str
    auth: str
    summary: str

    def matches(self, needle: str) -> bool:
        blob = f'{self.method} {self.path} {self.tag} {self.summary}'.lower()
        return all(word in blob for word in needle.lower().split())


class ShopdbClient:
    """Talks to one ShopDB instance with one credential."""

    def __init__(self, base_url: str | None = None, token: str | None = None,
                 verify_tls: bool | None = None, timeout: int = DEFAULT_TIMEOUT):
        self.base_url = (base_url or os.environ.get('SHOPDB_URL') or '').rstrip('/')
        self.token = token or os.environ.get('SHOPDB_TOKEN') or ''
        if verify_tls is None:
            # Sites commonly run a self-signed certificate on an internal name,
            # so this has to be settable - but it defaults to verifying, and the
            # caller has to say otherwise deliberately.
            verify_tls = (os.environ.get('SHOPDB_VERIFY_TLS', '1').strip().lower()
                          not in ('0', 'false', 'no'))
        self.verify_tls = verify_tls
        self.timeout = timeout
        self._lock = threading.Lock()
        self._cache: dict[str, tuple[float, Any]] = {}

        if not self.base_url:
            raise ShopdbError(
                'SHOPDB_URL is not set. Point it at the instance, including any '
                'mount path, for example https://shopdb.example.net/shopdb')

    # ---- plumbing --------------------------------------------------------

    def _opener(self):
        if self.verify_tls:
            return request.build_opener()
        import ssl
        ctx = ssl.create_default_context()
        ctx.check_hostname = False
        ctx.verify_mode = ssl.CERT_NONE
        return request.build_opener(request.HTTPSHandler(context=ctx))

    def request(self, method: str, path: str, params: dict | None = None,
                body: Any = None) -> tuple[int, Any]:
        """One HTTP call. Returns (status, parsed-body-or-text)."""
        # Encode what a caller typed. A model filling an id into a templated
        # path can easily produce a space or an accent; leaving those raw turns
        # a legitimate request into an exception rather than a 404. '/' and the
        # query delimiters stay literal so an already-formed path still works.
        url = self.base_url + '/' + parse.quote(path.lstrip('/'), safe='/?=&%+,')
        if params:
            # Drop empty values rather than sending key= , which several list
            # endpoints treat as a filter on the empty string.
            clean = {k: v for k, v in params.items() if v not in (None, '')}
            if clean:
                url += ('&' if '?' in url else '?') + parse.urlencode(clean, doseq=True)

        data = None
        headers = {'Accept': 'application/json'}
        if self.token:
            headers['Authorization'] = f'Bearer {self.token}'
        if body is not None:
            data = json.dumps(body).encode('utf-8')
            headers['Content-Type'] = 'application/json'

        try:
            req = request.Request(url, data=data, headers=headers,
                                  method=method.upper())
            with self._opener().open(req, timeout=self.timeout) as response:
                return response.status, _decode(response.read(),
                                                response.headers.get('Content-Type'))
        except error.HTTPError as exc:
            # An error body from this API is the useful part (it carries the
            # error code and message), so return it rather than raising.
            return exc.code, _decode(exc.read(), exc.headers.get('Content-Type'))
        except error.URLError as exc:
            raise ShopdbError(f'could not reach {self.base_url}: {exc.reason}') from exc
        except (OSError, ValueError, http_client.HTTPException) as exc:
            # Catching only the two urllib errors let ordinary mistakes escape
            # as raw exceptions: a space in a path raises InvalidURL, a non
            # ASCII path raises UnicodeEncodeError, a read that outruns the
            # timeout raises TimeoutError. The caller then saw a bare "error
            # executing tool" with no cause and nothing to try differently.
            raise ShopdbError(
                f'{method.upper()} {path} could not be sent: '
                f'{type(exc).__name__}: {exc}') from exc

    def _cached(self, key: str, fetch):
        with self._lock:
            hit = self._cache.get(key)
            if hit and (time.time() - hit[0]) < CACHE_SECONDS:
                return hit[1]
        value = fetch()
        with self._lock:
            self._cache[key] = (time.time(), value)
        return value

    # ---- what the instance publishes about itself ------------------------

    def operations(self) -> list[Operation]:
        """The published operation index, parsed.

        Read from /api/docs/operations.txt rather than the full spec: the index
        is around 70 KB against 700 KB, and listing what exists does not need
        the schemas.
        """
        def fetch():
            status, text = self.request('GET', '/api/docs/operations.txt')
            if status != 200:
                raise ShopdbError(
                    f'operation index unavailable (HTTP {status}). This instance '
                    'may predate /api/docs/operations.txt.')
            return _parse_index(text if isinstance(text, str) else str(text))
        return self._cached('operations', fetch)

    def spec(self, tag: str | None = None) -> dict:
        """The OpenAPI spec, optionally narrowed to one tag group."""
        def fetch():
            path = '/api/docs/openapi.json'
            status, payload = self.request('GET', path, params={'tag': tag} if tag else None)
            if status != 200:
                raise ShopdbError(f'spec unavailable (HTTP {status}): {payload}')
            return payload
        return self._cached(f'spec:{tag or "*"}', fetch)

    def find_operation(self, method: str, path: str) -> dict | None:
        """The spec entry for one operation, fetched by its tag to stay small."""
        method = method.lower()
        for op in self.operations():
            if op.method.lower() == method and op.path == path:
                spec = self.spec(op.tag)
                return (spec.get('paths', {}).get(path, {}) or {}).get(method)
        return None


_INDEX_LINE = re.compile(
    r'^(?P<method>[A-Z]+)\s+(?P<path>\S+)\s+\[(?P<tag>[^\]]*)\]\s+'
    r'\((?P<auth>[^)]*)\)\s*(?P<summary>.*)$')


def _parse_index(text: str) -> list[Operation]:
    out = []
    for line in text.splitlines():
        line = line.strip()
        if not line or line.startswith('#'):
            continue
        m = _INDEX_LINE.match(line)
        if m:
            out.append(Operation(m.group('method'), m.group('path'),
                                 m.group('tag'), m.group('auth'),
                                 m.group('summary').strip()))
    return out


def _decode(raw: bytes, content_type: str | None):
    text = raw.decode('utf-8', errors='replace')
    if content_type and 'json' in content_type.lower():
        try:
            return json.loads(text)
        except json.JSONDecodeError:
            return text
    return text