create_task

Instructions

Implementation Reference

• src/lunatask_mcp/tools/tasks_create.py:23-319 (handler)

  Main handler function 'create_task_tool' that validates inputs, coerces types, builds a TaskCreate model, calls the API client, and returns success/error responses.

  async def create_task_tool(  # noqa: PLR0913, PLR0911, PLR0915, PLR0912, C901
      lunatask_client: LunaTaskClient,
      ctx: Context,
      name: str,
      note: str | None = None,
      area_id: str | None = None,
      status: str = "later",
      priority: int | str = 0,
      motivation: str = "unknown",
      eisenhower: int | str | None = None,
      estimate: int | str | None = None,
      progress: int | str | None = None,
      goal_id: str | None = None,
      scheduled_on: str | None = None,
      source: str | None = None,
      source_id: str | None = None,
  ) -> dict[str, Any]:
      """Create a new task in LunaTask.
  
      This MCP tool creates a new task using the LunaTask API. All task fields
      are supported, with only the name being required.
  
      Args:
          ctx: MCP context for logging and communication
          name: Task name (required)
          note: Optional task note
          area_id: Optional area ID the task belongs to
          status: Task status (default: "later")
          priority: Optional task priority level (accepts int or numeric string)
          motivation: Optional task motivation (must, should, want, unknown)
          eisenhower: Optional eisenhower matrix quadrant (0-4; accepts int or numeric string)
          estimate: Optional estimated duration in minutes (accepts int or numeric string)
          progress: Optional task completion percentage (accepts int or numeric string)
          goal_id: Optional goal ID the task belongs to
          scheduled_on: Optional scheduled date in YYYY-MM-DD format
          source: Optional external system label for the task origin
          source_id: Optional external record identifier in the source system
  
      Returns:
          dict[str, Any]: Response containing task creation result with task_id
  
      Raises:
          LunaTaskValidationError: When task validation fails (422)
          LunaTaskSubscriptionRequiredError: When subscription required (402)
          LunaTaskAuthenticationError: When authentication fails (401)
          LunaTaskRateLimitError: When rate limit exceeded (429)
          LunaTaskServerError: When server error occurs (5xx)
          LunaTaskAPIError: For other API errors
      """
      await ctx.info(f"Creating new task: {name}")
  
      # Coerce string priority values to integers when possible for client UX
      coerced_priority: int
      if isinstance(priority, int):
          coerced_priority = priority
      else:
          try:
              coerced_priority = int(priority)
          except (TypeError, ValueError):
              error_msg = "Invalid priority: must be an integer between -2 and 2"
              result = {
                  "success": False,
                  "error": "validation_error",
                  "message": f"Validation failed for priority: {error_msg}",
              }
              await ctx.error(error_msg)
              logger.warning("Invalid priority type for create_task: %r", priority)
              return result
  
      # Coerce string eisenhower values to integers when possible for client UX
      coerced_eisenhower: int | None = None
      if eisenhower is not None:
          if isinstance(eisenhower, int):
              coerced_eisenhower = eisenhower
          else:
              try:
                  coerced_eisenhower = int(eisenhower)
              except (TypeError, ValueError):
                  error_msg = "Invalid eisenhower: must be an integer between 0 and 4"
                  result = {
                      "success": False,
                      "error": "validation_error",
                      "message": f"Validation failed for eisenhower: {error_msg}",
                  }
                  await ctx.error(error_msg)
                  logger.warning("Invalid eisenhower type for create_task: %r", eisenhower)
                  return result
  
      # Coerce string estimate values to integers when possible for client UX
      coerced_estimate: int | None = None
      if estimate is not None:
          if isinstance(estimate, int):
              coerced_estimate = estimate
          else:
              try:
                  coerced_estimate = int(estimate)
              except (TypeError, ValueError):
                  error_msg = "Invalid estimate: must be an integer (minutes)"
                  result = {
                      "success": False,
                      "error": "validation_error",
                      "message": f"Validation failed for estimate: {error_msg}",
                  }
                  await ctx.error(error_msg)
                  logger.warning("Invalid estimate type for create_task: %r", estimate)
                  return result
  
      # Coerce string progress values to integers when possible for client UX
      coerced_progress: int | None = None
      if progress is not None:
          if isinstance(progress, int):
              coerced_progress = progress
          else:
              try:
                  coerced_progress = int(progress)
              except (TypeError, ValueError):
                  error_msg = "Invalid progress: must be an integer (percentage)"
                  result = {
                      "success": False,
                      "error": "validation_error",
                      "message": f"Validation failed for progress: {error_msg}",
                  }
                  await ctx.error(error_msg)
                  logger.warning("Invalid progress type for create_task: %r", progress)
                  return result
  
      # Parse and validate scheduled_on if provided
      parsed_scheduled_on = None
      if scheduled_on is not None:
          try:
              parsed_scheduled_on = date.fromisoformat(scheduled_on)
          except (ValueError, TypeError) as e:
              error_msg = f"Invalid scheduled_on format. Expected YYYY-MM-DD format: {e}"
              result = {
                  "success": False,
                  "error": "validation_error",
                  "message": error_msg,
              }
              await ctx.error(error_msg)
              logger.warning("Invalid scheduled_on format for create_task: %s", scheduled_on)
              return result
  
      try:
          # Create TaskCreate object from parameters
          # Cast string parameters to proper Literal types
          task_status = (
              status if status in ("later", "next", "started", "waiting", "completed") else "later"
          )
          task_motivation = (
              motivation if motivation in ("must", "should", "want", "unknown") else "unknown"
          )
  
          task_data = TaskCreate(
              name=name,
              note=note,
              area_id=area_id,
              status=task_status,  # type: ignore[arg-type]
              priority=coerced_priority,
              motivation=task_motivation,  # type: ignore[arg-type]
              eisenhower=coerced_eisenhower,
              estimate=coerced_estimate,
              progress=coerced_progress,
              goal_id=goal_id,
              scheduled_on=parsed_scheduled_on,
              source=source,
              source_id=source_id,
          )
  
          # Use LunaTask client to create the task
          async with lunatask_client as client:
              created_task = await client.create_task(task_data)
  
          # Return success response with task ID
          result = {
              "success": True,
              "task_id": created_task.id,
              "message": "Task created successfully",
          }
  
      except LunaTaskValidationError as e:
          # Handle validation errors (422)
          error_msg = f"Task validation failed: {e}"
          result = {
              "success": False,
              "error": "validation_error",
              "message": error_msg,
          }
          await ctx.error(error_msg)
          logger.warning("Task validation error: %s", e)
          return result
  
      except LunaTaskSubscriptionRequiredError as e:
          # Handle subscription required errors (402)
          error_msg = f"Subscription required: {e}"
          result = {
              "success": False,
              "error": "subscription_required",
              "message": error_msg,
          }
          await ctx.error(error_msg)
          logger.warning("Subscription required for task creation: %s", e)
          return result
  
      except LunaTaskAuthenticationError as e:
          # Handle authentication errors (401)
          error_msg = f"Authentication failed: {e}"
          result = {
              "success": False,
              "error": "authentication_error",
              "message": error_msg,
          }
          await ctx.error(error_msg)
          logger.warning("Authentication error during task creation: %s", e)
          return result
  
      except LunaTaskRateLimitError as e:
          # Handle rate limit errors (429)
          error_msg = f"Rate limit exceeded: {e}"
          result = {
              "success": False,
              "error": "rate_limit_error",
              "message": error_msg,
          }
          await ctx.error(error_msg)
          logger.warning("Rate limit exceeded during task creation: %s", e)
          return result
  
      except LunaTaskServerError as e:
          # Handle server errors (5xx)
          error_msg = f"Server error: {e}"
          result = {
              "success": False,
              "error": "server_error",
              "message": error_msg,
          }
          await ctx.error(error_msg)
          logger.warning("Server error during task creation: %s", e)
          return result
  
      except LunaTaskAPIError as e:
          # Handle other API errors
          error_msg = f"API error: {e}"
          result = {
              "success": False,
              "error": "api_error",
              "message": error_msg,
          }
          await ctx.error(error_msg)
          logger.warning("API error during task creation: %s", e)
          return result
  
      except Exception as e:
          # Handle Pydantic validation errors specifically
          if "ValidationError" in str(type(e)) and hasattr(e, "errors"):
              # Handle Pydantic validation errors with structured MCP response
              error_details: list[str] = []
              for error in e.errors():  # type: ignore[attr-defined]
                  field = error.get("loc", ["unknown"])[0] if error.get("loc") else "unknown"  # type: ignore[misc]
                  msg = error.get("msg", "Invalid value")  # type: ignore[misc]
                  if field == "motivation":
                      msg = "Must be one of: must, should, want, unknown"
                  elif field == "eisenhower":
                      msg = "Must be between 0 and 4"
                  elif field == "priority":
                      msg = "Must be between -2 and 2"
                  elif field == "status":
                      msg = "Must be one of: later, next, started, waiting, completed"
                  elif field == "estimate":
                      msg = "Must be a positive integer (minutes)"
                  elif field == "progress":
                      msg = "Must be an integer between 0 and 100 (percentage)"
                  elif field == "scheduled_on":
                      msg = "Must be in YYYY-MM-DD format"
                  error_details.append(f"{field}: {msg}")
  
              error_msg = f"Validation failed for {', '.join(error_details)}"
              result = {
                  "success": False,
                  "error": "validation_error",
                  "message": error_msg,
              }
              await ctx.error(error_msg)
              logger.warning("Task validation error: %s", error_msg)
              return result
          # Handle unexpected errors
          error_msg = f"Unexpected error creating task: {e}"
          result = {
              "success": False,
              "error": "unexpected_error",
              "message": error_msg,
          }
          await ctx.error(error_msg)
          logger.exception("Unexpected error during task creation")
          return result
      else:
          await ctx.info(f"Successfully created task {created_task.id}")
          return result

