monday.com MCP Server¶

# Python example: Connect to monday.com via MCP Gateway
import asyncio
from mcp_gateway_client import MCPGatewayClient

async def connect_monday():
    # Initialize gateway client
    gateway = MCPGatewayClient(
        base_url="http://localhost:4444",
        bearer_token=os.getenv("MCPGATEWAY_BEARER_TOKEN")
    )

    # Connect to monday.com server
    await gateway.connect_server("monday-official")

    # List available tools
    tools = await gateway.list_tools("monday-official")
    print(f"Available monday.com tools: {len(tools)}")

    return gateway

# Run connection
gateway = asyncio.run(connect_monday())

import asyncio
from mcp_gateway_client import MCPGatewayClient

async def setup_project_board():
    """Create a new project board with items and team assignments"""
    gateway = MCPGatewayClient(
        base_url="http://localhost:4444",
        bearer_token=os.getenv("MCPGATEWAY_BEARER_TOKEN")
    )

    # Create a new board
    board = await gateway.call_tool(
        server="monday-official",
        tool="create_board",
        arguments={
            "workspace_id": "12345678",
            "board_name": "Website Redesign Project",
            "board_kind": "public",
            "description": "Complete website redesign for Q1 2025"
        }
    )

    board_id = board["id"]
    print(f"Created board: {board_id}")

    # Add custom columns
    await gateway.call_tool(
        server="monday-official",
        tool="create_column",
        arguments={
            "board_id": board_id,
            "title": "Priority",
            "column_type": "status",
            "defaults": {
                "labels": {
                    "0": "Low",
                    "1": "Medium",
                    "2": "High"
                }
            }
        }
    )

    # Create project tasks
    tasks = [
        "Design mockups",
        "Develop frontend",
        "Backend API integration",
        "Testing and QA",
        "Deployment"
    ]

    for task_name in tasks:
        item = await gateway.call_tool(
            server="monday-official",
            tool="create_item",
            arguments={
                "board_id": board_id,
                "item_name": task_name,
                "column_values": {
                    "status": "Not Started",
                    "priority": "Medium"
                }
            }
        )
        print(f"Created task: {item['name']}")

    return board_id

# Run the setup
board_id = asyncio.run(setup_project_board())

async def automate_status_workflow(board_id):
    """Set up automated notifications when tasks are completed"""
    gateway = MCPGatewayClient(
        base_url="http://localhost:4444",
        bearer_token=os.getenv("MCPGATEWAY_BEARER_TOKEN")
    )

    # Create automation: notify team when status changes to "Done"
    automation = await gateway.call_tool(
        server="monday-official",
        tool="create_automation",
        arguments={
            "board_id": board_id,
            "trigger": {
                "type": "status_changed",
                "column_id": "status",
                "value": "Done"
            },
            "action": {
                "type": "notify_team",
                "message": "🎉 Task completed: {item_name} by {person}"
            }
        }
    )

    print(f"Automation created: {automation['id']}")

    # Create automation: move to "Completed" group when done
    await gateway.call_tool(
        server="monday-official",
        tool="create_automation",
        arguments={
            "board_id": board_id,
            "trigger": {
                "type": "status_changed",
                "column_id": "status",
                "value": "Done"
            },
            "action": {
                "type": "move_item_to_group",
                "group_id": "completed_tasks"
            }
        }
    )

async def track_team_progress(board_id):
    """Monitor team progress and log time"""
    gateway = MCPGatewayClient(
        base_url="http://localhost:4444",
        bearer_token=os.getenv("MCPGATEWAY_BEARER_TOKEN")
    )

    # Get all items from the board
    board = await gateway.call_tool(
        server="monday-official",
        tool="get_board",
        arguments={
            "board_id": board_id,
            "include_items": True
        }
    )

    # Track time for each in-progress item
    for item in board["items"]:
        if item["column_values"]["status"] == "Working on it":
            # Log time
            await gateway.call_tool(
                server="monday-official",
                tool="log_time",
                arguments={
                    "item_id": item["id"],
                    "hours": 2.0,
                    "date": "2024-12-10",
                    "notes": "Development work"
                }
            )

            # Update progress
            await gateway.call_tool(
                server="monday-official",
                tool="update_item",
                arguments={
                    "item_id": item["id"],
                    "column_values": {
                        "progress": 50
                    }
                }
            )

    # Generate progress report
    report = await gateway.call_tool(
        server="monday-official",
        tool="generate_report",
        arguments={
            "board_id": board_id,
            "report_type": "status_summary",
            "include_charts": True
        }
    )

    print(f"Team Progress Report:")
    print(f"- Total Items: {report['total_items']}")
    print(f"- Completed: {report['completed_items']}")
    print(f"- In Progress: {report['in_progress_items']}")
    print(f"- Not Started: {report['not_started_items']}")

from fastapi import FastAPI, Request, HTTPException
import hmac
import hashlib

app = FastAPI()

async def verify_monday_signature(request: Request, secret: str) -> bool:
    """Verify webhook signature from monday.com"""
    signature = request.headers.get("X-Monday-Signature")
    if not signature:
        return False

    body = await request.body()
    expected_signature = hmac.new(
        secret.encode(),
        body,
        hashlib.sha256
    ).hexdigest()

    return hmac.compare_digest(signature, expected_signature)

