liuren_mcp.py•9.5 kB
#!/usr/bin/env python3
"""
最简单的六壬MCP工具
直接调用kinliuren库并返回JSON结果
参考MCPQimenJson实现重构
"""
import json
import sys
from datetime import datetime
try:
import config
from kinliuren import kinliuren
except ImportError:
print("错误: 请先安装kinliuren库: pip install kinliuren")
sys.exit(1)
def get_current_time():
"""获取当前时间并返回格式化结果"""
try:
now = datetime.now()
result = {
"year": now.year,
"month": now.month,
"day": now.day,
"hour": now.hour,
"minute": now.minute,
"second": now.second,
"datetime_str": now.strftime("%Y-%m-%d %H:%M:%S"),
"weekday": now.strftime("%A"),
"weekday_cn": ["星期一", "星期二", "星期三", "星期四", "星期五", "星期六", "星期日"][now.weekday()],
"timestamp": int(now.timestamp())
}
return {"success": True, "data": result}
except Exception as e:
return {"success": False, "error": str(e)}
def calculate_liuren(year, month, day, hour, minute):
"""计算六壬排盘并返回完整JSON结果
Args:
year, month, day, hour, minute: 时间参数
"""
try:
# 使用config模块获取节气、农历月份和干支
solar_term = config.get_solar_term(year, month, day)
lunar_month = config.get_lunar_month(year, month, day)
day_gz, hour_gz = config.get_ganzhi(year, month, day, hour)
# 使用kinliuren库计算六壬排盘
liuren_result = kinliuren.Liuren(solar_term, lunar_month, day_gz, hour_gz).result(0)
result = {
"year": year,
"month": month,
"day": day,
"hour": hour,
"minute": minute,
"solar_term": solar_term,
"lunar_month": lunar_month,
"day_ganzhi": day_gz,
"hour_ganzhi": hour_gz,
"liuren_data": liuren_result
}
return {"success": True, "data": result}
except Exception as e:
return {"success": False, "error": str(e)}
def main():
"""主函数 - 处理MCP请求"""
# 读取标准输入
for line in sys.stdin:
try:
request = json.loads(line.strip())
method = request.get("method")
params = request.get("params", {})
if method == "initialize":
# 初始化响应
response = {
"jsonrpc": "2.0",
"id": request.get("id"),
"result": {
"protocolVersion": "2024-11-05",
"capabilities": {"tools": {}},
"serverInfo": {
"name": "liuren-mcp-server",
"version": "1.0.0",
"description": "六壬排盘服务器 - 基于kinliuren库"
}
}
}
elif method == "tools/list":
# 返回可用工具列表
response = {
"jsonrpc": "2.0",
"id": request.get("id"),
"result": {
"tools": [
{
"name": "get_current_time",
"description": "获取当前系统时间，返回详细的时间信息",
"inputSchema": {
"type": "object",
"properties": {},
"required": []
}
},
{
"name": "calculate_liuren",
"description": "计算指定时间的六壬排盘，返回完整的六壬数据（包含天地盘、三传、四课等）",
"inputSchema": {
"type": "object",
"properties": {
"datetime_str": {
"type": "string",
"description": "时间字符串，格式: YYYY-MM-DD HH:MM:SS",
"examples": ["2024-01-15 14:30:00", "2024-12-25 08:00:00", "2025-06-15 12:00:00"]
}
},
"required": ["datetime_str"]
}
}
]
}
}
elif method == "tools/call":
# 处理工具调用
tool_name = params.get("name")
arguments = params.get("arguments", {})
if tool_name == "get_current_time":
# 获取当前时间，无需参数
try:
result = get_current_time()
response = {
"jsonrpc": "2.0",
"id": request.get("id"),
"result": {
"content": [{"type": "text", "text": json.dumps(result, ensure_ascii=False, indent=2)}],
"isError": False
}
}
except Exception as e:
response = {
"jsonrpc": "2.0",
"id": request.get("id"),
"result": {
"content": [{"type": "text", "text": f"获取时间错误: {str(e)}"}],
"isError": True
}
}
elif tool_name == "calculate_liuren":
# 解析时间参数
datetime_str = arguments.get("datetime_str")
if datetime_str:
try:
dt = datetime.strptime(datetime_str, "%Y-%m-%d %H:%M:%S")
result = calculate_liuren(dt.year, dt.month, dt.day, dt.hour, dt.minute)
response = {
"jsonrpc": "2.0",
"id": request.get("id"),
"result": {
"content": [{"type": "text", "text": json.dumps(result, ensure_ascii=False, indent=2)}],
"isError": False
}
}
except ValueError as e:
response = {
"jsonrpc": "2.0",
"id": request.get("id"),
"result": {
"content": [{"type": "text", "text": f"时间格式错误: {str(e)}"}],
"isError": True
}
}
except Exception as e:
response = {
"jsonrpc": "2.0",
"id": request.get("id"),
"result": {
"content": [{"type": "text", "text": f"计算错误: {str(e)}"}],
"isError": True
}
}
else:
response = {
"jsonrpc": "2.0",
"id": request.get("id"),
"result": {
"content": [{"type": "text", "text": "错误: 需要提供 datetime_str 参数"}],
"isError": True
}
}
else:
response = {
"jsonrpc": "2.0",
"id": request.get("id"),
"result": {
"content": [{"type": "text", "text": f"未知工具: {tool_name}"}],
"isError": True
}
}
else:
response = {
"jsonrpc": "2.0",
"id": request.get("id"),
"error": {"code": -32601, "message": "Method not found"}
}
# 输出响应
print(json.dumps(response, ensure_ascii=False))
sys.stdout.flush()
except json.JSONDecodeError:
print(json.dumps({
"jsonrpc": "2.0",
"id": None,
"error": {"code": -32700, "message": "Parse error"}
}, ensure_ascii=False))
sys.stdout.flush()
except Exception as e:
print(json.dumps({
"jsonrpc": "2.0",
"id": None,
"error": {"code": -32603, "message": f"Internal error: {str(e)}"}
}, ensure_ascii=False))
sys.stdout.flush()
if __name__ == "__main__":
main()