• src/lunatask_mcp/api/models.py:244-269 (schema)

  TaskCreate Pydantic model defining the request schema for creating tasks, with fields like area_id, name, status, priority, motivation, eisenhower, estimate, progress, goal_id, scheduled_on, source, source_id.

  class TaskCreate(TaskPayload):
      """Request model for creating new tasks in LunaTask.
  
      Inherits shared fields and validation from `TaskPayload` and applies
      create-time defaults and requirements. Overrides specific fields with
      proper non-None defaults to ensure explicit validation behavior.
      """
  
      # Ensure outbound JSON uses enum string values
      model_config = ConfigDict(use_enum_values=True)
  
      area_id: str = Field(description="Area ID the task belongs to")
      source: str | None = Field(
          default=None,
          description=(
              "Identification of external system where the task originated (e.g., 'github')."
          ),
      )
      source_id: str | None = Field(
          default=None,
          description="Identifier of the record in the external system (e.g., issue ID)",
      )
  
      def __init__(self, **data: object) -> None:
          """Pydantic-compatible initializer with permissive typing for tools/tests."""
          super().__init__(**data)  # type: ignore[arg-type]

• src/lunatask_mcp/api/models.py:20-97 (schema)

  Supporting schema definitions: TaskStatus enum, TaskMotivation enum, TaskPayload base model with validation constraints for all task fields.

  class TaskStatus(StrEnum):
      """Status values accepted by LunaTask task creation/update."""
  
      LATER = "later"
      NEXT = "next"
      STARTED = "started"
      WAITING = "waiting"
      COMPLETED = "completed"
  
  
  class TaskMotivation(StrEnum):
      """Motivation values accepted by LunaTask task creation/update."""
  
      MUST = "must"
      SHOULD = "should"
      WANT = "want"
      UNKNOWN = "unknown"
  
  
  # Constants for validation bounds
  MIN_PRIORITY = -2
  MAX_PRIORITY = 2
  MIN_EISENHOWER = 0
  MAX_EISENHOWER = 4
  
  
  class TaskPayload(BaseModel):
      """Shared request payload fields for task create/update.
  
      This base model centralizes field declarations and validation constraints
      common to both `TaskCreate` and `TaskUpdate`. Fields that need defaults
      for creation are overridden in `TaskCreate` with proper non-None defaults.
  
      Notes:
          - Enum and bounds validation live here so both subclasses stay in sync.
          - Outbound serialization uses enum string values (`use_enum_values=True`).
      """
  
      # Ensure outbound JSON uses enum string values and reject unsupported fields
      model_config = ConfigDict(use_enum_values=True, extra="forbid")
  
      # Shared relational/context fields
      goal_id: str | None = Field(
          default=None,
          description=(
              "ID of the goal where the task should belong to (optional, "
              "can be found in our apps in the goal's settings)"
          ),
      )
  
      # State and prioritization - kept as None for TaskUpdate's PATCH semantics
      # TaskCreate will override these with proper defaults
      status: TaskStatus | None = Field(default=None, description="Task status")
      estimate: int | None = Field(default=None, description="Estimated duration in minutes")
      priority: int | None = Field(
          default=None,
          ge=MIN_PRIORITY,
          le=MAX_PRIORITY,
          description=f"Task priority level [{MIN_PRIORITY}, {MAX_PRIORITY}]",
      )
      progress: int | None = Field(default=None, description="Task completion percentage")
      motivation: TaskMotivation | None = Field(
          default=None, description="Task motivation level (must, should, want, unknown)"
      )
      eisenhower: int | None = Field(
          default=None,
          ge=MIN_EISENHOWER,
          le=MAX_EISENHOWER,
          description=f"Eisenhower matrix quadrant [{MIN_EISENHOWER}, {MAX_EISENHOWER}]",
      )
      scheduled_on: date | None = Field(
          default=None, description="Date when task is scheduled (YYYY-MM-DD format, date-only)"
      )
      # Optional encrypted content fields
      name: str | None = Field(default=None, description="Task name (gets encrypted client-side)")
      note: str | None = Field(
          default=None, description="Task note in Markdown (encrypted client-side)"
      )

