How to Integrate AI Agents with Asana

pip install asana langchain langchain-openai python-dotenv

export ASANA_ACCESS_TOKEN="your-asana-access-token"
export ASANA_WORKSPACE_GID="your-workspace-gid"

import asana
import os

configuration = asana.Configuration()
configuration.access_token = os.getenv("ASANA_ACCESS_TOKEN")
api_client = asana.ApiClient(configuration)
workspaces_api = asana.WorkspacesApi(api_client)
workspaces = workspaces_api.get_workspaces({})
for ws in workspaces:
    print(ws)  # Lists your workspaces

import os
import asana
from datetime import datetime, timedelta
from langchain.tools import tool
from dotenv import load_dotenv

load_dotenv()

configuration = asana.Configuration()
configuration.access_token = os.getenv("ASANA_ACCESS_TOKEN")
api_client = asana.ApiClient(configuration)
WORKSPACE_GID = os.getenv("ASANA_WORKSPACE_GID")

tasks_api = asana.TasksApi(api_client)
projects_api = asana.ProjectsApi(api_client)
users_api = asana.UsersApi(api_client)


@tool
def list_projects() -> str:
    """List all active projects in the Asana workspace."""
    projects = projects_api.get_projects({"workspace": WORKSPACE_GID, "archived": False})
    result = ["Active projects:"]
    for project in projects:
        result.append(f"  - {project['name']} (GID: {project['gid']})")
    return "\n".join(result) if len(result) > 1 else "No active projects found"


@tool
def get_project_tasks(project_gid: str, completed: bool = False) -> str:
    """Get tasks from a specific Asana project. Set completed=True to include finished tasks."""
    tasks = tasks_api.get_tasks_for_project(
        project_gid,
        {"completed_since": "now" if not completed else "1970-01-01",
         "opt_fields": "name,assignee.name,due_on,completed,notes"}
    )
    task_list = list(tasks)
    if not task_list:
        return f"No {'incomplete' if not completed else ''} tasks in project {project_gid}"

    result = [f"Tasks in project ({len(task_list)} found):"]
    for task in task_list[:20]:
        assignee = task.get("assignee", {}).get("name", "Unassigned") if task.get("assignee") else "Unassigned"
        due = task.get("due_on", "No due date")
        status = "✅" if task.get("completed") else "⬜"
        result.append(f"  {status} {task['name']} | Assignee: {assignee} | Due: {due}")
    return "\n".join(result)


@tool
def create_task(name: str, project_gid: str, assignee_email: str = None,
                due_days_from_now: int = None, notes: str = "") -> str:
    """
    Create a new task in an Asana project.
    assignee_email: optional email of the assignee.
    due_days_from_now: optional days until the due date.
    """
    task_data = {
        "name": name,
        "notes": notes,
        "projects": [project_gid],
        "workspace": WORKSPACE_GID
    }

    if assignee_email:
        # Find user by email
        users = users_api.get_users({"workspace": WORKSPACE_GID, "opt_fields": "name,email"})
        for user in users:
            if user.get("email") == assignee_email:
                task_data["assignee"] = user["gid"]
                break

    if due_days_from_now is not None:
        due_date = (datetime.now() + timedelta(days=due_days_from_now)).strftime("%Y-%m-%d")
        task_data["due_on"] = due_date

    task = tasks_api.create_task({"data": task_data})
    return f"Task created: '{task['name']}' (GID: {task['gid']}) in project {project_gid}"


@tool
def get_overdue_tasks(project_gid: str = None) -> str:
    """Get all overdue incomplete tasks, optionally filtered by project."""
    today = datetime.now().strftime("%Y-%m-%d")
    params = {
        "workspace": WORKSPACE_GID,
        "completed_since": "now",
        "due_before": today,
        "opt_fields": "name,assignee.name,due_on,projects.name"
    }
    if project_gid:
        tasks = tasks_api.get_tasks_for_project(project_gid, {
            "completed_since": "now",
            "opt_fields": "name,assignee.name,due_on"
        })
    else:
        tasks = tasks_api.get_tasks(params)

    overdue_list = list(tasks)
    if not overdue_list:
        return "No overdue tasks found"

    result = [f"Overdue tasks ({len(overdue_list)} total):"]
    for task in overdue_list[:15]:
        assignee = task.get("assignee", {}).get("name", "Unassigned") if task.get("assignee") else "Unassigned"
        days_overdue = (datetime.now() - datetime.strptime(task.get("due_on", today), "%Y-%m-%d")).days
        result.append(f"  {task['name']} | Assignee: {assignee} | {days_overdue} days overdue")
    return "\n".join(result)


@tool
def update_task_status(task_gid: str, completed: bool = True, notes: str = "") -> str:
    """Mark a task as complete or incomplete. Optionally add completion notes."""
    update_data = {"completed": completed}
    if notes:
        update_data["notes"] = notes
    task = tasks_api.update_task(task_gid, {"data": update_data})
    status = "completed" if completed else "reopened"
    return f"Task '{task['name']}' {status} successfully"

from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o", temperature=0.1)
tools = [list_projects, get_project_tasks, create_task,
         get_overdue_tasks, update_task_status]

prompt = ChatPromptTemplate.from_messages([
    ("system", """You are a project management assistant with access to Asana.

When processing meeting notes or requests:
1. Parse all action items with clear owners and deadlines
2. Create tasks with specific, actionable names (verb + object, e.g., "Review API proposal")
3. Set realistic due dates based on priority and complexity
4. Always list the projects available before creating tasks

When reporting project health:
1. Check overdue tasks and quantify the backlog
2. Calculate completion rate for tasks due this week
3. Identify team members with concentrated overdue items
4. Provide concise recommendations, not lengthy reports"""),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True, max_iterations=8)