Custom Monday.com action script

Hi all, I wasn’t able to set up the monday.com action that is in the Pickaxe marketplace because I’m not an admin of my client’s Monday account, but built my own action that is working, thought I’d share.

Trigger prompt:

When a user is working through any task, project, action item, or 
follow-up that would benefit from tracking, proactively suggest 
logging it to Monday.com. For example, if the user mentions they 
need to follow up with someone, create a deliverable, assign work 
to a team member, or track a deadline, say something like "Would 
you like me to add this to your Monday.com board?"

If they say yes, ask:
1. Which workspace to use — do they want to use an existing one or 
   create a new one? If existing, what is the name?
2. Which board to use — do they want to use an existing board or 
   create a new one? If existing, what is the name?
3. Which section/group to add it to (optional) — existing or new?
4. What to name the task
5. Who to assign it to (optional)
6. Due date (optional)

Once you have at least a workspace name, board name, and task name, 
use the create_monday_task tool. The tool will automatically create 
any workspace, board, or group that doesn't already exist, so it is 
safe to proceed with either new or existing names.

Function inputs

  • workspace_name
  • board_name
  • group_name (optional)
  • task_name
  • assignee_name (optional)
  • due_date (optional)

Environment variables

MONDAY_API_TOKEN (find this in Monday > Account > Developers)

Action Code

import os
import requests
import json

def create_monday_task(workspace_name: str, board_name: str, group_name: str, task_name: str, assignee_name: str, due_date: str):
    """
    Create Monday Task in Karsh Monday Account

    Args:
        workspace_name (string): workspace_name
        board_name (string): board_name
        group_name (string): group_name (optional)
        task_name (string): task_name
        assignee_name (string): assignee_name (optional)
        due_date (string): due_date (optional, YYYY-MM-DD format)
    Envs:
        MONDAY_API_TOKEN (string): Find on Monday > Profile > Developers
    """



    MONDAY_API_TOKEN = os.environ["MONDAY_API_TOKEN"]

    def run_query(query, variables=None):
        headers = {
            "Authorization": MONDAY_API_TOKEN,
            "Content-Type": "application/json",
            "API-Version": "2024-01"
        }
        payload = {"query": query}
        if variables:
            payload["variables"] = variables
        response = requests.post(
            "https://api.monday.com/v2",
            json=payload,
            headers=headers
        )
        return response.json()

    def get_workspace_id(name):
        query = """
        query {
            workspaces {
                id
                name
            }
        }
        """
        result = run_query(query)
        workspaces = result.get("data", {}).get("workspaces", [])
        for w in workspaces:
            if w["name"].lower() == name.lower():
                return w["id"]
        return None

    def create_workspace(name):
        query = """
        mutation {
            create_workspace(
                name: "%s",
                kind: open
            ) {
                id
                name
            }
        }
        """ % name
        return run_query(query)

    def get_board_id(workspace_id, name):
        query = """
        query {
            boards(workspace_ids: [%s]) {
                id
                name
            }
        }
        """ % workspace_id
        result = run_query(query)
        boards = result.get("data", {}).get("boards", [])
        for b in boards:
            if b["name"].lower() == name.lower():
                return b["id"]
        return None

    def create_board(name, workspace_id):
        query = """
        mutation {
            create_board(
                board_name: "%s",
                board_kind: public,
                workspace_id: %s
            ) {
                id
                name
            }
        }
        """ % (name, workspace_id)
        return run_query(query)

    def get_group_id(board_id, name):
        query = """
        query {
            boards(ids: [%s]) {
                groups {
                    id
                    title
                }
            }
        }
        """ % board_id
        result = run_query(query)
        groups = result.get("data", {}).get("boards", [{}])[0].get("groups", [])
        for g in groups:
            if g["title"].lower() == name.lower():
                return g["id"]
        return None

    def create_group(board_id, name):
        query = """
        mutation {
            create_group(
                board_id: %s,
                group_name: "%s"
            ) {
                id
                title
            }
        }
        """ % (board_id, name)
        return run_query(query)

    def get_user_id(name):
        query = """
        query {
            users {
                id
                name
            }
        }
        """
        result = run_query(query)
        users = result.get("data", {}).get("users", [])
        for u in users:
            if u["name"].lower() == name.lower():
                return u["id"]
        return None

    def create_item(board_id, item_name, group_id=None, assignee_id=None, due_date=None):
        column_values = {}
        if assignee_id:
            column_values["person"] = {"personsAndTeams": [{"id": int(assignee_id), "kind": "person"}]}
        if due_date and due_date != "":
            column_values["date"] = {"date": due_date}

        col_values_str = json.dumps(json.dumps(column_values))
        group_line = 'group_id: "%s",' % group_id if group_id else ""

        query = """
        mutation {
            create_item(
                board_id: %s,
                %s
                item_name: "%s",
                column_values: %s
            ) {
                id
                name
            }
        }
        """ % (board_id, group_line, item_name, col_values_str)
        return run_query(query)

    # --- RUN ---
    notes = []

    # Workspace lookup or create
    workspace_id = get_workspace_id(workspace_name)
    if not workspace_id:
        ws_result = create_workspace(workspace_name)
        new_ws = ws_result.get("data", {}).get("create_workspace", {})
        if new_ws:
            workspace_id = new_ws["id"]
            notes.append("Created new workspace '%s'." % workspace_name)
        else:
            print("Could not find or create workspace '%s': %s" % (workspace_name, str(ws_result)))
            return

    # Board lookup or create
    board_id = get_board_id(workspace_id, board_name)
    if not board_id:
        board_result = create_board(board_name, workspace_id)
        new_board = board_result.get("data", {}).get("create_board", {})
        if new_board:
            board_id = new_board["id"]
            notes.append("Created new board '%s'." % board_name)
        else:
            print("Could not find or create board '%s': %s" % (board_name, str(board_result)))
            return

    # Group lookup or create
    group_id = None
    if group_name and group_name != "":
        group_id = get_group_id(board_id, group_name)
        if not group_id:
            group_result = create_group(board_id, group_name)
            new_group = group_result.get("data", {}).get("create_group", {})
            if new_group:
                group_id = new_group["id"]
                notes.append("Created new group '%s'." % group_name)
            else:
                notes.append("Could not create group '%s' — task will be in default group." % group_name)

    # User lookup
    assignee_id = None
    if assignee_name and assignee_name != "":
        assignee_id = get_user_id(assignee_name)
        if not assignee_id:
            notes.append("User '%s' not found — task created unassigned." % assignee_name)

    # Create the task
    result = create_item(board_id, task_name, group_id, assignee_id, due_date)
    item = result.get("data", {}).get("create_item", {})

    if item:
        msg = "Task '%s' created successfully in %s > %s" % (item["name"], workspace_name, board_name)
        if group_id:
            msg += " > %s" % group_name
        if notes:
            msg += "\n\nNotes:\n" + "\n".join(["• " + n for n in notes])
        print(msg)
    else:
        print("Task could not be created: " + str(result))
2 Likes