"""Asana project management — in-process MCP so Jarvis runs Ahmed's projects by voice. THIN WRAPPER — deliberately. All the REST logic lives in ONE place, `jarvis-tools-service/asana.py`, which this module imports and re-exposes as in-process SDK-MCP tools. The hosted connector (mcp_server.py) wraps the SAME functions. So voice Jarvis and the claude.ai connector have IDENTICAL Asana capability, and one bug fix fixes both. Never re-implement a call here. Asana (app.asana.com) is where Ahmed's PROJECT work lives: projects, their tasks, due dates, assignees and comment threads. Distinct from mcp__tasks__* (his personal to-do list on the Railway brain). Auth: ASANA_PAT in .env, sent as `Authorization: Bearer `. MCP_ASANA=0 disables the whole server; a missing jarvis-tools-service dir just disables it too (enabled() returns False — the engine keeps booting). Optional ASANA_WORKSPACE (gid or name) pins the workspace. No extra deps — stdlib urllib. Everything a tool returns is SPOKEN ALOUD, so results are short plain prose / plain "- name — due …" lines. Never markdown, never a JSON dump, never a traceback: failures come back as one human sentence. (The one exception is asana_api, the escape hatch, whose raw JSON is for the model, not for speech.) Tools (Claude sees them as mcp__asana__): asana_workspaces list workspaces (and the default one) asana_projects list projects, optional name filter, optional archived asana_project_create new project (team needed only in an organization) asana_project_update rename / re-date / re-own / ARCHIVE (reversible) asana_project_members list, add or remove the people on a project asana_project_status post a status update (on_track / at_risk / …) asana_task FULL detail of one task (notes, section, subtasks, comments) asana_tasks open (or completed) tasks: a project, a section, or his own asana_task_create new task (project / section / parent / assignee / due date) asana_task_update rename, re-note, re-date, re-assign, (un)complete asana_task_move to another project and/or section (add + remove, a real move) asana_task_complete shortcut for "mark it done" asana_task_delete delete a task (recoverable for 30 days) asana_task_people assignee, followers, tags asana_subtasks list or create subtasks asana_comment post a comment asana_comments read the comment thread asana_search find tasks (premium search, typeahead fallback) asana_api escape hatch — any Asana REST endpoint """ from __future__ import annotations import os import sys from pathlib import Path # The one implementation, shared with the hosted connector. _SERVICE = Path(__file__).resolve().parent.parent / "jarvis-tools-service" asana = None try: # a missing dir must not kill the engine if _SERVICE.is_dir(): if str(_SERVICE) not in sys.path: sys.path.insert(0, str(_SERVICE)) import asana # type: ignore # noqa: E402 (jarvis-tools-service/asana.py) except Exception: # noqa: BLE001 asana = None def enabled() -> bool: """On when the shared module imported and a Personal Access Token is present.""" if os.environ.get("MCP_ASANA", "1") == "0": return False return bool(asana) and asana.configured() def _text(msg: str) -> dict: return {"content": [{"type": "text", "text": str(msg)}]} def _s(args: dict, key: str, default: str = "") -> str: return str(args.get(key, default) or default).strip() def _b(args: dict, key: str): """Tri-state: absent/'' → None, else true/false. (MCP may hand us a string.)""" v = args.get(key) if v is None or v == "": return None if isinstance(v, bool): return v return str(v).strip().lower() in ("1", "true", "yes", "y", "done", "complete") def _schema(props: dict, required: list[str] | None = None) -> dict: """A FULL JSON Schema for @tool. Why not the {"name": str} shorthand the SDK also accepts: it marks EVERY declared parameter as required, so `asana_projects` (all-optional) would be rejected with "'query' is a required property". A full schema (type + properties) is passed through verbatim, so we control `required` — and each property carries its own description, which is what the model actually reads. props: {name: "description"} → string, or {name: ("integer", "description")}. """ out: dict = {} for k, v in props.items(): kind, desc = ("string", v) if isinstance(v, str) else v out[k] = {"type": kind, "description": desc} return {"type": "object", "properties": out, "required": list(required or [])} # --------------------------------------------------------------------------- # In-process MCP server (Claude sees mcp__asana__) # --------------------------------------------------------------------------- def build_server(): import asyncio from claude_agent_sdk import create_sdk_mcp_server, tool async def _run(fn, *a, **kw) -> dict: """Every asana.* function is blocking and NEVER raises — it returns a sentence. Run it off the event loop and speak whatever comes back.""" return _text(await asyncio.to_thread(fn, *a, **kw)) @tool("asana_workspaces", "List the Asana workspaces this account can see (and which one the other " "asana tools default to). Rarely needed — only for 'which Asana am I on' " "or when a project can't be found.", _schema({})) async def asana_workspaces(args: dict) -> dict: return await _run(asana.workspaces) @tool("asana_projects", "List Ahmed's Asana projects. `query` optionally narrows by name " "('redesign', 'client'). `archived`='true' lists archived ones instead. " "Use for 'what projects do I have', 'what's in Asana'.", _schema({"query": "a word or two to narrow the list by name", "archived": ("boolean", "true = list ARCHIVED projects instead")})) async def asana_projects(args: dict) -> dict: return await _run(asana.projects, _s(args, "query"), bool(_b(args, "archived"))) @tool("asana_project_create", "Create an Asana project. `name` required. `team` is only needed if the " "workspace is an ORGANIZATION (the tool tells you, and lists the teams, " "if so). Optional `notes`, `due_on` (YYYY-MM-DD), `privacy` " "(public_to_workspace | private_to_team | private).", _schema({"name": "the project name", "team": "only needed in an ORGANIZATION workspace", "notes": "description of the project", "due_on": "YYYY-MM-DD", "privacy": "public_to_workspace | private_to_team | private"}, ["name"])) async def asana_project_create(args: dict) -> dict: return await _run(asana.project_create, _s(args, "name"), _s(args, "team"), _s(args, "notes"), _s(args, "due_on"), _s(args, "privacy")) @tool("asana_project_update", "Change an Asana project (`project` = its name or gid): `name` (rename), " "`notes`, `due_on`, `owner`, or `archive`='true' to ARCHIVE it (hides it, " "keeps everything, reversible with archive='false'). Archiving is how you " "'get rid of' a project — deleting one is NOT offered here because it " "can't be undone.", _schema({"project": "the project's name or gid", "name": "a new name (rename)", "notes": "new description", "due_on": "YYYY-MM-DD", "owner": "new owner: a name, email or 'me'", "archive": ("boolean", "true = archive it, false = un-archive")}, ["project"])) async def asana_project_update(args: dict) -> dict: return await _run(asana.project_update, _s(args, "project"), _s(args, "name"), _s(args, "notes"), _s(args, "due_on"), _s(args, "owner"), _b(args, "archive")) @tool("asana_project_members", "Who's on a project — and invite/remove them ('add Sara to the launch " "project'). `add` / `remove` = comma-separated emails, names or gids ('me' " "works). `access_level` (admin|editor|commenter|viewer) sets their rights " "when adding. No add/remove = just list the members. NOTE: this shares an " "existing project with someone already in the workspace; inviting a " "brand-new person to Asana itself is asana_api POST " "/workspaces/{gid}/addUser.", _schema({"project": "the project's name or gid", "add": "comma-separated people to add (emails, names, gids, 'me')", "remove": "comma-separated people to remove", "access_level": "admin | editor | commenter | viewer"}, ["project"])) async def asana_project_members(args: dict) -> dict: return await _run(asana.project_members, _s(args, "project"), _s(args, "add"), _s(args, "remove"), _s(args, "access_level")) @tool("asana_project_status", "Post a status update on a project so the team sees where it stands. " "`text` = the update. `status_type` = on_track (default), at_risk, " "off_track, on_hold, or complete. `title` optional.", _schema({"project": "the project's name or gid", "text": "the status update itself", "status_type": "on_track (default) | at_risk | off_track | on_hold | complete", "title": "optional headline"}, ["project", "text"])) async def asana_project_status(args: dict) -> dict: return await _run(asana.project_status, _s(args, "project"), _s(args, "text"), _s(args, "status_type", "on_track"), _s(args, "title")) @tool("asana_task", "FULL detail of ONE Asana task (`task` = its name or gid): notes, " "assignee, due date, project, section, tags, followers, subtasks and the " "latest comments. Use it for 'tell me about that task', 'what's the story " "on X', or before changing something.", _schema({"task": "the task's name or gid"}, ["task"])) async def asana_task(args: dict) -> dict: return await _run(asana.task, _s(args, "task")) @tool("asana_tasks", "List Asana tasks. `project` (name or gid) = that project's tasks; add " "`section` for one board column; no project = Ahmed's OWN tasks ('what's " "on my plate'). Open tasks only unless `completed`='true'. `limit` " "defaults to 50.", _schema({"project": "a project name or gid; empty = Ahmed's own tasks", "section": "a board column inside that project", "completed": ("boolean", "true = list COMPLETED tasks instead"), "limit": ("integer", "max tasks to return (default 50)")})) async def asana_tasks(args: dict) -> dict: proj = _s(args, "project") return await _run(asana.tasks, proj, not proj, _s(args, "section"), bool(_b(args, "completed")), int(args.get("limit") or 50)) @tool("asana_task_create", "Create an Asana task. `name` (short, imperative) is required. `project` = " "the project name/gid to file it under (without one it lands in his My " "Tasks). `section` = the board column inside that project. `parent` = " "another task, to make this a SUBTASK of it. `notes` = detail. `assignee` " "= a person's name/email, or 'me' (the default). `due_on` = YYYY-MM-DD — " "resolve 'tomorrow' / 'next Thursday' to a real date yourself (Asia/Riyadh).", _schema({"name": "the task name (short, imperative)", "project": "project name or gid to file it under", "section": "board column inside that project", "parent": "another task — makes this a SUBTASK of it", "notes": "detail / description", "assignee": "a person's name or email, or 'me' (default)", "due_on": "YYYY-MM-DD"}, ["name"])) async def asana_task_create(args: dict) -> dict: return await _run(asana.task_create, _s(args, "name"), _s(args, "notes"), _s(args, "project"), _s(args, "assignee", "me"), _s(args, "due_on"), _s(args, "section"), _s(args, "parent")) @tool("asana_task_update", "Change an existing Asana task (`task` = its gid or name, matched " "fuzzily). Set only what changes: `name` (rename), `notes`, `due_on` " "(YYYY-MM-DD), `start_on` (needs a due date too, and a PAID Asana plan — it " "fails on his free one), `assignee` (a name " "or 'me'), `complete` ('true' to close, 'false' to reopen). Use for 'push " "that to Friday', 'give it to Sara', 'reopen the launch task'.", _schema({"task": "the task's name or gid", "name": "a new name (rename)", "notes": "new description", "due_on": "YYYY-MM-DD", "start_on": "YYYY-MM-DD (paid plans only)", "assignee": "a person's name or email, or 'me'", "complete": ("boolean", "true = close it, false = reopen")}, ["task"])) async def asana_task_update(args: dict) -> dict: return await _run(asana.task_update, _s(args, "task"), _s(args, "name"), _s(args, "notes"), _s(args, "due_on"), _s(args, "start_on"), _s(args, "assignee"), _b(args, "complete")) @tool("asana_task_move", "Move a task to another `project` and/or `section` (board column). This is " "a REAL move — the task is taken out of the project it was in (Asana would " "otherwise leave it in both). `section` alone moves it between columns of " "its current project. `remove_from` optionally names just one project to " "drop. Use for 'move it to Done', 'that belongs on the client board'.", _schema({"task": "the task's name or gid", "project": "the project to move it INTO", "section": "the board column to put it in", "remove_from": "optionally, the one project to drop it from"}, ["task"])) async def asana_task_move(args: dict) -> dict: return await _run(asana.task_move, _s(args, "task"), _s(args, "project"), _s(args, "section"), _s(args, "remove_from")) @tool("asana_task_complete", "Mark an Asana task done. `task` = its gid or name. Use for 'tick off the " "landing page task', 'that Asana task is finished'.", _schema({"task": "the task's name or gid"}, ["task"])) async def asana_task_complete(args: dict) -> dict: return await _run(asana.task_complete, _s(args, "task")) @tool("asana_task_delete", "Delete an Asana task ('remove that task', 'get rid of it'). It goes to " "Asana's Deleted Items and can be restored for 30 days, so it's safe — but " "only do it when he clearly means DELETE, not 'mark it done' " "(asana_task_complete).", _schema({"task": "the task's name or gid"}, ["task"])) async def asana_task_delete(args: dict) -> dict: return await _run(asana.task_delete, _s(args, "task")) @tool("asana_task_people", "Who and what is on a task: set the `assignee`, add/remove followers " "(`add_followers` / `remove_followers`, comma-separated names, emails or " "'me'), add/remove tags (`add_tag` / `remove_tag` — a new tag is created " "if it doesn't exist). With no changes it just reports who's on it.", _schema({"task": "the task's name or gid", "assignee": "a person's name or email, or 'me'", "add_followers": "comma-separated people to add as followers", "remove_followers": "comma-separated people to un-follow", "add_tag": "comma-separated tags (created if new)", "remove_tag": "comma-separated tags to remove"}, ["task"])) async def asana_task_people(args: dict) -> dict: return await _run(asana.task_people, _s(args, "task"), _s(args, "assignee"), _s(args, "add_followers"), _s(args, "remove_followers"), _s(args, "add_tag"), _s(args, "remove_tag")) @tool("asana_subtasks", "List a task's subtasks, or create them: `add` = comma-separated subtask " "names ('draft the copy, get sign-off'). Use for 'break that task down'.", _schema({"task": "the parent task's name or gid", "add": "comma-separated subtask names to create"}, ["task"])) async def asana_subtasks(args: dict) -> dict: return await _run(asana.subtasks, _s(args, "task"), _s(args, "add")) @tool("asana_comment", "Add a comment to an Asana task so the team sees it. `task` = its gid or " "name, `text` = the comment. Use for 'comment on the design task that the " "client approved'.", _schema({"task": "the task's name or gid", "text": "the comment"}, ["task", "text"])) async def asana_comment(args: dict) -> dict: return await _run(asana.comment, _s(args, "task"), _s(args, "text")) @tool("asana_comments", "READ the comment thread on a task (system events are skipped) — 'what did " "they say on that task', 'any update from the team'. `limit` = how many of " "the latest to show (default 10).", _schema({"task": "the task's name or gid", "limit": ("integer", "how many of the latest (default 10)")}, ["task"])) async def asana_comments(args: dict) -> dict: return await _run(asana.comments, _s(args, "task"), int(args.get("limit") or 10)) @tool("asana_search", "Find Asana tasks when you don't know where they live ('find the invoice " "task', 'anything about the Riyadh launch'). `query` = a word or two; " "optional filters `project`, `assignee` ('me' or a name), `due_before` / " "`due_after` (YYYY-MM-DD), `completed` ('true'/'false'). On a free Asana " "plan it falls back to name matching (descriptions aren't searched), and " "the index lags up to a minute — for a task you JUST created use " "asana_task / asana_tasks instead.", _schema({"query": "a word or two to look for", "project": "limit to one project (name or gid)", "assignee": "'me' or a person's name", "due_before": "YYYY-MM-DD", "due_after": "YYYY-MM-DD", "completed": ("boolean", "true = only done, false = only open")})) async def asana_search(args: dict) -> dict: return await _run(asana.search, _s(args, "query"), _s(args, "project"), _s(args, "assignee"), _s(args, "due_before"), _s(args, "due_after"), _b(args, "completed")) @tool("asana_api", "Escape hatch — call ANY Asana REST endpoint (base " "https://app.asana.com/api/1.0) for things the other asana tools don't " "cover: sections, tags, teams, attachments, dependencies, templates, " "goals, portfolios, custom fields, webhooks, inviting a NEW person to the " "workspace, deleting a project. `method` = GET/POST/PUT/DELETE, `path` = " "everything after /api/1.0 (e.g. '/projects/123/sections'). Do NOT wrap " "`body_json` in {\"data\":…} — that's added for you. IMPORTANT: Asana " "returns only gid/name unless you ask for fields, so pass e.g. " "params_json={\"opt_fields\":\"name,due_on,assignee.name\",\"limit\":50}. " "Returns raw JSON — summarise it, never read it out.", _schema({"method": "GET | POST | PUT | DELETE", "path": "everything after /api/1.0, e.g. /projects/123/sections", "body_json": "JSON body, WITHOUT the {\"data\":…} wrapper", "params_json": "JSON query params — put opt_fields here"}, ["method", "path"])) async def asana_api(args: dict) -> dict: return await _run(asana.raw, _s(args, "method", "GET"), _s(args, "path"), _s(args, "body_json"), _s(args, "params_json")) return create_sdk_mcp_server( name="asana", version="2.0.0", tools=[asana_workspaces, asana_projects, asana_project_create, asana_project_update, asana_project_members, asana_project_status, asana_task, asana_tasks, asana_task_create, asana_task_update, asana_task_move, asana_task_complete, asana_task_delete, asana_task_people, asana_subtasks, asana_comment, asana_comments, asana_search, asana_api], )