e4b5ea649f
Fork of https://github.com/thinkitpossible/CalcMCP with: - tools/list now publishes inputSchema (upstream used non-standard 'params' key, leaving clients with no parameter information) - tool descriptions and parameter docs rewritten in English with examples - JSON-RPC notifications are no longer answered; tool failures set isError - raw eval() of tool arguments removed (json.loads/ast.literal_eval for matrices, sympy.lambdify for the numeric ODE right-hand side) - complex eigenvalues serialize as 'a+bi' strings instead of crashing - solve accepts '=' equations; ttest accepts {"popmean": ...} params - new symbolic_tool operation 'dsolve' for symbolic ODE solutions Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
326 lines
16 KiB
Python
326 lines
16 KiB
Python
"""
|
||
Scientific Calculator MCP Server - Production Ready
|
||
完全符合 MCP 规范的科学计算器服务器
|
||
支持通过标准输入输出（STDIO）进行 JSON-RPC 2.0 通信
|
||
|
||
This implementation strictly follows:
|
||
https://modelcontextprotocol.io/docs/develop/build-server
|
||
"""
|
||
|
||
import json
|
||
import sys
|
||
import logging
|
||
from typing import Any, Dict, List
|
||
from dataclasses import dataclass, asdict
|
||
from calculator import CALCULATOR_TOOLS
|
||
import numpy as np
|
||
|
||
# 配置日志到 stderr（不污染 stdout JSON-RPC 输出）
|
||
logging.basicConfig(
|
||
level=logging.WARNING,
|
||
format='[%(levelname)s] %(message)s',
|
||
stream=sys.stderr,
|
||
force=True
|
||
)
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
@dataclass
|
||
class JSONRPCRequest:
|
||
"""JSON-RPC 2.0 请求"""
|
||
jsonrpc: str = "2.0"
|
||
method: str = None
|
||
params: Dict[str, Any] = None
|
||
id: Any = None
|
||
|
||
|
||
@dataclass
|
||
class JSONRPCResponse:
|
||
"""JSON-RPC 2.0 响应"""
|
||
jsonrpc: str = "2.0"
|
||
result: Any = None
|
||
error: Dict[str, Any] = None
|
||
id: Any = None
|
||
|
||
|
||
class MCPCalculatorServer:
|
||
"""符合 MCP 规范的计算器服务器"""
|
||
|
||
def __init__(self):
|
||
self.version = "1.0.0"
|
||
self.name = "Scientific Calculator"
|
||
self.tools = self._build_tool_definitions()
|
||
|
||
def _build_tool_definitions(self) -> List[Dict[str, Any]]:
|
||
"""构建工具定义列表"""
|
||
tools = []
|
||
|
||
# Consolidated tool definitions (3 tools)
|
||
tool_definitions = {
|
||
"symbolic_tool": {
|
||
"description": (
|
||
"Exact symbolic math (SymPy). Set operation to one of: simplify, expand, factor, "
|
||
"derivative, integral, limit, solve, dsolve, taylor, matrix. "
|
||
"Write expressions in Python/SymPy syntax, e.g. 'x**2 * exp(-x)'; use 'oo' for infinity. "
|
||
"integral: optional lower_limit/upper_limit for definite integrals. "
|
||
"solve: pass the equation, '=' is allowed (e.g. 'x**2 = 4' or 'x**3 - 6*x**2 + 11*x - 6'). "
|
||
"dsolve: solves an ODE symbolically — write the unknown function as y(x), "
|
||
"e.g. equation='Derivative(y(x), x, 2) + y(x) = 0'. "
|
||
"taylor: expression + variable + point + order. "
|
||
"matrix: matrix_data (JSON) with variable=determinant|inverse|rank|trace. "
|
||
"Results are exact (e.g. '2', 'E', 'C1*sin(x) + C2*cos(x)')."
|
||
),
|
||
"params": [
|
||
{"name": "operation", "type": "string", "description": "One of: simplify, expand, factor, derivative, integral, limit, solve, dsolve, taylor, matrix"},
|
||
{"name": "expression", "type": "string", "description": "Expression in SymPy syntax, e.g. 'sin(x)/x'", "optional": True},
|
||
{"name": "variable", "type": "string", "description": "Variable name, e.g. 'x' (for matrix: the operation name)", "optional": True},
|
||
{"name": "order", "type": "integer", "description": "Order for derivative/taylor (default 1 / 5)", "optional": True},
|
||
{"name": "lower_limit", "type": "string", "description": "Lower bound for definite integral, e.g. '0'", "optional": True},
|
||
{"name": "upper_limit", "type": "string", "description": "Upper bound for definite integral, e.g. 'oo'", "optional": True},
|
||
{"name": "equation", "type": "string", "description": "Equation for solve/dsolve; '=' allowed", "optional": True},
|
||
{"name": "point", "type": "string", "description": "Point for limit/taylor, e.g. '0' or 'oo'", "optional": True},
|
||
{"name": "matrix_data", "type": "string", "description": "Matrix as JSON, e.g. '[[1,2],[3,4]]'", "optional": True}
|
||
]
|
||
},
|
||
"numpy_tool": {
|
||
"description": (
|
||
"Numeric arrays and linear algebra (NumPy). Set operation to one of: "
|
||
"sum, mean, std, max, min (array_data, optional axis); "
|
||
"eigenvalues, eigenvectors, determinant, inverse, solve, norm, rank, trace, "
|
||
"svd, qr, cholesky, matmul, dot, hadamard (matrix_a, and matrix_b where two matrices are needed); "
|
||
"sin, cos, tan, arcsin, arccos, arctan, sinh, cosh, tanh (values, optional use_degrees); "
|
||
"poly_eval, poly_derivative, poly_integral (coefficients highest-degree first, optional x_values). "
|
||
"All array/matrix inputs are JSON strings, e.g. matrix_a='[[2,1],[1,2]]'. Results are numeric."
|
||
),
|
||
"params": [
|
||
{"name": "operation", "type": "string", "description": "Operation name, e.g. 'eigenvalues', 'matmul', 'mean', 'poly_eval'"},
|
||
{"name": "array_data", "type": "string", "description": "Array as JSON, e.g. '[1,2,3]' or '[[1,2],[3,4]]'", "optional": True},
|
||
{"name": "axis", "type": "string", "description": "Axis for reductions: '0' or '1'", "optional": True},
|
||
{"name": "matrix_a", "type": "string", "description": "First matrix as JSON", "optional": True},
|
||
{"name": "matrix_b", "type": "string", "description": "Second matrix as JSON (solve/matmul/dot/hadamard)", "optional": True},
|
||
{"name": "values", "type": "string", "description": "Input values for trig functions as JSON, e.g. '[0, 90, 180]'", "optional": True},
|
||
{"name": "use_degrees", "type": "boolean", "description": "Treat trig inputs as degrees (default radians)", "optional": True},
|
||
{"name": "coefficients", "type": "string", "description": "Polynomial coefficients as JSON, highest degree first", "optional": True},
|
||
{"name": "x_values", "type": "string", "description": "x values for poly_eval as JSON", "optional": True}
|
||
]
|
||
},
|
||
"scipy_tool": {
|
||
"description": (
|
||
"Numerical methods (SciPy). Set operation to one of: "
|
||
"integrate (numeric definite integral: expression, variable, lower, upper); "
|
||
"optimize_minimize, optimize_root (expression, initial_value, optional method); "
|
||
"interpolate_linear, interpolate_cubic, interpolate_spline (x_data, y_data, x_eval); "
|
||
"special (function name + parameters JSON, e.g. gamma, besselj); "
|
||
"solve_ode (NUMERIC simulation of dy/dt = f(t,y): expression e.g. '-2*y + t', "
|
||
"initial_conditions, t_values; for exact general solutions use symbolic_tool with operation='dsolve'); "
|
||
"statistics ops mean, std, describe, ttest, pearsonr (function + data, optional params); "
|
||
"fft, rfft (data_series or data); matrix_eigensystem (matrix_a). "
|
||
"All list inputs are JSON strings. Results are numeric."
|
||
),
|
||
"params": [
|
||
{"name": "operation", "type": "string", "description": "Operation name, e.g. 'integrate', 'optimize_minimize', 'solve_ode', 'fft'"},
|
||
{"name": "expression", "type": "string", "description": "Expression/function of the variable, e.g. 'x**2 * exp(-x)'", "optional": True},
|
||
{"name": "variable", "type": "string", "description": "Variable name, e.g. 'x'", "optional": True},
|
||
{"name": "lower", "type": "string", "description": "Lower integration bound (number)", "optional": True},
|
||
{"name": "upper", "type": "string", "description": "Upper integration bound (number or 'oo')", "optional": True},
|
||
{"name": "initial_value", "type": "string", "description": "Starting point for optimization/root finding", "optional": True},
|
||
{"name": "method", "type": "string", "description": "Algorithm, e.g. 'BFGS', 'brentq'", "optional": True},
|
||
{"name": "x_data", "type": "string", "description": "Interpolation x data as JSON", "optional": True},
|
||
{"name": "y_data", "type": "string", "description": "Interpolation y data as JSON", "optional": True},
|
||
{"name": "x_eval", "type": "string", "description": "Points to evaluate the interpolant at, JSON", "optional": True},
|
||
{"name": "function", "type": "string", "description": "Special-function or statistics operation name", "optional": True},
|
||
{"name": "parameters", "type": "string", "description": "Special-function parameters as JSON", "optional": True},
|
||
{"name": "initial_conditions", "type": "string", "description": "ODE initial value(s) as JSON, e.g. '[1.0]'", "optional": True},
|
||
{"name": "t_values", "type": "string", "description": "ODE time points as JSON, e.g. '[0, 0.1, 0.2]'", "optional": True},
|
||
{"name": "variables", "type": "string", "description": "ODE dependent variable name, default 'y'", "optional": True},
|
||
{"name": "data", "type": "string", "description": "Statistics/FFT data as JSON", "optional": True},
|
||
{"name": "params", "type": "string", "description": "Extra statistics parameter as JSON: for ttest the population mean ('2.3' or '{\"popmean\": 2.3}'), for pearsonr the second data series", "optional": True},
|
||
{"name": "data_series", "type": "string", "description": "FFT data as JSON", "optional": True},
|
||
{"name": "matrix_a", "type": "string", "description": "Matrix as JSON for matrix_eigensystem", "optional": True}
|
||
]
|
||
},
|
||
}
|
||
|
||
# 构建工具列表
|
||
for tool_name, definition in tool_definitions.items():
|
||
tools.append({
|
||
"name": tool_name,
|
||
"description": definition["description"],
|
||
"inputSchema": {
|
||
"type": "object",
|
||
"properties": {
|
||
param["name"]: {
|
||
"type": param["type"],
|
||
"description": param["description"]
|
||
}
|
||
for param in definition["params"]
|
||
},
|
||
"required": [
|
||
p["name"] for p in definition["params"]
|
||
if not p.get("optional", False)
|
||
]
|
||
}
|
||
})
|
||
|
||
return tools
|
||
|
||
def call_tool(self, tool_name: str, params: Dict[str, Any]) -> str:
|
||
"""调用计算器工具"""
|
||
if tool_name not in CALCULATOR_TOOLS:
|
||
raise ValueError(f"Unknown tool: {tool_name}")
|
||
|
||
try:
|
||
# 获取工具函数
|
||
tool_func = CALCULATOR_TOOLS[tool_name]
|
||
|
||
# 调用工具函数
|
||
result = tool_func(**params)
|
||
|
||
return str(result)
|
||
|
||
except TypeError as e:
|
||
return f"Parameter Error: {str(e)}"
|
||
except Exception as e:
|
||
return f"Execution Error: {str(e)}"
|
||
|
||
def handle_rpc_request(self, request_dict: Dict[str, Any]) -> Dict[str, Any]:
|
||
"""处理 JSON-RPC 请求"""
|
||
try:
|
||
# 验证 JSON-RPC 2.0 格式
|
||
if request_dict.get("jsonrpc") != "2.0":
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"error": {
|
||
"code": -32600,
|
||
"message": "Invalid Request"
|
||
},
|
||
"id": request_dict.get("id")
|
||
}
|
||
|
||
method = request_dict.get("method")
|
||
params = request_dict.get("params", {})
|
||
req_id = request_dict.get("id")
|
||
|
||
# JSON-RPC notifications (no id) must not be answered
|
||
if req_id is None and isinstance(method, str) and method.startswith("notifications/"):
|
||
return None
|
||
|
||
# 处理特殊方法
|
||
if method == "initialize":
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"result": {
|
||
"protocolVersion": "2024-11-05",
|
||
"capabilities": {},
|
||
"serverInfo": {
|
||
"name": self.name,
|
||
"version": self.version
|
||
}
|
||
},
|
||
"id": req_id
|
||
}
|
||
|
||
elif method == "tools/list":
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"result": {
|
||
"tools": self.tools
|
||
},
|
||
"id": req_id
|
||
}
|
||
|
||
elif method == "tools/call":
|
||
tool_name = params.get("name")
|
||
tool_params = params.get("arguments", {})
|
||
|
||
result = self.call_tool(tool_name, tool_params)
|
||
|
||
is_error = isinstance(result, str) and result.startswith(
|
||
("Error", "Parameter Error", "Execution Error")
|
||
)
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"result": {
|
||
"content": [
|
||
{
|
||
"type": "text",
|
||
"text": result
|
||
}
|
||
],
|
||
"isError": is_error
|
||
},
|
||
"id": req_id
|
||
}
|
||
|
||
else:
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"error": {
|
||
"code": -32601,
|
||
"message": f"Method not found: {method}"
|
||
},
|
||
"id": req_id
|
||
}
|
||
|
||
except Exception as e:
|
||
logger.error(f"RPC Error: {e}")
|
||
return {
|
||
"jsonrpc": "2.0",
|
||
"error": {
|
||
"code": -32603,
|
||
"message": f"Internal error: {str(e)}"
|
||
},
|
||
"id": request_dict.get("id")
|
||
}
|
||
|
||
def run(self):
|
||
"""运行 MCP 服务器（STDIO 模式）"""
|
||
logger.info(f"Starting {self.name} v{self.version}")
|
||
logger.info(f"Registered {len(self.tools)} tools")
|
||
|
||
try:
|
||
while True:
|
||
# 从标准输入读取 JSON-RPC 请求
|
||
line = sys.stdin.readline()
|
||
|
||
if not line:
|
||
# EOF
|
||
break
|
||
|
||
try:
|
||
request = json.loads(line)
|
||
response = self.handle_rpc_request(request)
|
||
|
||
# Notifications return None — nothing must be written back
|
||
if response is not None:
|
||
sys.stdout.write(json.dumps(response, ensure_ascii=False) + '\n')
|
||
sys.stdout.flush()
|
||
|
||
except json.JSONDecodeError as e:
|
||
# 发送 JSON 解析错误
|
||
error_response = {
|
||
"jsonrpc": "2.0",
|
||
"error": {
|
||
"code": -32700,
|
||
"message": "Parse error"
|
||
}
|
||
}
|
||
sys.stdout.write(json.dumps(error_response) + '\n')
|
||
sys.stdout.flush()
|
||
|
||
except KeyboardInterrupt:
|
||
logger.info("Server interrupted")
|
||
except Exception as e:
|
||
logger.error(f"Server error: {e}")
|
||
sys.exit(1)
|
||
|
||
|
||
def main():
|
||
"""启动 MCP 服务器"""
|
||
server = MCPCalculatorServer()
|
||
server.run()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|