ge-aerospace/shopdb-flask
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Make the API usable by an agent, and stop the contract going stale

Browse Source 
Someone writing a script against ShopDB had two options: read a 705 KB spec, or
guess. The spec is the reference, not an index, and handing an agent the whole
thing means either truncating it or spending a context window on 479 operations
nobody asked about.

Three additions, each feeding the next.

/api/docs/operations.txt is the index: every operation on one line with its
method, path, tag, auth and summary, at around 70 KB for all 486. A caller can
hold the entire API at once, find the four operations it needs, and only then
fetch their schemas.

/api/docs/openapi.json?tag=... narrows the spec to one of the 41 tag groups.
Components are kept intact deliberately: pruning the schema graph to match would
leave references resolving to nothing, and a generator that fails confusingly is
worse than a file that is larger than strictly necessary.

tools/mcp/ is an MCP server built on both. Five tools rather than 483, because a
tool per operation is unusable - clients degrade past a few dozen and models
start picking the wrong one. It reads everything from the live instance at call
time, so enabling a plugin adds operations it can already reach with no change
here. The token is the security boundary and the README says so plainly: every
operation the credential permits is reachable, so exploratory work wants a
read-scoped token.

The rest of this is about the contract not drifting, which is what makes any of
the above trustworthy.

Two CI jobs:

- api-contract runs the parity gates and then regenerates the spec and fails if
  anything changed. Those gates already existed and already worked - they caught
  both routes added here, unprompted - but nothing ran them except by hand, and
  the committed spec had silently drifted by three operations, including two
  setup endpoints added the day before.
- migrations-mysql is the job the header comment has been claiming for some
  time while not existing. It described "the REAL multi-site deploy path" and
  "what catches an Alembic regression on MySQL before it ships", and there was
  no such job. That is worse than no comment: it is the false confidence that
  let a migration ship which could not build a schema on a fresh MySQL database.
  The suite runs on SQLite, and SQLite accepts a foreign key onto a table that
  does not exist.

CONTRIBUTING.md now says to run gen_openapi.py --refresh after changing a route,
and that a route the sweep cannot classify belongs in PUBLIC_ENDPOINTS with a
reason rather than hand-edited into the inventory, which the next refresh would
overwrite. The new index route is declared there alongside the spec it indexes.

(cherry picked from commit 762d1b7210)
This commit is contained in:
cproudlock
2026-09-20 12:59:49 -04:00
parent 39aaa5d0cb
commit 1e42356aca
10 changed files with 862 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

191
tools/mcp/shopdb_mcp/client.py Normal file
View File

@@ -0,0 +1,191 @@
"""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
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)."""
url = self.base_url + '/' + path.lstrip('/')
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'
req = request.Request(url, data=data, headers=headers, method=method.upper())
try:
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
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