nexus-mcp/archive/Workday/workday-mcp/server.py

from mcp.server.fastmcp import FastMCP
from typing import Any
from lib.data import (
    MOCK_WORKERS,
    scan_department_drift,
    scan_job_title_mismatches,
    scan_name_variance,
    scan_status_reconciliation_mismatches,
)

mcp = FastMCP("Workday-Sync") #Server name for MCP registration, e.g. "Workday-Sync"

@mcp.tool()
def get_worker_status(employee_id: str) -> dict[str, Any]:
    """
    WIS-009: Fetch structured worker status.
    This replaces the previous string-based version.
    """
    worker_id = employee_id.upper()
    worker = MOCK_WORKERS.get(worker_id)

    # Structured Return (WIS-004 Allowlist)
    if worker:
        return {
            "employee_id": worker_id,
            "full_name": worker["name"],
            "status": worker["status"],
            "department": worker["dept"],
            "work_email": worker["email"],
            "manager_id": worker["manager_id"],
        }

    return {"error": f"Worker {employee_id} not found."}

@mcp.tool()
def get_worker_manager(employee_id: str) -> dict[str, Any]:
    """WIS-017 prep: resolve a worker's manager relationship from mock data."""
    worker_id = employee_id.upper()
    worker = MOCK_WORKERS.get(worker_id)
    if not worker:
        return {"error": f"Worker {employee_id} not found."}

    manager_id = worker.get("manager_id", "")
    if not manager_id:
        return {
            "employee_id": worker_id,
            "employee_name": worker["name"],
            "manager_id": "",
            "manager_name": None,
            "relationship_status": "no_manager_assigned",
        }

    manager = MOCK_WORKERS.get(manager_id)
    if not manager:
        return {
            "employee_id": worker_id,
            "employee_name": worker["name"],
            "manager_id": manager_id,
            "manager_name": None,
            "relationship_status": "manager_not_found",
        }

    return {
        "employee_id": worker_id,
        "employee_name": worker["name"],
        "manager_id": manager_id,
        "manager_name": manager["name"],
        "relationship_status": "ok",
    }

@mcp.tool()
def scan_manager_mismatches() -> dict[str, Any]:
    """
    WIS-017: Scan all workers and return a report of manager relationship errors.
    This provides the data for WIS-020 weekly drift reporting.
    """
    mismatches: list[dict[str, Any]] = []
    total_scanned = 0

    for emp_id, details in MOCK_WORKERS.items():
        total_scanned += 1
        manager_id = details.get("manager_id")
        
        # We only care about cases where a manager is assigned but cannot be found
        if manager_id and manager_id not in MOCK_WORKERS:
            mismatches.append({
                "employee_id": emp_id,
                "employee_name": details["name"],
                "invalid_manager_id": manager_id,
                "error_type": "manager_not_found"
            })

    return {
        "scan_summary": {
            "total_records_checked": total_scanned,
            "mismatches_found": len(mismatches),
            "status": "action_required" if mismatches else "clean"
        },
        "mismatches": mismatches
    }


@mcp.tool()
def scan_status_reconciliation() -> dict[str, Any]:
    """
    WIS-014: Flag workers terminated in Workday but still enabled in AD.
    """
    return scan_status_reconciliation_mismatches()


@mcp.tool()
def scan_job_title_drift() -> dict[str, Any]:
    """
    Detect workers whose Workday title differs from their AD title.
    """
    return scan_job_title_mismatches()


@mcp.tool()
def scan_department_mismatches() -> dict[str, Any]:
    """
    Detect workers whose Workday department differs from their AD department.
    """
    return scan_department_drift()


@mcp.tool()
def scan_name_variance_mismatches() -> dict[str, Any]:
    """
    Detect AD display names that do not align with Workday legal or preferred names.
    """
    return scan_name_variance()

if __name__ == "__main__":
    mcp.run()