ryan/simbarag
1
0
Fork 0
Code Issues Pull Requests 4 Actions Packages Projects Releases Wiki Activity

Return artifacts from Mealie, Obsidian and search tools

Browse Source 
Convert mealie_get_recipe, mealie_meal_plans, mealie_todays_meals,
mealie_shopping_lists, obsidian_read_note, journal_get_today,
obsidian_search_notes and simba_search to content_and_artifact. The text
the LLM sees is unchanged; the artifact is a JSON-safe ToolArtifact dict
per the Cowork UI contract (mealie.recipe, mealie.meal_plan,
mealie.shopping_lists, obsidian.note, search.documents). Error, empty and
not-configured paths return no artifact.

simba_search's artifact changes from a list of Documents (unused) to a
search.documents dict.

New optional env vars: MEALIE_GROUP_SLUG (default "home") for Mealie v2
recipe links, OBSIDIAN_VAULT_NAME for obsidian://open links.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Ryan Chen
2026-10-12 17:53:30 -04:00
parent 417d4061d7
commit 50d123e4d2
5 changed files with 954 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.example
+4
View File
@@ -70,6 +70,8 @@ YNAB_SYNC_INTERVAL=900
MEALIE_BASE_URL=http://192.168.1.5:9000
# Get your API token from Mealie's user settings page
MEALIE_API_TOKEN=your-mealie-api-token
# Group slug used in recipe page links (/g/<slug>/r/<recipe>); defaults to "home"
MEALIE_GROUP_SLUG=home
# Email Integration
# Email Encryption Key (32-byte URL-safe base64)
@@ -167,6 +169,8 @@ OBSIDIAN_DEVICE_NAME=simbarag
OBSIDIAN_CONTINUOUS_SYNC=false
# Local path to Obsidian vault (where files are synced)
OBSIDIAN_VAULT_PATH=/app/data/obsidian
# Vault name for obsidian://open links in the artifact panel (optional; no link if unset)
OBSIDIAN_VAULT_NAME=
# LangGraph checkpoint store (SQLite). Must be on a persistent volume.
# Both blue and green containers share this volume via docker-compose.
blueprints/conversation/agents.py
+188 -41
View File
@@ -11,6 +11,19 @@ from langchain_core.runnables import RunnableConfig, RunnableLambda
from langchain_openai import ChatOpenAI
from tavily import AsyncTavilyClient
from blueprints.conversation.artifacts_util import (
NOTE_MAX_CHARS,
format_date_range,
format_day,
json_safe,
make_artifact,
mealie_source_ref,
note_title,
obsidian_source_ref,
search_documents_artifact,
truncate,
unique,
)
from blueprints.conversation.memory import save_memory
from blueprints.conversation.prompts import SIMBA_SYSTEM_PROMPT
from blueprints.rag.logic import query_vector_store
@@ -131,20 +144,36 @@ async def simba_search(query: str):
print(f"[SIMBA SEARCH] Found {len(docs)} documents")
print(f"[SIMBA SEARCH] Serialized result length: {len(serialized)}")
print(f"[SIMBA SEARCH] First 200 chars: {serialized[:200]}")
return serialized, docs
return serialized, search_documents_artifact(query, "paperless", docs)
@tool
async def mealie_todays_meals() -> str:
MEALIE_NOT_CONFIGURED = "Mealie integration is not configured. Please set MEALIE_BASE_URL and MEALIE_API_TOKEN environment variables."
def _meal_entry(meal: dict) -> dict:
"""Meal plan entry in the `mealie.meal_plan` artifact shape."""
entry = {
"entry_type": meal["entry_type"],
"name": meal["recipe_name"] or meal["title"] or "Untitled",
}
if meal.get("recipe_slug"):
entry["recipe_slug"] = meal["recipe_slug"]
if meal.get("note"):
entry["note"] = meal["note"]
return entry
@tool(response_format="content_and_artifact")
async def mealie_todays_meals():
"""Get today's planned meals from Mealie."""
if not mealie_enabled:
return "Mealie integration is not configured. Please set MEALIE_BASE_URL and MEALIE_API_TOKEN environment variables."
return MEALIE_NOT_CONFIGURED, None
try:
result = await mealie_service.get_todays_meals()
if result["total_meals"] == 0:
return f"No meals planned for today ({result['today']})."
return f"No meals planned for today ({result['today']}).", None
lines = [f"Meals planned for today ({result['today']}):"]
for meal in result["meals"]:
@@ -152,13 +181,29 @@ async def mealie_todays_meals() -> str:
lines.append(f"- {meal['entry_type'].capitalize()}: {name}")
if meal["note"]:
lines.append(f" Note: {meal['note']}")
return "\n".join(lines)
today = result["today"]
artifact = make_artifact(
"mealie.meal_plan",
f"Today's meals · {format_day(today)}",
{
"start_date": today,
"end_date": today,
"days": [
{
"date": today,
"meals": [_meal_entry(m) for m in result["meals"]],
}
],
},
)
return "\n".join(lines), artifact
except Exception as e:
return f"Error fetching today's meals: {str(e)}"
return f"Error fetching today's meals: {str(e)}", None
@tool
async def mealie_meal_plans(start_date: str = "", end_date: str = "") -> str:
@tool(response_format="content_and_artifact")
async def mealie_meal_plans(start_date: str = "", end_date: str = ""):
"""Get Mealie meal plans for a date range.
Args:
@@ -166,7 +211,7 @@ async def mealie_meal_plans(start_date: str = "", end_date: str = "") -> str:
end_date: YYYY-MM-DD (default: 7 days out)
"""
if not mealie_enabled:
return "Mealie integration is not configured. Please set MEALIE_BASE_URL and MEALIE_API_TOKEN environment variables."
return MEALIE_NOT_CONFIGURED, None
try:
result = await mealie_service.get_meal_plans(
@@ -175,11 +220,15 @@ async def mealie_meal_plans(start_date: str = "", end_date: str = "") -> str:
)
if result["total_plans"] == 0:
return f"No meal plans found from {result['start_date']} to {result['end_date']}."
return (
f"No meal plans found from {result['start_date']} to {result['end_date']}.",
None,
)
lines = [
f"Meal plans from {result['start_date']} to {result['end_date']} ({result['total_plans']} meals):"
]
days = []
for date, plans in sorted(result["plans_by_date"].items()):
lines.append(f"\n{date}:")
for plan in plans:
@@ -187,20 +236,31 @@ async def mealie_meal_plans(start_date: str = "", end_date: str = "") -> str:
lines.append(f" - {plan['entry_type'].capitalize()}: {name}")
if plan["note"]:
lines.append(f" Note: {plan['note']}")
return "\n".join(lines)
days.append({"date": date, "meals": [_meal_entry(p) for p in plans]})
artifact = make_artifact(
"mealie.meal_plan",
f"Meal plan · {format_date_range(result['start_date'], result['end_date'])}",
{
"start_date": result["start_date"],
"end_date": result["end_date"],
"days": days,
},
)
return "\n".join(lines), artifact
except Exception as e:
return f"Error fetching meal plans: {str(e)}"
return f"Error fetching meal plans: {str(e)}", None
@tool
async def mealie_get_recipe(recipe_slug: str) -> str:
@tool(response_format="content_and_artifact")
async def mealie_get_recipe(recipe_slug: str):
"""Get full recipe details (ingredients, instructions, times) from Mealie.
Args:
recipe_slug: Recipe slug from meal plan data
"""
if not mealie_enabled:
return "Mealie integration is not configured. Please set MEALIE_BASE_URL and MEALIE_API_TOKEN environment variables."
return MEALIE_NOT_CONFIGURED, None
try:
recipe = await mealie_service.get_recipe(recipe_slug)
@@ -236,22 +296,49 @@ async def mealie_get_recipe(recipe_slug: str) -> str:
if recipe["tags"]:
lines.append(f"\nTags: {', '.join(recipe['tags'])}")
return "\n".join(lines)
instructions = []
for step in recipe["instructions"]:
instruction = {"text": step.get("text", "")}
if step.get("title"):
instruction["title"] = step["title"]
instructions.append(instruction)
slug = recipe.get("slug") or recipe_slug
artifact = make_artifact(
"mealie.recipe",
recipe["name"],
json_safe(
{
"slug": slug,
"name": recipe["name"],
"description": recipe["description"],
"servings": recipe["servings"],
"prep_time": recipe["prep_time"],
"cook_time": recipe["cook_time"],
"total_time": recipe["total_time"],
"ingredients": [ing["display"] for ing in recipe["ingredients"]],
"instructions": instructions,
"tags": recipe["tags"],
}
),
mealie_source_ref(slug),
)
return "\n".join(lines), artifact
except Exception as e:
return f"Error fetching recipe: {str(e)}"
return f"Error fetching recipe: {str(e)}", None
@tool
async def mealie_shopping_lists() -> str:
@tool(response_format="content_and_artifact")
async def mealie_shopping_lists():
"""Get Mealie shopping lists with needed and checked-off items."""
if not mealie_enabled:
return "Mealie integration is not configured. Please set MEALIE_BASE_URL and MEALIE_API_TOKEN environment variables."
return MEALIE_NOT_CONFIGURED, None
try:
result = await mealie_service.get_shopping_lists()
if result["total_lists"] == 0:
return "No shopping lists found."
return "No shopping lists found.", None
lines = [f"Shopping lists ({result['total_unchecked_items']} items needed):"]
for sl in result["shopping_lists"]:
@@ -270,9 +357,37 @@ async def mealie_shopping_lists() -> str:
lines.append(
f" ({len(sl['checked_items'])} items already checked off)"
)
return "\n".join(lines)
lists = [
{
"name": sl["name"],
"items": [
{
"name": item["name"],
"quantity": item["quantity"],
"checked": False,
}
for item in sl["unchecked_items"]
]
+ [
{
"name": item["name"],
"quantity": item["quantity"],
"checked": True,
}
for item in sl["checked_items"]
],
}
for sl in result["shopping_lists"]
]
artifact = make_artifact(
"mealie.shopping_lists",
"Shopping lists",
json_safe({"lists": lists}),
)
return "\n".join(lines), artifact
except Exception as e:
return f"Error fetching shopping lists: {str(e)}"
return f"Error fetching shopping lists: {str(e)}", None
@tool
@@ -309,34 +424,55 @@ async def mealie_create_meal_plan(
return f"Error creating meal plan: {str(e)}"
@tool
async def obsidian_search_notes(query: str) -> str:
OBSIDIAN_NOT_CONFIGURED = "Obsidian integration is not configured. Please set OBSIDIAN_VAULT_PATH environment variable."
def _note_artifact(path: str, parsed: dict, title: str | None = None) -> dict:
"""`obsidian.note` artifact from ObsidianService.parse_markdown output."""
frontmatter = json_safe(parsed["metadata"])
title = title or note_title(path, frontmatter)
return make_artifact(
"obsidian.note",
title,
{
"path": path,
"title": title,
"markdown": truncate(parsed["content"], NOTE_MAX_CHARS),
"tags": unique(parsed["tags"]),
"frontmatter": frontmatter,
},
obsidian_source_ref(path),
)
@tool(response_format="content_and_artifact")
async def obsidian_search_notes(query: str):
"""Search personal Obsidian vault notes.
Args:
query: Search query
"""
if not obsidian_enabled:
return "Obsidian integration is not configured. Please set OBSIDIAN_VAULT_PATH environment variable."
return OBSIDIAN_NOT_CONFIGURED, None
try:
# Query vector store filtered to obsidian source only
serialized, docs = await query_vector_store(query=query, source="obsidian")
return serialized
return serialized, search_documents_artifact(query, "obsidian", docs)
except Exception as e:
return f"Error searching Obsidian notes: {str(e)}"
return f"Error searching Obsidian notes: {str(e)}", None
@tool
async def obsidian_read_note(relative_path: str) -> str:
@tool(response_format="content_and_artifact")
async def obsidian_read_note(relative_path: str):
"""Read the full content of a specific Obsidian note by path.
Args:
relative_path: Path relative to vault root (e.g., "notes/my-note.md")
"""
if not obsidian_enabled:
return "Obsidian integration is not configured. Please set OBSIDIAN_VAULT_PATH environment variable."
return OBSIDIAN_NOT_CONFIGURED, None
try:
note = obsidian_service.read_note(relative_path)
@@ -348,12 +484,15 @@ async def obsidian_read_note(relative_path: str) -> str:
result += f"Tags: {', '.join(content_data['tags'])}\n"
result += f"Contains {len(content_data['wikilinks'])} wikilinks and {len(content_data['embeds'])} embeds"
return result
return result, _note_artifact(note["path"], content_data)
except FileNotFoundError:
return f"Note not found at '{relative_path}'. Please check the path is correct."
return (
f"Note not found at '{relative_path}'. Please check the path is correct.",
None,
)
except Exception as e:
return f"Error reading note: {str(e)}"
return f"Error reading note: {str(e)}", None
@tool
@@ -391,19 +530,27 @@ async def obsidian_create_note(
return f"Error creating note: {str(e)}"
@tool
def journal_get_today() -> str:
@tool(response_format="content_and_artifact")
def journal_get_today():
"""Get today's full daily journal note including tasks and log entries."""
if not obsidian_enabled:
return "Obsidian integration is not configured."
return "Obsidian integration is not configured.", None
try:
note = obsidian_service.get_daily_note()
if not note["found"]:
return f"No daily note found for {note['date']}. Use journal_add_task to create one."
return f"Daily note for {note['date']}:\n\n{note['content']}"
return (
f"No daily note found for {note['date']}. Use journal_add_task to create one.",
None,
)
artifact = _note_artifact(
note["path"],
obsidian_service.parse_markdown(note["content"]),
title=f"Daily note · {format_day(note['date'])}",
)
return f"Daily note for {note['date']}:\n\n{note['content']}", artifact
except Exception as e:
return f"Error reading daily note: {str(e)}"
return f"Error reading daily note: {str(e)}", None
@tool
blueprints/conversation/artifacts_util.py
+169
View File
@@ -0,0 +1,169 @@
"""Helpers for building ToolArtifact dicts returned by agent tools.
See docs/cowork-ui-plan.md ("Artifacts") for the contract. Artifacts must be
plain JSON-serialisable dicts: they ride along in ToolMessage.artifact and are
stored in the LangGraph checkpoint.
"""
import os
from datetime import date, datetime
from pathlib import Path
from urllib.parse import quote, urlencode
SNIPPET_MAX_CHARS = 400
NOTE_MAX_CHARS = 50_000
ELLIPSIS = "…"
def make_artifact(
kind: str, title: str, data: dict, source_ref: dict | None = None
) -> dict:
artifact = {"kind": kind, "title": title, "data": data}
if source_ref:
artifact["source_ref"] = source_ref
return artifact
def truncate(text: str, limit: int) -> str:
"""Cap text at `limit` chars, ending in an ellipsis when cut."""
if len(text) <= limit:
return text
return text[: limit - len(ELLIPSIS)].rstrip() + ELLIPSIS
def snippet(text: str, limit: int = SNIPPET_MAX_CHARS) -> str:
"""Collapse whitespace and cap at the snippet limit."""
return truncate(" ".join((text or "").split()), limit)
def json_safe(value):
"""Convert a value (e.g. parsed YAML frontmatter) to JSON-safe types."""
if isinstance(value, (str, int, float, bool, type(None))):
return value
if isinstance(value, (datetime, date)):
return value.isoformat()
if isinstance(value, (list, tuple, set)):
return [json_safe(v) for v in value]
if isinstance(value, dict):
return {str(k): json_safe(v) for k, v in value.items()}
return str(value)
def unique(items) -> list:
"""Deduplicate while keeping first-seen order."""
return list(dict.fromkeys(items))
def _parse_date(value: str) -> date | None:
try:
return datetime.strptime(value, "%Y-%m-%d").date()
except (TypeError, ValueError):
return None
def format_day(value: str) -> str:
"""'2026-10-12' -> 'Oct 12'. Unparseable input is returned unchanged."""
d = _parse_date(value)
return f"{d:%b} {d.day}" if d else str(value)
def format_date_range(start: str, end: str) -> str:
"""Human date range: 'Oct 12–19', 'Oct 28–Nov 3', 'Dec 28, 2026–Jan 3, 2027'."""
s, e = _parse_date(start), _parse_date(end)
if not s or not e:
return f"{start}–{end}"
if s == e:
return f"{s:%b} {s.day}"
if s.year != e.year:
return f"{s:%b} {s.day}, {s.year}–{e:%b} {e.day}, {e.year}"
if s.month == e.month:
return f"{s:%b} {s.day}–{e.day}"
return f"{s:%b} {s.day}–{e:%b} {e.day}"
def epoch_to_date(value) -> str | None:
"""Epoch seconds -> 'YYYY-MM-DD' (None if not a number)."""
try:
return datetime.fromtimestamp(float(value)).strftime("%Y-%m-%d")
except (TypeError, ValueError, OverflowError, OSError):
return None
def mealie_recipe_url(slug: str) -> str | None:
"""Recipe page URL for Mealie v2 (`/g/<group-slug>/r/<slug>`)."""
base_url = os.getenv("MEALIE_BASE_URL", "").rstrip("/")
if not base_url:
return None
group = os.getenv("MEALIE_GROUP_SLUG", "") or "home"
return f"{base_url}/g/{quote(group)}/r/{quote(slug)}"
def mealie_source_ref(slug: str) -> dict:
ref = {"system": "mealie", "id": slug}
url = mealie_recipe_url(slug)
if url:
ref["url"] = url
return ref
def obsidian_source_ref(path: str) -> dict:
"""Obsidian ref; the obsidian:// URL needs OBSIDIAN_VAULT_NAME."""
ref = {"system": "obsidian", "path": path}
vault = os.getenv("OBSIDIAN_VAULT_NAME", "")
if vault:
ref["url"] = "obsidian://open?" + urlencode(
{"vault": vault, "file": path}, quote_via=quote
)
return ref
def note_title(path: str, frontmatter: dict | None = None) -> str:
"""Frontmatter `title` if present, else the filename without extension."""
title = (frontmatter or {}).get("title")
if isinstance(title, str) and title.strip():
return title.strip()
return Path(path).stem or path
def vault_relative_path(filepath: str) -> str:
"""Strip OBSIDIAN_VAULT_PATH from an absolute indexed filepath."""
vault = os.getenv("OBSIDIAN_VAULT_PATH", "/app/data/obsidian")
try:
return str(Path(filepath).relative_to(vault))
except ValueError:
return filepath
def search_documents_artifact(query: str, source: str, docs) -> dict | None:
"""Build a `search.documents` artifact from vector-store Documents."""
if not docs:
return None
results = []
for doc in docs:
meta = doc.metadata or {}
result = {"snippet": snippet(doc.page_content)}
if source == "obsidian":
filepath = meta.get("filepath") or ""
path = vault_relative_path(filepath) if filepath else ""
result["title"] = note_title(path, meta) if path else "Untitled note"
if path:
result["path"] = path
created = meta.get("created_at")
if isinstance(created, str) and created:
result["date"] = created[:10]
else:
filename = meta.get("filename") or ""
result["title"] = Path(filename).stem or filename or "Document"
if filename:
result["filename"] = filename
if meta.get("document_type"):
result["document_type"] = str(meta["document_type"])
created = epoch_to_date(meta.get("created_date"))
if created:
result["date"] = created
results.append(result)
return make_artifact(
"search.documents",
f"Search: {query}",
{"query": query, "source": source, "results": results},
)
docker-compose.yml
+2
View File
@@ -27,6 +27,7 @@ x-raggr-common: &raggr-common
- YNAB_SYNC_INTERVAL=${YNAB_SYNC_INTERVAL:-900}
- MEALIE_BASE_URL=${MEALIE_BASE_URL}
- MEALIE_API_TOKEN=${MEALIE_API_TOKEN}
- MEALIE_GROUP_SLUG=${MEALIE_GROUP_SLUG:-home}
- TWILIO_ACCOUNT_SID=${TWILIO_ACCOUNT_SID}
- TWILIO_AUTH_TOKEN=${TWILIO_AUTH_TOKEN}
- TWILIO_WHATSAPP_NUMBER=${TWILIO_WHATSAPP_NUMBER}
@@ -40,6 +41,7 @@ x-raggr-common: &raggr-common
- OBSIDIAN_DEVICE_NAME=${OBSIDIAN_DEVICE_NAME}
- OBSIDIAN_CONTINUOUS_SYNC=${OBSIDIAN_CONTINUOUS_SYNC:-false}
- OBSIDIAN_VAULT_PATH=${OBSIDIAN_VAULT_PATH:-/app/data/obsidian}
- OBSIDIAN_VAULT_NAME=${OBSIDIAN_VAULT_NAME:-}
- S3_ENDPOINT_URL=${S3_ENDPOINT_URL}
- S3_ACCESS_KEY_ID=${S3_ACCESS_KEY_ID}
- S3_SECRET_ACCESS_KEY=${S3_SECRET_ACCESS_KEY}
tests/unit/test_tool_artifacts.py
+591
View File
@@ -0,0 +1,591 @@
"""Tests for tools returning (text, artifact) via content_and_artifact.
The text half must stay identical to the pre-artifact output (the LLM's view),
and the artifact must match the contract in docs/cowork-ui-plan.md.
"""
import datetime
import json
import os
from unittest.mock import AsyncMock, MagicMock
import pytest
from langchain_core.documents import Document
from langchain_core.messages import ToolMessage
os.environ.setdefault("TAVILY_API_KEY", "dummy")
from blueprints.conversation import agents # noqa: E402
from blueprints.conversation.artifacts_util import ( # noqa: E402
NOTE_MAX_CHARS,
SNIPPET_MAX_CHARS,
format_date_range,
search_documents_artifact,
)
from utils.obsidian_service import ObsidianService # noqa: E402
KIND_DATA_KEYS = {
"mealie.recipe": {
"slug",
"name",
"description",
"servings",
"prep_time",
"cook_time",
"total_time",
"ingredients",
"instructions",
"tags",
},
"mealie.meal_plan": {"start_date", "end_date", "days"},
"mealie.shopping_lists": {"lists"},
"obsidian.note": {"path", "title", "markdown", "tags", "frontmatter"},
"search.documents": {"query", "source", "results"},
}
async def call(tool, args=None) -> ToolMessage:
"""Invoke a tool the way the agent does: with a ToolCall dict."""
msg = await tool.ainvoke(
{"type": "tool_call", "id": "call-1", "name": tool.name, "args": args or {}}
)
assert isinstance(msg, ToolMessage)
return msg
def assert_artifact(artifact, kind):
assert artifact is not None
assert artifact["kind"] == kind
assert isinstance(artifact["title"], str) and artifact["title"]
assert set(artifact["data"]) == KIND_DATA_KEYS[kind]
json.dumps(artifact)
@pytest.fixture
def mealie(monkeypatch):
service = MagicMock()
monkeypatch.setattr(agents, "mealie_enabled", True)
monkeypatch.setattr(agents, "mealie_service", service, raising=False)
monkeypatch.setenv("MEALIE_BASE_URL", "http://mealie.local:9000/")
monkeypatch.delenv("MEALIE_GROUP_SLUG", raising=False)
return service
@pytest.fixture
def obsidian(monkeypatch, tmp_path):
(tmp_path / "seed.md").write_text("# seed")
monkeypatch.setenv("OBSIDIAN_VAULT_PATH", str(tmp_path))
monkeypatch.delenv("OBSIDIAN_VAULT_NAME", raising=False)
service = ObsidianService()
monkeypatch.setattr(agents, "obsidian_enabled", True)
monkeypatch.setattr(agents, "obsidian_service", service, raising=False)
return service
# --- Mealie -----------------------------------------------------------------
RECIPE = {
"name": "Chicken Tikka Masala",
"slug": "chicken-tikka-masala",
"description": "Creamy and spiced.",
"ingredients": [{"display": "1 lb chicken"}, {"display": "1 cup yogurt"}],
"instructions": [
{"title": "Marinate", "text": "Mix chicken and yogurt."},
{"title": "", "text": "Cook it."},
],
"prep_time": "20 minutes",
"cook_time": "30 minutes",
"total_time": "",
"servings": "4",
"tags": ["Indian", "Dinner"],
}
RECIPE_TEXT = """**Chicken Tikka Masala**
Creamy and spiced.
Servings: 4
Prep: 20 minutes | Cook: 30 minutes
Ingredients:
- 1 lb chicken
- 1 cup yogurt
Instructions:
1. Mix chicken and yogurt.
2. Cook it.
Tags: Indian, Dinner"""
async def test_mealie_get_recipe(mealie):
mealie.get_recipe = AsyncMock(return_value=RECIPE)
msg = await call(agents.mealie_get_recipe, {"recipe_slug": "chicken-tikka-masala"})
assert msg.content == RECIPE_TEXT
art = msg.artifact
assert_artifact(art, "mealie.recipe")
assert art["title"] == "Chicken Tikka Masala"
assert art["data"]["ingredients"] == ["1 lb chicken", "1 cup yogurt"]
assert art["data"]["instructions"] == [
{"title": "Marinate", "text": "Mix chicken and yogurt."},
{"text": "Cook it."},
]
assert art["source_ref"] == {
"system": "mealie",
"id": "chicken-tikka-masala",
"url": "http://mealie.local:9000/g/home/r/chicken-tikka-masala",
}
async def test_mealie_recipe_url_uses_group_slug(mealie, monkeypatch):
monkeypatch.setenv("MEALIE_GROUP_SLUG", "family")
mealie.get_recipe = AsyncMock(return_value=RECIPE)
msg = await call(agents.mealie_get_recipe, {"recipe_slug": "chicken-tikka-masala"})
assert msg.artifact["source_ref"]["url"].endswith(
"/g/family/r/chicken-tikka-masala"
)
async def test_mealie_get_recipe_error(mealie):
mealie.get_recipe = AsyncMock(side_effect=Exception("404"))
msg = await call(agents.mealie_get_recipe, {"recipe_slug": "nope"})
assert msg.content == "Error fetching recipe: 404"
assert msg.artifact is None
MEAL_PLANS = {
"start_date": "2026-10-12",
"end_date": "2026-10-19",
"total_plans": 3,
"plans_by_date": {
"2026-10-13": [
{
"entry_type": "dinner",
"title": "",
"recipe_name": "Tacos",
"recipe_slug": "tacos",
"note": "",
}
],
"2026-10-12": [
{
"entry_type": "lunch",
"title": "Leftovers",
"recipe_name": "",
"recipe_slug": "",
"note": "from Sunday",
},
{
"entry_type": "dinner",
"title": "",
"recipe_name": "Chicken Tikka Masala",
"recipe_slug": "chicken-tikka-masala",
"note": "",
},
],
},
}
MEAL_PLANS_TEXT = """Meal plans from 2026-10-12 to 2026-10-19 (3 meals):
2026-10-12:
- Lunch: Leftovers
Note: from Sunday
- Dinner: Chicken Tikka Masala
2026-10-13:
- Dinner: Tacos"""
async def test_mealie_meal_plans(mealie):
mealie.get_meal_plans = AsyncMock(return_value=MEAL_PLANS)
msg = await call(agents.mealie_meal_plans, {})
assert msg.content == MEAL_PLANS_TEXT
art = msg.artifact
assert_artifact(art, "mealie.meal_plan")
assert art["title"] == "Meal plan · Oct 12–19"
assert [d["date"] for d in art["data"]["days"]] == ["2026-10-12", "2026-10-13"]
assert art["data"]["days"][0]["meals"] == [
{"entry_type": "lunch", "name": "Leftovers", "note": "from Sunday"},
{
"entry_type": "dinner",
"name": "Chicken Tikka Masala",
"recipe_slug": "chicken-tikka-masala",
},
]
async def test_mealie_meal_plans_empty(mealie):
mealie.get_meal_plans = AsyncMock(
return_value={
"start_date": "2026-10-12",
"end_date": "2026-10-19",
"total_plans": 0,
"plans_by_date": {},
}
)
msg = await call(agents.mealie_meal_plans, {})
assert msg.content == "No meal plans found from 2026-10-12 to 2026-10-19."
assert msg.artifact is None
async def test_mealie_todays_meals(mealie):
mealie.get_todays_meals = AsyncMock(
return_value={
"today": "2026-10-12",
"total_meals": 1,
"meals": [
{
"entry_type": "dinner",
"title": "",
"recipe_name": "Tacos",
"recipe_slug": "tacos",
"note": "extra salsa",
}
],
}
)
msg = await call(agents.mealie_todays_meals)
assert msg.content == (
"Meals planned for today (2026-10-12):\n- Dinner: Tacos\n Note: extra salsa"
)
art = msg.artifact
assert_artifact(art, "mealie.meal_plan")
assert art["title"] == "Today's meals · Oct 12"
assert art["data"] == {
"start_date": "2026-10-12",
"end_date": "2026-10-12",
"days": [
{
"date": "2026-10-12",
"meals": [
{
"entry_type": "dinner",
"name": "Tacos",
"recipe_slug": "tacos",
"note": "extra salsa",
}
],
}
],
}
async def test_mealie_todays_meals_empty(mealie):
mealie.get_todays_meals = AsyncMock(
return_value={"today": "2026-10-12", "total_meals": 0, "meals": []}
)
msg = await call(agents.mealie_todays_meals)
assert msg.content == "No meals planned for today (2026-10-12)."
assert msg.artifact is None
async def test_mealie_shopping_lists(mealie):
mealie.get_shopping_lists = AsyncMock(
return_value={
"total_lists": 1,
"total_unchecked_items": 2,
"shopping_lists": [
{
"name": "Groceries",
"unchecked_items": [
{"name": "Milk", "quantity": 1, "note": "", "food": None},
{"name": "Eggs", "quantity": 12.0, "note": "", "food": None},
],
"checked_items": [
{"name": "Bread", "quantity": 1, "note": "", "food": None}
],
}
],
}
)
msg = await call(agents.mealie_shopping_lists)
assert msg.content == (
"Shopping lists (2 items needed):\n"
"\n**Groceries** (2 items needed)\n"
" - [ ] Milk\n"
" - [ ] Eggs x12.0\n"
" (1 items already checked off)"
)
art = msg.artifact
assert_artifact(art, "mealie.shopping_lists")
assert art["data"]["lists"] == [
{
"name": "Groceries",
"items": [
{"name": "Milk", "quantity": 1, "checked": False},
{"name": "Eggs", "quantity": 12.0, "checked": False},
{"name": "Bread", "quantity": 1, "checked": True},
],
}
]
async def test_mealie_shopping_lists_empty(mealie):
mealie.get_shopping_lists = AsyncMock(
return_value={
"total_lists": 0,
"total_unchecked_items": 0,
"shopping_lists": [],
}
)
msg = await call(agents.mealie_shopping_lists)
assert msg.content == "No shopping lists found."
assert msg.artifact is None
@pytest.mark.parametrize(
"tool,args",
[
(agents.mealie_get_recipe, {"recipe_slug": "x"}),
(agents.mealie_meal_plans, {}),
(agents.mealie_todays_meals, {}),
(agents.mealie_shopping_lists, {}),
],
)
async def test_mealie_not_configured(monkeypatch, tool, args):
monkeypatch.setattr(agents, "mealie_enabled", False)
msg = await call(tool, args)
assert msg.content == (
"Mealie integration is not configured. Please set MEALIE_BASE_URL and "
"MEALIE_API_TOKEN environment variables."
)
assert msg.artifact is None
# --- Obsidian ---------------------------------------------------------------
NOTE = """---
title: Vet visit
date: 2026-10-01
tags: [simba]
---
Simba saw the [[Dr Smith|vet]] today. #simba #health #simba
"""
async def test_obsidian_read_note(obsidian, monkeypatch):
monkeypatch.setenv("OBSIDIAN_VAULT_NAME", "Home Vault")
os.makedirs(os.path.join(obsidian.vault_path, "notes"))
with open(os.path.join(obsidian.vault_path, "notes", "vet visit.md"), "w") as f:
f.write(NOTE)
msg = await call(agents.obsidian_read_note, {"relative_path": "notes/vet visit.md"})
assert msg.content == (
"File: notes/vet visit.md\n\n"
"Frontmatter:\n{'title': 'Vet visit', 'date': datetime.date(2026, 10, 1), "
"'tags': ['simba']}\n\n"
"Content:\nSimba saw the vet today. #simba #health #simba\n\n"
"Tags: simba, health, simba\n"
"Contains 1 wikilinks and 0 embeds"
)
art = msg.artifact
assert_artifact(art, "obsidian.note")
assert art["title"] == "Vet visit"
assert art["data"]["frontmatter"] == {
"title": "Vet visit",
"date": "2026-10-01",
"tags": ["simba"],
}
assert art["data"]["tags"] == ["simba", "health"]
assert art["data"]["markdown"] == "Simba saw the vet today. #simba #health #simba"
assert art["source_ref"] == {
"system": "obsidian",
"path": "notes/vet visit.md",
"url": "obsidian://open?vault=Home%20Vault&file=notes%2Fvet%20visit.md",
}
async def test_obsidian_read_note_caps_markdown_and_omits_url(obsidian):
with open(os.path.join(obsidian.vault_path, "big.md"), "w") as f:
f.write("x" * (NOTE_MAX_CHARS + 500))
msg = await call(agents.obsidian_read_note, {"relative_path": "big.md"})
art = msg.artifact
assert art["title"] == "big"
assert len(art["data"]["markdown"]) <= NOTE_MAX_CHARS
assert art["source_ref"] == {"system": "obsidian", "path": "big.md"}
async def test_obsidian_read_note_missing(obsidian):
msg = await call(agents.obsidian_read_note, {"relative_path": "nope.md"})
assert msg.content == (
"Note not found at 'nope.md'. Please check the path is correct."
)
assert msg.artifact is None
async def test_journal_get_today(obsidian):
path = obsidian.get_daily_note_path()
full = os.path.join(obsidian.vault_path, path)
os.makedirs(os.path.dirname(full), exist_ok=True)
raw = "## Tasks\n- [ ] Feed Simba\n"
with open(full, "w") as f:
f.write(raw)
today = datetime.datetime.now().strftime("%Y-%m-%d")
msg = await call(agents.journal_get_today)
assert msg.content == f"Daily note for {today}:\n\n{raw}"
art = msg.artifact
assert_artifact(art, "obsidian.note")
assert art["title"].startswith("Daily note · ")
assert art["data"]["path"] == path
assert art["data"]["markdown"] == "## Tasks\n- [ ] Feed Simba"
async def test_journal_get_today_missing(obsidian):
today = datetime.datetime.now().strftime("%Y-%m-%d")
msg = await call(agents.journal_get_today)
assert msg.content == (
f"No daily note found for {today}. Use journal_add_task to create one."
)
assert msg.artifact is None
async def test_obsidian_not_configured(monkeypatch):
monkeypatch.setattr(agents, "obsidian_enabled", False)
for tool, args in [
(agents.obsidian_read_note, {"relative_path": "a.md"}),
(agents.obsidian_search_notes, {"query": "a"}),
]:
msg = await call(tool, args)
assert msg.content == (
"Obsidian integration is not configured. Please set "
"OBSIDIAN_VAULT_PATH environment variable."
)
assert msg.artifact is None
msg = await call(agents.journal_get_today)
assert msg.content == "Obsidian integration is not configured."
assert msg.artifact is None
# --- Search -----------------------------------------------------------------
PAPERLESS_DOCS = [
Document(
page_content="Rabies vaccine\nadministered. " + "y" * 600,
metadata={
"source": "paperless",
"created_date": 1760227200.0,
"filename": "vet_invoice.pdf",
"document_type": "Invoice",
},
)
]
async def test_simba_search(monkeypatch):
query_mock = AsyncMock(return_value=("Source: {...}\nContent: x", PAPERLESS_DOCS))
monkeypatch.setattr(agents, "query_vector_store", query_mock)
msg = await call(agents.simba_search, {"query": "vet records"})
assert msg.content == "Source: {...}\nContent: x"
query_mock.assert_awaited_once_with(query="vet records", source="paperless")
art = msg.artifact
assert_artifact(art, "search.documents")
assert art["title"] == "Search: vet records"
assert art["data"]["source"] == "paperless"
(result,) = art["data"]["results"]
assert result["title"] == "vet_invoice"
assert result["filename"] == "vet_invoice.pdf"
assert result["document_type"] == "Invoice"
assert len(result["date"]) == 10
assert result["snippet"].startswith("Rabies vaccine administered. ")
assert len(result["snippet"]) <= SNIPPET_MAX_CHARS
async def test_simba_search_empty(monkeypatch):
monkeypatch.setattr(agents, "query_vector_store", AsyncMock(return_value=("", [])))
msg = await call(agents.simba_search, {"query": "nothing"})
assert msg.content == ""
assert msg.artifact is None
async def test_obsidian_search_notes(obsidian, monkeypatch):
docs = [
Document(
page_content="Simba's vet is Dr Smith.",
metadata={
"source": "obsidian",
"filepath": os.path.join(obsidian.vault_path, "pets", "simba.md"),
"tags": ["simba"],
"created_at": "2026-01-02T10:00:00",
},
)
]
monkeypatch.setattr(
agents, "query_vector_store", AsyncMock(return_value=("serialized", docs))
)
msg = await call(agents.obsidian_search_notes, {"query": "vet"})
assert msg.content == "serialized"
art = msg.artifact
assert_artifact(art, "search.documents")
assert art["data"]["results"] == [
{
"snippet": "Simba's vet is Dr Smith.",
"title": "simba",
"path": "pets/simba.md",
"date": "2026-01-02",
}
]
async def test_obsidian_search_notes_error(obsidian, monkeypatch):
monkeypatch.setattr(
agents, "query_vector_store", AsyncMock(side_effect=Exception("db down"))
)
msg = await call(agents.obsidian_search_notes, {"query": "vet"})
assert msg.content == "Error searching Obsidian notes: db down"
assert msg.artifact is None
def test_search_documents_artifact_empty():
assert search_documents_artifact("q", "paperless", []) is None
@pytest.mark.parametrize(
"start,end,expected",
[
("2026-10-12", "2026-10-19", "Oct 12–19"),
("2026-10-28", "2026-11-03", "Oct 28–Nov 3"),
("2026-12-28", "2027-01-03", "Dec 28, 2026–Jan 3, 2027"),
("2026-10-12", "2026-10-12", "Oct 12"),
("bad", "2026-10-12", "bad–2026-10-12"),
],
)
def test_format_date_range(start, end, expected):
assert format_date_range(start, end) == expected