"""Task CRUD MCP tools. Tasks are notes with a non-null `status` — same model, different filter. Wrappers call services/notes.py for CRUD with is_task=True and add the task-specific fields (status, priority, due_date, parent_id), plus services/task_logs.py for add_task_log. There is no delete_task — matches the existing fable-mcp surface. Cancel by updating status to "cancelled". Sentinels (preserved from existing fable-mcp): - status="" / priority="" / title="" / body="" → "leave unchanged" on update - status="todo" is the default on create (creates a task; non-null status is what makes a Note a Task) - priority="none" sets explicit no-priority; priority="" is "leave unchanged" - project_id=0 / milestone_id=0 / parent_id=0 → "no association" on create, "leave unchanged" on update; on update, -1 clears the FK (sets it NULL) """ from __future__ import annotations from scribe.mcp._context import current_user_id from scribe.services import notes as notes_svc from scribe.services import planning as planning_svc from scribe.services import rulebooks as rulebooks_svc from scribe.services import task_logs as task_logs_svc from scribe.services import trash as trash_svc async def list_tasks( limit: int = 20, offset: int = 0, status: str = "", project_id: int = 0, kind: str = "", ) -> dict: """List tasks in Scribe. Args: status: Filter by status — one of: todo, in_progress, done, cancelled. Omit for all. project_id: Filter to a specific project. PASS THE ACTIVE PROJECT'S ID whenever a project is in scope so you list that project's tasks, not every project's. 0 = no filter (all projects — use only for a deliberate cross-project view). kind: Filter by task kind — 'work' or 'plan'. Omit (empty) for all kinds. Results are ordered by last-updated descending. """ uid = current_user_id() rows, total = await notes_svc.list_notes( uid, is_task=True, status=status or None, project_id=project_id or None, task_kind=kind or None, limit=max(1, min(limit, 100)), offset=max(0, offset), ) return {"tasks": [n.to_dict() for n in rows], "total": total} async def get_task(task_id: int) -> dict: """Fetch a single Scribe task by ID. Returns id, title, body, status, priority, tags, project_id, milestone_id, parent_id, parent_title, due_date, created_at, updated_at. For kind=plan tasks, the response also includes applicable_rules + subscribed_rulebooks from the task's project's rulebook subscriptions. """ uid = current_user_id() note = await notes_svc.get_note(uid, task_id) if note is None: raise ValueError(f"task {task_id} not found") data = note.to_dict() parent_title = None if note.parent_id: parent = await notes_svc.get_note(uid, note.parent_id) if parent is not None: parent_title = parent.title data["parent_title"] = parent_title if data.get("task_kind") == "plan" and note.project_id: applicable = await rulebooks_svc.get_applicable_rules( project_id=note.project_id, user_id=uid, ) data["applicable_rules"] = applicable["rules"] data["subscribed_rulebooks"] = applicable["subscribed_rulebooks"] data["applicable_rules_truncated"] = applicable["truncated"] data["project_rules"] = applicable.get("project_rules", []) data["suppressed_rules"] = applicable.get("suppressed_rules", []) data["suppressed_topics"] = applicable.get("suppressed_topics", []) return data async def create_task( title: str, body: str = "", status: str = "todo", priority: str = "", project_id: int = 0, milestone_id: int = 0, parent_id: int = 0, tags: list[str] | None = None, kind: str = "work", ) -> dict: """Create a new task in Scribe. Args: title: Task title (required). body: Markdown description / notes for the task. status: Initial status — one of: todo (default), in_progress, done, cancelled. priority: One of: low, medium, high, or 'none'. Omit (empty string) to leave unset. project_id: Associate with a project (0 = no project). milestone_id: Place within a project milestone (0 = no milestone). parent_id: Make this a sub-task of another task (0 = top-level). tags: List of plain-string tags without # prefix. kind: 'work' (default) or 'plan'. Prefer the start_planning tool to create plans. """ uid = current_user_id() note = await notes_svc.create_note( uid, title=title, body=body, status=status, priority=priority or None, project_id=project_id or None, milestone_id=milestone_id or None, parent_id=parent_id or None, tags=tags, task_kind=kind, ) return note.to_dict() async def update_task( task_id: int, title: str = "", body: str = "", status: str = "", priority: str = "", project_id: int = 0, milestone_id: int = 0, ) -> dict: """Update an existing Scribe task. Only explicitly provided fields are changed. Args: task_id: ID of the task to update. title: New title, or omit to leave unchanged. body: New markdown body, or omit to leave unchanged. status: New status — one of: todo, in_progress, done, cancelled. Drive the lifecycle: set in_progress when you start, done when complete — don't leave finished work at todo. priority: New priority — one of: none, low, medium, high. project_id: New project. 0 = leave unchanged, -1 = clear (remove from its project; also clears the milestone), positive = set. milestone_id: New milestone. 0 = leave unchanged, -1 = clear (remove from its milestone), positive = set. """ uid = current_user_id() fields: dict = {} if title: fields["title"] = title if body: fields["body"] = body if status: fields["status"] = status if priority: fields["priority"] = priority # Optional FKs: 0 = leave unchanged, -1 = clear (set NULL), positive = set. if project_id == -1: fields["project_id"] = None fields["milestone_id"] = None # a milestone can't outlive its project elif project_id: fields["project_id"] = project_id if milestone_id == -1: fields["milestone_id"] = None elif milestone_id: fields["milestone_id"] = milestone_id note = await notes_svc.update_note(uid, task_id, **fields) if note is None: raise ValueError(f"task {task_id} not found") return note.to_dict() async def add_task_log(task_id: int, content: str) -> dict: """Append a timestamped progress log entry to a Scribe task. Use this to record work sessions, decisions, or status updates over time without overwriting the task's main body. Each entry is stored separately and shown chronologically in the task view. """ uid = current_user_id() log = await task_logs_svc.create_log(uid, task_id, content) return log.to_dict() if hasattr(log, "to_dict") else { "id": log.id, "task_id": log.task_id, "content": log.content, "created_at": log.created_at.isoformat() if log.created_at else None, } async def start_planning(project_id: int, title: str) -> dict: """Begin a plan in Scribe (the preferred home for plans — not a local .md file). Creates a plan-task (a task with kind=plan) seeded with a plan template under the given project, and returns it together with the project's applicable Rulebook rules and brief context. Maintain the plan afterwards with the normal task tools (update_task to edit the body, add_task_log to record progress). Args: project_id: The project this plan is for. title: A short title for the plan. """ uid = current_user_id() return await planning_svc.start_planning( user_id=uid, project_id=project_id, title=title, ) async def delete_task(task_id: int) -> dict: """Move a Scribe task (or plan) to the trash (recoverable). Sub-tasks go with it. Restore via restore(batch_id).""" uid = current_user_id() batch = await trash_svc.delete(uid, "task", task_id) if batch is None: raise ValueError(f"task {task_id} not found") return {"deleted_batch_id": batch, "message": f"Task {task_id} moved to trash. Restore with restore('{batch}')."} def register(mcp) -> None: for fn in ( list_tasks, get_task, create_task, update_task, add_task_log, start_planning, delete_task, ): mcp.tool(name=fn.__name__)(fn)