-
-
Notifications
You must be signed in to change notification settings - Fork 1.4k
Expand file tree
/
Copy pathmcp_server.py
More file actions
59 lines (49 loc) · 1.88 KB
/
Copy pathmcp_server.py
File metadata and controls
59 lines (49 loc) · 1.88 KB
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
"""Naive MCP server implementation that crashes on missing tools and lacks schema validation.
Also ignores the stateless 2026-07-28 specification: it has no input_required
flow, so tools that need client input cannot finish safely.
"""
from __future__ import annotations
from collections.abc import Callable
from typing import Any
class MCPServer:
def __init__(self) -> None:
self.tools: dict[str, Any] = {}
def register_tool(
self,
name: str,
description: str,
schema: dict[str, Any],
handler: Callable[[dict[str, Any]], str],
) -> None:
self.tools[name] = {
"name": name,
"description": description,
"inputSchema": schema,
"handler": handler,
}
def handle_request(self, request: Any) -> dict[str, Any]:
# Naive: does not check jsonrpc version or error handling
req_id = request.get("id")
method = request["method"]
if method == "tools/list":
tools_list = [
{
"name": t["name"],
"description": t["description"],
"inputSchema": t["inputSchema"],
}
for t in self.tools.values()
]
return {"jsonrpc": "2.0", "id": req_id, "result": {"tools": tools_list}}
if method == "tools/call":
params = request["params"]
tool_name = params["name"]
# Will crash if tool doesn't exist or handler raises
tool = self.tools[tool_name]
output = tool["handler"](params["arguments"])
return {
"jsonrpc": "2.0",
"id": req_id,
"result": {"content": [{"type": "text", "text": str(output)}]},
}
return {"jsonrpc": "2.0", "id": req_id, "error": {"code": -32601, "message": "Method not found"}}