• src/lunatask_mcp/api/client_tasks.py:159-200 (helper)

  API client method 'create_task' in TasksClientMixin that serializes TaskCreate, makes POST request to /v1/tasks, and parses TaskResponse from the wrapped response.

  async def create_task(self, task_data: TaskCreate) -> TaskResponse:
      """Create a new task in the LunaTask API.
  
      Args:
          task_data: TaskCreate object containing task data to create
  
      Returns:
          TaskResponse: Created task object from the API with assigned ID
  
      Raises:
          LunaTaskValidationError: Validation error (422)
          LunaTaskSubscriptionRequiredError: Subscription required (402)
          LunaTaskAuthenticationError: Invalid bearer token (401)
          LunaTaskRateLimitError: Rate limit exceeded (429)
          LunaTaskServerError: Server error occurred (5xx)
          LunaTaskNetworkError: Network connectivity error
          LunaTaskAPIError: Other API errors
      """
      # Convert TaskCreate model to JSON data
      # Use model_dump_json to properly serialize date objects, then parse back to dict
      json_data = json.loads(task_data.model_dump_json(exclude_none=True))
  
      # Make authenticated request to POST /v1/tasks endpoint
      response_data = await self._get_base_client().make_request("POST", "tasks", data=json_data)
  
      # Parse response JSON into TaskResponse model instance
      # The create task API returns a wrapped response in format {"task": {...}}
      try:
          task = TaskResponse(**response_data["task"])
      except KeyError as e:
          logger.exception("Failed to extract task from wrapped response format")
          task_name = json_data.get("name", "unknown")
          raise LunaTaskAPIError.create_parse_error(
              "tasks", task_name=f"{task_name} - missing 'task' key"
          ) from e
      except Exception as e:
          logger.exception("Failed to parse created task response data")
          task_name = json_data.get("name", "unknown")
          raise LunaTaskAPIError.create_parse_error("tasks", task_name=task_name) from e
      else:
          logger.debug("Successfully created task: %s", task.id)
          return task

