自然进化

1. 修复mcp服务器信息无法修改问题;
2.logseq mcp服务器补充read_page工具。
This commit is contained in:
guo zebin
2025-10-21 11:18:37 +08:00
parent ca63d30824
commit 28811485fa
2 changed files with 154 additions and 0 deletions

View File

@@ -561,6 +561,118 @@ def connect_server(server_id):
}), 500
return jsonify({"error": "服务器未找到"}), 404
# API路由 - 更新MCP服务器配置
@app.route('/api/mcp/servers/<int:server_id>', methods=['PUT'])
def update_mcp_server(server_id):
"""更新MCP服务器配置"""
global mcp_servers, mcp_clients
data = request.json
auto_reconnect = data.get('auto_reconnect', False)
# 查找服务器
server = None
server_index = None
for i, s in enumerate(mcp_servers):
if s['id'] == server_id:
server = s
server_index = i
break
if not server:
return jsonify({"success": False, "message": "服务器未找到"}), 404
# 更新基本信息
server['name'] = data.get('name', server['name'])
transport = data.get('transport', server.get('transport', 'sse'))
server['transport'] = transport
# 根据传输类型更新配置
if transport == 'stdio':
server['command'] = data.get('command', '')
args_input = data.get('args', [])
# 如果args是字符串拆分为列表
if isinstance(args_input, str):
# 简单按空格拆分,支持引号
import shlex
try:
server['args'] = shlex.split(args_input)
except:
# 如果shlex失败简单按空格拆分
server['args'] = args_input.split() if args_input else []
else:
server['args'] = args_input
server['cwd'] = data.get('cwd', '')
# 处理环境变量
env_input = data.get('env', {})
if isinstance(env_input, str):
try:
server['env'] = json.loads(env_input)
except:
util.log(1, f"环境变量JSON解析失败: {env_input}")
server['env'] = {}
else:
server['env'] = env_input
# 清空SSE相关字段
server['ip'] = ''
server['key'] = ''
else:
# SSE模式
server['ip'] = data.get('ip', '')
server['key'] = data.get('key', '')
# 清空本地命令字段
server['command'] = ''
server['args'] = []
server['cwd'] = ''
server['env'] = {}
# 保存配置
save_mcp_servers(mcp_servers)
# 如果需要自动重连
if auto_reconnect:
# 先断开
if server_id in mcp_clients:
try:
del mcp_clients[server_id]
tool_registry.mark_all_unavailable(server_id)
except Exception as e:
util.log(1, f"断开连接失败: {e}")
# 重新连接
try:
success, updated_server, tools = connect_to_real_mcp(server)
if success:
# 更新服务器信息
mcp_servers[server_index] = updated_server
save_mcp_servers(mcp_servers)
return jsonify({
"success": True,
"message": "配置已更新并重新连接",
"server": updated_server
})
else:
return jsonify({
"success": True,
"message": "配置已更新,但重新连接失败",
"server": server
})
except Exception as e:
util.log(1, f"重新连接失败: {e}")
return jsonify({
"success": True,
"message": f"配置已更新,但重新连接失败: {str(e)}",
"server": server
})
return jsonify({
"success": True,
"message": "配置已更新",
"server": server
})
# API路由 - 删除MCP服务器
@app.route('/api/mcp/servers/<int:server_id>', methods=['DELETE'])
def delete_server(server_id):

View File

@@ -198,6 +198,32 @@ class LogseqGraph:
pages.add(name)
return {"success": True, "pages": sorted(pages)}
def read_page(self, page_name: str) -> Dict[str, Any]:
"""读取指定页面的完整内容"""
self.ensure_dirs()
path = self.page_path(page_name)
if not os.path.exists(path):
return {
"success": False,
"message": f"页面不存在: {page_name}",
"path": path
}
try:
content = self._read_text(path)
return {
"success": True,
"page": page_name,
"path": path,
"content": content,
"lines": len(content.splitlines())
}
except Exception as e:
return {
"success": False,
"message": f"读取失败: {str(e)}",
"path": path
}
def _ensure_root(self) -> str:
if not self.root:
@@ -344,6 +370,18 @@ async def list_tools() -> List[Tool]:
"required": ["tag"]
}
),
Tool(
name="read_page",
description="读取指定页面的完整内容",
inputSchema={
"type": "object",
"properties": {
"page": {"type": "string", "description": "页面名不含扩展名例如Fay"},
"graph_dir": {"type": "string", "description": "可选,覆盖 LOGSEQ_GRAPH_DIR"}
},
"required": ["page"]
}
),
Tool(
name="append_todo_to_page",
description="在指定 page 末尾追加一个 TODO 列表项",
@@ -449,6 +487,10 @@ async def handle_call_tool(name: str, arguments: Dict[str, Any]) -> List[TextCon
res = g.get_pages_by_tag(arguments["tag"])
return [TextContent(type="text", text=json.dumps(res, ensure_ascii=False))]
elif name == "read_page":
res = g.read_page(arguments["page"])
return [TextContent(type="text", text=json.dumps(res, ensure_ascii=False))]
elif name == "append_todo_to_page":
level = max(int(arguments.get("level", 0)), 0)
indent = " " * level