Compare commits

..
Author SHA1 Message Date
Brooklyn Nicholson 2e1dc5c4b8 feat(skills): add optional Rive MCP skill
Add an optional creative/rive-mcp skill for Rive animation workflows. The skill
supports the official desktop Rive HTTP MCP at 127.0.0.1:9791 and documents the
third-party headless RiveMCP stdio path for .riv/.rev generation.

The skill stays opt-in under optional-skills and ships only a small preflight
doctor plus research notes; no core tool or dependency is added.
2026-06-25 15:34:19 -05:00
10 changed files with 336 additions and 564 deletions
-144
View File
@@ -1,144 +0,0 @@
---
name: ableton
description: "Control Ableton Live through AbletonMCP."
version: 1.0.0
author: Hermes Agent
license: MIT
platforms: [macos, windows]
metadata:
hermes:
category: creative
tags: [Ableton, Music, DAW, MCP, MIDI, Audio, Production]
related_skills: []
prerequisites:
commands: [hermes, uvx]
---
# AbletonMCP Skill
Optional skill — **not active until installed**:
```bash
hermes skills install official/creative/ableton
```
After install, scripts live under `~/.hermes/skills/creative/ableton/`.
In prose below, `{SKILL_DIR}` means that directory (or this repo path before
install).
Connect Hermes to Ableton Live through
[ahujasid/ableton-mcp](https://github.com/ahujasid/ableton-mcp): a stdio MCP
server (`uvx ableton-mcp`) that talks to an Ableton MIDI Remote Script over a
local socket.
## When to Use
- The user wants Hermes to inspect or control an Ableton Live session.
- The user wants tracks, clips, MIDI notes, instruments/effects, transport, or
Arrangement View edits driven through MCP tools.
- The user has Ableton Live 10+ installed and can configure a MIDI Remote
Script.
## Prerequisites
- Ableton Live 10 or newer.
- `uv` / `uvx`.
- AbletonMCP Remote Script installed inside Ableton.
- Telemetry opt-out configured for the MCP server.
Run the doctor first:
```bash
python {SKILL_DIR}/scripts/ableton_doctor.py
```
## How to Run
### 1. Install the Ableton Remote Script
Download upstream `AbletonMCP_Remote_Script/__init__.py` from
<https://github.com/ahujasid/ableton-mcp> and place it at:
```text
<Ableton Remote Scripts>/AbletonMCP/__init__.py
```
Common locations:
- macOS app bundle:
`/Applications/Ableton Live*.app/Contents/App-Resources/MIDI Remote Scripts/`
- macOS user preferences:
`~/Library/Preferences/Ableton/Live XX/User Remote Scripts/`
- Windows user preferences:
`C:\Users\<User>\AppData\Roaming\Ableton\Live x.x.x\Preferences\User Remote Scripts`
- Windows install folders:
`C:\ProgramData\Ableton\Live XX\Resources\MIDI Remote Scripts\`
or `C:\Program Files\Ableton\Live XX\Resources\MIDI Remote Scripts\`
Then launch/restart Ableton Live, open Preferences → Link, Tempo & MIDI, select
`AbletonMCP` as a Control Surface, and set Input/Output to `None`.
### 2. Add the MCP server to Hermes
Use upstream's canonical stdio command (`uvx ableton-mcp`) and disable upstream
telemetry by default:
```bash
hermes mcp add ableton --command uvx --env ABLETON_MCP_DISABLE_TELEMETRY=true --args ableton-mcp
```
Only run one AbletonMCP server instance at a time (Hermes, Claude Desktop, or
Cursor — not multiple). Start a new Hermes session after adding it.
### 3. Use the tools
Ask Hermes for session/track info before making changes. Keep edits small:
- "Get information about the current Ableton session."
- "Create a 4-bar MIDI clip with a simple melody."
- "Set the tempo to 120 BPM."
- "Load a drum rack, then create a basic kick/snare pattern."
## Quick Reference
| Goal | Command / action |
| --- | --- |
| Check setup | `python {SKILL_DIR}/scripts/ableton_doctor.py` |
| Add Hermes MCP | `hermes mcp add ableton --command uvx --env ABLETON_MCP_DISABLE_TELEMETRY=true --args ableton-mcp` |
| Test tools | `hermes mcp test ableton` |
| Reconfigure tool subset | `hermes mcp configure ableton` |
| Disable telemetry | `ABLETON_MCP_DISABLE_TELEMETRY=true` |
## Procedure
1. Run `ableton_doctor.py`.
2. Install / enable the Ableton Remote Script if the socket is not reachable.
3. Add the Hermes MCP server with telemetry disabled.
4. Restart Hermes so tools load.
5. Inspect session state first (`get_session_info`, `get_track_info`) before
creating or modifying clips.
6. Save the Ableton set before broad arrangement generation.
## Pitfalls
- **Manual Ableton setup is required.** `uvx ableton-mcp` alone is not enough;
Ableton must load the Remote Script.
- **Only one MCP client instance.** Upstream warns not to run Cursor and Claude
Desktop and Hermes against the same server at once.
- **Telemetry exists upstream.** Always pass
`ABLETON_MCP_DISABLE_TELEMETRY=true` unless the user explicitly wants to
enable it.
- **It edits the live set.** Save before broad generation and keep early
requests narrow.
- **Browser paths can be large.** Ask for specific categories/paths before
loading instruments/effects.
## Verification
- `ableton_doctor.py` shows `uvx` present and socket `127.0.0.1:9877`
reachable after Ableton loads the Remote Script.
- `hermes mcp test ableton` connects and lists tools.
- A harmless read-only call (`get_session_info`) works before any mutating
operation.
See `references/research.md` for source links, tool names, and setup details.
@@ -1,103 +0,0 @@
# AbletonMCP research
Primary source: <https://github.com/ahujasid/ableton-mcp>
## Shape
AbletonMCP has two parts:
1. `AbletonMCP_Remote_Script/__init__.py` — Ableton MIDI Remote Script. It
creates a local socket server inside Ableton Live.
2. `MCP_Server/server.py` — Python stdio MCP server. It connects MCP tool calls
to the Remote Script.
Upstream's canonical MCP command is:
```json
{
"mcpServers": {
"AbletonMCP": {
"command": "uvx",
"args": ["ableton-mcp"]
}
}
}
```
For Hermes:
```bash
hermes mcp add ableton --command uvx --env ABLETON_MCP_DISABLE_TELEMETRY=true --args ableton-mcp
```
`--env` must appear before `--args` because Hermes's `--args` consumes the
remaining command line.
## Remote Script setup
Upstream locations:
- macOS:
- `Contents/App-Resources/MIDI Remote Scripts/` inside the Ableton app bundle
- `~/Library/Preferences/Ableton/Live XX/User Remote Scripts`
- Windows:
- `C:\Users\<User>\AppData\Roaming\Ableton\Live x.x.x\Preferences\User Remote Scripts`
- `C:\ProgramData\Ableton\Live XX\Resources\MIDI Remote Scripts\`
- `C:\Program Files\Ableton\Live XX\Resources\MIDI Remote Scripts\`
Create `AbletonMCP/` under the Remote Scripts dir, copy `__init__.py`, restart
Ableton, then Preferences → Link, Tempo & MIDI → Control Surface:
`AbletonMCP`, Input/Output: `None`.
The Remote Script socket is observed in community docs as `127.0.0.1:9877`.
## Tool surface
Tool names from upstream `MCP_Server/server.py` include:
- `get_session_info`
- `get_track_info`
- `create_midi_track`
- `set_track_name`
- `create_clip`
- `create_audio_clip`
- `add_notes_to_clip`
- `set_clip_name`
- `set_tempo`
- `load_instrument_or_effect`
- `fire_clip`
- `stop_clip`
- `stop_playback`
- `get_browser_tree`
- `get_browser_items_at_path`
- `load_drum_kit`
- `set_arrangement_time`
- `get_arrangement_clips`
- `duplicate_to_arrangement`
Capabilities advertised by upstream: track manipulation, MIDI/audio clips,
instruments/effects, Arrangement View composition, session/transport control.
## Telemetry
Upstream collects anonymous telemetry unless disabled. Disable by default in
Hermes instructions with one of:
- `ABLETON_MCP_DISABLE_TELEMETRY=true`
- `DISABLE_TELEMETRY=true`
- `MCP_DISABLE_TELEMETRY=true`
The skill uses `ABLETON_MCP_DISABLE_TELEMETRY=true`.
## GitHub usage checks
Observed real-world usage agrees that `uvx ableton-mcp` is the canonical entry:
- `ahujasid/ableton-mcp`
- `dnkrow/claude-code-ableton-mac`
- `ParkerRex/synthia`
- `patrickking67/producer`
Some forks warn about package/version confusion, but for the upstream repo the
published PyPI package exposes console script `ableton-mcp =
MCP_Server.server:main`.
@@ -1,95 +0,0 @@
#!/usr/bin/env python3
"""Preflight checks for AbletonMCP.
Checks the lazy runtime bits (`uvx`) and the Ableton Remote Script socket. The
script never installs packages and never starts the MCP server.
"""
from __future__ import annotations
import argparse
import json
import os
import shutil
import socket
from pathlib import Path
ABLETON_HOST = "127.0.0.1"
ABLETON_PORT = 9877
def _port_open(host: str, port: int, timeout: float = 0.35) -> bool:
try:
with socket.create_connection((host, port), timeout=timeout):
return True
except OSError:
return False
def _mac_remote_script_candidates(home: Path) -> list[str]:
root = home / "Library" / "Preferences" / "Ableton"
if not root.exists():
return []
return [
str(p / "User Remote Scripts" / "AbletonMCP")
for p in sorted(root.glob("Live *"))
if p.is_dir()
]
def check(*, which=shutil.which, port_open=_port_open, home=None) -> dict:
home_path = Path(home or Path.home())
telemetry_disabled = any(
os.environ.get(key, "").lower() == "true"
for key in (
"ABLETON_MCP_DISABLE_TELEMETRY",
"DISABLE_TELEMETRY",
"MCP_DISABLE_TELEMETRY",
)
)
return {
"uvx": bool(which("uvx")),
"remote_script_socket": {
"host": ABLETON_HOST,
"port": ABLETON_PORT,
"reachable": port_open(ABLETON_HOST, ABLETON_PORT),
},
"telemetry_disabled_in_env": telemetry_disabled,
"mac_remote_script_candidates": _mac_remote_script_candidates(home_path),
}
def _summary(s: dict) -> str:
sock = s["remote_script_socket"]
lines = [
"✓ uvx found" if s["uvx"] else "✗ uvx not found — install uv first",
(
f"✓ Ableton Remote Script socket: {sock['host']}:{sock['port']} reachable"
if sock["reachable"]
else f"✗ Ableton Remote Script socket: {sock['host']}:{sock['port']} not reachable"
),
(
"✓ telemetry disabled in current env"
if s["telemetry_disabled_in_env"]
else "⚠ telemetry not disabled in current env; configure MCP env opt-out"
),
]
candidates = s["mac_remote_script_candidates"]
if candidates:
lines.append("macOS candidate Remote Script dirs:")
lines.extend(f" - {p}" for p in candidates)
return "\n".join(lines)
def main(argv=None) -> int:
ap = argparse.ArgumentParser(description="Check AbletonMCP prerequisites.")
ap.add_argument("--json", action="store_true", help="Emit JSON")
args = ap.parse_args(argv)
status = check()
print(json.dumps(status, indent=2) if args.json else _summary(status))
return 0 if status["uvx"] else 1
if __name__ == "__main__":
raise SystemExit(main())
+140
View File
@@ -0,0 +1,140 @@
---
name: rive-mcp
description: "Create and edit Rive animations through MCP."
version: 1.0.0
author: Hermes Agent
license: MIT
platforms: [linux, macos, windows]
metadata:
hermes:
category: creative
tags: [Rive, Animation, MCP, Motion, State Machines, Design-to-Code]
related_skills: [blender-mcp]
prerequisites:
commands: [hermes]
---
# Rive Skill
Optional skill — **not active until installed**:
```bash
hermes skills install official/creative/rive-mcp
```
After install, scripts live under `~/.hermes/skills/creative/rive-mcp/`. In prose
below, `{SKILL_DIR}` means that directory (or this repo path before install).
Use Rive from Hermes through MCP. There are two real integration paths:
- **Official Rive desktop MCP** — local HTTP server in the Rive desktop editor
at `http://127.0.0.1:9791/mcp`.
- **RiveMCP** — third-party standalone stdio MCP (`npx -y rivemcp`) that writes
`.riv` / `.rev` files headlessly.
Do not invent a Rive CLI export path. Official Rive exports runtime `.riv`
files from the editor UI; official MCP requires the desktop editor to be open.
## When to Use
- The user wants to create or edit Rive artboards, shapes, animations, state
machines, View Models, scripts, shaders, or runtime `.riv` files.
- The user already has the Rive desktop app open and wants AI to manipulate the
current file.
- The user wants headless `.riv` / `.rev` generation and accepts RiveMCP's
third-party licensing model.
## Prerequisites
Run the doctor first:
```bash
python {SKILL_DIR}/scripts/rive_doctor.py
```
Official path:
- Rive desktop editor / Early Access app on macOS or Windows.
- Rive app open, with a file and artboard created.
- MCP endpoint reachable at `http://127.0.0.1:9791/mcp`.
RiveMCP path:
- `node` / `npx`, or a downloaded RiveMCP binary.
- RiveMCP has 3 free exports per machine; after that it requires
`RIVEMCP_LICENSE_KEY`.
## How to Run
### Path A — Official Rive desktop MCP
Use this when the user is working in the Rive editor and wants live changes.
```bash
hermes mcp add rive --url http://127.0.0.1:9791/mcp
```
When prompted whether the server requires auth, answer **no**. The Rive app
must be open before probing. Start a new Hermes session after adding the MCP so
the tools load.
Rive's official docs say to finish a prompt by typing `End Prompt` so the Rive
editor applies the AI changes. Treat this as part of the interaction contract.
### Path B — RiveMCP headless stdio
Use this when the user wants automation, CI, or file generation without the
Rive desktop app.
```bash
hermes mcp add rivemcp --command npx --args -y rivemcp
```
If the user has a paid key:
```bash
export RIVEMCP_LICENSE_KEY=...
hermes mcp add rivemcp --command npx --env RIVEMCP_LICENSE_KEY=$RIVEMCP_LICENSE_KEY --args -y rivemcp
```
Start a new Hermes session after adding the MCP.
## Quick Reference
| Goal | Path |
| --- | --- |
| Live-edit current Rive file | official `rive` HTTP MCP |
| Headless `.riv` / `.rev` creation | third-party `rivemcp` stdio MCP |
| Inspect official endpoint | `python {SKILL_DIR}/scripts/rive_doctor.py` |
| Configure official MCP | `hermes mcp add rive --url http://127.0.0.1:9791/mcp` |
| Configure RiveMCP | `hermes mcp add rivemcp --command npx --args -y rivemcp` |
## Procedure
1. Ask which path the user wants: **official desktop** or **headless RiveMCP**.
2. Run `rive_doctor.py`.
3. Configure the selected MCP using the commands above.
4. Start a new Hermes session so MCP tools load.
5. For official Rive, keep the Rive app open and use `End Prompt` when the
editor asks for confirmation.
6. For RiveMCP, export `.riv` for runtime or `.rev` for editor handoff.
## Pitfalls
- **Official Rive MCP is desktop-bound.** It is not a headless server and is
currently documented for macOS / Windows desktop editor only.
- **RiveMCP is third-party and licensed.** It offers free exports, then needs a
license key. Make that explicit before choosing it.
- **Do not confuse `.riv` and `.rev`.** `.riv` is runtime output; `.rev` opens
in the Rive editor for further manual editing.
- **Tool surfaces evolve.** After connecting, inspect the actual MCP tools with
`hermes mcp configure rive` or `hermes mcp configure rivemcp`.
## Verification
- Official path: `rive_doctor.py` shows port `9791` reachable, and Hermes MCP
probe lists Rive tools.
- RiveMCP path: Hermes MCP probe lists RiveMCP tools, then export a trivial
`.riv` and open it in a Rive runtime/editor.
See `references/research.md` for source links and verified constraints.
@@ -0,0 +1,60 @@
# Rive integration research
## Official Rive desktop MCP
Source: <https://rive.mintlify.dev/docs/editor/ai/mcp>
- Official MCP is currently documented as available in the desktop Editor for
Windows and macOS.
- The endpoint is local HTTP:
```json
{
"mcpServers": {
"rive": {
"url": "http://127.0.0.1:9791/mcp"
}
}
}
```
- Rive desktop / Early Access app must be open.
- User should have a Rive file open and an artboard created.
- Rive docs describe a confirmation flow: after prompt processing, type
`End Prompt` to allow changes.
- Supported feature areas include files/artboards, scene hierarchy edits,
shapes/layout/components/assets, animations, state machines, View Models,
data binding, Luau scripts, and WGSL shaders.
## RiveMCP
Source: <https://github.com/paradoxsyn/rivemcp-releases>
- Third-party standalone MCP server for programmatically creating/editing Rive
animations.
- Recommended config:
```json
{
"mcpServers": {
"rivemcp": {
"command": "npx",
"args": ["-y", "rivemcp"]
}
}
}
```
- Exposes 139+ tools across project, shapes, styling, animation, state
machines, text, images, bones, nested artboards, events, data binding,
advanced layout/scroll/audio/scripting, HLAPI, sprites, import/export, edit.
- Can export runtime `.riv` and editor `.rev`.
- 3 free exports per machine; license required after that via
`RIVEMCP_LICENSE_KEY`.
## Non-paths
- Official Rive runtime export is an editor action, not an official CLI export
command.
- Do not claim headless official Rive support; use RiveMCP if the user needs
headless generation.
@@ -0,0 +1,76 @@
#!/usr/bin/env python3
"""Preflight checks for the optional Rive MCP skill.
No installs, no MCP process spawning. Reports whether the official Rive
desktop MCP port is reachable and whether the optional headless RiveMCP path
has the Node/npx bits needed to launch `npx -y rivemcp`.
"""
from __future__ import annotations
import argparse
import json
import shutil
import socket
RIVE_HOST = "127.0.0.1"
RIVE_PORT = 9791
def _port_open(host: str, port: int, timeout: float = 0.35) -> bool:
try:
with socket.create_connection((host, port), timeout=timeout):
return True
except OSError:
return False
def check(*, which=shutil.which, port_open=_port_open) -> dict:
node, npx, rivemcp = which("node"), which("npx"), which("rivemcp")
return {
"official_rive": {
"url": f"http://{RIVE_HOST}:{RIVE_PORT}/mcp",
"reachable": port_open(RIVE_HOST, RIVE_PORT),
},
"rivemcp": {
"node": bool(node),
"npx": bool(npx),
"global_binary": bool(rivemcp),
"launch": "npx -y rivemcp",
},
}
def _summary(s: dict) -> str:
official, headless = s["official_rive"], s["rivemcp"]
lines = [
(
f"✓ official Rive MCP: {official['url']} reachable"
if official["reachable"]
else f"✗ official Rive MCP: {official['url']} not reachable "
"(open the Rive desktop editor / Early Access app)"
)
]
if headless["npx"]:
lines.append("✓ RiveMCP launch path: npx -y rivemcp")
elif headless["global_binary"]:
lines.append("✓ RiveMCP global binary found: rivemcp")
else:
lines.append("✗ RiveMCP: need npx or a downloaded rivemcp binary")
if not headless["node"] and not headless["global_binary"]:
lines.append("⚠ node not found; npm/npx RiveMCP install path unavailable")
return "\n".join(lines)
def main(argv=None) -> int:
ap = argparse.ArgumentParser(description="Check Rive MCP integration paths.")
ap.add_argument("--json", action="store_true", help="Emit JSON")
args = ap.parse_args(argv)
status = check()
print(json.dumps(status, indent=2) if args.json else _summary(status))
return 0 if status["official_rive"]["reachable"] or status["rivemcp"]["npx"] else 1
if __name__ == "__main__":
raise SystemExit(main())
@@ -1,72 +0,0 @@
"""Tests for the optional Ableton skill."""
from __future__ import annotations
import importlib.util
import re
from pathlib import Path
ROOT = Path(__file__).resolve().parents[2]
SKILL_DIR = ROOT / "optional-skills" / "creative" / "ableton"
def _load(path: Path, name: str):
spec = importlib.util.spec_from_file_location(name, path)
assert spec and spec.loader
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
return mod
def _description(skill_dir: Path) -> str:
text = (skill_dir / "SKILL.md").read_text()
match = re.search(r'^description:\s*"?([^"\n]+)"?\s*$', text, re.MULTILINE)
assert match
return match.group(1)
class TestAbletonSkillShape:
def test_skill_lives_under_optional_skills(self):
assert (SKILL_DIR / "SKILL.md").is_file()
assert not (ROOT / "skills" / "creative" / "ableton").exists()
def test_description_is_catalog_sized(self):
desc = _description(SKILL_DIR)
assert len(desc) <= 60, desc
assert desc.endswith(".")
def test_optional_skill_source_fetches_skill(self):
from tools.skills_hub import OptionalSkillSource
source = OptionalSkillSource()
assert source.fetch("official/creative/ableton").name == "ableton"
class TestAbletonDoctor:
def test_uvx_socket_and_telemetry(self, monkeypatch, tmp_path):
doctor = _load(SKILL_DIR / "scripts" / "ableton_doctor.py", "ableton_doctor")
monkeypatch.setenv("ABLETON_MCP_DISABLE_TELEMETRY", "true")
live = tmp_path / "Library" / "Preferences" / "Ableton" / "Live 12"
live.mkdir(parents=True)
status = doctor.check(
which={"uvx": "/bin/uvx"}.get,
port_open=lambda host, port: (host, port) == ("127.0.0.1", 9877),
home=tmp_path,
)
assert status["uvx"] is True
assert status["remote_script_socket"]["reachable"] is True
assert status["telemetry_disabled_in_env"] is True
assert status["mac_remote_script_candidates"] == [
str(live / "User Remote Scripts" / "AbletonMCP")
]
def test_uvx_missing(self):
doctor = _load(
SKILL_DIR / "scripts" / "ableton_doctor.py", "ableton_doctor_none"
)
status = doctor.check(which=lambda _name: None, port_open=lambda *_a: False)
assert status["uvx"] is False
assert status["remote_script_socket"]["reachable"] is False
@@ -0,0 +1,60 @@
"""Tests for the optional Rive MCP skill."""
from __future__ import annotations
import importlib.util
import re
from pathlib import Path
ROOT = Path(__file__).resolve().parents[2]
SKILL_DIR = ROOT / "optional-skills" / "creative" / "rive-mcp"
def _load(path: Path, name: str):
spec = importlib.util.spec_from_file_location(name, path)
assert spec and spec.loader
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
return mod
def _description(skill_dir: Path) -> str:
text = (skill_dir / "SKILL.md").read_text()
match = re.search(r'^description:\s*"?([^"\n]+)"?\s*$', text, re.MULTILINE)
assert match
return match.group(1)
class TestRiveMcpSkillShape:
def test_skill_lives_under_optional_skills(self):
assert (SKILL_DIR / "SKILL.md").is_file()
assert not (ROOT / "skills" / "creative" / "rive-mcp").exists()
def test_description_is_catalog_sized(self):
desc = _description(SKILL_DIR)
assert len(desc) <= 60, desc
assert desc.endswith(".")
def test_optional_skill_source_fetches_skill(self):
from tools.skills_hub import OptionalSkillSource
source = OptionalSkillSource()
assert source.fetch("official/creative/rive-mcp").name == "rive-mcp"
class TestRiveDoctor:
def test_official_port_reachable(self):
doctor = _load(SKILL_DIR / "scripts" / "rive_doctor.py", "rive_doctor")
status = doctor.check(
which={"node": "/bin/node", "npx": "/bin/npx"}.get,
port_open=lambda host, port: (host, port) == ("127.0.0.1", 9791),
)
assert status["official_rive"]["reachable"] is True
assert status["rivemcp"]["npx"] is True
def test_no_paths_available(self):
doctor = _load(SKILL_DIR / "scripts" / "rive_doctor.py", "rive_doctor_none")
status = doctor.check(which=lambda _name: None, port_open=lambda *_a: False)
assert status["official_rive"]["reachable"] is False
assert status["rivemcp"]["npx"] is False
-133
View File
@@ -1,133 +0,0 @@
"""Tests for defensive type coercion in todo_tool (issue #14185).
Covers three crash patterns:
1. todos is a JSON string instead of a list
2. todos list contains non-dict items (e.g., bare strings)
3. Well-formed input continues to work unchanged
"""
import json
from tools.todo_tool import TodoStore, todo_tool
class TestJsonStringCoercion:
"""Guard 1: todo_tool() recovers when LLM sends todos as a JSON string."""
def test_json_string_is_parsed_into_list(self):
store = TodoStore()
todos_str = json.dumps([
{"id": "t1", "content": "Do A", "status": "pending"},
{"id": "t2", "content": "Do B", "status": "in_progress"},
])
result = json.loads(todo_tool(todos=todos_str, store=store))
assert "error" not in result
assert result["summary"]["total"] == 2
assert result["todos"][0]["id"] == "t1"
assert result["todos"][1]["status"] == "in_progress"
def test_unparseable_string_returns_error(self):
store = TodoStore()
result = json.loads(todo_tool(todos="not valid json [", store=store))
assert "error" in result
def test_json_string_that_parses_to_non_list_returns_error(self):
store = TodoStore()
# Valid JSON, but a dict instead of a list
result = json.loads(todo_tool(todos='{"id": "1"}', store=store))
assert "error" in result
def test_non_list_non_string_returns_error(self):
store = TodoStore()
result = json.loads(todo_tool(todos=42, store=store))
assert "error" in result
class TestNonDictListItems:
"""Guards 2 & 3: _validate and _dedupe_by_id handle non-dict items."""
def test_string_item_in_list_does_not_crash(self):
store = TodoStore()
result = store.write(["not-a-dict"])
assert len(result) == 1
assert result[0]["id"] == "?"
assert result[0]["content"] == "(invalid item)"
assert result[0]["status"] == "pending"
def test_mixed_valid_and_invalid_items(self):
store = TodoStore()
result = store.write([
{"id": "1", "content": "Real task", "status": "pending"},
"garbage",
42,
{"id": "2", "content": "Another task", "status": "completed"},
])
assert len(result) == 4
# Valid items are preserved
assert result[0]["id"] == "1"
assert result[0]["content"] == "Real task"
assert result[3]["id"] == "2"
# Invalid items get placeholder values
assert result[1]["content"] == "(invalid item)"
assert result[2]["content"] == "(invalid item)"
def test_none_item_in_list(self):
store = TodoStore()
result = store.write([None])
assert len(result) == 1
assert result[0]["id"] == "?"
def test_integer_item_in_list(self):
store = TodoStore()
result = store.write([123])
assert len(result) == 1
assert result[0]["content"] == "(invalid item)"
def test_non_dict_items_via_todo_tool(self):
"""End-to-end: non-dict list items produce valid output, not a crash."""
store = TodoStore()
result = json.loads(todo_tool(todos=["bad", "also bad"], store=store))
assert "error" not in result
assert result["summary"]["total"] == 2
assert result["summary"]["pending"] == 2
class TestWellFormedInputUnchanged:
"""Regression: normal usage must not be affected by the guards."""
def test_normal_write_and_read(self):
store = TodoStore()
items = [
{"id": "a", "content": "First", "status": "pending"},
{"id": "b", "content": "Second", "status": "in_progress"},
]
result = json.loads(todo_tool(todos=items, store=store))
assert result["summary"]["total"] == 2
assert result["summary"]["pending"] == 1
assert result["summary"]["in_progress"] == 1
def test_merge_mode_still_works(self):
store = TodoStore()
store.write([{"id": "1", "content": "Original", "status": "pending"}])
result = json.loads(todo_tool(
todos=[{"id": "1", "status": "completed"}],
merge=True,
store=store,
))
assert result["summary"]["completed"] == 1
assert result["todos"][0]["content"] == "Original"
def test_read_mode_still_works(self):
store = TodoStore()
store.write([{"id": "x", "content": "Task", "status": "pending"}])
result = json.loads(todo_tool(store=store))
assert result["summary"]["total"] == 1
def test_dedup_still_works(self):
store = TodoStore()
result = store.write([
{"id": "1", "content": "v1", "status": "pending"},
{"id": "1", "content": "v2", "status": "in_progress"},
])
assert len(result) == 1
assert result[0]["content"] == "v2"
-17
View File
@@ -158,9 +158,6 @@ class TodoStore:
Ensures required fields exist and status is valid.
Returns a clean dict with only {id, content, status}.
"""
if not isinstance(item, dict):
return {"id": "?", "content": "(invalid item)", "status": "pending"}
item_id = str(item.get("id", "")).strip()
if not item_id:
item_id = "?"
@@ -182,10 +179,6 @@ class TodoStore:
"""Collapse duplicate ids, keeping the last occurrence in its position."""
last_index: Dict[str, int] = {}
for i, item in enumerate(todos):
if not isinstance(item, dict):
# Non-dict items get a synthetic key so _validate can handle them
last_index[f"__invalid_{i}"] = i
continue
item_id = str(item.get("id", "")).strip() or "?"
last_index[item_id] = i
return [todos[i] for i in sorted(last_index.values())]
@@ -211,16 +204,6 @@ def todo_tool(
return tool_error("TodoStore not initialized")
if todos is not None:
# Guard: LLM sometimes sends todos as a JSON string instead of a list
if isinstance(todos, str):
try:
todos = json.loads(todos)
except (json.JSONDecodeError, TypeError):
return tool_error("todos must be a list of objects, got unparseable string")
if not isinstance(todos, list):
return tool_error(
f"todos must be a list, got {type(todos).__name__}"
)
items = store.write(todos, merge)
else:
items = store.read()