fix: revert Asana to httpx-based implementation (asana SDK not available)
This commit is contained in:
Timothy
@@ -1,55 +1,35 @@
|
||||
from .base import CredentialSpec
|
||||
"""
|
||||
Asana credentials.
|
||||
|
||||
ASANA_TOOLS = [
|
||||
"asana_create_task",
|
||||
"asana_update_task",
|
||||
"asana_get_task",
|
||||
"asana_search_tasks",
|
||||
"asana_delete_task",
|
||||
"asana_add_task_comment",
|
||||
"asana_complete_task",
|
||||
"asana_add_subtask",
|
||||
"asana_create_project",
|
||||
"asana_update_project",
|
||||
"asana_get_project",
|
||||
"asana_list_projects",
|
||||
"asana_get_project_tasks",
|
||||
"asana_add_task_to_project",
|
||||
"asana_get_workspace",
|
||||
"asana_list_workspaces",
|
||||
"asana_get_user",
|
||||
"asana_list_team_members",
|
||||
"asana_create_section",
|
||||
"asana_list_sections",
|
||||
"asana_move_task_to_section",
|
||||
"asana_create_tag",
|
||||
"asana_add_tag_to_task",
|
||||
"asana_list_tags",
|
||||
"asana_update_custom_field",
|
||||
]
|
||||
Contains credentials for Asana task and project management.
|
||||
"""
|
||||
|
||||
from .base import CredentialSpec
|
||||
|
||||
ASANA_CREDENTIALS = {
|
||||
"asana": CredentialSpec(
|
||||
env_var="ASANA_ACCESS_TOKEN",
|
||||
tools=ASANA_TOOLS,
|
||||
tools=[
|
||||
"asana_list_workspaces",
|
||||
"asana_list_projects",
|
||||
"asana_list_tasks",
|
||||
"asana_get_task",
|
||||
"asana_create_task",
|
||||
"asana_search_tasks",
|
||||
],
|
||||
required=True,
|
||||
startup_required=False,
|
||||
help_url="https://app.asana.com/0/my-apps",
|
||||
description="Asana Personal Access Token (PAT)",
|
||||
# Auth method support
|
||||
aden_supported=False,
|
||||
help_url="https://developers.asana.com/docs/personal-access-token",
|
||||
description="Asana personal access token for task and project management",
|
||||
direct_api_key_supported=True,
|
||||
api_key_instructions="""To get an Asana Personal Access Token:
|
||||
api_key_instructions="""To get an Asana personal access token:
|
||||
1. Go to https://app.asana.com/0/my-apps
|
||||
2. Click "+ New access token"
|
||||
3. Name your token (e.g. "Aden Agent")
|
||||
4. Copy the token string (starts with "1/...")
|
||||
5. Store it securely.""",
|
||||
# Health check configuration
|
||||
2. Click 'Create new token'
|
||||
3. Give it a name and copy the token
|
||||
4. Set the environment variable:
|
||||
export ASANA_ACCESS_TOKEN=your-pat""",
|
||||
health_check_endpoint="https://app.asana.com/api/1.0/users/me",
|
||||
health_check_method="GET",
|
||||
# Credential store mapping
|
||||
credential_id="asana",
|
||||
credential_key="access_token",
|
||||
credential_key="api_key",
|
||||
),
|
||||
}
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
from .tool import register_tools
|
||||
"""Asana project management tool package for Aden Tools."""
|
||||
|
||||
from .asana_tool import register_tools
|
||||
|
||||
__all__ = ["register_tools"]
|
||||
|
||||
@@ -0,0 +1,315 @@
|
||||
"""
|
||||
Asana Tool - Task and project management.
|
||||
|
||||
Supports:
|
||||
- Asana personal access token (ASANA_ACCESS_TOKEN)
|
||||
- Tasks, Projects, Workspaces, Sections, Tags
|
||||
|
||||
API Reference: https://developers.asana.com/docs
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
import httpx
|
||||
from fastmcp import FastMCP
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from aden_tools.credentials import CredentialStoreAdapter
|
||||
|
||||
ASANA_API = "https://app.asana.com/api/1.0"
|
||||
|
||||
|
||||
def _get_token(credentials: CredentialStoreAdapter | None) -> str | None:
|
||||
if credentials is not None:
|
||||
return credentials.get("asana")
|
||||
return os.getenv("ASANA_ACCESS_TOKEN")
|
||||
|
||||
|
||||
def _headers(token: str) -> dict[str, str]:
|
||||
return {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
|
||||
|
||||
|
||||
def _get(endpoint: str, token: str, params: dict | None = None) -> dict[str, Any]:
|
||||
try:
|
||||
resp = httpx.get(f"{ASANA_API}/{endpoint}", headers=_headers(token), params=params, timeout=30.0)
|
||||
if resp.status_code == 401:
|
||||
return {"error": "Unauthorized. Check your ASANA_ACCESS_TOKEN."}
|
||||
if resp.status_code == 403:
|
||||
return {"error": f"Forbidden: {resp.text[:300]}"}
|
||||
if resp.status_code == 404:
|
||||
return {"error": "Not found"}
|
||||
if resp.status_code != 200:
|
||||
return {"error": f"Asana API error {resp.status_code}: {resp.text[:500]}"}
|
||||
return resp.json()
|
||||
except httpx.TimeoutException:
|
||||
return {"error": "Request to Asana timed out"}
|
||||
except Exception as e:
|
||||
return {"error": f"Asana request failed: {e!s}"}
|
||||
|
||||
|
||||
def _post(endpoint: str, token: str, body: dict | None = None) -> dict[str, Any]:
|
||||
try:
|
||||
resp = httpx.post(f"{ASANA_API}/{endpoint}", headers=_headers(token), json={"data": body or {}}, timeout=30.0)
|
||||
if resp.status_code == 401:
|
||||
return {"error": "Unauthorized. Check your ASANA_ACCESS_TOKEN."}
|
||||
if resp.status_code not in (200, 201):
|
||||
return {"error": f"Asana API error {resp.status_code}: {resp.text[:500]}"}
|
||||
return resp.json()
|
||||
except httpx.TimeoutException:
|
||||
return {"error": "Request to Asana timed out"}
|
||||
except Exception as e:
|
||||
return {"error": f"Asana request failed: {e!s}"}
|
||||
|
||||
|
||||
def _auth_error() -> dict[str, Any]:
|
||||
return {
|
||||
"error": "ASANA_ACCESS_TOKEN not set",
|
||||
"help": "Create a PAT at https://app.asana.com/0/my-apps",
|
||||
}
|
||||
|
||||
|
||||
def register_tools(
|
||||
mcp: FastMCP,
|
||||
credentials: CredentialStoreAdapter | None = None,
|
||||
) -> None:
|
||||
"""Register Asana tools with the MCP server."""
|
||||
|
||||
@mcp.tool()
|
||||
def asana_list_workspaces() -> dict[str, Any]:
|
||||
"""
|
||||
List all workspaces accessible to the authenticated user.
|
||||
|
||||
Returns:
|
||||
Dict with workspaces list (gid, name)
|
||||
"""
|
||||
token = _get_token(credentials)
|
||||
if not token:
|
||||
return _auth_error()
|
||||
|
||||
data = _get("workspaces", token)
|
||||
if "error" in data:
|
||||
return data
|
||||
|
||||
workspaces = []
|
||||
for w in data.get("data", []):
|
||||
workspaces.append({"gid": w.get("gid", ""), "name": w.get("name", "")})
|
||||
return {"workspaces": workspaces}
|
||||
|
||||
@mcp.tool()
|
||||
def asana_list_projects(
|
||||
workspace_gid: str,
|
||||
limit: int = 50,
|
||||
) -> dict[str, Any]:
|
||||
"""
|
||||
List projects in an Asana workspace.
|
||||
|
||||
Args:
|
||||
workspace_gid: Workspace GID
|
||||
limit: Number of results (1-100, default 50)
|
||||
|
||||
Returns:
|
||||
Dict with projects list (gid, name, color, archived)
|
||||
"""
|
||||
token = _get_token(credentials)
|
||||
if not token:
|
||||
return _auth_error()
|
||||
if not workspace_gid:
|
||||
return {"error": "workspace_gid is required"}
|
||||
|
||||
params = {
|
||||
"workspace": workspace_gid,
|
||||
"limit": max(1, min(limit, 100)),
|
||||
"opt_fields": "name,color,archived,created_at",
|
||||
}
|
||||
data = _get("projects", token, params)
|
||||
if "error" in data:
|
||||
return data
|
||||
|
||||
projects = []
|
||||
for p in data.get("data", []):
|
||||
projects.append({
|
||||
"gid": p.get("gid", ""),
|
||||
"name": p.get("name", ""),
|
||||
"color": p.get("color", ""),
|
||||
"archived": p.get("archived", False),
|
||||
})
|
||||
return {"projects": projects}
|
||||
|
||||
@mcp.tool()
|
||||
def asana_list_tasks(
|
||||
project_gid: str = "",
|
||||
assignee: str = "me",
|
||||
workspace_gid: str = "",
|
||||
limit: int = 50,
|
||||
) -> dict[str, Any]:
|
||||
"""
|
||||
List tasks from Asana, filtered by project or assignee.
|
||||
|
||||
Args:
|
||||
project_gid: Project GID to filter by (optional)
|
||||
assignee: Assignee: "me" or user GID (used with workspace_gid)
|
||||
workspace_gid: Workspace GID (required when filtering by assignee without project)
|
||||
limit: Number of results (1-100, default 50)
|
||||
|
||||
Returns:
|
||||
Dict with tasks list (gid, name, completed, due_on, assignee_name)
|
||||
"""
|
||||
token = _get_token(credentials)
|
||||
if not token:
|
||||
return _auth_error()
|
||||
if not project_gid and not workspace_gid:
|
||||
return {"error": "Either project_gid or workspace_gid is required"}
|
||||
|
||||
params: dict[str, Any] = {
|
||||
"limit": max(1, min(limit, 100)),
|
||||
"opt_fields": "name,completed,due_on,assignee.name",
|
||||
}
|
||||
if project_gid:
|
||||
params["project"] = project_gid
|
||||
else:
|
||||
params["workspace"] = workspace_gid
|
||||
params["assignee"] = assignee
|
||||
|
||||
data = _get("tasks", token, params)
|
||||
if "error" in data:
|
||||
return data
|
||||
|
||||
tasks = []
|
||||
for t in data.get("data", []):
|
||||
assignee_obj = t.get("assignee") or {}
|
||||
tasks.append({
|
||||
"gid": t.get("gid", ""),
|
||||
"name": t.get("name", ""),
|
||||
"completed": t.get("completed", False),
|
||||
"due_on": t.get("due_on", ""),
|
||||
"assignee_name": assignee_obj.get("name", ""),
|
||||
})
|
||||
return {"tasks": tasks, "count": len(tasks)}
|
||||
|
||||
@mcp.tool()
|
||||
def asana_get_task(task_gid: str) -> dict[str, Any]:
|
||||
"""
|
||||
Get details of a specific Asana task.
|
||||
|
||||
Args:
|
||||
task_gid: Task GID
|
||||
|
||||
Returns:
|
||||
Dict with task details: name, notes, completed, due_on, assignee, projects, tags
|
||||
"""
|
||||
token = _get_token(credentials)
|
||||
if not token:
|
||||
return _auth_error()
|
||||
if not task_gid:
|
||||
return {"error": "task_gid is required"}
|
||||
|
||||
params = {"opt_fields": "name,notes,completed,due_on,assignee.name,projects.name,tags.name,created_at,modified_at"}
|
||||
data = _get(f"tasks/{task_gid}", token, params)
|
||||
if "error" in data:
|
||||
return data
|
||||
|
||||
t = data.get("data", {})
|
||||
assignee_obj = t.get("assignee") or {}
|
||||
return {
|
||||
"gid": t.get("gid", ""),
|
||||
"name": t.get("name", ""),
|
||||
"notes": (t.get("notes", "") or "")[:500],
|
||||
"completed": t.get("completed", False),
|
||||
"due_on": t.get("due_on", ""),
|
||||
"assignee_name": assignee_obj.get("name", ""),
|
||||
"projects": [p.get("name", "") for p in t.get("projects", [])],
|
||||
"tags": [tag.get("name", "") for tag in t.get("tags", [])],
|
||||
"created_at": t.get("created_at", ""),
|
||||
"modified_at": t.get("modified_at", ""),
|
||||
}
|
||||
|
||||
@mcp.tool()
|
||||
def asana_create_task(
|
||||
workspace_gid: str,
|
||||
name: str,
|
||||
notes: str = "",
|
||||
project_gid: str = "",
|
||||
assignee: str = "",
|
||||
due_on: str = "",
|
||||
) -> dict[str, Any]:
|
||||
"""
|
||||
Create a new task in Asana.
|
||||
|
||||
Args:
|
||||
workspace_gid: Workspace GID (required)
|
||||
name: Task name (required)
|
||||
notes: Task description/notes (optional)
|
||||
project_gid: Add to this project (optional)
|
||||
assignee: Assignee GID or "me" (optional)
|
||||
due_on: Due date YYYY-MM-DD (optional)
|
||||
|
||||
Returns:
|
||||
Dict with created task gid, name, and status
|
||||
"""
|
||||
token = _get_token(credentials)
|
||||
if not token:
|
||||
return _auth_error()
|
||||
if not workspace_gid or not name:
|
||||
return {"error": "workspace_gid and name are required"}
|
||||
|
||||
body: dict[str, Any] = {"workspace": workspace_gid, "name": name}
|
||||
if notes:
|
||||
body["notes"] = notes
|
||||
if project_gid:
|
||||
body["projects"] = [project_gid]
|
||||
if assignee:
|
||||
body["assignee"] = assignee
|
||||
if due_on:
|
||||
body["due_on"] = due_on
|
||||
|
||||
data = _post("tasks", token, body)
|
||||
if "error" in data:
|
||||
return data
|
||||
|
||||
t = data.get("data", {})
|
||||
return {"gid": t.get("gid", ""), "name": t.get("name", ""), "status": "created"}
|
||||
|
||||
@mcp.tool()
|
||||
def asana_search_tasks(
|
||||
workspace_gid: str,
|
||||
query: str,
|
||||
limit: int = 20,
|
||||
) -> dict[str, Any]:
|
||||
"""
|
||||
Search tasks in an Asana workspace.
|
||||
|
||||
Args:
|
||||
workspace_gid: Workspace GID
|
||||
query: Search text
|
||||
limit: Number of results (1-100, default 20)
|
||||
|
||||
Returns:
|
||||
Dict with matching tasks (gid, name, completed)
|
||||
"""
|
||||
token = _get_token(credentials)
|
||||
if not token:
|
||||
return _auth_error()
|
||||
if not workspace_gid or not query:
|
||||
return {"error": "workspace_gid and query are required"}
|
||||
|
||||
params = {
|
||||
"text": query,
|
||||
"limit": max(1, min(limit, 100)),
|
||||
"opt_fields": "name,completed,due_on",
|
||||
}
|
||||
data = _get(f"workspaces/{workspace_gid}/tasks/search", token, params)
|
||||
if "error" in data:
|
||||
return data
|
||||
|
||||
tasks = []
|
||||
for t in data.get("data", []):
|
||||
tasks.append({
|
||||
"gid": t.get("gid", ""),
|
||||
"name": t.get("name", ""),
|
||||
"completed": t.get("completed", False),
|
||||
"due_on": t.get("due_on", ""),
|
||||
})
|
||||
return {"query": query, "tasks": tasks}
|
||||
@@ -1,687 +0,0 @@
|
||||
"""
|
||||
Asana Tool - Manage tasks, projects, and workspaces via Asana API.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
import asana # type: ignore
|
||||
from asana.rest import ApiException # type: ignore
|
||||
from fastmcp import FastMCP
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from aden_tools.credentials import CredentialStoreAdapter
|
||||
|
||||
|
||||
def to_dict(obj: Any) -> Any:
|
||||
"""Helper to convert API response objects to dicts."""
|
||||
if hasattr(obj, "to_dict"):
|
||||
return obj.to_dict()
|
||||
if isinstance(obj, list):
|
||||
return [to_dict(x) for x in obj]
|
||||
return obj
|
||||
|
||||
|
||||
def get_workspace_gid(
|
||||
client: asana.ApiClient, workspace_name_or_gid: str | None = None
|
||||
) -> str:
|
||||
"""Helper to resolve workspace GID from name or ID, or return default."""
|
||||
workspaces_api = asana.WorkspacesApi(client)
|
||||
|
||||
try:
|
||||
# Request name,gid to be efficient
|
||||
# Note: In Asana python-asana v5+, opt_fields are kwargs
|
||||
response = workspaces_api.get_workspaces(opt_fields="name,gid")
|
||||
|
||||
# Handle response format (v5/data collection)
|
||||
workspaces = []
|
||||
if hasattr(response, "data"):
|
||||
workspaces = response.data
|
||||
else:
|
||||
workspaces = list(response)
|
||||
|
||||
if not workspaces:
|
||||
# If we successfully listed but found nothing
|
||||
raise ValueError("No accessible workspaces found for this token")
|
||||
|
||||
# If no specific workspace requested, return the first one
|
||||
if not workspace_name_or_gid:
|
||||
first = workspaces[0]
|
||||
return str(first["gid"] if isinstance(first, dict) else first.gid)
|
||||
|
||||
# Search by name or GID in the accessible list
|
||||
search_target = str(workspace_name_or_gid)
|
||||
for ws in workspaces:
|
||||
gid = str(ws["gid"] if isinstance(ws, dict) else ws.gid)
|
||||
name = ws["name"] if isinstance(ws, dict) else ws.name
|
||||
if gid == search_target or name == search_target:
|
||||
return gid
|
||||
|
||||
except ApiException as e:
|
||||
# Don't swallow auth errors
|
||||
if e.status in (401, 403):
|
||||
raise ValueError(f"Asana authentication failed: {e}") from e
|
||||
|
||||
# For other errors, if a GID was provided, we might try to blindly use it
|
||||
# (e.g. if listing workspaces is restricted but accessing a specific one isn't)
|
||||
if workspace_name_or_gid and str(workspace_name_or_gid).isdigit():
|
||||
return str(workspace_name_or_gid)
|
||||
|
||||
raise ValueError(f"Failed to resolve workspace: {e}") from e
|
||||
|
||||
# If we looked through the list and didn't find it
|
||||
# Fallback: if it looks like a GID, trust it (maybe it's a hidden workspace)
|
||||
if workspace_name_or_gid and str(workspace_name_or_gid).isdigit():
|
||||
return str(workspace_name_or_gid)
|
||||
|
||||
raise ValueError(f"Workspace '{workspace_name_or_gid}' not found")
|
||||
|
||||
|
||||
def register_tools(
|
||||
mcp: FastMCP,
|
||||
credentials: CredentialStoreAdapter | None = None,
|
||||
) -> None:
|
||||
"""Register Asana tools with the MCP server."""
|
||||
|
||||
def _get_token() -> str | None:
|
||||
"""Get Asana access token from credential manager or environment."""
|
||||
if credentials is not None:
|
||||
token = credentials.get("asana")
|
||||
# Defensive check: ensure we get a string
|
||||
if token is not None and not isinstance(token, str):
|
||||
return None
|
||||
return token
|
||||
return os.getenv("ASANA_ACCESS_TOKEN")
|
||||
|
||||
def _get_client() -> asana.ApiClient | dict[str, str]:
|
||||
"""Get an Asana client, or return an error dict if no credentials."""
|
||||
token = _get_token()
|
||||
if not token:
|
||||
return {
|
||||
"error": "Asana credentials not configured",
|
||||
"help": (
|
||||
"Set ASANA_ACCESS_TOKEN environment variable "
|
||||
"or configure via credential store"
|
||||
),
|
||||
}
|
||||
|
||||
configuration = asana.Configuration()
|
||||
configuration.access_token = token
|
||||
return asana.ApiClient(configuration)
|
||||
|
||||
# --- Task Tools ---
|
||||
|
||||
@mcp.tool()
|
||||
def asana_create_task(
|
||||
name: str,
|
||||
workspace: str | None = None,
|
||||
notes: str | None = None,
|
||||
assignee: str | None = None,
|
||||
projects: list[str] | None = None,
|
||||
due_on: str | None = None,
|
||||
start_on: str | None = None,
|
||||
tags: list[str] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Create a new task in Asana."""
|
||||
try:
|
||||
client = _get_client()
|
||||
if isinstance(client, dict):
|
||||
return client
|
||||
|
||||
tasks_api = asana.TasksApi(client)
|
||||
workspace_gid = get_workspace_gid(client, workspace)
|
||||
|
||||
task_data: dict[str, Any] = {
|
||||
"workspace": workspace_gid,
|
||||
"name": name,
|
||||
}
|
||||
if notes:
|
||||
task_data["notes"] = notes
|
||||
if assignee:
|
||||
task_data["assignee"] = assignee
|
||||
if projects:
|
||||
task_data["projects"] = projects
|
||||
if due_on:
|
||||
task_data["due_on"] = due_on
|
||||
if start_on:
|
||||
task_data["start_on"] = start_on
|
||||
if tags:
|
||||
task_data["tags"] = tags
|
||||
|
||||
body = {"data": task_data}
|
||||
result = tasks_api.create_task(body)
|
||||
return to_dict(result)
|
||||
except Exception as e:
|
||||
return {"error": str(e)}
|
||||
|
||||
@mcp.tool()
|
||||
def asana_update_task(
|
||||
task_gid: str,
|
||||
name: str | None = None,
|
||||
notes: str | None = None,
|
||||
assignee: str | None = None,
|
||||
due_on: str | None = None,
|
||||
completed: bool | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Update an existing task."""
|
||||
try:
|
||||
client = _get_client()
|
||||
if isinstance(client, dict):
|
||||
return client
|
||||
|
||||
tasks_api = asana.TasksApi(client)
|
||||
|
||||
task_data: dict[str, Any] = {}
|
||||
if name:
|
||||
task_data["name"] = name
|
||||
if notes:
|
||||
task_data["notes"] = notes
|
||||
if assignee:
|
||||
task_data["assignee"] = assignee
|
||||
if due_on:
|
||||
task_data["due_on"] = due_on
|
||||
if completed is not None:
|
||||
task_data["completed"] = completed
|
||||
|
||||
if not task_data:
|
||||
return {"error": "No fields to update"}
|
||||
|
||||
# Note: task_gid must be string
|
||||
body = {"data": task_data}
|
||||
result = tasks_api.update_task(body, str(task_gid))
|
||||
return to_dict(result)
|
||||
except Exception as e:
|
||||
return {"error": str(e)}
|
||||
|
||||
@mcp.tool()
|
||||
def asana_get_task(task_gid: str) -> dict[str, Any]:
|
||||
"""Get details of a task."""
|
||||
try:
|
||||
client = _get_client()
|
||||
if isinstance(client, dict):
|
||||
return client
|
||||
|
||||
tasks_api = asana.TasksApi(client)
|
||||
result = tasks_api.get_task(str(task_gid))
|
||||
return to_dict(result)
|
||||
except Exception as e:
|
||||
return {"error": str(e)}
|
||||
|
||||
@mcp.tool()
|
||||
def asana_search_tasks(
|
||||
workspace: str | None = None,
|
||||
text: str | None = None,
|
||||
assignee: str | None = None,
|
||||
project: str | None = None,
|
||||
completed: bool | None = None,
|
||||
limit: int = 20,
|
||||
) -> dict[str, Any]:
|
||||
"""Search for tasks in a workspace."""
|
||||
try:
|
||||
client = _get_client()
|
||||
if isinstance(client, dict):
|
||||
return client
|
||||
|
||||
tasks_api = asana.TasksApi(client)
|
||||
workspace_gid = get_workspace_gid(client, workspace)
|
||||
|
||||
opts = {
|
||||
"limit": limit,
|
||||
"opt_fields": "name,gid,completed,assignee.name,projects.name",
|
||||
}
|
||||
if text:
|
||||
opts["text"] = text
|
||||
if assignee:
|
||||
opts["assignee.any"] = assignee
|
||||
if project:
|
||||
opts["projects.any"] = project
|
||||
if completed is not None:
|
||||
opts["completed"] = str(completed).lower()
|
||||
|
||||
result = tasks_api.search_tasks_for_workspace(workspace_gid, **opts)
|
||||
data = list(result.data) if hasattr(result, "data") else list(result)
|
||||
return {"data": to_dict(data)}
|
||||
except Exception as e:
|
||||
return {"error": str(e)}
|
||||
|
||||
@mcp.tool()
|
||||
def asana_delete_task(task_gid: str) -> dict[str, Any]:
|
||||
"""Delete a task."""
|
||||
try:
|
||||
client = _get_client()
|
||||
if isinstance(client, dict):
|
||||
return client
|
||||
|
||||
tasks_api = asana.TasksApi(client)
|
||||
result = tasks_api.delete_task(str(task_gid))
|
||||
return {"success": True, "result": to_dict(result)}
|
||||
except Exception as e:
|
||||
return {"error": str(e)}
|
||||
|
||||
@mcp.tool()
|
||||
def asana_add_task_comment(task_gid: str, text: str) -> dict[str, Any]:
|
||||
"""Add a comment to a task."""
|
||||
try:
|
||||
client = _get_client()
|
||||
if isinstance(client, dict):
|
||||
return client
|
||||
|
||||
stories_api = asana.StoriesApi(client)
|
||||
body = {"data": {"text": text}}
|
||||
result = stories_api.create_story_for_task(body, str(task_gid))
|
||||
return to_dict(result)
|
||||
except Exception as e:
|
||||
return {"error": str(e)}
|
||||
|
||||
@mcp.tool()
|
||||
def asana_complete_task(task_gid: str) -> dict[str, Any]:
|
||||
"""Mark a task as complete."""
|
||||
# Reuse update logic
|
||||
return asana_update_task(task_gid, completed=True)
|
||||
|
||||
@mcp.tool()
|
||||
def asana_add_subtask(
|
||||
parent_task_gid: str,
|
||||
name: str,
|
||||
notes: str | None = None,
|
||||
assignee: str | None = None,
|
||||
due_on: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Create a subtask."""
|
||||
try:
|
||||
client = _get_client()
|
||||
if isinstance(client, dict):
|
||||
return client
|
||||
|
||||
tasks_api = asana.TasksApi(client)
|
||||
|
||||
task_data: dict[str, Any] = {"name": name}
|
||||
if notes:
|
||||
task_data["notes"] = notes
|
||||
if assignee:
|
||||
task_data["assignee"] = assignee
|
||||
if due_on:
|
||||
task_data["due_on"] = due_on
|
||||
|
||||
body = {"data": task_data}
|
||||
result = tasks_api.create_subtask_for_task(body, str(parent_task_gid))
|
||||
return to_dict(result)
|
||||
except Exception as e:
|
||||
return {"error": str(e)}
|
||||
|
||||
# --- Project Tools ---
|
||||
|
||||
@mcp.tool()
|
||||
def asana_create_project(
|
||||
name: str,
|
||||
workspace: str | None = None,
|
||||
team: str | None = None,
|
||||
notes: str | None = None,
|
||||
public: bool = True,
|
||||
) -> dict[str, Any]:
|
||||
"""Create a new project."""
|
||||
try:
|
||||
client = _get_client()
|
||||
if isinstance(client, dict):
|
||||
return client
|
||||
|
||||
projects_api = asana.ProjectsApi(client)
|
||||
workspace_gid = get_workspace_gid(client, workspace)
|
||||
|
||||
project_data: dict[str, Any] = {
|
||||
"name": name,
|
||||
"workspace": workspace_gid,
|
||||
"public": public,
|
||||
}
|
||||
if team:
|
||||
project_data["team"] = team
|
||||
if notes:
|
||||
project_data["notes"] = notes
|
||||
|
||||
body = {"data": project_data}
|
||||
result = projects_api.create_project(body)
|
||||
return to_dict(result)
|
||||
except Exception as e:
|
||||
return {"error": str(e)}
|
||||
|
||||
@mcp.tool()
|
||||
def asana_update_project(
|
||||
project_gid: str,
|
||||
name: str | None = None,
|
||||
notes: str | None = None,
|
||||
public: bool | None = None,
|
||||
owner: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Update a project."""
|
||||
try:
|
||||
client = _get_client()
|
||||
if isinstance(client, dict):
|
||||
return client
|
||||
|
||||
projects_api = asana.ProjectsApi(client)
|
||||
|
||||
project_data: dict[str, Any] = {}
|
||||
if name:
|
||||
project_data["name"] = name
|
||||
if notes:
|
||||
project_data["notes"] = notes
|
||||
if public is not None:
|
||||
project_data["public"] = public
|
||||
if owner:
|
||||
project_data["owner"] = owner
|
||||
|
||||
if not project_data:
|
||||
return {"error": "No fields to update"}
|
||||
|
||||
body = {"data": project_data}
|
||||
result = projects_api.update_project(body, str(project_gid))
|
||||
return to_dict(result)
|
||||
except Exception as e:
|
||||
return {"error": str(e)}
|
||||
|
||||
@mcp.tool()
|
||||
def asana_get_project(project_gid: str) -> dict[str, Any]:
|
||||
"""Get project details."""
|
||||
try:
|
||||
client = _get_client()
|
||||
if isinstance(client, dict):
|
||||
return client
|
||||
|
||||
projects_api = asana.ProjectsApi(client)
|
||||
result = projects_api.get_project(str(project_gid))
|
||||
return to_dict(result)
|
||||
except Exception as e:
|
||||
return {"error": str(e)}
|
||||
|
||||
@mcp.tool()
|
||||
def asana_list_projects(
|
||||
workspace: str | None = None,
|
||||
team: str | None = None,
|
||||
archived: bool | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""List projects in a workspace or team."""
|
||||
try:
|
||||
client = _get_client()
|
||||
if isinstance(client, dict):
|
||||
return client
|
||||
|
||||
projects_api = asana.ProjectsApi(client)
|
||||
workspace_gid = get_workspace_gid(client, workspace)
|
||||
|
||||
opts: dict[str, Any] = {"workspace": workspace_gid}
|
||||
if team:
|
||||
opts["team"] = team
|
||||
if archived is not None:
|
||||
opts["archived"] = archived
|
||||
|
||||
result = projects_api.get_projects(**opts)
|
||||
data = list(result.data) if hasattr(result, "data") else list(result)
|
||||
return {"data": to_dict(data)}
|
||||
except Exception as e:
|
||||
return {"error": str(e)}
|
||||
|
||||
@mcp.tool()
|
||||
def asana_get_project_tasks(
|
||||
project_gid: str,
|
||||
completed_since: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Get tasks in a project."""
|
||||
try:
|
||||
client = _get_client()
|
||||
if isinstance(client, dict):
|
||||
return client
|
||||
|
||||
tasks_api = asana.TasksApi(client)
|
||||
|
||||
opts = {"project": str(project_gid)}
|
||||
if completed_since:
|
||||
opts["completed_since"] = completed_since
|
||||
|
||||
# get_tasks_for_project is cleaner than get_tasks with project filter
|
||||
result = tasks_api.get_tasks_for_project(str(project_gid), **opts)
|
||||
data = list(result.data) if hasattr(result, "data") else list(result)
|
||||
return {"data": to_dict(data)}
|
||||
except Exception as e:
|
||||
return {"error": str(e)}
|
||||
|
||||
@mcp.tool()
|
||||
def asana_add_task_to_project(
|
||||
task_gid: str,
|
||||
project_gid: str,
|
||||
section_gid: str | None = None,
|
||||
insert_after: str | None = None,
|
||||
insert_before: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Add a task to a project."""
|
||||
try:
|
||||
client = _get_client()
|
||||
if isinstance(client, dict):
|
||||
return client
|
||||
|
||||
tasks_api = asana.TasksApi(client)
|
||||
|
||||
data: dict[str, Any] = {"project": project_gid}
|
||||
if section_gid:
|
||||
data["section"] = section_gid
|
||||
if insert_after:
|
||||
data["insert_after"] = insert_after
|
||||
if insert_before:
|
||||
data["insert_before"] = insert_before
|
||||
|
||||
body = {"data": data}
|
||||
result = tasks_api.add_project_for_task(body, str(task_gid))
|
||||
return to_dict(result)
|
||||
except Exception as e:
|
||||
return {"error": str(e)}
|
||||
|
||||
# --- Workspace / User Tools ---
|
||||
|
||||
@mcp.tool()
|
||||
def asana_get_workspace(workspace_gid: str | None = None) -> dict[str, Any]:
|
||||
"""Get workspace details."""
|
||||
try:
|
||||
client = _get_client()
|
||||
if isinstance(client, dict):
|
||||
return client
|
||||
|
||||
workspaces_api = asana.WorkspacesApi(client)
|
||||
gid = get_workspace_gid(client, workspace_gid)
|
||||
result = workspaces_api.get_workspace(gid)
|
||||
return to_dict(result)
|
||||
except Exception as e:
|
||||
return {"error": str(e)}
|
||||
|
||||
@mcp.tool()
|
||||
def asana_list_workspaces() -> dict[str, Any]:
|
||||
"""List all accessible workspaces."""
|
||||
try:
|
||||
client = _get_client()
|
||||
if isinstance(client, dict):
|
||||
return client
|
||||
|
||||
workspaces_api = asana.WorkspacesApi(client)
|
||||
result = workspaces_api.get_workspaces()
|
||||
data = list(result.data) if hasattr(result, "data") else list(result)
|
||||
return {"data": to_dict(data)}
|
||||
except Exception as e:
|
||||
return {"error": str(e)}
|
||||
|
||||
@mcp.tool()
|
||||
def asana_get_user(user_gid: str) -> dict[str, Any]:
|
||||
"""Get user information."""
|
||||
try:
|
||||
client = _get_client()
|
||||
if isinstance(client, dict):
|
||||
return client
|
||||
|
||||
users_api = asana.UsersApi(client)
|
||||
result = users_api.get_user(user_gid)
|
||||
return to_dict(result)
|
||||
except Exception as e:
|
||||
return {"error": str(e)}
|
||||
|
||||
@mcp.tool()
|
||||
def asana_list_team_members(
|
||||
workspace: str | None = None,
|
||||
team_gid: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""List users in a workspace or team."""
|
||||
try:
|
||||
client = _get_client()
|
||||
if isinstance(client, dict):
|
||||
return client
|
||||
|
||||
users_api = asana.UsersApi(client)
|
||||
workspace_gid = get_workspace_gid(client, workspace)
|
||||
|
||||
opts = {"workspace": workspace_gid}
|
||||
if team_gid:
|
||||
opts["team"] = team_gid
|
||||
|
||||
result = users_api.get_users(**opts)
|
||||
data = list(result.data) if hasattr(result, "data") else list(result)
|
||||
return {"data": to_dict(data)}
|
||||
except Exception as e:
|
||||
return {"error": str(e)}
|
||||
|
||||
# --- Section Tools ---
|
||||
|
||||
@mcp.tool()
|
||||
def asana_create_section(
|
||||
project_gid: str,
|
||||
name: str,
|
||||
) -> dict[str, Any]:
|
||||
"""Create a section in a project."""
|
||||
try:
|
||||
client = _get_client()
|
||||
if isinstance(client, dict):
|
||||
return client
|
||||
|
||||
sections_api = asana.SectionsApi(client)
|
||||
body = {"data": {"name": name}}
|
||||
result = sections_api.create_section_for_project(body, str(project_gid))
|
||||
return to_dict(result)
|
||||
except Exception as e:
|
||||
return {"error": str(e)}
|
||||
|
||||
@mcp.tool()
|
||||
def asana_list_sections(project_gid: str) -> dict[str, Any]:
|
||||
"""List sections in a project."""
|
||||
try:
|
||||
client = _get_client()
|
||||
if isinstance(client, dict):
|
||||
return client
|
||||
|
||||
sections_api = asana.SectionsApi(client)
|
||||
result = sections_api.get_sections_for_project(str(project_gid))
|
||||
data = list(result.data) if hasattr(result, "data") else list(result)
|
||||
return {"data": to_dict(data)}
|
||||
except Exception as e:
|
||||
return {"error": str(e)}
|
||||
|
||||
@mcp.tool()
|
||||
def asana_move_task_to_section(
|
||||
task_gid: str,
|
||||
section_gid: str,
|
||||
) -> dict[str, Any]:
|
||||
"""Move a task to a specific section."""
|
||||
try:
|
||||
client = _get_client()
|
||||
if isinstance(client, dict):
|
||||
return client
|
||||
|
||||
sections_api = asana.SectionsApi(client)
|
||||
body = {"data": {"task": str(task_gid)}}
|
||||
result = sections_api.add_task_for_section(body, str(section_gid))
|
||||
return to_dict(result)
|
||||
except Exception as e:
|
||||
return {"error": str(e)}
|
||||
|
||||
# --- Tag / Custom Field Tools ---
|
||||
|
||||
@mcp.tool()
|
||||
def asana_create_tag(
|
||||
name: str,
|
||||
workspace: str | None = None,
|
||||
color: str | None = None,
|
||||
notes: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Create a new tag."""
|
||||
try:
|
||||
client = _get_client()
|
||||
if isinstance(client, dict):
|
||||
return client
|
||||
|
||||
tags_api = asana.TagsApi(client)
|
||||
workspace_gid = get_workspace_gid(client, workspace)
|
||||
|
||||
data: dict[str, Any] = {"workspace": workspace_gid, "name": name}
|
||||
if color:
|
||||
data["color"] = color
|
||||
if notes:
|
||||
data["notes"] = notes
|
||||
|
||||
body = {"data": data}
|
||||
result = tags_api.create_tag(body)
|
||||
return to_dict(result)
|
||||
except Exception as e:
|
||||
return {"error": str(e)}
|
||||
|
||||
@mcp.tool()
|
||||
def asana_add_tag_to_task(
|
||||
task_gid: str,
|
||||
tag_gid: str,
|
||||
) -> dict[str, Any]:
|
||||
"""Add a tag to a task."""
|
||||
try:
|
||||
client = _get_client()
|
||||
if isinstance(client, dict):
|
||||
return client
|
||||
|
||||
tasks_api = asana.TasksApi(client)
|
||||
body = {"data": {"tag": tag_gid}}
|
||||
result = tasks_api.add_tag_for_task(body, str(task_gid))
|
||||
return to_dict(result)
|
||||
except Exception as e:
|
||||
return {"error": str(e)}
|
||||
|
||||
@mcp.tool()
|
||||
def asana_list_tags(workspace: str | None = None) -> dict[str, Any]:
|
||||
"""List tags in a workspace."""
|
||||
try:
|
||||
client = _get_client()
|
||||
if isinstance(client, dict):
|
||||
return client
|
||||
|
||||
tags_api = asana.TagsApi(client)
|
||||
workspace_gid = get_workspace_gid(client, workspace)
|
||||
|
||||
# Note: tags API often requires workspace filter
|
||||
result = tags_api.get_tags(workspace=workspace_gid)
|
||||
data = list(result.data) if hasattr(result, "data") else list(result)
|
||||
return {"data": to_dict(data)}
|
||||
except Exception as e:
|
||||
return {"error": str(e)}
|
||||
|
||||
@mcp.tool()
|
||||
def asana_update_custom_field(
|
||||
task_gid: str,
|
||||
custom_field_gid: str,
|
||||
value: str | float | int,
|
||||
) -> dict[str, Any]:
|
||||
"""Update a custom field value on a task."""
|
||||
try:
|
||||
client = _get_client()
|
||||
if isinstance(client, dict):
|
||||
return client
|
||||
|
||||
tasks_api = asana.TasksApi(client)
|
||||
|
||||
params = {"custom_fields": {custom_field_gid: value}}
|
||||
body = {"data": params}
|
||||
result = tasks_api.update_task(body, str(task_gid))
|
||||
return to_dict(result)
|
||||
except Exception as e:
|
||||
return {"error": str(e)}
|
||||
@@ -1,260 +1,170 @@
|
||||
"""
|
||||
Tests for Asana tools.
|
||||
"""
|
||||
"""Tests for asana_tool - Asana task and project management."""
|
||||
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
from unittest.mock import MagicMock, patch
|
||||
from fastmcp import FastMCP
|
||||
from aden_tools.tools.asana_tool.tool import register_tools
|
||||
from aden_tools.credentials import CredentialStoreAdapter
|
||||
import os
|
||||
|
||||
# Try to import ApiException for side_effect, or create a mock exception
|
||||
try:
|
||||
from asana.rest import ApiException
|
||||
except ImportError:
|
||||
from aden_tools.tools.asana_tool.asana_tool import register_tools
|
||||
|
||||
class ApiException(Exception):
|
||||
def __init__(self, status=None, reason=None, http_resp=None):
|
||||
self.status = status
|
||||
self.reason = reason
|
||||
self.body = http_resp
|
||||
ENV = {"ASANA_ACCESS_TOKEN": "test-token"}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_credentials():
|
||||
return CredentialStoreAdapter.for_testing({"asana": "test-token"})
|
||||
def tool_fns(mcp: FastMCP):
|
||||
register_tools(mcp, credentials=None)
|
||||
tools = mcp._tool_manager._tools
|
||||
return {name: tools[name].fn for name in tools}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_apis():
|
||||
with patch("aden_tools.tools.asana_tool.tool.asana") as mock_asana:
|
||||
# Mock Configuration and ApiClient
|
||||
mock_asana.Configuration.return_value = MagicMock()
|
||||
mock_asana.ApiClient.return_value = MagicMock()
|
||||
|
||||
# Mock API classes
|
||||
mock_tasks = MagicMock()
|
||||
mock_asana.TasksApi.return_value = mock_tasks
|
||||
|
||||
mock_projects = MagicMock()
|
||||
mock_asana.ProjectsApi.return_value = mock_projects
|
||||
|
||||
mock_workspaces = MagicMock()
|
||||
mock_asana.WorkspacesApi.return_value = mock_workspaces
|
||||
|
||||
mock_users = MagicMock()
|
||||
mock_asana.UsersApi.return_value = mock_users
|
||||
|
||||
mock_stories = MagicMock()
|
||||
mock_asana.StoriesApi.return_value = mock_stories
|
||||
|
||||
mock_sections = MagicMock()
|
||||
mock_asana.SectionsApi.return_value = mock_sections
|
||||
|
||||
mock_tags = MagicMock()
|
||||
mock_asana.TagsApi.return_value = mock_tags
|
||||
|
||||
# Setup specific return values matches typical Asana response structure
|
||||
|
||||
# Tasks
|
||||
# Need to return objects that have .to_dict or are dicts
|
||||
# Our tool implementation handles both but let's be consistent
|
||||
t1 = MagicMock()
|
||||
t1.to_dict.return_value = {"gid": "123", "name": "Task"}
|
||||
mock_tasks.create_task.return_value = t1
|
||||
mock_tasks.update_task.return_value = t1
|
||||
mock_tasks.get_task.return_value = t1
|
||||
|
||||
# Search returns iterable
|
||||
t2 = MagicMock()
|
||||
t2.to_dict.return_value = {"gid": "123", "name": "Found Task"}
|
||||
# search_tasks_for_workspace returns request response which is iterable or has data
|
||||
search_res = MagicMock()
|
||||
search_res.data = [t2]
|
||||
search_res.__iter__.return_value = [t2]
|
||||
mock_tasks.search_tasks_for_workspace.return_value = search_res
|
||||
|
||||
mock_tasks.delete_task.return_value = {}
|
||||
mock_tasks.create_subtask_for_task.return_value = {
|
||||
"gid": "sub1",
|
||||
"name": "Subtask",
|
||||
}
|
||||
|
||||
# Projects
|
||||
p1 = MagicMock()
|
||||
p1.to_dict.return_value = {"gid": "p1", "name": "Project"}
|
||||
mock_projects.create_project.return_value = p1
|
||||
mock_projects.get_projects.return_value = [p1]
|
||||
|
||||
# Workspaces (default for get_workspace_gid)
|
||||
# Note: workspace resolution expects an iterable of dict-like objects or objects with .gid
|
||||
ws1 = MagicMock()
|
||||
ws1.gid = "ws1"
|
||||
ws1.name = "Workspace 1"
|
||||
# Ensure __getitem__ works for dict-like access if needed
|
||||
ws1.__getitem__ = (
|
||||
lambda name: "ws1"
|
||||
if name == "gid"
|
||||
else "Workspace 1"
|
||||
if name == "name"
|
||||
else None
|
||||
)
|
||||
|
||||
ws_res = MagicMock()
|
||||
ws_res.data = [ws1]
|
||||
ws_res.__iter__.return_value = [ws1]
|
||||
mock_workspaces.get_workspaces.return_value = ws_res
|
||||
|
||||
mock_workspaces.get_workspace.return_value = {
|
||||
"gid": "ws1",
|
||||
"name": "Workspace 1",
|
||||
}
|
||||
|
||||
# Stories
|
||||
story = MagicMock()
|
||||
story.to_dict.return_value = {"gid": "story1", "text": "Comment"}
|
||||
mock_stories.create_story_for_task.return_value = story
|
||||
|
||||
yield {
|
||||
"tasks": mock_tasks,
|
||||
"projects": mock_projects,
|
||||
"workspaces": mock_workspaces,
|
||||
"stories": mock_stories,
|
||||
"asana_module": mock_asana,
|
||||
}
|
||||
|
||||
|
||||
def test_missing_credentials():
|
||||
"""Test that missing credentials return an error dict with help."""
|
||||
mcp = FastMCP("test")
|
||||
# Empty credentials and ensure env var is unset
|
||||
creds = CredentialStoreAdapter.for_testing({})
|
||||
with patch.dict(os.environ, {}, clear=True):
|
||||
register_tools(mcp, credentials=creds)
|
||||
|
||||
# Access tool directly via _tool_manager
|
||||
tool = mcp._tool_manager._tools["asana_create_task"].fn
|
||||
result = tool(name="Task")
|
||||
|
||||
assert isinstance(result, dict)
|
||||
class TestAsanaListWorkspaces:
|
||||
def test_missing_token(self, tool_fns):
|
||||
with patch.dict("os.environ", {}, clear=True):
|
||||
result = tool_fns["asana_list_workspaces"]()
|
||||
assert "error" in result
|
||||
assert "Asana credentials not configured" in result["error"]
|
||||
assert "help" in result
|
||||
|
||||
def test_successful_list(self, tool_fns):
|
||||
mock_resp = {"data": [{"gid": "ws-1", "name": "My Workspace"}]}
|
||||
with (
|
||||
patch.dict("os.environ", ENV),
|
||||
patch("aden_tools.tools.asana_tool.asana_tool.httpx.get") as mock_get,
|
||||
):
|
||||
mock_get.return_value.status_code = 200
|
||||
mock_get.return_value.json.return_value = mock_resp
|
||||
result = tool_fns["asana_list_workspaces"]()
|
||||
|
||||
assert len(result["workspaces"]) == 1
|
||||
assert result["workspaces"][0]["name"] == "My Workspace"
|
||||
|
||||
|
||||
def test_invalid_credential_type():
|
||||
"""Test that non-string credential value is handled safely."""
|
||||
mcp = FastMCP("test")
|
||||
creds = CredentialStoreAdapter.for_testing({})
|
||||
# Mock get to return an object (non-string)
|
||||
with patch.object(creds, "get", return_value=123):
|
||||
register_tools(mcp, credentials=creds)
|
||||
tool = mcp._tool_manager._tools["asana_create_task"].fn
|
||||
result = tool(name="Task")
|
||||
class TestAsanaListProjects:
|
||||
def test_missing_workspace(self, tool_fns):
|
||||
with patch.dict("os.environ", ENV):
|
||||
result = tool_fns["asana_list_projects"](workspace_gid="")
|
||||
assert "error" in result
|
||||
|
||||
assert isinstance(result, dict)
|
||||
assert "error" in result # Should fallback to missing creds error path
|
||||
def test_successful_list(self, tool_fns):
|
||||
mock_resp = {
|
||||
"data": [
|
||||
{"gid": "proj-1", "name": "Website Redesign", "color": "blue", "archived": False}
|
||||
]
|
||||
}
|
||||
with (
|
||||
patch.dict("os.environ", ENV),
|
||||
patch("aden_tools.tools.asana_tool.asana_tool.httpx.get") as mock_get,
|
||||
):
|
||||
mock_get.return_value.status_code = 200
|
||||
mock_get.return_value.json.return_value = mock_resp
|
||||
result = tool_fns["asana_list_projects"](workspace_gid="ws-1")
|
||||
|
||||
assert len(result["projects"]) == 1
|
||||
assert result["projects"][0]["name"] == "Website Redesign"
|
||||
|
||||
|
||||
def test_api_error_handling(mock_apis, mock_credentials):
|
||||
"""Test standard API error handling."""
|
||||
mcp = FastMCP("test")
|
||||
register_tools(mcp, credentials=mock_credentials)
|
||||
tool = mcp._tool_manager._tools["asana_create_task"].fn
|
||||
class TestAsanaListTasks:
|
||||
def test_missing_params(self, tool_fns):
|
||||
with patch.dict("os.environ", ENV):
|
||||
result = tool_fns["asana_list_tasks"]()
|
||||
assert "error" in result
|
||||
|
||||
# Mock exception
|
||||
mock_apis["tasks"].create_task.side_effect = ApiException(
|
||||
status=400, reason="Bad Request"
|
||||
)
|
||||
def test_successful_list(self, tool_fns):
|
||||
mock_resp = {
|
||||
"data": [
|
||||
{
|
||||
"gid": "task-1",
|
||||
"name": "Design homepage",
|
||||
"completed": False,
|
||||
"due_on": "2024-06-15",
|
||||
"assignee": {"name": "Alice"},
|
||||
}
|
||||
]
|
||||
}
|
||||
with (
|
||||
patch.dict("os.environ", ENV),
|
||||
patch("aden_tools.tools.asana_tool.asana_tool.httpx.get") as mock_get,
|
||||
):
|
||||
mock_get.return_value.status_code = 200
|
||||
mock_get.return_value.json.return_value = mock_resp
|
||||
result = tool_fns["asana_list_tasks"](project_gid="proj-1")
|
||||
|
||||
result = tool(name="Fail Task", workspace="ws1")
|
||||
assert "error" in result
|
||||
assert "Bad Request" in str(result["error"])
|
||||
assert len(result["tasks"]) == 1
|
||||
assert result["tasks"][0]["name"] == "Design homepage"
|
||||
|
||||
|
||||
def test_create_task(mock_apis, mock_credentials):
|
||||
mcp = FastMCP("test")
|
||||
register_tools(mcp, credentials=mock_credentials)
|
||||
tool = mcp._tool_manager._tools["asana_create_task"].fn
|
||||
class TestAsanaGetTask:
|
||||
def test_missing_gid(self, tool_fns):
|
||||
with patch.dict("os.environ", ENV):
|
||||
result = tool_fns["asana_get_task"](task_gid="")
|
||||
assert "error" in result
|
||||
|
||||
result = tool(name="New Task", workspace="ws1")
|
||||
def test_successful_get(self, tool_fns):
|
||||
mock_resp = {
|
||||
"data": {
|
||||
"gid": "task-1",
|
||||
"name": "Design homepage",
|
||||
"notes": "Create the new homepage design",
|
||||
"completed": False,
|
||||
"due_on": "2024-06-15",
|
||||
"assignee": {"name": "Alice"},
|
||||
"projects": [{"name": "Website Redesign"}],
|
||||
"tags": [{"name": "urgent"}],
|
||||
"created_at": "2024-01-01T00:00:00Z",
|
||||
"modified_at": "2024-06-01T00:00:00Z",
|
||||
}
|
||||
}
|
||||
with (
|
||||
patch.dict("os.environ", ENV),
|
||||
patch("aden_tools.tools.asana_tool.asana_tool.httpx.get") as mock_get,
|
||||
):
|
||||
mock_get.return_value.status_code = 200
|
||||
mock_get.return_value.json.return_value = mock_resp
|
||||
result = tool_fns["asana_get_task"](task_gid="task-1")
|
||||
|
||||
assert result["gid"] == "123"
|
||||
|
||||
# Check args
|
||||
call_args = mock_apis["tasks"].create_task.call_args[0][0]
|
||||
assert call_args["data"]["name"] == "New Task"
|
||||
assert call_args["data"]["workspace"] == "ws1"
|
||||
assert result["name"] == "Design homepage"
|
||||
assert result["projects"] == ["Website Redesign"]
|
||||
assert result["tags"] == ["urgent"]
|
||||
|
||||
|
||||
def test_search_tasks_format(mock_apis, mock_credentials):
|
||||
"""Test that list results are wrapped in a dict."""
|
||||
mcp = FastMCP("test")
|
||||
register_tools(mcp, credentials=mock_credentials)
|
||||
tool = mcp._tool_manager._tools["asana_search_tasks"].fn
|
||||
class TestAsanaCreateTask:
|
||||
def test_missing_name(self, tool_fns):
|
||||
with patch.dict("os.environ", ENV):
|
||||
result = tool_fns["asana_create_task"](workspace_gid="ws-1", name="")
|
||||
assert "error" in result
|
||||
|
||||
result = tool(text="hello", workspace="ws1")
|
||||
def test_successful_create(self, tool_fns):
|
||||
mock_resp = {"data": {"gid": "task-new", "name": "New Task"}}
|
||||
with (
|
||||
patch.dict("os.environ", ENV),
|
||||
patch("aden_tools.tools.asana_tool.asana_tool.httpx.post") as mock_post,
|
||||
):
|
||||
mock_post.return_value.status_code = 201
|
||||
mock_post.return_value.json.return_value = mock_resp
|
||||
result = tool_fns["asana_create_task"](
|
||||
workspace_gid="ws-1", name="New Task", due_on="2024-07-01"
|
||||
)
|
||||
|
||||
# Expect dict wrapper {"data": [...]}
|
||||
assert isinstance(result, dict)
|
||||
assert "data" in result
|
||||
assert isinstance(result["data"], list)
|
||||
assert len(result["data"]) == 1
|
||||
assert result["data"][0]["gid"] == "123"
|
||||
assert result["status"] == "created"
|
||||
assert result["gid"] == "task-new"
|
||||
|
||||
|
||||
def test_workspace_resolution(mock_apis, mock_credentials):
|
||||
"""Test workspace resolution logic."""
|
||||
mcp = FastMCP("test")
|
||||
register_tools(mcp, credentials=mock_credentials)
|
||||
tool = mcp._tool_manager._tools["asana_create_task"].fn
|
||||
class TestAsanaSearchTasks:
|
||||
def test_missing_params(self, tool_fns):
|
||||
with patch.dict("os.environ", ENV):
|
||||
result = tool_fns["asana_search_tasks"](workspace_gid="", query="")
|
||||
assert "error" in result
|
||||
|
||||
# 1. By Name (Workspace 1 -> ws1)
|
||||
tool(name="T1", workspace="Workspace 1")
|
||||
args1 = mock_apis["tasks"].create_task.call_args[0][0]["data"]["workspace"]
|
||||
assert args1 == "ws1"
|
||||
def test_successful_search(self, tool_fns):
|
||||
mock_resp = {
|
||||
"data": [
|
||||
{"gid": "task-1", "name": "Design homepage", "completed": False, "due_on": "2024-06-15"}
|
||||
]
|
||||
}
|
||||
with (
|
||||
patch.dict("os.environ", ENV),
|
||||
patch("aden_tools.tools.asana_tool.asana_tool.httpx.get") as mock_get,
|
||||
):
|
||||
mock_get.return_value.status_code = 200
|
||||
mock_get.return_value.json.return_value = mock_resp
|
||||
result = tool_fns["asana_search_tasks"](workspace_gid="ws-1", query="design")
|
||||
|
||||
# 2. By GID (explicit)
|
||||
# If not found in list, fallback to using it as GID because it's digits
|
||||
tool(name="T2", workspace="999")
|
||||
args2 = mock_apis["tasks"].create_task.call_args[0][0]["data"]["workspace"]
|
||||
assert args2 == "999"
|
||||
|
||||
|
||||
def test_registration_completeness(mock_credentials):
|
||||
"""Verify all 25 tools are registered."""
|
||||
mcp = FastMCP("test")
|
||||
register_tools(mcp, credentials=mock_credentials)
|
||||
|
||||
expected_tools = [
|
||||
"asana_create_task",
|
||||
"asana_update_task",
|
||||
"asana_get_task",
|
||||
"asana_search_tasks",
|
||||
"asana_delete_task",
|
||||
"asana_add_task_comment",
|
||||
"asana_complete_task",
|
||||
"asana_add_subtask",
|
||||
"asana_create_project",
|
||||
"asana_update_project",
|
||||
"asana_get_project",
|
||||
"asana_list_projects",
|
||||
"asana_get_project_tasks",
|
||||
"asana_add_task_to_project",
|
||||
"asana_get_workspace",
|
||||
"asana_list_workspaces",
|
||||
"asana_get_user",
|
||||
"asana_list_team_members",
|
||||
"asana_create_section",
|
||||
"asana_list_sections",
|
||||
"asana_move_task_to_section",
|
||||
"asana_create_tag",
|
||||
"asana_add_tag_to_task",
|
||||
"asana_list_tags",
|
||||
"asana_update_custom_field",
|
||||
]
|
||||
|
||||
registered = mcp._tool_manager._tools.keys()
|
||||
for t in expected_tools:
|
||||
assert t in registered, f"Tool {t} not registered"
|
||||
assert len(result["tasks"]) == 1
|
||||
|
||||
Reference in New Issue
Block a user