1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
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()
|
