get_all_tasks

Instructions

Implementation Reference

• src/amazing_marvin_mcp/main.py:196-229 (handler)

  The MCP tool handler for get_all_tasks. Decorated with @mcp.tool(), it accepts an optional label filter, creates an API client, calls get_all_tasks_impl, and returns a StandardResponse.

  @mcp.tool()
  async def get_all_tasks(
      label: str | None = None, debug: bool = False
  ) -> StandardResponse:
      """Get all tasks across all projects with optional label filtering (comprehensive search).
  
      Use when you need to search/filter across your entire task system.
      For daily focus, use get_daily_productivity_overview() instead.
  
      Args:
          label: Optional label name to filter by. If None, returns all tasks.
  
      Note: This is a heavy operation that recursively searches all projects.
      """
      start_time = time.time()
      try:
          api_client = create_api_client()
          result = get_all_tasks_impl(api_client, label)
  
          # Estimate API calls based on typical project count
          estimated_api_calls = result.get("api_calls_made", 5)
  
          return create_simple_response(
              data=result,
              summary_text=f"Retrieved {result.get('total_tasks', 0)} tasks across all projects"
              + (f" with label '{label}'" if label else ""),
              api_endpoint="/categories + /children",
              api_calls_made=estimated_api_calls,
              debug=debug,
              start_time=start_time,
          )
      except Exception as e:
          logger.exception("Failed to get all tasks")
          return create_error_response(e, "/categories + /children", debug, start_time)

• src/amazing_marvin_mcp/tasks.py:211-267 (helper)

  The implementation helper (get_all_tasks_impl). Fetches all top-level items (today tasks, due items, projects, categories), then recursively collects all nested items via get_all_nested_items. Optionally filters by label. Finally filters to tasks only (excluding projects/categories).

  def get_all_tasks_impl(
      api_client: MarvinAPIClient, label: str | None = None
  ) -> dict[str, Any]:
      """Get all tasks and projects with optional label filtering, using recursive traversal."""
      try:
          # Get all top-level items
          today_items = api_client.get_tasks()
          due_items = api_client.get_due_items()
          projects = api_client.get_projects()
          categories = api_client.get_categories()
  
          all_items = get_all_nested_items(
              list(itertools.chain(*[today_items, due_items, projects, categories])),
              api_client=api_client,
          )
  
          # Filter by label if specified
          if label:
              # Get all labels to understand the label structure
              labels = api_client.get_labels()
  
              # Find the label ID
              label_id = None
              for lbl in labels:
                  if lbl.get("title", "").lower() == label.lower():
                      label_id = lbl.get("_id")
                      break
  
              if label_id:
                  filtered_items = []
                  for item in all_items:
                      item_labels = item.get("labelIds", [])
                      if label_id in item_labels:
                          filtered_items.append(item)
                  all_items = filtered_items
              else:
                  # If label not found, return empty results
                  all_items = []
  
          # Filter to only tasks (not projects or categories)
          tasks = [
              item
              for item in all_items
              if item.get("type") not in ["project", "category"]
          ]
  
          return {
              "tasks": tasks,
              "task_count": len(tasks),
              "filter_applied": label is not None,
              "label_filter": label,
              "source": "Recursive traversal of all items",
          }
  
      except Exception as e:
          logger.exception("Failed to get all tasks")
          return {"error": str(e), "tasks": [], "task_count": 0}

• src/amazing_marvin_mcp/tasks.py:184-208 (helper)

  Helper function get_all_nested_items that takes a list of top-level items and recursively expands projects/categories by fetching their children via _get_all_children_recursive, deduplicating by ID.

  def get_all_nested_items(
      items: list[dict[str, Any]], api_client: MarvinAPIClient
  ) -> list[dict[str, Any]]:
      all_items: list[dict[str, Any]] = []
      item_ids = set()
  
      for item in items:
          item_id = item.get("_id")
          if item_id and item_id not in item_ids:
              all_items.append(item)
              item_ids.add(item_id)
  
      # For each project/category, recursively get all children
      for item in list(all_items):  # Create a copy to avoid modifying while iterating
          item_id = item.get("_id")
          item_type = item.get("type")
          if not item_id or (item_type not in ["project", "category"]):
              continue
          children = _get_all_children_recursive(api_client, item_id)
          for child in children:
              child_id = child.get("_id")
              if child_id and child_id not in item_ids:
                  all_items.append(child)
                  item_ids.add(child_id)
      return all_items

• src/amazing_marvin_mcp/tasks.py:129-159 (helper)

  Recursive helper _get_all_children_recursive that fetches children of a parent item and recursively fetches grandchildren, with cycle detection via a visited set.

  def _get_all_children_recursive(
      api_client: MarvinAPIClient, parent_id: str, visited: set[str] | None = None
  ) -> list[dict[str, Any]]:
      """Recursively get all children of a parent item, avoiding infinite loops."""
      if visited is None:
          visited = set()
  
      if parent_id in visited:
          logger.warning("Circular reference detected for parent_id %s", parent_id)
          return []
  
      visited.add(parent_id)
  
      try:
          direct_children = api_client.get_children(parent_id)
          all_children = list(direct_children)  # Copy the list
  
          # Recursively get children of each child
          for child in direct_children:
              child_id = child.get("_id")
              if child_id:
                  grandchildren = _get_all_children_recursive(
                      api_client, child_id, visited.copy()
                  )
                  all_children.extend(grandchildren)
  
      except Exception as e:
          logger.warning("Failed to get children for parent_id %s: %s", parent_id, e)
          return []
      else:
          return all_children

• src/amazing_marvin_mcp/main.py:43-43 (registration)

  The FastMCP instance creation. The @mcp.tool() decorator on get_all_tasks (line 196) registers the tool with the MCP server.

  mcp: FastMCP = FastMCP(name="amazing-marvin-mcp")