"""MCP stdio server exposing Api01 (acme_client.py) as tools. Generated by mimicplus. Register with an MCP client, e.g. Claude Desktop / Cursor ``mcpServers`` config: {"acme_client": {"command": "python", "args": ["acme_client_mcp.py"]}} Stdlib only. Writes exposed: False. Logs go to stderr (stdout is the protocol channel). """ # ruff: noqa from __future__ import annotations import importlib.util import json import pathlib import sys import traceback from typing import Any PROTOCOL_VERSION = '2025-06-18' SUPPORTED_VERSIONS = ("2025-06-18", "2025-03-26", "2024-11-05") SERVER_INFO = {"name": "mimicplus-acme_client", "title": "Api01 (mimicplus)", "version": '0.4.0'} MAX_TEXT = 100_000 _here = pathlib.Path(__file__).resolve().parent _spec = importlib.util.spec_from_file_location('acme_client', _here / 'acme_client.py') assert _spec is not None and _spec.loader is not None _mod = importlib.util.module_from_spec(_spec) sys.modules['acme_client'] = _mod _spec.loader.exec_module(_mod) TOOLS: list[dict[str, Any]] = [{'name': 'get_me', 'title': 'Get me', 'description': 'GET /api/me on 127.0.0.1.', 'inputSchema': {'type': 'object', 'properties': {}, 'required': [], 'additionalProperties': False}, 'annotations': {'readOnlyHint': True, 'destructiveHint': False, 'idempotentHint': True, 'openWorldHint': True}}, {'name': 'list_products', 'title': 'List products', 'description': 'GET /api/products on 127.0.0.1. Returns results in `items`.', 'inputSchema': {'type': 'object', 'properties': {'page': {'type': 'integer', 'description': "query field 'page'", 'default': '1'}, 'per_page': {'type': 'integer', 'description': "query field 'per_page'", 'default': '2'}, 'max_items': {'type': 'integer', 'minimum': 1, 'maximum': 1000, 'description': 'fetch across pages until this many ' 'result items (paginated)'}}, 'required': [], 'additionalProperties': False}, 'annotations': {'readOnlyHint': True, 'destructiveHint': False, 'idempotentHint': True, 'openWorldHint': True}}, {'name': 'get_product', 'title': 'Get product', 'description': 'GET /api/products/{product_id} on 127.0.0.1.', 'inputSchema': {'type': 'object', 'properties': {'product_id': {'type': 'integer', 'description': "path segment, e.g. '102'"}}, 'required': ['product_id'], 'additionalProperties': False}, 'annotations': {'readOnlyHint': True, 'destructiveHint': False, 'idempotentHint': True, 'openWorldHint': True}}, {'name': 'search_products', 'title': 'Search products', 'description': 'GraphQL query SearchProducts on 127.0.0.1. Returns results in ' '`data.searchProducts.nodes`.', 'inputSchema': {'type': 'object', 'properties': {'q': {'type': 'string', 'description': "body field 'variables.q' (free-text " 'search)'}, 'first': {'type': 'integer', 'description': "body field 'variables.first'", 'default': 5}, 'max_items': {'type': 'integer', 'minimum': 1, 'maximum': 1000, 'description': 'fetch across pages until this many ' 'result items (paginated)'}}, 'required': ['q'], 'additionalProperties': False}, 'annotations': {'readOnlyHint': True, 'destructiveHint': False, 'idempotentHint': True, 'openWorldHint': True}}, {'name': 'get_order', 'title': 'Get order', 'description': 'GraphQL query GetOrder on 127.0.0.1.', 'inputSchema': {'type': 'object', 'properties': {'id': {'type': 'string', 'description': "body field 'variables.id'", 'default': 'o-1'}}, 'required': [], 'additionalProperties': False}, 'annotations': {'readOnlyHint': True, 'destructiveHint': False, 'idempotentHint': True, 'openWorldHint': True}}] PAGINATED = set(['list_products']) _BY_NAME = {t["name"]: t for t in TOOLS} _client: Any = None def _get_client() -> Any: global _client if _client is None: _client = _mod.Api01() return _client def _call_tool(name: str, args: dict[str, Any]) -> dict[str, Any]: tool = _BY_NAME.get(name) if tool is None: raise KeyError(f"unknown tool {name!r}") allowed = set(tool["inputSchema"]["properties"]) unknown = sorted(set(args) - allowed) if unknown: return {"content": [{"type": "text", "text": f"unknown arguments {unknown}"}], "isError": True} missing = [r for r in tool["inputSchema"]["required"] if r not in args] if missing: return {"content": [{"type": "text", "text": f"missing required arguments {missing}"}], "isError": True} args = dict(args) max_items = args.pop("max_items", None) client = _get_client() try: if max_items is not None and name in PAGINATED: result: Any = list(client.paginate(name, max_items=int(max_items), max_pages=50, **args)) elif max_items is not None: result = client.items(name, getattr(client, name)(**args))[: int(max_items)] else: result = getattr(client, name)(**args) except _mod.MimicError as e: return {"content": [{"type": "text", "text": f"{type(e).__name__}: {e}"}], "isError": True} text = json.dumps(result, default=str, ensure_ascii=False) out: dict[str, Any] = {"content": [{"type": "text", "text": text[:MAX_TEXT]}], "isError": False} if isinstance(result, dict): out["structuredContent"] = result return out def handle(msg: dict[str, Any]) -> dict[str, Any] | None: """One JSON-RPC message in, one response out (None for notifications).""" mid, method, params = msg.get("id"), msg.get("method"), msg.get("params") or {} if mid is None: # notification (e.g. notifications/initialized) return None try: if method == "initialize": want = params.get("protocolVersion") result: Any = {"protocolVersion": want if want in SUPPORTED_VERSIONS else PROTOCOL_VERSION, "capabilities": {"tools": {"listChanged": False}}, "serverInfo": SERVER_INFO, "instructions": "Tools call a learned HTTP API. Read-only tools are safe to retry."} elif method == "ping": result = {} elif method == "tools/list": result = {"tools": TOOLS} elif method == "tools/call": result = _call_tool(str(params.get("name")), dict(params.get("arguments") or {})) else: return {"jsonrpc": "2.0", "id": mid, "error": {"code": -32601, "message": f"method not found: {method}"}} except KeyError as e: return {"jsonrpc": "2.0", "id": mid, "error": {"code": -32602, "message": str(e.args[0] if e.args else e)}} except Exception as e: # never crash the server on one bad call traceback.print_exc(file=sys.stderr) return {"jsonrpc": "2.0", "id": mid, "error": {"code": -32603, "message": f"{type(e).__name__}: {e}"}} return {"jsonrpc": "2.0", "id": mid, "result": result} def serve(stdin: Any = None, stdout: Any = None) -> None: stdin, stdout = stdin or sys.stdin, stdout or sys.stdout for line in stdin: line = line.strip() if not line: continue try: msg = json.loads(line) except ValueError: resp: Any = {"jsonrpc": "2.0", "id": None, "error": {"code": -32700, "message": "parse error"}} else: if isinstance(msg, list): resp = [r for r in (handle(m) for m in msg if isinstance(m, dict)) if r is not None] or None else: resp = handle(msg) if isinstance(msg, dict) else {"jsonrpc": "2.0", "id": None, "error": {"code": -32600, "message": "invalid request"}} if resp is not None: stdout.write(json.dumps(resp, default=str) + "\n") stdout.flush() if __name__ == "__main__": serve()