bvandeusen/FabledScribe
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 50 Wiki Activity

feat(mcp): typed-entity tools (person/place/list)

Browse Source 
Nine tools — list/create/update for each of person, place, list.
Get and delete reuse fable_get_note / fable_delete_note (typed
entities share the Note model).

Lists: the wrappers accept an `items: list[str]` for ergonomics and
translate to the {text, checked} dict shape that
services/knowledge.py and KnowledgeView.vue expect. items=[] clears;
items=None leaves unchanged.

Updates do an explicit get → merge → update round trip so updating
one typed field doesn't clobber the others stored alongside it in
entity_meta (which is a single JSONB column).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Bryan Van Deusen
2026-05-26 20:53:36 -04:00
parent 6961144c3a
commit d4f3516552
3 changed files with 438 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/fabledassistant/mcp/tools/__init__.py
+2 -1
View File
@@ -5,7 +5,7 @@ to a FastMCP instance. `register_all(mcp)` is the single entry point called
from `mcp.server.build_mcp_server`.
"""
from fabledassistant.mcp.tools import (
events, milestones, notes, projects, recent, search, tags, tasks,
entities, events, milestones, notes, projects, recent, search, tags, tasks,
)
@@ -19,3 +19,4 @@ def register_all(mcp) -> None:
events.register(mcp)
tags.register(mcp)
recent.register(mcp)
entities.register(mcp)
src/fabledassistant/mcp/tools/entities.py
+245
View File
@@ -0,0 +1,245 @@
"""Typed-entity MCP tools: person, place, list.
These are notes with a non-'note' note_type and type-specific JSON metadata
stored in the entity_meta column. Three tools per type — list, create, update.
For get and delete, use fable_get_note / fable_delete_note (typed entities
share the Note model).
The wrappers translate typed-field kwargs into the entity_meta dict shape that
KnowledgeView.vue and services/knowledge.py expect.
Lists: a list entity stores its items in entity_meta["list_items"] as a list
of {text, checked} dicts. The create/update tools take a simpler `items` list
of plain strings for ergonomics; checked-state is reset to False on each call.
"""
from __future__ import annotations
from fabledassistant.mcp._context import current_user_id
from fabledassistant.services import knowledge as knowledge_svc
from fabledassistant.services import notes as notes_svc
async def _list_by_type(note_type: str, q: str, tag: str, limit: int) -> dict:
"""Common list query for a typed entity."""
uid = current_user_id()
items, total = await knowledge_svc.query_knowledge(
user_id=uid,
note_type=note_type,
tags=[tag] if tag else [],
sort="modified",
q=q or None,
limit=max(1, min(limit, 100)),
offset=0,
)
# Map "items" key to a type-specific key for caller clarity.
plural = {"person": "persons", "place": "places", "list": "lists"}[note_type]
return {plural: items, "total": total}
async def _create_entity(note_type: str, name: str, entity_meta: dict,
tags: list[str] | None = None) -> dict:
"""Create a typed note with the given metadata."""
uid = current_user_id()
note = await notes_svc.create_note(
uid,
title=name,
note_type=note_type,
entity_meta=entity_meta or None,
tags=tags,
)
return note.to_dict()
async def _update_entity(entity_id: int, note_type: str, name: str,
meta_updates: dict) -> dict:
"""Merge updates into entity_meta and (optionally) update the title."""
uid = current_user_id()
note = await notes_svc.get_note(uid, entity_id)
if note is None or note.note_type != note_type:
raise ValueError(f"{note_type} {entity_id} not found")
new_meta = dict(note.entity_meta or {})
new_meta.update(meta_updates)
fields: dict = {"entity_meta": new_meta}
if name:
fields["title"] = name
updated = await notes_svc.update_note(uid, entity_id, **fields)
return updated.to_dict()
# ─── Person ──────────────────────────────────────────────────────────────────
_PERSON_FIELDS = ("relationship", "email", "phone", "birthday",
"organization", "address")
async def fable_list_persons(q: str = "", tag: str = "", limit: int = 25) -> dict:
"""List people in the user's knowledge base.
Args:
q: Free-text search across name + body (optional).
tag: Filter to a single tag (optional).
limit: Max results (1-100).
"""
return await _list_by_type("person", q, tag, limit)
async def fable_create_person(
name: str,
relationship: str = "",
email: str = "",
phone: str = "",
birthday: str = "",
organization: str = "",
address: str = "",
tags: list[str] | None = None,
) -> dict:
"""Create a person in the user's knowledge base.
Args:
name: Person's name (required).
relationship: How the user knows them (e.g. "colleague", "friend").
email / phone / birthday (YYYY-MM-DD) / organization / address: optional.
tags: Plain-string tags, no # prefix.
"""
meta = {f: v for f, v in (
("relationship", relationship), ("email", email), ("phone", phone),
("birthday", birthday), ("organization", organization),
("address", address),
) if v}
return await _create_entity("person", name, meta, tags)
async def fable_update_person(
person_id: int,
name: str = "",
relationship: str = "",
email: str = "",
phone: str = "",
birthday: str = "",
organization: str = "",
address: str = "",
) -> dict:
"""Update a person. Only explicitly provided fields are changed.
To clear a field, pass an explicit space character (this preserves the
fable-mcp empty-string-means-omit convention).
"""
meta_updates = {f: v for f, v in (
("relationship", relationship), ("email", email), ("phone", phone),
("birthday", birthday), ("organization", organization),
("address", address),
) if v}
return await _update_entity(person_id, "person", name, meta_updates)
# ─── Place ───────────────────────────────────────────────────────────────────
async def fable_list_places(q: str = "", tag: str = "", limit: int = 25) -> dict:
"""List places (cafes, offices, addresses) in the user's knowledge base."""
return await _list_by_type("place", q, tag, limit)
async def fable_create_place(
name: str,
address: str = "",
phone: str = "",
hours: str = "",
website: str = "",
category: str = "",
tags: list[str] | None = None,
) -> dict:
"""Create a place in the user's knowledge base.
Args:
name: Place name (required).
address / phone / hours / website / category: optional.
tags: Plain-string tags, no # prefix.
"""
meta = {f: v for f, v in (
("address", address), ("phone", phone), ("hours", hours),
("website", website), ("category", category),
) if v}
return await _create_entity("place", name, meta, tags)
async def fable_update_place(
place_id: int,
name: str = "",
address: str = "",
phone: str = "",
hours: str = "",
website: str = "",
category: str = "",
) -> dict:
"""Update a place. Only explicitly provided fields are changed."""
meta_updates = {f: v for f, v in (
("address", address), ("phone", phone), ("hours", hours),
("website", website), ("category", category),
) if v}
return await _update_entity(place_id, "place", name, meta_updates)
# ─── List ────────────────────────────────────────────────────────────────────
async def fable_list_lists(q: str = "", tag: str = "", limit: int = 25) -> dict:
"""List checklists in the user's knowledge base."""
return await _list_by_type("list", q, tag, limit)
async def fable_create_list(
name: str,
category: str = "",
items: list[str] | None = None,
tags: list[str] | None = None,
) -> dict:
"""Create a checklist (a list-type entity).
Args:
name: List name (required).
category: Optional category label (e.g. "shopping", "packing").
items: Initial item texts. All items start unchecked.
tags: Plain-string tags, no # prefix.
"""
meta: dict = {}
if category:
meta["category"] = category
if items:
meta["list_items"] = [{"text": t, "checked": False} for t in items]
return await _create_entity("list", name, meta, tags)
async def fable_update_list(
list_id: int,
name: str = "",
category: str = "",
items: list[str] | None = None,
) -> dict:
"""Update a checklist.
Args:
list_id: ID of the list to update.
name: New title (optional).
category: New category (optional).
items: REPLACES the entire item set with these texts (all reset to
unchecked). Omit (None) to leave items unchanged. Pass an empty
list to clear all items.
"""
meta_updates: dict = {}
if category:
meta_updates["category"] = category
if items is not None:
meta_updates["list_items"] = [
{"text": t, "checked": False} for t in items
]
return await _update_entity(list_id, "list", name, meta_updates)
def register(mcp) -> None:
for fn in (
fable_list_persons, fable_create_person, fable_update_person,
fable_list_places, fable_create_place, fable_update_place,
fable_list_lists, fable_create_list, fable_update_list,
):
mcp.tool(name=fn.__name__)(fn)
tests/test_mcp_tool_entities.py
+191
View File
@@ -0,0 +1,191 @@
"""Tests for typed-entity tools (person/place/list).
Focuses on the metadata-shape translations and the entity_meta merge logic,
since that's where bugs would hide."""
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from fabledassistant.mcp._context import _user_id_ctx
from fabledassistant.mcp.tools.entities import (
fable_list_persons, fable_create_person, fable_update_person,
fable_create_place, fable_update_place,
fable_create_list, fable_update_list,
)
@pytest.fixture(autouse=True)
def _bind_user():
token = _user_id_ctx.set(7)
yield
_user_id_ctx.reset(token)
def _fake_note(*, note_type="note", entity_meta=None, **overrides) -> MagicMock:
n = MagicMock()
n.note_type = note_type
n.entity_meta = entity_meta
base = {"id": 1, "title": "t", "note_type": note_type,
"metadata": entity_meta or {}}
base.update(overrides)
n.to_dict.return_value = base
return n
# ─── list ────────────────────────────────────────────────────────────────────
@pytest.mark.asyncio
async def test_list_persons_calls_knowledge_with_person_type():
mock = AsyncMock(return_value=([{"id": 1, "title": "alice"}], 1))
with patch("fabledassistant.mcp.tools.entities.knowledge_svc.query_knowledge", mock):
out = await fable_list_persons(tag="work")
kwargs = mock.call_args.kwargs
assert kwargs["note_type"] == "person"
assert kwargs["tags"] == ["work"]
assert out["total"] == 1
assert "persons" in out # type-specific plural key
# ─── create: metadata building ───────────────────────────────────────────────
@pytest.mark.asyncio
async def test_create_person_only_includes_provided_fields_in_meta():
"""Empty-string fields must NOT pollute entity_meta."""
fake = _fake_note(note_type="person")
mock = AsyncMock(return_value=fake)
with patch("fabledassistant.mcp.tools.entities.notes_svc.create_note", mock):
await fable_create_person(name="Alice", email="a@x.com")
kwargs = mock.call_args.kwargs
assert kwargs["title"] == "Alice"
assert kwargs["note_type"] == "person"
assert kwargs["entity_meta"] == {"email": "a@x.com"}
@pytest.mark.asyncio
async def test_create_person_all_empty_meta_is_none():
"""Service is called with entity_meta=None when no typed fields were given."""
fake = _fake_note(note_type="person")
mock = AsyncMock(return_value=fake)
with patch("fabledassistant.mcp.tools.entities.notes_svc.create_note", mock):
await fable_create_person(name="Bob")
assert mock.call_args.kwargs["entity_meta"] is None
@pytest.mark.asyncio
async def test_create_place_categories_into_meta():
fake = _fake_note(note_type="place")
mock = AsyncMock(return_value=fake)
with patch("fabledassistant.mcp.tools.entities.notes_svc.create_note", mock):
await fable_create_place(name="Cafe X", address="123 Main", category="coffee")
meta = mock.call_args.kwargs["entity_meta"]
assert meta == {"address": "123 Main", "category": "coffee"}
@pytest.mark.asyncio
async def test_create_list_translates_strings_to_unchecked_items():
fake = _fake_note(note_type="list")
mock = AsyncMock(return_value=fake)
with patch("fabledassistant.mcp.tools.entities.notes_svc.create_note", mock):
await fable_create_list(name="shopping", items=["milk", "bread"])
meta = mock.call_args.kwargs["entity_meta"]
assert meta["list_items"] == [
{"text": "milk", "checked": False},
{"text": "bread", "checked": False},
]
# ─── update: meta merge ──────────────────────────────────────────────────────
@pytest.mark.asyncio
async def test_update_person_merges_meta_preserving_other_fields():
"""Updating one field must NOT clobber the others stored in entity_meta."""
existing = _fake_note(
id=5, note_type="person",
entity_meta={"email": "old@x.com", "phone": "555-1234"},
)
get_mock = AsyncMock(return_value=existing)
updated = _fake_note(note_type="person")
update_mock = AsyncMock(return_value=updated)
with patch(
"fabledassistant.mcp.tools.entities.notes_svc.get_note", get_mock,
), patch(
"fabledassistant.mcp.tools.entities.notes_svc.update_note", update_mock,
):
await fable_update_person(person_id=5, email="new@x.com")
new_meta = update_mock.call_args.kwargs["entity_meta"]
assert new_meta == {"email": "new@x.com", "phone": "555-1234"}
@pytest.mark.asyncio
async def test_update_person_no_typed_fields_keeps_meta_unchanged():
existing = _fake_note(
id=5, note_type="person",
entity_meta={"email": "a@x.com"},
)
updated = _fake_note(note_type="person")
with patch(
"fabledassistant.mcp.tools.entities.notes_svc.get_note",
AsyncMock(return_value=existing),
), patch(
"fabledassistant.mcp.tools.entities.notes_svc.update_note",
AsyncMock(return_value=updated),
) as update_mock:
await fable_update_person(person_id=5, name="New Name")
# entity_meta unchanged, but still passed (service gets the full new dict)
assert update_mock.call_args.kwargs["entity_meta"] == {"email": "a@x.com"}
assert update_mock.call_args.kwargs["title"] == "New Name"
@pytest.mark.asyncio
async def test_update_person_rejects_wrong_type():
"""Trying to update a 'place' as a 'person' must fail."""
wrong_type = _fake_note(id=5, note_type="place")
with patch(
"fabledassistant.mcp.tools.entities.notes_svc.get_note",
AsyncMock(return_value=wrong_type),
):
with pytest.raises(ValueError, match="person 5 not found"):
await fable_update_person(person_id=5, email="x@x.com")
@pytest.mark.asyncio
async def test_update_list_items_empty_list_clears_items():
"""items=[] clears all items; items=None leaves unchanged."""
existing = _fake_note(
id=5, note_type="list",
entity_meta={"list_items": [{"text": "old", "checked": True}]},
)
updated = _fake_note(note_type="list")
with patch(
"fabledassistant.mcp.tools.entities.notes_svc.get_note",
AsyncMock(return_value=existing),
), patch(
"fabledassistant.mcp.tools.entities.notes_svc.update_note",
AsyncMock(return_value=updated),
) as update_mock:
await fable_update_list(list_id=5, items=[])
assert update_mock.call_args.kwargs["entity_meta"]["list_items"] == []
@pytest.mark.asyncio
async def test_update_list_items_none_leaves_items_unchanged():
existing = _fake_note(
id=5, note_type="list",
entity_meta={"list_items": [{"text": "keep", "checked": False}]},
)
updated = _fake_note(note_type="list")
with patch(
"fabledassistant.mcp.tools.entities.notes_svc.get_note",
AsyncMock(return_value=existing),
), patch(
"fabledassistant.mcp.tools.entities.notes_svc.update_note",
AsyncMock(return_value=updated),
) as update_mock:
await fable_update_list(list_id=5, name="renamed")
# Existing items intact in the merged meta
assert update_mock.call_args.kwargs["entity_meta"]["list_items"] == [
{"text": "keep", "checked": False},
]