377 lines
8.1 KiB
Python
377 lines
8.1 KiB
Python
"""
|
|
Minimal MCP stdio server
|
|
Only uses Python standard library: sys + json
|
|
|
|
Tool:
|
|
calculate_bmi(height, weight)
|
|
"""
|
|
|
|
import sys
|
|
import json
|
|
|
|
|
|
PROTOCOL_VERSION = "2026-07-28"
|
|
|
|
SERVER_INFO = {
|
|
"name": "bmi-mcp-server",
|
|
"version": "0.1.0"
|
|
}
|
|
|
|
|
|
# --------------------------------------------------
|
|
# JSON-RPC helpers
|
|
# --------------------------------------------------
|
|
|
|
def rpc_result(request_id, result):
|
|
return {
|
|
"jsonrpc": "2.0",
|
|
"id": request_id,
|
|
"result": result
|
|
}
|
|
|
|
|
|
def rpc_error(request_id, code, message, data=None):
|
|
error = {
|
|
"code": code,
|
|
"message": message
|
|
}
|
|
|
|
if data is not None:
|
|
error["data"] = data
|
|
|
|
return {
|
|
"jsonrpc": "2.0",
|
|
"id": request_id,
|
|
"error": error
|
|
}
|
|
|
|
|
|
def result_meta():
|
|
return {
|
|
"io.modelcontextprotocol/serverInfo": SERVER_INFO
|
|
}
|
|
|
|
|
|
def tool_error(request_id, message):
|
|
"""
|
|
MCP request itself is valid,
|
|
but the tool cannot execute successfully.
|
|
"""
|
|
return rpc_result(
|
|
request_id,
|
|
{
|
|
"resultType": "complete",
|
|
"_meta": result_meta(),
|
|
"content": [
|
|
{
|
|
"type": "text",
|
|
"text": message
|
|
}
|
|
],
|
|
"isError": True
|
|
}
|
|
)
|
|
|
|
|
|
# --------------------------------------------------
|
|
# MCP tool definition
|
|
# --------------------------------------------------
|
|
|
|
BMI_TOOL = {
|
|
"name": "calculate_bmi",
|
|
"description": "Calculate BMI from height in meters and weight in kilograms.",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"height": {
|
|
"type": "number",
|
|
"description": "Height in meters"
|
|
},
|
|
"weight": {
|
|
"type": "number",
|
|
"description": "Weight in kilograms"
|
|
}
|
|
},
|
|
"required": [
|
|
"height",
|
|
"weight"
|
|
],
|
|
"additionalProperties": False
|
|
}
|
|
}
|
|
|
|
|
|
# --------------------------------------------------
|
|
# MCP methods
|
|
# --------------------------------------------------
|
|
|
|
def handle_discover(request):
|
|
return rpc_result(
|
|
request["id"],
|
|
{
|
|
"resultType": "complete",
|
|
"supportedVersions": [
|
|
PROTOCOL_VERSION
|
|
],
|
|
"capabilities": {
|
|
"tools": {
|
|
"listChanged": False
|
|
}
|
|
},
|
|
"_meta": result_meta(),
|
|
"instructions": "Provides a BMI calculation tool.",
|
|
"ttlMs": 3600000,
|
|
"cacheScope": "public"
|
|
}
|
|
)
|
|
|
|
|
|
def handle_tools_list(request):
|
|
return rpc_result(
|
|
request["id"],
|
|
{
|
|
"resultType": "complete",
|
|
"_meta": result_meta(),
|
|
"tools": [
|
|
BMI_TOOL
|
|
],
|
|
"ttlMs": 300000,
|
|
"cacheScope": "public"
|
|
}
|
|
)
|
|
|
|
|
|
def handle_tools_call(request):
|
|
request_id = request["id"]
|
|
|
|
params = request.get("params", {})
|
|
|
|
name = params.get("name")
|
|
arguments = params.get("arguments", {})
|
|
|
|
# -------------------------------
|
|
# Protocol-level error
|
|
# -------------------------------
|
|
|
|
if name != "calculate_bmi":
|
|
return rpc_error(
|
|
request_id,
|
|
-32602,
|
|
f"Unknown tool: {name}"
|
|
)
|
|
|
|
# -------------------------------
|
|
# Tool input validation
|
|
# -------------------------------
|
|
|
|
height = arguments.get("height")
|
|
weight = arguments.get("weight")
|
|
|
|
if height is None:
|
|
return tool_error(
|
|
request_id,
|
|
"height is required"
|
|
)
|
|
|
|
if weight is None:
|
|
return tool_error(
|
|
request_id,
|
|
"weight is required"
|
|
)
|
|
|
|
# bool is a subclass of int in Python,
|
|
# so explicitly reject True / False.
|
|
if (
|
|
not isinstance(height, (int, float))
|
|
or isinstance(height, bool)
|
|
):
|
|
return tool_error(
|
|
request_id,
|
|
"height must be a number"
|
|
)
|
|
|
|
if (
|
|
not isinstance(weight, (int, float))
|
|
or isinstance(weight, bool)
|
|
):
|
|
return tool_error(
|
|
request_id,
|
|
"weight must be a number"
|
|
)
|
|
|
|
if height <= 0:
|
|
return tool_error(
|
|
request_id,
|
|
"height must be greater than 0"
|
|
)
|
|
|
|
if weight <= 0:
|
|
return tool_error(
|
|
request_id,
|
|
"weight must be greater than 0"
|
|
)
|
|
|
|
# -------------------------------
|
|
# Actual business logic
|
|
# -------------------------------
|
|
|
|
bmi = weight / (height ** 2)
|
|
bmi = round(bmi, 2)
|
|
|
|
return rpc_result(
|
|
request_id,
|
|
{
|
|
"resultType": "complete",
|
|
"_meta": result_meta(),
|
|
"content": [
|
|
{
|
|
"type": "text",
|
|
"text": str(bmi)
|
|
}
|
|
],
|
|
"isError": False
|
|
}
|
|
)
|
|
|
|
|
|
# --------------------------------------------------
|
|
# Protocol validation / dispatch
|
|
# --------------------------------------------------
|
|
|
|
def process(request):
|
|
|
|
# JSON-RPC message must be an object
|
|
if not isinstance(request, dict):
|
|
return rpc_error(
|
|
None,
|
|
-32600,
|
|
"Invalid Request"
|
|
)
|
|
|
|
if request.get("jsonrpc") != "2.0":
|
|
return rpc_error(
|
|
request.get("id"),
|
|
-32600,
|
|
"Invalid Request"
|
|
)
|
|
|
|
# No id means notification.
|
|
# This simple server does not need to respond.
|
|
if "id" not in request:
|
|
return None
|
|
|
|
method = request.get("method")
|
|
|
|
if not isinstance(method, str):
|
|
return rpc_error(
|
|
request.get("id"),
|
|
-32600,
|
|
"Invalid Request"
|
|
)
|
|
|
|
# Modern MCP request metadata
|
|
params = request.get("params", {})
|
|
meta = params.get("_meta", {}) if isinstance(params, dict) else {}
|
|
|
|
requested_version = meta.get(
|
|
"io.modelcontextprotocol/protocolVersion"
|
|
)
|
|
|
|
if requested_version != PROTOCOL_VERSION:
|
|
return rpc_error(
|
|
request.get("id"),
|
|
-32022,
|
|
"Unsupported protocol version",
|
|
{
|
|
"supported": [
|
|
PROTOCOL_VERSION
|
|
],
|
|
"requested": requested_version
|
|
}
|
|
)
|
|
|
|
# -------------------------------
|
|
# MCP method dispatcher
|
|
# -------------------------------
|
|
|
|
if method == "server/discover":
|
|
return handle_discover(request)
|
|
|
|
if method == "tools/list":
|
|
return handle_tools_list(request)
|
|
|
|
if method == "tools/call":
|
|
return handle_tools_call(request)
|
|
|
|
return rpc_error(
|
|
request.get("id"),
|
|
-32601,
|
|
f"Method not found: {method}"
|
|
)
|
|
|
|
|
|
# --------------------------------------------------
|
|
# stdio transport
|
|
# --------------------------------------------------
|
|
|
|
def send(message):
|
|
text = json.dumps(
|
|
message,
|
|
ensure_ascii=False,
|
|
separators=(",", ":")
|
|
)
|
|
|
|
sys.stdout.write(text + "\n")
|
|
sys.stdout.flush()
|
|
|
|
|
|
def main():
|
|
|
|
# stdin remains open while the MCP server is alive
|
|
for line in sys.stdin:
|
|
|
|
line = line.strip()
|
|
|
|
if not line:
|
|
continue
|
|
|
|
try:
|
|
request = json.loads(line)
|
|
|
|
except json.JSONDecodeError:
|
|
send(
|
|
rpc_error(
|
|
None,
|
|
-32700,
|
|
"Parse error"
|
|
)
|
|
)
|
|
continue
|
|
|
|
try:
|
|
response = process(request)
|
|
|
|
if response is not None:
|
|
send(response)
|
|
|
|
except Exception as e:
|
|
# Logs MUST go to stderr, never stdout
|
|
print(
|
|
f"Internal error: {e}",
|
|
file=sys.stderr,
|
|
flush=True
|
|
)
|
|
|
|
send(
|
|
rpc_error(
|
|
request.get("id")
|
|
if isinstance(request, dict)
|
|
else None,
|
|
-32603,
|
|
"Internal error"
|
|
)
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |