| 1 | import time |
| 2 | from helpers.api import ApiHandler, Request, Response |
| 3 | |
| 4 | from typing import Any |
| 5 | |
| 6 | from helpers.mcp_handler import MCPConfig |
| 7 | from helpers.settings import set_settings_delta |
| 8 | from helpers import projects |
| 9 | |
| 10 | |
| 11 | class McpServersApply(ApiHandler): |
| 12 | async def process(self, input: dict[Any, Any], request: Request) -> dict[Any, Any] | Response: |
| 13 | mcp_servers = input["mcp_servers"] |
| 14 | project_name = str(input.get("project_name", "") or "").strip() |
| 15 | try: |
| 16 | if project_name: |
| 17 | projects.save_project_mcp_servers(project_name, mcp_servers) |
| 18 | config = MCPConfig.refresh_project(project_name) |
| 19 | else: |
| 20 | # MCPConfig.update(mcp_servers) # done in settings automatically |
| 21 | set_settings_delta({"mcp_servers": "[]"}) # to force reinitialization |
| 22 | set_settings_delta({"mcp_servers": mcp_servers}) |
| 23 | |
| 24 | time.sleep(1) # wait at least a second |
| 25 | # MCPConfig.wait_for_lock() # wait until config lock is released |
| 26 | config = MCPConfig.get_instance() |
| 27 | status = config.get_servers_status() |
| 28 | return {"success": True, "status": status, "mcp_servers": mcp_servers, "project_name": project_name} |
| 29 | |
| 30 | except Exception as e: |
| 31 | return {"success": False, "error": str(e)} |