aaronrene / gabriel-muse public
server.py python
185 lines 7.0 KB
Raw
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8 docs(KD-STAGING): sync governance after KD-6b DONE Human 13 days ago
1 """Muse MCP server — JSON-RPC 2.0 engine with domain activation gate (§KD-5a.E)."""
2
3 from __future__ import annotations
4
5 import json
6 import logging
7 import pathlib
8 from typing import Any
9
10 from muse.core.repo import find_repo_root
11 from muse.mcp.bootstrap import bootstrap_knowtation_mcp
12 from muse.mcp.errors import (
13 INTERNAL_ERROR,
14 INVALID_PARAMS,
15 INVALID_REQUEST,
16 METHOD_NOT_FOUND,
17 MCPApplicationError,
18 TOOL_ERROR,
19 error_response,
20 success_response,
21 )
22 from muse.mcp.namespaces.knowtation import (
23 PROMPT_NAMES,
24 TOOL_NAMES,
25 KnowtationNamespace,
26 )
27 from muse.plugins.knowtation.detector import classify_repo
28 from muse.plugins.registry import read_domain
29
30 logger = logging.getLogger(__name__)
31
32 _PROTOCOL_VERSION = "2024-11-05"
33 _SERVER_NAME = "muse-mcp"
34 _SERVER_VERSION = "0.2.0"
35
36
37 class MuseMCPServer:
38 """Lightweight MCP server bound to a single Muse repository root."""
39
40 def __init__(self, root: pathlib.Path | None = None) -> None:
41 self.root = root or find_repo_root() or pathlib.Path.cwd()
42 self._domain = read_domain(self.root)
43 self._knowtation: KnowtationNamespace | None = None
44 self._detected = classify_repo(self.root)
45 if self._domain == "knowtation":
46 bootstrap_knowtation_mcp(self.root)
47 self._knowtation = KnowtationNamespace(self.root)
48
49 @property
50 def capabilities(self) -> dict[str, Any]:
51 """Advisory detection hint — never auto-mounts namespaces."""
52 return {
53 "active_domain": self._domain,
54 "detected_domain": self._detected.domain,
55 "detected_confidence": self._detected.confidence,
56 "knowtation_mounted": self._knowtation is not None,
57 }
58
59 def handle_request(self, raw: dict[str, Any]) -> dict[str, Any] | None:
60 """Process one JSON-RPC request dict; return response or None (notification)."""
61 req_id = raw.get("id")
62 is_notification = "id" not in raw
63 method = raw.get("method")
64 if not isinstance(method, str):
65 if is_notification:
66 return None
67 return error_response(req_id, INVALID_REQUEST, "Missing method")
68
69 params = raw.get("params")
70 if params is None:
71 params_obj: dict[str, Any] = {}
72 elif isinstance(params, dict):
73 params_obj = params
74 else:
75 if is_notification:
76 return None
77 return error_response(req_id, INVALID_PARAMS, "params must be an object")
78
79 try:
80 result = self._dispatch(method, params_obj)
81 if is_notification:
82 return None
83 return success_response(req_id, result)
84 except MCPApplicationError as exc:
85 if is_notification:
86 return None
87 return error_response(
88 req_id,
89 TOOL_ERROR,
90 exc.message,
91 data=exc.to_data(),
92 )
93 except Exception:
94 logger.exception("MCP internal error on method %s", method)
95 if is_notification:
96 return None
97 return error_response(req_id, INTERNAL_ERROR, "Internal error")
98
99 def _dispatch(self, method: str, params: dict[str, Any]) -> Any:
100 match method:
101 case "initialize":
102 return {
103 "protocolVersion": _PROTOCOL_VERSION,
104 "capabilities": {
105 "tools": {"listChanged": False},
106 "resources": {"subscribe": False, "listChanged": False},
107 "prompts": {"listChanged": False},
108 },
109 "serverInfo": {"name": _SERVER_NAME, "version": _SERVER_VERSION},
110 "meta": self.capabilities,
111 }
112 case "tools/list":
113 if self._knowtation is None:
114 return {"tools": []}
115 return {"tools": self._knowtation.list_tools()}
116 case "tools/call":
117 name = params.get("name")
118 if not isinstance(name, str):
119 raise MCPApplicationError("INVALID_PARAMS", "tool name required")
120 if self._knowtation is None:
121 raise MCPApplicationError("DOMAIN_INACTIVE", "knowtation namespace not mounted")
122 arguments = params.get("arguments")
123 if arguments is None:
124 arguments = {}
125 if not isinstance(arguments, dict):
126 raise MCPApplicationError("INVALID_PARAMS", "arguments must be object")
127 content = self._knowtation.call_tool(name, arguments)
128 return {
129 "content": [{"type": "text", "text": json.dumps(content, default=str)}],
130 "isError": False,
131 }
132 case "resources/list":
133 if self._knowtation is None:
134 return {"resources": []}
135 return {"resources": self._knowtation.list_resources()}
136 case "resources/read":
137 uri = params.get("uri")
138 if not isinstance(uri, str):
139 raise MCPApplicationError("INVALID_PARAMS", "uri required")
140 if self._knowtation is None:
141 raise MCPApplicationError("DOMAIN_INACTIVE", "knowtation namespace not mounted")
142 body = self._knowtation.read_resource(uri)
143 return {
144 "contents": [
145 {
146 "uri": uri,
147 "mimeType": "application/json",
148 "text": json.dumps(body, default=str),
149 }
150 ]
151 }
152 case "prompts/list":
153 if self._knowtation is None:
154 return {"prompts": []}
155 return {"prompts": self._knowtation.list_prompts()}
156 case "prompts/get":
157 name = params.get("name")
158 if not isinstance(name, str):
159 raise MCPApplicationError("INVALID_PARAMS", "prompt name required")
160 if self._knowtation is None:
161 raise MCPApplicationError("DOMAIN_INACTIVE", "knowtation namespace not mounted")
162 arguments = params.get("arguments") or {}
163 if not isinstance(arguments, dict):
164 raise MCPApplicationError("INVALID_PARAMS", "arguments must be object")
165 return self._knowtation.get_prompt(name, arguments)
166 case "ping":
167 return {}
168 case _:
169 raise MCPApplicationError("METHOD_NOT_FOUND", f"Unknown method: {method}")
170
171
172 def expected_tool_count(domain: str) -> int:
173 """Return expected mounted tool count for tests."""
174 return len(TOOL_NAMES) if domain == "knowtation" else 0
175
176
177 def expected_prompt_count(domain: str) -> int:
178 return len(PROMPT_NAMES) if domain == "knowtation" else 0
179
180
181 __all__ = [
182 "MuseMCPServer",
183 "expected_prompt_count",
184 "expected_tool_count",
185 ]
File History 1 commit
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8 docs(KD-STAGING): sync governance after KD-6b DONE Human 13 days ago