summaryrefslogtreecommitdiff
path: root/src/sailfish_devel_mcp/server.py
diff options
context:
space:
mode:
authorAndrew Branson <andrew.branson@jolla.com>2026-05-15 12:21:39 +0200
committerAndrew Branson <andrew.branson@jolla.com>2026-05-15 12:21:39 +0200
commit1f14c5483ee111105f94d66fb3b82208946d914a (patch)
treed931715d7a3335324fbcdc51fcef1e0cef590235 /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.py196
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()
+