Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-mcp/src/lexigram/ai/mcp/server/core.py: 20%
102 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
1"""MCPServer — JSON-RPC message router for the Model Context Protocol."""
3from __future__ import annotations
5from typing import Any
7from lexigram.ai.mcp.types import (
8 MCPServerCapabilities,
9 MCPServerInfo,
10)
11from lexigram.contracts.mcp.protocols import MCPAuthorizerProtocol
12from lexigram.logging import (
13 get_logger,
14)
15from lexigram.result import Result
17logger = get_logger(__name__)
19JSONRPC_VERSION = "2.0"
20MCP_PROTOCOL_VERSION = "2024-11-05"
22_PRE_INIT_METHODS: frozenset[str] = frozenset(
23 {"initialize", "ping", "notifications/initialized"}
24)
27class MCPServer:
28 """Core MCP server — routes JSON-RPC messages to handlers.
30 The server is transport-agnostic. It receives parsed JSON-RPC
31 messages and returns JSON-RPC responses. The transport layer
32 (stdio or HTTP+SSE) handles serialization and I/O.
34 Usage::
36 server = MCPServer(
37 name="my-app",
38 tool_handler=tool_handler,
39 resource_handler=resource_handler,
40 )
41 response = await server.handle_message(request)
42 """
44 def __init__(
45 self,
46 config: Any | None = None,
47 name: str | None = None,
48 version: str | None = None,
49 tool_handler: Any | None = None,
50 resource_handler: Any | None = None,
51 prompt_handler: Any | None = None,
52 sampling_handler: Any | None = None,
53 logging_handler: Any | None = None,
54 authorizer: MCPAuthorizerProtocol | None = None,
55 allow_unauthenticated: bool = False,
56 ) -> None:
57 """Initialize the MCP server.
59 Args:
60 config: Optional MCP server configuration.
61 name: Server name (overrides config if provided).
62 version: Server version (overrides config if provided).
63 tool_handler: Handler for tool-related methods.
64 resource_handler: Handler for resource-related methods.
65 prompt_handler: Handler for prompt-related methods.
66 sampling_handler: Handler for sampling/createMessage (optional).
67 logging_handler: Handler for logging/setLevel (optional).
68 authorizer: Optional authz hook consulted once per
69 non-handshake request post-initialize. ``None`` means
70 requests are denied with ``-32000`` unless
71 ``allow_unauthenticated`` is set.
72 allow_unauthenticated: When ``True`` (and no ``authorizer``)
73 restore the open posture: non-handshake methods dispatch
74 without authorization. Defaults to ``False`` (fail-closed).
75 """
76 if config is None:
77 from lexigram.ai.mcp.config import MCPConfig
79 config = MCPConfig()
81 self.config = config
82 self._name = name or config.server_name
83 self._version = version or config.server_version
84 self._tool_handler = tool_handler
85 self._resource_handler = resource_handler
86 self._prompt_handler = prompt_handler
87 self._sampling_handler = sampling_handler
88 self._logging_handler = logging_handler
89 self._authorizer = authorizer
90 self._allow_unauthenticated = allow_unauthenticated
91 self._initialized = False
92 self._client_info: dict[str, Any] = {}
94 self._handlers: dict[str, Any] = {}
95 self._register_handlers()
97 def _register_handlers(self) -> None:
98 """Register method handlers for MCP protocol methods."""
99 self._handlers["initialize"] = self._handle_initialize
100 self._handlers["ping"] = self._handle_ping
101 self._handlers["notifications/initialized"] = (
102 self._handle_initialized_notification
103 )
105 if self._tool_handler:
106 self._handlers["tools/list"] = self._tool_handler.list_tools
107 self._handlers["tools/call"] = self._tool_handler.call_tool
109 if self._resource_handler:
110 self._handlers["resources/list"] = self._resource_handler.list_resources
111 self._handlers["resources/read"] = self._resource_handler.read_resource
112 self._handlers["resources/templates/list"] = (
113 self._resource_handler.list_templates
114 )
116 if self._prompt_handler:
117 self._handlers["prompts/list"] = self._prompt_handler.list_prompts
118 self._handlers["prompts/get"] = self._prompt_handler.get_prompt
120 if self._sampling_handler:
121 self._handlers["sampling/createMessage"] = (
122 self._sampling_handler.create_message
123 )
125 if self._logging_handler:
126 self._handlers["logging/setLevel"] = self._logging_handler.set_level
128 async def handle_message(
129 self,
130 message: dict[str, Any],
131 ) -> dict[str, Any] | None:
132 """Handle a JSON-RPC message and return the response.
134 Args:
135 message: Parsed JSON-RPC request message.
137 Returns:
138 JSON-RPC response dict, or None for notifications.
139 """
140 try:
141 # Validate JSON-RPC structure
142 if message.get("jsonrpc") != JSONRPC_VERSION:
143 return self._error_response(
144 None,
145 -32600,
146 "Invalid JSON-RPC version",
147 )
149 method = message.get("method")
150 if not method:
151 return self._error_response(
152 message.get("id"),
153 -32600,
154 "Missing method",
155 )
157 # Check if it's a notification (no id)
158 is_notification = "id" not in message
159 request_id = message.get("id")
161 # Handle the method
162 if method not in self._handlers:
163 return self._error_response(
164 request_id,
165 -32601,
166 f"Method not found: {method}",
167 )
169 handler = self._handlers[method]
170 params = message.get("params", {})
172 if method not in _PRE_INIT_METHODS:
173 if not self._initialized:
174 return self._error_response(
175 request_id, -32002, "Server not initialized"
176 )
177 if self._authorizer is not None:
178 allowed = await self._authorizer.authorize(
179 method=method,
180 params=params,
181 client_info=self._client_info,
182 )
183 if not allowed:
184 return self._error_response(
185 request_id, -32000, "Request not authorized"
186 )
187 elif not self._allow_unauthenticated:
188 return self._error_response(
189 request_id, -32000, "Request not authorized"
190 )
192 try:
193 handler_result = await handler(**params) if params else await handler()
195 # Handlers may return Result[T, E] (e.g. ToolHandler.call_tool).
196 # Unwrap here so the transport layer always receives plain dicts.
197 if isinstance(handler_result, Result):
198 if handler_result.is_ok():
199 handler_result = handler_result.unwrap()
200 else:
201 error = handler_result.unwrap_err()
202 logger.error(
203 "mcp_handler_result_error",
204 method=method,
205 error=str(error),
206 )
207 return self._error_response(
208 request_id,
209 -32603,
210 str(error),
211 )
213 result = handler_result
215 # Notifications don't get responses
216 if is_notification:
217 return None
219 return self._success_response(request_id, result)
221 except (RuntimeError, TypeError, AttributeError, LookupError, OSError) as e:
222 logger.error("mcp_handler_error", method=method, error=str(e))
223 return self._error_response(
224 request_id,
225 -32603,
226 f"Internal error: {e!s}",
227 )
229 except (
230 RuntimeError,
231 TypeError,
232 AttributeError,
233 LookupError,
234 OSError,
235 ValueError,
236 ) as e:
237 logger.error("mcp_message_error", error=str(e))
238 return self._error_response(
239 message.get("id") if isinstance(message, dict) else None,
240 -32603,
241 f"Internal error: {e!s}",
242 )
244 def _success_response(
245 self,
246 request_id: int | str | None,
247 result: Any,
248 ) -> dict[str, Any]:
249 """Create a successful JSON-RPC response."""
250 return {
251 "jsonrpc": JSONRPC_VERSION,
252 "id": request_id,
253 "result": result,
254 }
256 def _error_response(
257 self,
258 request_id: int | str | None,
259 code: int,
260 message: str,
261 ) -> dict[str, Any]:
262 """Create an error JSON-RPC response."""
263 error: dict[str, Any] = {
264 "code": code,
265 "message": message,
266 }
267 return {
268 "jsonrpc": JSONRPC_VERSION,
269 "id": request_id,
270 "error": error,
271 }
273 async def _handle_initialize(
274 self,
275 **params: Any,
276 ) -> dict[str, Any]:
277 """Handle the initialize method."""
278 # Client can send capabilities in params
279 self._client_info = params.get("clientInfo", {})
281 logger.info(
282 "mcp_initialized",
283 client_info=self._client_info,
284 server_name=self._name,
285 )
287 self._initialized = True
289 return {
290 "protocolVersion": MCP_PROTOCOL_VERSION,
291 "capabilities": MCPServerCapabilities(
292 tools=self._tool_handler is not None,
293 resources=self._resource_handler is not None,
294 prompts=self._prompt_handler is not None,
295 logging=self._logging_handler is not None,
296 sampling=self._sampling_handler is not None,
297 ).to_dict(),
298 "serverInfo": MCPServerInfo(
299 name=self._name,
300 version=self._version,
301 ).to_dict(),
302 }
304 async def _handle_ping(self) -> dict[str, Any]:
305 """Handle the ping method."""
306 return {}
308 async def _handle_initialized_notification(self) -> None:
309 """Handle the notifications/initialized notification."""
310 logger.info("mcp_client_initialized")
313__all__ = ["MCPServer"]