• src/lunatask_mcp/tools/tasks.py:188-302 (registration)

  Registration of the 'create_task' tool on the MCP server via self.mcp.tool('create_task')(_create_task_tool) at line 302, and the wrapper function _create_task_tool that delegates to the main handler.

  async def _create_task_tool(  # noqa: PLR0913
      ctx: ServerContext,
      name: str,
      note: str | None = None,
      area_id: str | None = None,
      status: str = "later",
      priority: int | str = 0,
      motivation: str = "unknown",
      eisenhower: int | str | None = None,
      estimate: int | str | None = None,
      progress: int | str | None = None,
      goal_id: str | None = None,
      scheduled_on: str | None = None,
      source: str | None = None,
      source_id: str | None = None,
  ) -> dict[str, Any]:
      """Create a new task in LunaTask."""
      return await create_task_tool_fn(
          self.lunatask_client,
          ctx,
          name,
          note,
          area_id,
          status,
          priority,
          motivation,
          eisenhower,
          estimate,
          progress,
          goal_id,
          scheduled_on,
          source,
          source_id,
      )
  
  # TODO: Refactor _update_task_tool with`TypedDict` to avoid too many arguments
  async def _update_task_tool(  # noqa: PLR0913
      ctx: ServerContext,
      id: str,  # noqa: A002
      name: str | None = None,
      note: str | None = None,
      area_id: str | None = None,
      status: str | None = None,
      priority: int | str | None = None,
      scheduled_on: str | None = None,
      motivation: str | None = None,
      eisenhower: int | str | None = None,
      estimate: int | str | None = None,
      progress: int | str | None = None,
      goal_id: str | None = None,
  ) -> dict[str, Any]:
      """Update an existing task in LunaTask."""
      return await update_task_tool_fn(
          self.lunatask_client,
          ctx,
          id,
          name,
          note,
          area_id,
          status,
          priority,
          scheduled_on,
          motivation,
          eisenhower,
          estimate,
          progress,
          goal_id,
      )
  
  async def _delete_task_tool(ctx: ServerContext, id: str) -> dict[str, Any]:  # noqa: A002
      """Delete an existing task in LunaTask."""
      return await delete_task_tool_fn(self.lunatask_client, ctx, id)
  
  # Keep a non-breaking explicit discovery URI that maps to the same handler.
  self.mcp.resource("lunatask://tasks")(_tasks_discovery)
  self.mcp.resource("lunatask://tasks/discovery")(_tasks_discovery)
  self.mcp.resource("lunatask://tasks/{task_id}")(_task_single_resource)
  # Register area alias templates
  self.mcp.resource("lunatask://area/{area_id}/now")(_area_now)
  self.mcp.resource("lunatask://area/{area_id}/today")(_area_today)
  self.mcp.resource("lunatask://area/{area_id}/overdue")(_area_overdue)
  self.mcp.resource("lunatask://area/{area_id}/next-7-days")(_area_next7)
  self.mcp.resource("lunatask://area/{area_id}/high-priority")(_area_high)
  self.mcp.resource("lunatask://area/{area_id}/recent-completions")(_area_recent)
  
  # Register global alias resources
  async def _global_now(ctx: ServerContext) -> dict[str, Any]:
      return await list_tasks_global_alias_fn(self.lunatask_client, ctx, alias="now")
  
  async def _global_today(ctx: ServerContext) -> dict[str, Any]:
      return await list_tasks_global_alias_fn(self.lunatask_client, ctx, alias="today")
  
  async def _global_overdue(ctx: ServerContext) -> dict[str, Any]:
      return await list_tasks_global_alias_fn(self.lunatask_client, ctx, alias="overdue")
  
  async def _global_next7(ctx: ServerContext) -> dict[str, Any]:
      return await list_tasks_global_alias_fn(self.lunatask_client, ctx, alias="next_7_days")
  
  async def _global_high(ctx: ServerContext) -> dict[str, Any]:
      return await list_tasks_global_alias_fn(
          self.lunatask_client, ctx, alias="high_priority"
      )
  
  async def _global_recent(ctx: ServerContext) -> dict[str, Any]:
      return await list_tasks_global_alias_fn(
          self.lunatask_client, ctx, alias="recent_completions"
      )
  
  self.mcp.resource("lunatask://global/now")(_global_now)
  self.mcp.resource("lunatask://global/today")(_global_today)
  self.mcp.resource("lunatask://global/overdue")(_global_overdue)
  self.mcp.resource("lunatask://global/next-7-days")(_global_next7)
  self.mcp.resource("lunatask://global/high-priority")(_global_high)
  self.mcp.resource("lunatask://global/recent-completions")(_global_recent)
  self.mcp.tool("create_task")(_create_task_tool)