echo-agent / src /logic /tools_manager.py
Kuk1's picture
Deploy Echo Universal Host
85dd3af
"""
Tools Manager Module
Handles MCP Server installation, configuration, and status display.
Completely isolated from Chat logic to prevent side effects.
"""
import os
import json
from typing import Dict, List, Any, Optional, Tuple
# Path to mcp.json (relative to app.py)
MCP_CONFIG_PATH = os.path.join(os.path.dirname(__file__), "../../mcp.json")
def load_mcp_config() -> Dict[str, Any]:
"""Loads the mcp.json configuration file."""
abs_path = os.path.abspath(MCP_CONFIG_PATH)
if not os.path.exists(abs_path):
return {"mcpServers": {}}
try:
with open(abs_path, "r") as f:
return json.load(f)
except Exception as e:
print(f"Error loading mcp.json: {e}")
return {"mcpServers": {}}
def save_mcp_config(config: Dict[str, Any]) -> bool:
"""Saves the configuration to mcp.json."""
abs_path = os.path.abspath(MCP_CONFIG_PATH)
try:
with open(abs_path, "w") as f:
json.dump(config, f, indent=2)
return True
except Exception as e:
print(f"Error saving mcp.json: {e}")
return False
def parse_mcp_command(full_command: str) -> Tuple[str, str, List[str]]:
"""
Parses a full MCP command string into components.
Examples:
"npx -y @modelcontextprotocol/server-memory" -> ("npx", "memory", ["-y", "@modelcontextprotocol/server-memory"])
"uvx mcp-server-filesystem /path" -> ("uvx", "filesystem", ["mcp-server-filesystem", "/path"])
"python server.py" -> ("python", "server", ["server.py"])
Returns:
(command, suggested_name, args_list)
"""
parts = full_command.strip().split()
if not parts:
return "", "", []
command = parts[0] # npx, uvx, python, etc.
args_list = parts[1:] if len(parts) > 1 else []
# Try to extract a meaningful name
suggested_name = ""
for arg in args_list:
# Skip flags
if arg.startswith("-"):
continue
# Handle @org/package-name format
if arg.startswith("@") and "/" in arg:
# @modelcontextprotocol/server-memory -> memory
package_name = arg.split("/")[-1]
# server-memory -> memory
if package_name.startswith("server-"):
suggested_name = package_name[7:] # Remove "server-" prefix
elif package_name.startswith("mcp-server-"):
suggested_name = package_name[11:] # Remove "mcp-server-" prefix
else:
suggested_name = package_name
break
# Handle mcp-server-xxx format
if "mcp-server-" in arg:
suggested_name = arg.split("mcp-server-")[-1].split()[0]
break
# Handle server-xxx format
if arg.startswith("server-"):
suggested_name = arg[7:]
break
# Handle .py files
if arg.endswith(".py"):
suggested_name = arg[:-3].split("/")[-1]
break
# Use the first non-flag argument as fallback
if not suggested_name:
suggested_name = arg.split("/")[-1].split(".")[0]
return command, suggested_name, args_list
def add_server_to_config(
name: str,
command: str,
args: str,
description: str = ""
) -> Tuple[bool, str]:
"""
Adds a new MCP Server to the configuration.
Now supports parsing full command strings.
Args:
name: Server identifier (optional, will be auto-detected)
command: Full command string (e.g., "npx -y @modelcontextprotocol/server-memory")
args: Extra arguments (optional, appended to parsed args)
description: Optional description
Returns:
(success: bool, message: str)
"""
if not command:
return False, "Command is required."
# Parse the command
parsed_cmd, suggested_name, parsed_args = parse_mcp_command(command)
if not parsed_cmd:
return False, "Invalid command format."
# Use provided name or auto-detected name
final_name = name.strip() if name.strip() else suggested_name
if not final_name:
return False, "Could not detect server name. Please provide one."
# Sanitize name (no spaces, lowercase)
final_name = final_name.lower().replace(" ", "-")
# Combine parsed args with extra args
extra_args = args.strip().split() if args.strip() else []
final_args = parsed_args + extra_args
# Load existing config
config = load_mcp_config()
# Check if already exists
if final_name in config.get("mcpServers", {}):
return False, f"Server '{final_name}' already exists. Remove it first."
# Add new server
config.setdefault("mcpServers", {})[final_name] = {
"command": parsed_cmd,
"args": final_args,
"_description": description.strip() or f"MCP Server: {final_name}"
}
# Save
if save_mcp_config(config):
return True, f"Server '{final_name}' installed successfully!"
else:
return False, "Failed to save configuration."
def remove_server_from_config(name: str) -> Tuple[bool, str]:
"""Removes an MCP Server from the configuration."""
config = load_mcp_config()
if name not in config.get("mcpServers", {}):
return False, f"Server '{name}' not found."
del config["mcpServers"][name]
if save_mcp_config(config):
return True, f"Server '{name}' removed."
else:
return False, "Failed to save configuration."
def get_installed_servers() -> List[Dict[str, Any]]:
"""
Returns a list of installed servers with their details.
Each item: {name, command, args, description, is_connected}
"""
config = load_mcp_config()
servers = []
for name, details in config.get("mcpServers", {}).items():
servers.append({
"name": name,
"command": details.get("command", ""),
"args": details.get("args", []),
"description": details.get("_description", ""),
"is_connected": False # Will be updated by connection logic later
})
return servers
def get_server_details_html(server_name: str, is_connected: bool = False) -> str:
"""
Generates HTML for the server detail view.
Args:
server_name: Name of the server
is_connected: Whether the server is currently connected
"""
# Import here to avoid circular imports
from ui.components import get_server_emoji
config = load_mcp_config()
server = config.get("mcpServers", {}).get(server_name)
if not server:
return f"<div style='color: #FF3333; font-family: var(--font-mono);'>Server '{server_name}' not found.</div>"
command = server.get('command', '')
args_list = server.get("args", [])
args_str = " ".join(args_list)
description = server.get("_description", "No description provided.")
env_vars = server.get("env", {})
# Get emoji from centralized mapping
icon = get_server_emoji(server_name)
# Environment variables section (only show if exists)
env_html = ""
if env_vars:
env_items = "<br>".join([
f"<code style='color: #000;'>{k}={v[:20]}...</code>" if len(str(v)) > 20
else f"<code style='color: #000;'>{k}={v}</code>"
for k, v in env_vars.items()
])
env_html = f"""
<div class="detail-row">
<div class="detail-label">ENV</div>
<div class="detail-value">{env_items}</div>
</div>
"""
# Status indicator removed - now shown in top status bar
return f"""
<div style="font-family: var(--font-mono); color: #000000;">
<h2 style="margin: 0 0 24px 0; display: flex; align-items: center; gap: 12px; color: #000000;">
<span style="font-size: 32px;">{icon}</span>
<span style="color: #000000; font-weight: 700;">{server_name}</span>
</h2>
<div class="detail-row">
<div class="detail-label">COMMAND</div>
<div class="detail-value">
<code style="background: #1a1a1a; color: #00FF94; padding: 8px 12px; display: inline-block; border-radius: 4px;">{command}</code>
</div>
</div>
<div class="detail-row">
<div class="detail-label">ARGUMENTS</div>
<div class="detail-value">
<code style="background: #1a1a1a; color: #00FF94; padding: 8px 12px; display: block; word-break: break-all; border-radius: 4px;">{args_str or '(none)'}</code>
</div>
</div>
<div class="detail-row">
<div class="detail-label">DESCRIPTION</div>
<div class="detail-value" style="color: #333;">{description or 'No description provided.'}</div>
</div>
{env_html}
</div>
"""