Surface task location on reads and add parent_task_id to update_task

format_task now returns the Location property on all task reads (search,
snapshot, bulk, complete, etc.) when the Tasks DB defines one. update_task
gains a parent_task_id parameter, matching create_task and bulk_update_tasks.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: bobbyhyam

CLAUDE.md

@@ -45,7 +45,7 @@ Six required env vars: `NOTION_SECRET`, `TASKS_DS_ID`, `PROJECTS_DS_ID`, `NOTES_
 
 **Source layout:** `src/ultimate_brain_mcp/` with four modules:
 
-- **`server.py`** — All 30 MCP tool definitions using `@mcp.tool()` decorators. Tools are grouped: Tasks (6), Projects (4), Notes (4), Tags (3), Goals (4), Cross-cutting (3: `daily_summary`, `archive_item`, `set_page_content`), Workflow consolidators (2: `daily_review_snapshot`, `bulk_update_tasks`), Generic (4: `query_database`, `get_page`, `get_page_content`, `update_page`). The 4 create tools (`create_task`, `create_note`, `create_project`, `create_goal`) accept an optional `content` parameter for page body content. `create_task` and `update_task` also accept optional `tag_ids` (Tag relation) and `location` (auto-detected from the live Tasks schema). `bulk_update_tasks` applies up to N task patches concurrently with per-row results, never raising on a single failure. Each tool uses `ToolAnnotations` to declare read-only vs destructive. Property builder helpers (`_prop_title`, `_prop_select`, `_build_location_payload`, etc.) construct Notion API property payloads. `_coerce_property()` auto-converts Python types to Notion property format for the generic `update_page` tool.
+- **`server.py`** — All 30 MCP tool definitions using `@mcp.tool()` decorators. Tools are grouped: Tasks (6), Projects (4), Notes (4), Tags (3), Goals (4), Cross-cutting (3: `daily_summary`, `archive_item`, `set_page_content`), Workflow consolidators (2: `daily_review_snapshot`, `bulk_update_tasks`), Generic (4: `query_database`, `get_page`, `get_page_content`, `update_page`). The 4 create tools (`create_task`, `create_note`, `create_project`, `create_goal`) accept an optional `content` parameter for page body content. `create_task` and `update_task` also accept optional `parent_task_id` (Parent Task relation), `tag_ids` (Tag relation), and `location` (auto-detected from the live Tasks schema). `format_task` surfaces `location` on every task read (search/get/snapshot/bulk results) when the Tasks DB has a Location property — callers pass the discovered property name from `tasks_schema`. `bulk_update_tasks` applies up to N task patches concurrently with per-row results, never raising on a single failure. Each tool uses `ToolAnnotations` to declare read-only vs destructive. Property builder helpers (`_prop_title`, `_prop_select`, `_build_location_payload`, etc.) construct Notion API property payloads. `_coerce_property()` auto-converts Python types to Notion property format for the generic `update_page` tool.
 
 - **`notion_client.py`** — Async httpx wrapper around Notion API v2025-09-03. Uses the data_sources query endpoint (not legacy database queries). Core methods: `query_all()` (paginated), `create_page()` (supports `children` for inline body content), `get_page()`, `update_page()`, `get_blocks()`, `append_blocks()`, `delete_block()`. Raises `NotionAPIError` with status-specific hints.
 

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "ultimate-brain-mcp"
-version = "0.5.0"
+version = "0.5.1"
 description = "MCP server for Thomas Frank's Ultimate Brain Notion system"
 readme = "README.md"
 license = "MIT"

src/ultimate_brain_mcp/formatters.py

@@ -137,6 +137,7 @@ def format_task(
     *,
     project_lookup: dict[str, dict] | None = None,
     tag_lookup: dict[str, dict] | None = None,
+    location_property_name: str | None = None,
 ) -> dict:
     """Format a Task page into an agent-friendly dict.
 
@@ -145,6 +146,12 @@ def format_task(
     *tag_lookup* is supplied, ``area_tag_names`` is populated for any Tag
     relations whose IDs appear in the lookup. Existing callers that pass
     nothing keep the legacy shape — IDs only, no resolved names.
+
+    When *location_property_name* is supplied (the dynamically-discovered Tasks
+    Location property name), the result carries ``location``. The shape mirrors
+    how it is stored: a string for ``select`` / ``status`` properties, a list
+    for ``multi_select``. Absent when no name is passed, the property is missing
+    from the page, or its value is empty.
     """
     props = page.get("properties", {})
     result: dict = {
@@ -186,6 +193,22 @@ def format_task(
     labels = _multi_select(props.get("Labels", {}))
     if labels:
         result["labels"] = labels
+    # Location — dynamic property name discovered at boot. Dispatch on the
+    # property's embedded `type` so we don't need the TasksSchema here.
+    if location_property_name:
+        loc_prop = props.get(location_property_name)
+        if loc_prop:
+            ptype = loc_prop.get("type")
+            if ptype == "select":
+                val = _select(loc_prop)
+            elif ptype == "status":
+                val = _status(loc_prop)
+            elif ptype == "multi_select":
+                val = _multi_select(loc_prop)  # list
+            else:
+                val = None
+            if val:
+                result["location"] = val
     # Recurrence (Recur Unit select + Recur Interval number)
     recur_unit = _select(props.get("Recur Unit", {}))
     recur_interval = _number(props.get("Recur Interval", {}))

src/ultimate_brain_mcp/server.py

@@ -453,7 +453,8 @@ async def search_tasks(
         pages = await app.client.query_all(
             app.config.tasks_ds_id, filter=query_filter, sorts=sorts
         )
-        return [format_task(p) for p in pages[:limit]]
+        loc_name = app.tasks_schema.location_property_name
+        return [format_task(p, location_property_name=loc_name) for p in pages[:limit]]
     except NotionAPIError as e:
         return _handle_api_error(e)
 
@@ -475,7 +476,8 @@ async def get_my_day(
     }
     try:
         pages = await app.client.query_all(app.config.tasks_ds_id, filter=query_filter)
-        tasks = [format_task(p) for p in pages]
+        loc_name = app.tasks_schema.location_property_name
+        tasks = [format_task(p, location_property_name=loc_name) for p in pages]
         priority_order = {"High": 0, "Medium": 1, "Low": 2, None: 3}
         tasks.sort(key=lambda t: priority_order.get(t.get("priority"), 3))
         return tasks
@@ -501,7 +503,8 @@ async def get_inbox_tasks(
     }
     try:
         pages = await app.client.query_all(app.config.tasks_ds_id, filter=query_filter)
-        return [format_task(p) for p in pages]
+        loc_name = app.tasks_schema.location_property_name
+        return [format_task(p, location_property_name=loc_name) for p in pages]
     except NotionAPIError as e:
         return _handle_api_error(e)
 
@@ -601,7 +604,9 @@ async def create_task(
 
     try:
         page = await app.client.create_page(app.config.tasks_ds_id, props, children=children)
-        result = format_task(page)
+        result = format_task(
+            page, location_property_name=app.tasks_schema.location_property_name
+        )
         if location_warning:
             result["_warning"] = location_warning
         return result
@@ -630,6 +635,10 @@ async def update_task(
     project_id: Annotated[str | None, Field(description="New project page ID.")] = None,
     labels: Annotated[list[str] | None, Field(description="New labels (replaces existing).")] = None,
     my_day: Annotated[bool | None, Field(description="Set My Day flag.")] = None,
+    parent_task_id: Annotated[
+        str | None,
+        Field(description="New parent task page ID (for sub-tasks)."),
+    ] = None,
     tag_ids: Annotated[
         list[str] | None,
         Field(description=(
@@ -671,6 +680,8 @@ async def update_task(
         props["Labels"] = _prop_multi_select(labels)
     if my_day is not None:
         props["My Day"] = _prop_checkbox(my_day)
+    if parent_task_id is not None:
+        props["Parent Task"] = _prop_relation([parent_task_id])
     if tag_ids is not None:
         props["Tag"] = _prop_relation(tag_ids)
     if location is not None:
@@ -685,7 +696,9 @@ async def update_task(
 
     try:
         page = await app.client.update_page(task_id, props)
-        result = format_task(page)
+        result = format_task(
+            page, location_property_name=app.tasks_schema.location_property_name
+        )
         if location_warning:
             result["_warning"] = location_warning
         return result
@@ -704,9 +717,10 @@ async def complete_task(
     Handles recurring tasks: resets status to To Do and advances due date by the recurrence interval.
     Use search_tasks to find task IDs."""
     app = _ctx(ctx)
+    loc_name = app.tasks_schema.location_property_name
     try:
         page = await app.client.get_page(task_id)
-        task = format_task(page)
+        task = format_task(page, location_property_name=loc_name)
 
         # Check for recurrence
         recurrence = task.get("recurrence", "")
@@ -719,7 +733,7 @@ async def complete_task(
                 props["Due"] = _prop_date(new_due)
             props["My Day"] = _prop_checkbox(False)
             page = await app.client.update_page(task_id, props)
-            result = format_task(page)
+            result = format_task(page, location_property_name=loc_name)
             result["_note"] = f"Recurring task reset. Next due: {new_due or 'unchanged'}"
             return result
         else:
@@ -730,7 +744,7 @@ async def complete_task(
                 "My Day": _prop_checkbox(False),
             }
             page = await app.client.update_page(task_id, props)
-            return format_task(page)
+            return format_task(page, location_property_name=loc_name)
     except NotionAPIError as e:
         return _handle_api_error(e, "Use search_tasks to find valid task IDs.")
 
@@ -890,7 +904,8 @@ async def get_project_detail(
         )
 
         project = format_project(project_page)
-        tasks = [format_task(t) for t in task_pages]
+        loc_name = app.tasks_schema.location_property_name
+        tasks = [format_task(t, location_property_name=loc_name) for t in task_pages]
         notes = [format_note(n) for n in note_pages[:10]]
 
         # Task breakdown by status
@@ -1539,7 +1554,8 @@ async def daily_summary(
             my_day_fut, overdue_fut, inbox_fut, projects_fut, goals_fut
         )
 
-        my_day_tasks = [format_task(p) for p in my_day]
+        loc_name = app.tasks_schema.location_property_name
+        my_day_tasks = [format_task(p, location_property_name=loc_name) for p in my_day]
         priority_order = {"High": 0, "Medium": 1, "Low": 2, None: 3}
         my_day_tasks.sort(key=lambda t: priority_order.get(t.get("priority"), 3))
 
@@ -1551,7 +1567,7 @@ async def daily_summary(
             },
             "overdue": {
                 "count": len(overdue),
-                "tasks": [format_task(p) for p in overdue],
+                "tasks": [format_task(p, location_property_name=loc_name) for p in overdue],
             },
             "inbox_count": len(inbox),
             "active_projects_count": len(projects),
@@ -1748,12 +1764,19 @@ async def daily_review_snapshot(
         formatted = format_tag(t)
         tag_lookup[formatted["id"]] = {"name": formatted.get("name", "")}
 
+    loc_name = app.tasks_schema.location_property_name
+
     def _fmt_bucket(pages: list[dict], cap: int) -> tuple[list[dict], bool]:
         truncated = len(pages) > cap
         sliced = pages[:cap]
         return (
             [
-                format_task(p, project_lookup=project_lookup, tag_lookup=tag_lookup)
+                format_task(
+                    p,
+                    project_lookup=project_lookup,
+                    tag_lookup=tag_lookup,
+                    location_property_name=loc_name,
+                )
                 for p in sliced
             ],
             truncated,
@@ -1928,7 +1951,10 @@ async def _apply_one(idx: int, update: BulkTaskUpdate) -> dict:
                 row: dict = {
                     "task_id": update.task_id,
                     "ok": True,
-                    "task": format_task(page),
+                    "task": format_task(
+                        page,
+                        location_property_name=app.tasks_schema.location_property_name,
+                    ),
                 }
                 if warnings:
                     row["_warnings"] = warnings

tests/test_format_task_lookups.py

@@ -107,3 +107,73 @@ def test_format_task_no_relations_no_resolved_fields_even_with_lookups():
     assert "project_name" not in task
     assert "tag_ids" not in task
     assert "area_tag_names" not in task
+
+
+def _with_location(page: dict, prop_name: str, prop: dict) -> dict:
+    """Attach a Location-style property to a task page under *prop_name*."""
+    page["properties"][prop_name] = prop
+    return page
+
+
+def test_format_task_location_select():
+    """A select-typed Location property is surfaced as a string."""
+    page = _with_location(
+        _task_page(), "Location", {"type": "select", "select": {"name": "Home"}}
+    )
+
+    task = format_task(page, location_property_name="Location")
+
+    assert task["location"] == "Home"
+
+
+def test_format_task_location_status():
+    """A status-typed Location property is surfaced as a string."""
+    page = _with_location(
+        _task_page(), "Location", {"type": "status", "status": {"name": "Office"}}
+    )
+
+    task = format_task(page, location_property_name="Location")
+
+    assert task["location"] == "Office"
+
+
+def test_format_task_location_multi_select():
+    """A multi_select-typed Location property is surfaced as a list."""
+    page = _with_location(
+        _task_page(),
+        "Where",
+        {"type": "multi_select", "multi_select": [{"name": "Home"}, {"name": "Errands"}]},
+    )
+
+    task = format_task(page, location_property_name="Where")
+
+    assert task["location"] == ["Home", "Errands"]
+
+
+def test_format_task_location_omitted_without_name():
+    """Without location_property_name, location is never added (legacy shape)."""
+    page = _with_location(
+        _task_page(), "Location", {"type": "select", "select": {"name": "Home"}}
+    )
+
+    task = format_task(page)
+
+    assert "location" not in task
+
+
+def test_format_task_location_absent_when_property_missing():
+    """A name that isn't present on the page simply yields no location field."""
+    task = format_task(_task_page(), location_property_name="Location")
+
+    assert "location" not in task
+
+
+def test_format_task_location_absent_when_empty():
+    """An empty Location value yields no location field."""
+    page = _with_location(
+        _task_page(), "Location", {"type": "select", "select": None}
+    )
+
+    task = format_task(page, location_property_name="Location")
+
+    assert "location" not in task