"""MCP server for a ShopDB instance. WHY IT IS SHAPED LIKE THIS. ShopDB publishes 483 operations. Exposing one tool per operation would be useless: MCP clients degrade badly past a few dozen tools and a model starts picking the wrong one. So the surface here is small and generic - find an operation, read its contract, call it - and the instance's own published index is what makes that navigable. Adding a plugin adds operations that this server can already reach, with no change here. SECURITY MODEL: the token IS the boundary. Every operation the credential is allowed to perform is reachable, including writes and deletes. That is a deliberate choice, and it means the scope of the personal access token decides the blast radius, not this code. Create a read-scoped token for anything exploratory. See README.md. """ from __future__ import annotations import json import os from typing import Any # The SDK renamed FastMCP to MCPServer in 2.0. The two have the same shape for # what this server uses (a .tool() decorator and .run()), so support both rather # than forcing a pin: a site may already have either installed. try: from mcp.server.mcpserver import MCPServer as _Server # mcp >= 2 except ImportError: # pragma: no cover from mcp.server.fastmcp import FastMCP as _Server # mcp 1.x from .client import (DESTRUCTIVE_METHODS, WRITE_METHODS, ShopdbClient, ShopdbError) mcp = _Server('shopdb') _client: ShopdbClient | None = None def client() -> ShopdbClient: global _client if _client is None: _client = ShopdbClient() return _client def _fail(message: str) -> str: """Errors come back as text the model can read and act on. Raising would surface a stack trace the model cannot do anything with; a sentence telling it what went wrong and what to try next is more useful. """ return f'ERROR: {message}' @mcp.tool() def shopdb_whoami() -> str: """Check which ShopDB instance this server targets and who the token is. Call this first: it confirms the instance is reachable and the credential is valid. IMPORTANT: the permissions reported are the ACCOUNT's, not the token's. A scoped personal access token carries its own narrower scope list, which the server enforces but does not publish, so a call can still be refused with 403 even though the account shown here holds the permission. Treat this as "who am I", not "what may I do". """ try: c = client() except ShopdbError as exc: return _fail(str(exc)) try: status, body = c.request('GET', '/api/auth/me') except ShopdbError as exc: return _fail(str(exc)) if status == 401: return _fail('token rejected (HTTP 401). Check SHOPDB_TOKEN.') if status != 200: return _fail(f'HTTP {status} from /api/auth/me: {body}') user = (body or {}).get('data') or {} try: count = len(c.operations()) except ShopdbError as exc: count = f'unknown ({exc})' return json.dumps({ 'instance': c.base_url, 'username': user.get('username'), 'roles': user.get('roles'), # Named account_permissions, not permissions, because that is what it # is. Calling it "permissions" next to a token invited exactly the # wrong reading: that an admin account means every call will succeed, # when a read-scoped token on that same account is refused a write. 'account_permissions': len(user.get('permissions') or []), 'note': ('permissions shown are the account\'s; a scoped token may ' 'allow fewer, and a 403 from shopdb_call means the TOKEN ' 'lacks the scope even if the account holds the permission'), 'operations_published': count, 'tls_verified': c.verify_tls, }, indent=2) @mcp.tool() def shopdb_search_operations(query: str = '', tag: str = '', limit: int = 40) -> str: """Find API operations by keyword. START HERE to discover what exists. Searches method, path, tag and summary. `query` words are ANDed, so "printer supplies" matches an operation mentioning both. `tag` narrows to one group (for example plugin-printers, core-assets). Returns one line per operation. Use shopdb_describe_operation for the parameters and body schema of anything you intend to call. """ try: ops = client().operations() except ShopdbError as exc: return _fail(str(exc)) if tag: ops = [o for o in ops if o.tag == tag] if query: ops = [o for o in ops if o.matches(query)] if not ops: tags = sorted({o.tag for o in client().operations()}) return (f'No operations matched query={query!r} tag={tag!r}.\n' f'Available tags: {", ".join(tags)}') lines = [f'{o.method} {o.path} [{o.tag}] ({o.auth}) {o.summary}' for o in ops[:limit]] more = len(ops) - len(lines) if more > 0: lines.append(f'... and {more} more; narrow the query or raise limit') return '\n'.join(lines) @mcp.tool() def shopdb_describe_operation(method: str, path: str) -> str: """Full contract for one operation: parameters, request body, responses. `path` must be the templated form exactly as shopdb_search_operations printed it, for example /api/machines/{machineid} - not a filled-in URL. Read this before calling anything you have not called before. It also warns when the operation writes or deletes. """ try: op = client().find_operation(method, path) except ShopdbError as exc: return _fail(str(exc)) if op is None: return _fail(f'no operation {method.upper()} {path}. ' 'Use shopdb_search_operations to find the exact path.') out = { 'method': method.upper(), 'path': path, 'summary': op.get('summary'), 'description': op.get('description'), 'parameters': op.get('parameters'), 'requestBody': op.get('requestBody'), 'responses': {k: {'description': v.get('description')} for k, v in (op.get('responses') or {}).items()}, } warning = '' if method.upper() in DESTRUCTIVE_METHODS: warning = ('\n\nWARNING: this DELETES data and cannot be undone. ' 'Confirm with the user before calling it.') elif method.upper() in WRITE_METHODS: warning = '\n\nNote: this MODIFIES data.' return json.dumps(out, indent=2, default=str) + warning @mcp.tool() def shopdb_call(method: str, path: str, params: dict[str, Any] | None = None, body: dict[str, Any] | None = None) -> str: """Call any ShopDB operation. Use shopdb_describe_operation first. `path` may contain filled-in ids (/api/machines/42). `params` become the query string; `body` is sent as JSON. Every response is the envelope {status, data, meta} - the payload you want is under "data", not at the top level. List endpoints paginate with page and perpage, and perpage is silently capped at 100, so asking for 500 returns 100 without an error. This can write and delete if the token allows it. For DELETE, confirm with the user first. """ try: c = client() except ShopdbError as exc: return _fail(str(exc)) verb = method.upper() try: status, payload = c.request(verb, path, params=params, body=body) except ShopdbError as exc: return _fail(str(exc)) if status == 401: return _fail('HTTP 401. The token is missing, expired or rejected.') if status == 403: return _fail('HTTP 403. The token lacks the scope or role for this ' 'operation - check shopdb_whoami.') if status == 404 and isinstance(payload, str): return _fail(f'HTTP 404 and a non-JSON body, which usually means the ' f'path is wrong rather than the record missing: {path}') text = payload if isinstance(payload, str) else json.dumps(payload, indent=2, default=str) if len(text) > 100_000: text = text[:100_000] + '\n... truncated; narrow with params or paginate' return f'HTTP {status}\n{text}' @mcp.tool() def shopdb_llms_guide() -> str: """The instance's own orientation guide for agents. Short. Covers the response envelope, authentication and how the rest of the documentation is laid out. Worth reading once at the start of a session. """ try: status, text = client().request('GET', '/api/docs/llms.txt') except ShopdbError as exc: return _fail(str(exc)) if status != 200: return _fail(f'guide unavailable (HTTP {status})') return text if isinstance(text, str) else str(text) def main() -> None: if not os.environ.get('SHOPDB_URL'): raise SystemExit( 'SHOPDB_URL is not set.\n' ' SHOPDB_URL=https://shopdb.example.net/shopdb\n' ' SHOPDB_TOKEN=\n' 'See tools/mcp/README.md.') mcp.run() if __name__ == '__main__': main()