Private
Public Access
0
0

feat(mcp-server): add aggregate_directives tool to scripts/mcp_server.py

Adds a new MCP tool that exposes scripts/aggregate_directives.py as an
LLM-callable endpoint. The tool reads a presets markdown file, resolves
each directive's v1.md, and returns the concatenated clean directive
bodies. metadata (meta.md) is never read.

Tool contract:
  name: aggregate_directives
  params: preset_path (string, default current_baseline.md), max_chars (int, default 0)
  result: text content (clean bodies) or 'ERROR: ...' string on failure

Dispatch runs the impl in asyncio.to_thread so the 66+ v1.md reads do not
block the MCP stdio loop. Errors from aggregate_directives are caught and
formatted as 'ERROR: aggregate_directives(<path>) failed: <type>: <msg>' so
the LLM can read the failure mode inline. Tool count: 45 (MCP_TOOL_SPECS)
+ 2 (run_powershell, aggregate_directives) = 47.

Added scripts/ to sys.path so 'from aggregate_directives import ...' works
inside the MCP server process; idempotent with existing project_root and src
path inserts.
This commit is contained in:
2026-07-03 10:48:44 -04:00
parent 5df938805e
commit a527f6224f
+66 -3
View File
@@ -1,12 +1,15 @@
"""
MCP server exposing Manual Slop's custom tools (mcp_client.py) to Claude Code.
All 26 tools from mcp_client.MCP_TOOL_SPECS are served, plus run_powershell.
Delegates to mcp_client.dispatch() for all tools except run_powershell,
which routes through shell_runner.run_powershell() directly.
All 45 tools from mcp_client.MCP_TOOL_SPECS are served, plus run_powershell
and aggregate_directives (47 total). Delegates to mcp_client.dispatch() for
all tools except run_powershell (which routes through shell_runner directly)
and aggregate_directives (which routes through scripts/aggregate_directives.py).
Usage (in .claude/settings.json mcpServers):
"command": "uv", "args": ["run", "python", "scripts/mcp_server.py"]
# 2026-07-02: added aggregate_directives tool (Tier 1 directive hotswap harness expansion).
"""
import asyncio
@@ -17,11 +20,14 @@ import sys
project_root = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
sys.path.insert(0, project_root)
sys.path.insert(0, os.path.join(project_root, "src"))
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import mcp_client
import mcp_tool_specs
import shell_runner
from aggregate_directives import aggregate_directives as _do_aggregate
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
@@ -47,6 +53,52 @@ RUN_POWERSHELL_SPEC = {
}
}
# aggregate_directives is handled by scripts/aggregate_directives.py, not
# mcp_client.dispatch(). Define its spec here since it is not in MCP_TOOL_SPECS.
AGGREGATE_DIRECTIVES_SPEC = {
"name": "aggregate_directives",
"description": (
"Read a presets markdown file (e.g., conductor/directives/presets/current_baseline.md), "
"resolve each directive path to its v1.md file, and return the concatenated clean "
"directive bodies as a single text response. The metadata files (meta.md) are "
"NEVER read - only the v1.md bodies are emitted.\n\n"
"Useful when:\n"
"- Session context was reset and the LLM needs to rehydrate the directives\n"
"- The user said 'warm with: <preset>' and you need the actual text to read\n"
"- A preset was just changed and you want to inspect what the agent sees\n\n"
"For most sessions you should NOT call this - the warm-with: bootstrap "
"already instructs the LLM to read the preset's v1.md files directly."
),
"parameters": {
"type": "object",
"properties": {
"preset_path": {
"type": "string",
"description": "Path to the presets markdown file, relative to the project root or absolute. Defaults to 'conductor/directives/presets/current_baseline.md'."
},
"max_chars": {
"type": "integer",
"description": "Optional cap on total response size (chars). 0 or null = no cap. Useful for huge presets."
}
},
"required": []
}
}
def _aggregate_directives_impl(preset_path: str, max_chars: int = 0) -> str:
"""
Synchronous implementation backing the aggregate_directives MCP tool.
Resolves preset_path against the script's project_root and delegates to
scripts/aggregate_directives.py:aggregate_directives. Any exception is
converted to a single ERROR:... string so the LLM can read it inline.
"""
root = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
try:
return _do_aggregate(preset_path, max_chars=max_chars, project_root=root)
except Exception as e:
return "ERROR: aggregate_directives(" + str(preset_path) + ") failed: " + type(e).__name__ + ": " + str(e)
server = Server("manual-slop-tools")
@server.list_tools()
@@ -64,6 +116,12 @@ async def list_tools() -> list[Tool]:
description=RUN_POWERSHELL_SPEC["description"],
inputSchema=RUN_POWERSHELL_SPEC["parameters"],
))
# Add aggregate_directives
tools.append(Tool(
name=AGGREGATE_DIRECTIVES_SPEC["name"],
description=AGGREGATE_DIRECTIVES_SPEC["description"],
inputSchema=AGGREGATE_DIRECTIVES_SPEC["parameters"],
))
return tools
@server.call_tool()
@@ -73,6 +131,11 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]:
script = arguments.get("script", "")
# run_powershell is synchronous, so we run it in a thread to avoid blocking the loop
result = await asyncio.to_thread(shell_runner.run_powershell, script, os.getcwd())
elif name == "aggregate_directives":
preset_path = arguments.get("preset_path") or "conductor/directives/presets/current_baseline.md"
max_chars = int(arguments.get("max_chars") or 0)
# aggregate_directives is synchronous and may read 66+ files; offload to a thread
result = await asyncio.to_thread(_aggregate_directives_impl, preset_path, max_chars)
else:
result = await mcp_client.async_dispatch(name, arguments)
return [TextContent(type="text", text=str(result))]