"""Asana Guard MCP Server -- MVP with two-step delete protection and audit logging.
Cursor config: "command": "python", "args": ["${workspaceFolder}/mcp-local-server/src/server.py"]
"""
from __future__ import annotations
import json
import sys
from pathlib import Path
_SRC_DIR = str(Path(__file__).resolve().parent)
if _SRC_DIR not in sys.path:
sys.path.insert(0, _SRC_DIR)
from mcp.server.fastmcp import FastMCP
from tools import delete_guard
import audit_logger
import config
server = FastMCP(
name="asana-guard",
instructions="Asana Guard MCP Server v0.1.0 - Two-step delete protection with audit logging",
)
@server.tool(
name="guard_delete_task",
description=(
"Request deletion of an Asana task. Returns a preview of the task "
"and a single-use confirm_token (valid for 60 seconds). "
"You MUST show the preview to the user and obtain explicit approval "
"before calling guard_confirm_delete."
),
)
def guard_delete_task(task_gid: str) -> str:
"""Preview a task deletion and issue a confirm token."""
try:
result = delete_guard.preview_delete_task(task_gid)
return json.dumps(result, ensure_ascii=False, indent=2)
except Exception as exc:
audit_logger.log_operation(
tool="guard_delete_task",
params={"task_gid": task_gid},
error=str(exc),
)
return json.dumps({"error": str(exc)})
@server.tool(
name="guard_delete_project",
description=(
"Request deletion of an Asana project. Returns a preview of the project "
"and a single-use confirm_token (valid for 60 seconds). "
"You MUST show the preview to the user and obtain explicit approval "
"before calling guard_confirm_delete."
),
)
def guard_delete_project(project_gid: str) -> str:
"""Preview a project deletion and issue a confirm token."""
try:
result = delete_guard.preview_delete_project(project_gid)
return json.dumps(result, ensure_ascii=False, indent=2)
except Exception as exc:
audit_logger.log_operation(
tool="guard_delete_project",
params={"project_gid": project_gid},
error=str(exc),
)
return json.dumps({"error": str(exc)})
@server.tool(
name="guard_confirm_delete",
description=(
"Execute a previously previewed deletion. Requires the confirm_token "
"returned by guard_delete_task or guard_delete_project. "
"The token expires after 60 seconds. "
"entity_type must be 'task' or 'project'."
),
)
def guard_confirm_delete(
entity_type: str,
gid: str,
confirm_token: str,
) -> str:
"""Validate the token and perform the actual deletion."""
result = delete_guard.confirm_delete(entity_type, gid, confirm_token)
return json.dumps(result, ensure_ascii=False, indent=2)
@server.tool(
name="guard_audit_log",
description=(
"Search recent audit log entries. "
"Optional filter by tool name or gid. Returns the last N entries."
),
)
def guard_audit_log(
tool_filter: str = "",
gid_filter: str = "",
limit: int = 20,
) -> str:
"""Return recent audit log entries, optionally filtered."""
entries: list[dict] = []
log_files = sorted(config.LOG_DIR.glob("audit_*.jsonl"), reverse=True)
for log_file in log_files:
if len(entries) >= limit:
break
with open(log_file, "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line:
continue
try:
entry = json.loads(line)
except json.JSONDecodeError:
continue
if tool_filter and entry.get("tool") != tool_filter:
continue
if gid_filter and entry.get("asana_gid") != gid_filter:
continue
entries.append(entry)
if len(entries) >= limit:
break
return json.dumps(entries[-limit:], ensure_ascii=False, indent=2)
@server.tool(
name="guard_check_connection",
description="Verify that the Asana API connection is working. Returns the authenticated user info.",
)
def guard_check_connection() -> str:
"""Test connectivity to Asana API."""
import asana_client
try:
user = asana_client.get_user_me()
return json.dumps({
"status": "ok",
"user_name": user.get("name"),
"user_gid": user.get("gid"),
"user_email": user.get("email"),
}, ensure_ascii=False, indent=2)
except Exception as exc:
return json.dumps({"status": "error", "message": str(exc)})
if __name__ == "__main__":
server.run(transport="stdio")
