import argparse
import json
import shutil
import sys
import time
from pathlib import Path
from .server import run, mcp
from .core import (
get_context_bus_dir,
ensure_directory_exists,
ensure_gitignore,
get_state_file,
get_tracking_file,
read_with_lock
)
from .config import ConfigManager
from .monitor import HeartbeatMonitor
def init_project() -> None:
"""Initialize a new Amicus project in the current directory."""
print("Initializing Amicus project...")
context_dir = get_context_bus_dir()
ensure_directory_exists(context_dir)
ensure_gitignore(context_dir)
config = ConfigManager(context_dir.parent)
config.save_workspace()
print(f"✓ Created {context_dir}")
print(f"✓ Updated .gitignore")
print(f"✓ Created {config.workspace_file_json}")
print("\nAmicus is ready for this workspace.")
def list_tools() -> None:
"""List all available MCP tools."""
import asyncio
async def _list():
tools = await mcp.get_tools()
print("Available Tools:")
print("=" * 50)
for i, tool in enumerate(tools):
tool_name = tool if isinstance(tool, str) else (tool.name if hasattr(tool, 'name') else str(tool))
print(f"{i+1}. {tool_name}")
print()
asyncio.run(_list())
def list_prompts() -> None:
print("Available Prompts:")
print("=" * 50)
print("1. catch_up")
print("2. join_synapse")
print()
def validate_env() -> None:
"""Validate environment configuration and show diagnostics."""
print("Environment Validation:")
print("=" * 50)
print()
# Check CONTEXT_BUS_DIR
context_dir = get_context_bus_dir()
print(f"Context Bus Directory: {context_dir}")
if context_dir.exists():
print(f" ✓ Directory exists")
state_file = get_state_file()
if state_file.exists():
print(f" ✓ State file exists: {state_file}")
try:
state_data = read_with_lock(state_file)
print(f" ✓ State file is valid JSON")
except Exception as e:
print(f" ✗ Error reading state file: {e}")
else:
print(f" ⚠ State file does not exist yet")
else:
print(f" ⚠ Directory does not exist")
print()
print("✓ Environment validation complete")
def show_state() -> None:
"""Show current state without starting the server."""
# We can import read_state from server module but that's a tool function
# Easier to just invoke it
from .server import read_state
print(read_state())
def show_audit_prompt() -> None:
"""Display the comprehensive audit prompt."""
from pathlib import Path
# Try to find the prompt relative to the package installation
# This might need adjustment based on how package data is installed
# For now, assume development environment structure relative to CWD
# Check 1: In local prompts/ dir
prompt_file = Path("prompts/comprehensive-audit.md")
if not prompt_file.exists():
# Check 2: Relative to this file
prompt_file = Path(__file__).parent.parent.parent / "prompts" / "comprehensive-audit.md"
if prompt_file.exists():
with open(prompt_file, "r") as f:
print(f.read())
else:
print("Error: Audit prompt not found.")
def show_join_synapse_prompt() -> None:
"""Display the synapse protocol prompt for joining the cluster."""
from pathlib import Path
# Check 1: In local prompts/ dir
prompt_file = Path("prompts/SYNAPSE_PROTOCOL.md")
if not prompt_file.exists():
# Check 2: Relative to this file
prompt_file = Path(__file__).parent.parent.parent / "prompts" / "SYNAPSE_PROTOCOL.md"
if prompt_file.exists():
with open(prompt_file, "r") as f:
print(f.read())
else:
print("Error: Synapse protocol prompt not found.")
MCP_CLIENTS = {
"copilot": {
"name": "GitHub Copilot CLI",
"global_path": lambda: Path.home() / ".copilot" / "mcp-config.json",
"workspace_path": lambda: Path.cwd() / ".github" / "mcp.json",
"config_template": lambda cmd, use_full_path: {
"mcpServers": {
"amicus": {
"command": cmd if use_full_path else "amicus-mcp",
"args": [],
"env": {},
"tools": ["*"]
}
}
}
},
"claude-desktop": {
"name": "Claude Desktop App",
"global_path": lambda: (
Path.home() / "Library" / "Application Support" / "Claude" / "claude_desktop_config.json"
if sys.platform == "darwin"
else Path.home() / ".config" / "Claude" / "claude_desktop_config.json"
),
"workspace_path": None, # Claude Desktop doesn't support workspace configs
"config_template": lambda cmd, use_full_path: {
"mcpServers": {
"amicus": {
"command": "uvx",
"args": ["amicus-mcp"]
}
}
}
},
"claude-code": {
"name": "Claude Code CLI",
"global_path": None, # Claude Code uses `claude mcp add` command
"workspace_path": lambda: Path.cwd() / ".mcp.json",
"config_template": lambda cmd, use_full_path: {
"mcpServers": {
"amicus": {
"command": "uvx",
"args": ["amicus-mcp"]
}
}
},
"install_command": "claude mcp add amicus --transport stdio -- uvx amicus-mcp"
},
"vscode": {
"name": "VS Code",
"global_path": None, # VS Code uses settings.json
"workspace_path": lambda: Path.cwd() / ".vscode" / "mcp.json",
"config_template": lambda cmd, use_full_path: {
"mcpServers": {
"amicus": {
"command": "uvx",
"args": ["amicus-mcp"],
"env": {}
}
}
},
"install_command": 'code --add-mcp \'{"type":"stdio","name":"amicus","command":"uvx","args":["amicus-mcp"]}\''
},
"cursor": {
"name": "Cursor",
"global_path": None,
"workspace_path": lambda: Path.cwd() / ".cursor" / "mcp.json",
"config_template": lambda cmd, use_full_path: {
"mcpServers": {
"amicus": {
"command": "uvx",
"args": ["amicus-mcp"]
}
}
}
},
"windsurf": {
"name": "Windsurf",
"global_path": lambda: Path.home() / ".codeium" / "windsurf" / "mcp_config.json",
"workspace_path": lambda: Path.cwd() / ".windsurf" / "mcp.json",
"config_template": lambda cmd, use_full_path: {
"mcpServers": {
"amicus": {
"command": "uvx",
"args": ["amicus-mcp"],
"env": {}
}
}
}
},
"zed": {
"name": "Zed Editor",
"global_path": None, # Zed uses settings.json
"workspace_path": lambda: Path.cwd() / ".zed" / "settings.json",
"config_template": lambda cmd, use_full_path: {
"context_servers": {
"amicus": {
"source": "custom",
"command": "uvx",
"args": ["amicus-mcp"],
"env": {}
}
}
}
}
}
def generate_mcp_config(
client: str = "copilot",
target: str = "stdout",
force: bool = False
) -> None:
"""Generate MCP configuration for various AI coding clients.
Args:
client: The MCP client type (copilot, claude-desktop, claude-code, vscode, cursor, windsurf, zed)
target: Where to write config - "stdout", "global", or "workspace"
force: If True, overwrite existing files without prompting
"""
if client not in MCP_CLIENTS:
print(f"Error: Unknown client '{client}'", file=sys.stderr)
print(f"Supported clients: {', '.join(MCP_CLIENTS.keys())}", file=sys.stderr)
sys.exit(1)
client_info = MCP_CLIENTS[client]
# Detect amicus-mcp installation location
amicus_path = shutil.which("amicus-mcp")
if not amicus_path:
print("Error: amicus-mcp not found in PATH", file=sys.stderr)
print("Please ensure amicus-mcp is installed and in your PATH", file=sys.stderr)
sys.exit(1)
# Check for install command shortcut
if target in ("global", "install") and "install_command" in client_info:
print(f"💡 {client_info['name']} recommends using this command instead:")
print(f"\n {client_info['install_command']}\n")
if target == "install":
return
# Generate configuration
use_full_path = (target == "global")
config = client_info["config_template"](amicus_path, use_full_path)
config_json = json.dumps(config, indent=2)
if target == "stdout":
print(f"# {client_info['name']} MCP Configuration")
print(config_json)
elif target == "global":
path_fn = client_info.get("global_path")
if path_fn is None:
print(f"Error: {client_info['name']} doesn't support global configuration files.", file=sys.stderr)
if "install_command" in client_info:
print(f"Use this command instead: {client_info['install_command']}", file=sys.stderr)
sys.exit(1)
config_path = path_fn()
_write_config_file(config_path, config_json, force, client_info["name"])
elif target == "workspace":
path_fn = client_info.get("workspace_path")
if path_fn is None:
print(f"Error: {client_info['name']} doesn't support workspace configuration files.", file=sys.stderr)
sys.exit(1)
config_path = path_fn()
_write_config_file(config_path, config_json, force, client_info["name"])
else:
print(f"Error: Unknown target '{target}'", file=sys.stderr)
sys.exit(1)
def _write_config_file(config_path: Path, config_json: str, force: bool, client_name: str) -> None:
"""Helper to write config file with confirmation."""
config_path.parent.mkdir(parents=True, exist_ok=True)
# Check if file already exists
if config_path.exists() and not force:
print(f"Configuration file already exists: {config_path}")
print("Do you want to overwrite it? [y/N]: ", end="")
try:
response = input().strip().lower()
if response not in ["y", "yes"]:
print("Aborted.")
return
except EOFError:
print("\nUse --force to overwrite without prompting.")
return
config_path.write_text(config_json + "\n")
print(f"✓ Generated {client_name} MCP configuration: {config_path}")
print("\nConfiguration written successfully!")
def main():
"""Main entry point."""
parser = argparse.ArgumentParser(
description="Amicus MCP Server - State persistence for AI coding agents",
formatter_class=argparse.RawDescriptionHelpFormatter
)
parser.add_argument("--version", "-V", action="store_true", help="Show version")
parser.add_argument("--init", action="store_true", help="Initialize project")
parser.add_argument("--list-tools", action="store_true", help="List tools")
parser.add_argument("--list-prompts", action="store_true", help="List prompts")
parser.add_argument("--validate-env", action="store_true", help="Validate env")
parser.add_argument("--show-state", action="store_true", help="Show state")
parser.add_argument("--audit-prompt", action="store_true", help="Show audit prompt")
parser.add_argument("--join-synapse", action="store_true", help="Show synapse protocol prompt")
parser.add_argument(
"--generate-mcp-config",
choices=["stdout", "global", "workspace"],
help="Generate MCP configuration (stdout=print, global=user config, workspace=project config)"
)
parser.add_argument(
"--client",
choices=list(MCP_CLIENTS.keys()),
default="copilot",
help="MCP client to generate config for (default: copilot)"
)
parser.add_argument(
"--list-clients",
action="store_true",
help="List all supported MCP clients"
)
parser.add_argument(
"--force", "-f",
action="store_true",
help="Force overwrite existing configuration files without prompting"
)
args = parser.parse_args()
if args.version:
# Get version from pyproject.toml or package metadata
try:
from importlib.metadata import version
print(f"amicus-mcp version {version('amicus-mcp')}")
except Exception:
print("amicus-mcp version 0.4.0")
elif args.init:
init_project()
elif args.list_tools:
list_tools()
elif args.list_prompts:
list_prompts()
elif args.validate_env:
validate_env()
elif args.show_state:
show_state()
elif args.audit_prompt:
show_audit_prompt()
elif args.join_synapse:
show_join_synapse_prompt()
elif args.list_clients:
print("Supported MCP Clients:")
print("=" * 50)
for key, info in MCP_CLIENTS.items():
print(f" {key:15} - {info['name']}")
if info.get("global_path"):
print(f" Global: {info['global_path']()}")
if info.get("workspace_path"):
print(f" Workspace: {info['workspace_path']()}")
if info.get("install_command"):
print(f" Install: {info['install_command']}")
print()
elif args.generate_mcp_config:
generate_mcp_config(client=args.client, target=args.generate_mcp_config, force=args.force)
else:
# Start background monitor
monitor = HeartbeatMonitor()
monitor.start()
try:
run()
finally:
monitor.stop()
if __name__ == "__main__":
main()
