diff options
| author | Andrew Branson <andrew.branson@jolla.com> | 2026-05-15 12:21:39 +0200 |
|---|---|---|
| committer | Andrew Branson <andrew.branson@jolla.com> | 2026-05-15 12:21:39 +0200 |
| commit | 1f14c5483ee111105f94d66fb3b82208946d914a (patch) | |
| tree | d931715d7a3335324fbcdc51fcef1e0cef590235 /src/sailfish_devel_mcp/server.py | |
Initial Sailfish devel MCP
Diffstat (limited to 'src/sailfish_devel_mcp/server.py')
| -rw-r--r-- | src/sailfish_devel_mcp/server.py | 196 |
1 files changed, 196 insertions, 0 deletions
diff --git a/src/sailfish_devel_mcp/server.py b/src/sailfish_devel_mcp/server.py new file mode 100644 index 0000000..4fb2f91 --- /dev/null +++ b/src/sailfish_devel_mcp/server.py @@ -0,0 +1,196 @@ +from __future__ import annotations + +import argparse +import json +import sys +from typing import Any, TextIO + +from . import __version__ +from .config import Config, load_config +from .tools import build_registry, tool_error + + +PROTOCOL_VERSIONS = [ + "2025-11-25", + "2025-06-18", + "2025-03-26", + "2024-11-05", +] + + +class JsonRpcError(Exception): + def __init__(self, code: int, message: str, data: Any | None = None): + super().__init__(message) + self.code = code + self.message = message + self.data = data + + +class McpServer: + def __init__(self, config: Config): + self.config = config + self.registry = build_registry(config) + + def handle(self, message: dict[str, Any]) -> dict[str, Any] | None: + if not isinstance(message, dict): + raise JsonRpcError(-32600, "JSON-RPC message must be an object") + + request_id = message.get("id") + method = message.get("method") + if not method: + raise JsonRpcError(-32600, "JSON-RPC message is missing method") + + if request_id is None: + self._handle_notification(method) + return None + + try: + result = self._dispatch(method, message.get("params") or {}) + return {"jsonrpc": "2.0", "id": request_id, "result": result} + except JsonRpcError as exc: + error: dict[str, Any] = {"code": exc.code, "message": exc.message} + if exc.data is not None: + error["data"] = exc.data + return {"jsonrpc": "2.0", "id": request_id, "error": error} + except Exception as exc: # pragma: no cover - defensive protocol boundary + return { + "jsonrpc": "2.0", + "id": request_id, + "error": {"code": -32603, "message": str(exc)}, + } + + def _handle_notification(self, method: str) -> None: + if method in {"notifications/initialized", "notifications/cancelled"}: + return + + def _dispatch(self, method: str, params: Any) -> dict[str, Any]: + if method == "initialize": + return self._initialize(params) + if method == "ping": + return {} + if method == "tools/list": + return {"tools": [tool.spec for tool in self.registry.values()]} + if method == "tools/call": + return self._call_tool(params) + if method == "resources/list": + return self._resources_list() + if method == "resources/read": + return self._resources_read(params) + if method == "prompts/list": + return {"prompts": []} + if method == "logging/setLevel": + return {} + raise JsonRpcError(-32601, f"method not found: {method}") + + def _initialize(self, params: Any) -> dict[str, Any]: + requested = "" + if isinstance(params, dict): + requested = str(params.get("protocolVersion") or "") + protocol = requested if requested in PROTOCOL_VERSIONS else PROTOCOL_VERSIONS[0] + return { + "protocolVersion": protocol, + "capabilities": { + "tools": {"listChanged": False}, + "resources": {"subscribe": False, "listChanged": False}, + "prompts": {"listChanged": False}, + }, + "serverInfo": { + "name": "sailfish-devel-mcp", + "version": __version__, + }, + "instructions": ( + "Host-side Sailfish OS development tools for devices, builds, " + "OBS, packaging, repositories, and QML checks." + ), + } + + def _call_tool(self, params: Any) -> dict[str, Any]: + if not isinstance(params, dict): + raise JsonRpcError(-32602, "tools/call params must be an object") + name = params.get("name") + if not isinstance(name, str): + raise JsonRpcError(-32602, "tools/call requires a tool name") + if name not in self.registry: + raise JsonRpcError(-32602, f"unknown tool: {name}") + args = params.get("arguments") or {} + if not isinstance(args, dict): + return tool_error("tool arguments must be an object") + try: + return self.registry[name].handler(args) + except ValueError as exc: + return tool_error(str(exc)) + + def _resources_list(self) -> dict[str, Any]: + return { + "resources": [ + { + "uri": "sailfish-devel-mcp://config/effective", + "name": "Effective configuration", + "mimeType": "application/json", + "description": "Resolved server configuration without secrets.", + }, + { + "uri": "sailfish-devel-mcp://help/tools", + "name": "Tool summary", + "mimeType": "text/plain", + "description": "Names and descriptions of exposed tools.", + }, + ] + } + + def _resources_read(self, params: Any) -> dict[str, Any]: + if not isinstance(params, dict) or not isinstance(params.get("uri"), str): + raise JsonRpcError(-32602, "resources/read requires a uri") + uri = params["uri"] + if uri == "sailfish-devel-mcp://config/effective": + text = json.dumps(self.config.public_dict(), indent=2, sort_keys=True) + return {"contents": [{"uri": uri, "mimeType": "application/json", "text": text}]} + if uri == "sailfish-devel-mcp://help/tools": + text = "\n".join( + f"{tool.spec['name']}: {tool.spec.get('description', '')}" + for tool in self.registry.values() + ) + return {"contents": [{"uri": uri, "mimeType": "text/plain", "text": text}]} + raise JsonRpcError(-32602, f"unknown resource: {uri}") + + +def run_stdio(server: McpServer, stdin: TextIO = sys.stdin, stdout: TextIO = sys.stdout) -> None: + for line in stdin: + if not line.strip(): + continue + try: + message = json.loads(line) + except json.JSONDecodeError as exc: + response = { + "jsonrpc": "2.0", + "id": None, + "error": {"code": -32700, "message": f"parse error: {exc}"}, + } + else: + response = server.handle(message) + if response is not None: + stdout.write(json.dumps(response, separators=(",", ":")) + "\n") + stdout.flush() + + +def main(argv: list[str] | None = None) -> None: + parser = argparse.ArgumentParser(description="Sailfish OS development MCP server") + parser.add_argument("--config", help="Path to config.json") + parser.add_argument( + "--dump-config", + action="store_true", + help="Print the resolved config and exit", + ) + args = parser.parse_args(argv) + + config = load_config(args.config) + if args.dump_config: + print(json.dumps(config.public_dict(), indent=2, sort_keys=True)) + return + + run_stdio(McpServer(config)) + + +if __name__ == "__main__": + main() + |
