Skip to content

Sync Engine

sync

Sync engine — read crux.json + registry, generate .mcp.json.

Launcher scripts are no longer generated per-MCP. Instead, shared launcher scripts installed by crux setup handle keychain lookups at runtime, driven by env vars in the .mcp.json entries.

sync_project

sync_project(project_dir, registry=None)

Sync a single project: generate .mcp.json, copy skills.

Source code in src/crux_cli/sync.py
def sync_project(
    project_dir: Path,
    registry: dict[str, Any] | None = None,
) -> tuple[bool, list[str]]:
    """Sync a single project: generate .mcp.json, copy skills."""
    crux_json_path = project_dir / "crux.json"
    if not crux_json_path.exists():
        return False, ["No crux.json found"]

    with open(crux_json_path) as f:
        crux_json = json.load(f)

    if registry is None:
        registry = _load_registry_for_sync()

    mcp_defs = registry.get("mcp_definitions", {})
    skill_defs = registry.get("skill_definitions", {})
    issues: list[str] = []

    # Build .mcp.json
    declared_mcps = crux_json.get("mcps", [])
    mcp_servers: dict[str, Any] = {}
    for mcp_entry in declared_mcps:
        if isinstance(mcp_entry, str):
            mcp_name, extra_args = mcp_entry, []
        else:
            mcp_name = mcp_entry.get("name", "")
            extra_args = mcp_entry.get("args", [])
        if mcp_name not in mcp_defs:
            issues.append(f"MCP '{mcp_name}' not found in registry")
            continue
        mcp_servers[mcp_name] = _build_server_entry(
            mcp_name, mcp_defs[mcp_name], extra_args or None
        )

    mcp_file = project_dir / ".mcp.json"
    new_config = {"mcpServers": mcp_servers}
    _atomic_json_write(mcp_file, new_config)

    # Copy skills
    declared_skills = crux_json.get("skills", [])
    for skill_name in declared_skills:
        if skill_name not in skill_defs:
            issues.append(f"Skill '{skill_name}' not found in registry")
            continue
        skill_data = skill_defs[skill_name]
        source_dir_str = skill_data.get("source_dir", "")
        source_path = Path(source_dir_str)
        if not source_path.is_absolute():
            candidate = skills_dir() / skill_name
            if candidate.exists():
                source_path = candidate
            elif source_dir_str:
                source_path = crux_home().parent / source_dir_str
            else:
                source_path = skills_dir() / skill_name

        safe_skill = Path(skill_name).name
        skills_parent = project_dir / ".claude" / "skills"
        dest_path = skills_parent / safe_skill
        if source_path.exists():
            import shutil

            if not dest_path.resolve().is_relative_to(skills_parent.resolve()):
                issues.append(f"Skill '{skill_name}' resolves outside skills directory")
                continue
            dest_path.parent.mkdir(parents=True, exist_ok=True)
            if dest_path.exists():
                shutil.rmtree(dest_path)
            shutil.copytree(source_path, dest_path)
        else:
            issues.append(f"Skill '{skill_name}' source missing at {source_path}")

    return True, issues