@app.post("/webhooks/monday")
async def handle_monday_webhook(request: Request):
    """Handle incoming webhooks from monday.com"""
    # Verify signature
    secret = os.getenv("MONDAY_WEBHOOK_SECRET")
    if not await verify_monday_signature(request, secret):
        raise HTTPException(status_code=401, detail="Invalid signature")

    # Parse webhook payload
    payload = await request.json()
    event_type = payload.get("event", {}).get("type")

    # Handle different event types
    if event_type == "item_created":
        item = payload["event"]["pulseId"]
        print(f"New item created: {item}")

        # Automatically assign to team member
        gateway = MCPGatewayClient(
            base_url="http://localhost:4444",
            bearer_token=os.getenv("MCPGATEWAY_BEARER_TOKEN")
        )

        await gateway.call_tool(
            server="monday-official",
            tool="assign_user_to_item",
            arguments={
                "item_id": item,
                "user_id": "12345"
            }
        )

    elif event_type == "status_changed":
        item_id = payload["event"]["pulseId"]
        new_status = payload["event"]["value"]["label"]["text"]
        print(f"Status changed to: {new_status}")

        # Send notification or trigger other actions
        if new_status == "Done":
            # Archive completed item
            pass

    return {"status": "processed"}

async def generate_monthly_report(workspace_id: str, month: str):
    """Generate comprehensive monthly report across all boards"""
    gateway = MCPGatewayClient(
        base_url="http://localhost:4444",
        bearer_token=os.getenv("MCPGATEWAY_BEARER_TOKEN")
    )

    # Get all boards in workspace
    boards = await gateway.call_tool(
        server="monday-official",
        tool="get_boards",
        arguments={
            "workspace_id": workspace_id
        }
    )

    report_data = {
        "month": month,
        "boards": [],
        "total_items": 0,
        "completed_items": 0,
        "total_hours": 0
    }

    # Analyze each board
    for board in boards:
        board_report = await gateway.call_tool(
            server="monday-official",
            tool="generate_report",
            arguments={
                "board_id": board["id"],
                "report_type": "status_summary",
                "date_range": {
                    "start": f"{month}-01",
                    "end": f"{month}-31"
                }
            }
        )

        # Get time tracking data
        time_data = await gateway.call_tool(
            server="monday-official",
            tool="get_time_tracking",
            arguments={
                "board_id": board["id"],
                "start_date": f"{month}-01",
                "end_date": f"{month}-31"
            }
        )

        report_data["boards"].append({
            "name": board["name"],
            "items": board_report["total_items"],
            "completed": board_report["completed_items"],
            "hours": sum(entry["hours"] for entry in time_data)
        })

        report_data["total_items"] += board_report["total_items"]
        report_data["completed_items"] += board_report["completed_items"]
        report_data["total_hours"] += sum(entry["hours"] for entry in time_data)

    # Calculate metrics
    report_data["completion_rate"] = (
        report_data["completed_items"] / report_data["total_items"] * 100
        if report_data["total_items"] > 0 else 0
    )

    print(f"\n📊 Monthly Report for {month}")
    print(f"{'='*50}")
    print(f"Total Items: {report_data['total_items']}")
    print(f"Completed: {report_data['completed_items']}")
    print(f"Completion Rate: {report_data['completion_rate']:.1f}%")
    print(f"Total Hours: {report_data['total_hours']:.1f}")
    print(f"\nBoard Breakdown:")
    for board in report_data["boards"]:
        print(f"  - {board['name']}: {board['completed']}/{board['items']} items, {board['hours']:.1f} hours")

    return report_data

# Generate report
report = asyncio.run(generate_monthly_report("12345678", "2024-12"))

# Implement reconnection logic
async def connect_with_retry(gateway, server_id, max_retries=5):
    """Connect to server with exponential backoff"""
    for attempt in range(max_retries):
        try:
            await gateway.connect_server(server_id)
            print(f"Connected to {server_id}")
            return True
        except ConnectionError as e:
            if attempt < max_retries - 1:
                wait_time = 2 ** attempt
                print(f"Connection failed, retrying in {wait_time}s...")
                await asyncio.sleep(wait_time)
            else:
                print(f"Failed to connect after {max_retries} attempts")
                raise

from asyncio import Semaphore
import time

class RateLimitedMondayClient:
    """Client with built-in rate limiting"""

    def __init__(self, gateway, requests_per_minute=60):
        self.gateway = gateway
        self.semaphore = Semaphore(requests_per_minute)
        self.request_times = []

    async def call_tool(self, tool, arguments):
        """Call tool with rate limiting"""
        async with self.semaphore:
            # Remove old timestamps
            current_time = time.time()
            self.request_times = [
                t for t in self.request_times
                if current_time - t < 60
            ]

            # Wait if at limit
            if len(self.request_times) >= 60:
                wait_time = 60 - (current_time - self.request_times[0])
                if wait_time > 0:
                    await asyncio.sleep(wait_time)

            # Make request
            self.request_times.append(time.time())
            return await self.gateway.call_tool(
                server="monday-official",
                tool=tool,
                arguments=arguments
            )

async def verify_board_access(gateway, board_id):
    """Check if current user can access a board"""
    try:
        board = await gateway.call_tool(
            server="monday-official",
            tool="get_board",
            arguments={"board_id": board_id}
        )
        print(f"✓ Access granted to board: {board['name']}")
        return True
    except Exception as e:
        if "not_found" in str(e):
            print(f"✗ Board {board_id} not found or no access")
        elif "access_denied" in str(e):
            print(f"✗ Access denied to board {board_id}")
        return False