mcphub/plugins/n8n/handlers/projects.py

"""Projects Handler - manages n8n projects (Enterprise/Pro)."""

import json
from typing import Any

from plugins.n8n.client import N8nApiError, N8nClient


def _error_json(exc: Exception) -> str:
    if isinstance(exc, N8nApiError):
        return json.dumps({"success": False, **exc.to_dict()}, indent=2)
    return json.dumps({"success": False, "error": str(exc)}, indent=2)


def get_tool_specifications() -> list[dict[str, Any]]:
    """Return tool specifications for ToolGenerator"""
    return [
        {
            "name": "list_projects",
            "method_name": "list_projects",
            "description": "[Enterprise] List all projects. Requires n8n Enterprise/Pro license. All parameters are OPTIONAL.",
            "schema": {
                "type": "object",
                "properties": {
                    "limit": {"type": "integer", "default": 100, "minimum": 1, "maximum": 250},
                    "cursor": {"type": "string", "description": "OPTIONAL: Pagination cursor."},
                },
            },
            "scope": "read",
        },
        {
            "name": "get_project",
            "method_name": "get_project",
            "description": "[Enterprise] Get project details by ID. Requires n8n Enterprise/Pro license.",
            "schema": {
                "type": "object",
                "properties": {"project_id": {"type": "string", "minLength": 1}},
                "required": ["project_id"],
            },
            "scope": "read",
        },
        {
            "name": "create_project",
            "method_name": "create_project",
            "description": "[Enterprise] Create a new project. Requires n8n Enterprise/Pro license.",
            "schema": {
                "type": "object",
                "properties": {
                    "name": {"type": "string", "minLength": 1, "description": "Project name"}
                },
                "required": ["name"],
            },
            "scope": "admin",
        },
        {
            "name": "update_project",
            "method_name": "update_project",
            "description": "[Enterprise] Update project metadata. Requires n8n Enterprise/Pro license.",
            "schema": {
                "type": "object",
                "properties": {
                    "project_id": {"type": "string", "minLength": 1},
                    "name": {"type": "string", "minLength": 1, "description": "New project name"},
                },
                "required": ["project_id", "name"],
            },
            "scope": "admin",
        },
        {
            "name": "delete_project",
            "method_name": "delete_project",
            "description": "[Enterprise] Delete a project. Requires n8n Enterprise/Pro license.",
            "schema": {
                "type": "object",
                "properties": {"project_id": {"type": "string", "minLength": 1}},
                "required": ["project_id"],
            },
            "scope": "admin",
        },
        {
            "name": "add_project_users",
            "method_name": "add_project_users",
            "description": "[Enterprise] Add users to a project with roles. Requires n8n Enterprise/Pro license.",
            "schema": {
                "type": "object",
                "properties": {
                    "project_id": {"type": "string", "minLength": 1},
                    "users": {
                        "type": "array",
                        "items": {
                            "type": "object",
                            "properties": {
                                "user_id": {"type": "string"},
                                "role": {
                                    "type": "string",
                                    "enum": ["project:admin", "project:editor", "project:viewer"],
                                },
                            },
                            "required": ["user_id", "role"],
                        },
                        "minItems": 1,
                    },
                },
                "required": ["project_id", "users"],
            },
            "scope": "admin",
        },
        {
            "name": "change_project_user_role",
            "method_name": "change_project_user_role",
            "description": "[Enterprise] Change a user's role in a project. Requires n8n Enterprise/Pro license.",
            "schema": {
                "type": "object",
                "properties": {
                    "project_id": {"type": "string", "minLength": 1},
                    "user_id": {"type": "string", "minLength": 1},
                    "role": {
                        "type": "string",
                        "enum": ["project:admin", "project:editor", "project:viewer"],
                    },
                },
                "required": ["project_id", "user_id", "role"],
            },
            "scope": "admin",
        },
        {
            "name": "remove_project_user",
            "method_name": "remove_project_user",
            "description": "[Enterprise] Remove a user from a project. Requires n8n Enterprise/Pro license.",
            "schema": {
                "type": "object",
                "properties": {
                    "project_id": {"type": "string", "minLength": 1},
                    "user_id": {"type": "string", "minLength": 1},
                },
                "required": ["project_id", "user_id"],
            },
            "scope": "admin",
        },
    ]


async def list_projects(client: N8nClient, limit: int = 100, cursor: str | None = None) -> str:
    try:
        response = await client.list_projects(limit=limit, cursor=cursor)
        projects = response.get("data", [])
        result = {
            "success": True,
            "count": len(projects),
            "projects": [{"id": p.get("id"), "name": p.get("name")} for p in projects],
            "next_cursor": response.get("nextCursor"),
        }
        return json.dumps(result, indent=2)
    except Exception as e:
        return _error_json(e)


async def get_project(client: N8nClient, project_id: str) -> str:
    try:
        project = await client.get_project(project_id)
        return json.dumps(
            {"success": True, "project": {"id": project.get("id"), "name": project.get("name")}},
            indent=2,
        )
    except Exception as e:
        return _error_json(e)


async def create_project(client: N8nClient, name: str) -> str:
    try:
        project = await client.create_project(name)
        return json.dumps(
            {
                "success": True,
                "message": f"Project '{name}' created",
                "project": {"id": project.get("id"), "name": project.get("name")},
            },
            indent=2,
        )
    except Exception as e:
        return _error_json(e)


async def update_project(client: N8nClient, project_id: str, name: str) -> str:
    try:
        project = await client.update_project(project_id, name)
        return json.dumps(
            {
                "success": True,
                "message": f"Project updated to '{name}'",
                "project": {"id": project.get("id"), "name": project.get("name")},
            },
            indent=2,
        )
    except Exception as e:
        return _error_json(e)


async def delete_project(client: N8nClient, project_id: str) -> str:
    try:
        await client.delete_project(project_id)
        return json.dumps({"success": True, "message": f"Project {project_id} deleted"}, indent=2)
    except Exception as e:
        return _error_json(e)


async def add_project_users(client: N8nClient, project_id: str, users: list[dict[str, str]]) -> str:
    try:
        relations = [{"userId": u["user_id"], "role": u["role"]} for u in users]
        await client.add_project_users(project_id, relations)
        return json.dumps(
            {"success": True, "message": f"Added {len(users)} user(s) to project {project_id}"},
            indent=2,
        )
    except Exception as e:
        return _error_json(e)


async def change_project_user_role(
    client: N8nClient, project_id: str, user_id: str, role: str
) -> str:
    try:
        await client.change_project_user_role(project_id, user_id, role)
        return json.dumps(
            {
                "success": True,
                "message": f"User {user_id} role changed to {role} in project {project_id}",
            },
            indent=2,
        )
    except Exception as e:
        return _error_json(e)


async def remove_project_user(client: N8nClient, project_id: str, user_id: str) -> str:
    try:
        await client.remove_project_user(project_id, user_id)
        return json.dumps(
            {"success": True, "message": f"User {user_id} removed from project {project_id}"},
            indent=2,
        )
    except Exception as e:
        return _error_json(e)