manual_slop/scripts/mcp_server.py

"""
MCP server exposing Manual Slop's custom tools (mcp_client.py) to Claude Code.

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
import os
import sys

# Add project root and src/ to sys.path
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

# run_powershell is handled by shell_runner, not mcp_client.dispatch()
# Define its spec here since it's not in MCP_TOOL_SPECS
RUN_POWERSHELL_SPEC = {
 "name": "run_powershell",
 "description": (
  "Run a PowerShell script within the project base directory. "
  "Returns combined stdout, stderr, and exit code. "
  "60-second timeout. Use for builds, tests, and system commands."
 ),
 "parameters": {
  "type": "object",
  "properties": {
   "script": {
    "type": "string",
    "description": "PowerShell script content to execute."
   }
  },
  "required": ["script"]
 }
}

# 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()
async def list_tools() -> list[Tool]:
 tools = []
 for spec in [t.to_dict() for t in mcp_tool_specs.get_tool_schemas()]:
  tools.append(Tool(
    name=spec["name"],
    description=spec["description"],
    inputSchema=spec["parameters"],
   ))
  # Add run_powershell
 tools.append(Tool(
   name=RUN_POWERSHELL_SPEC["name"],
   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()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
 try:
  if name == "run_powershell":
   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))]
 except Exception as e:
  return [TextContent(type="text", text=f"ERROR: {e}")]

async def main() -> None:
 # Robust context detection: project_root is os.getcwd() (the directory
 # the user is actually working in), not just where the script lives.
 # The script's own home is a secondary fallback. This handles the case
 # where opencode launches the MCP from a sibling clone (e.g., main repo
 # launches the tier2 clone's MCP via a hardcoded path in opencode.json)
 # — the MCP should allow access to the user's working directory too.
 cwd = os.getcwd()
 script_root = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))

 extra_dirs: list[str] = []
 for d in (cwd, script_root):
  if d and d not in extra_dirs:
   extra_dirs.append(d)

 # Read mcp_paths.toml from cwd first (the user's working dir takes
 # precedence), then fall back to the script's home dir.
 for mcp_paths_toml in (os.path.join(cwd, "mcp_paths.toml"),
                          os.path.join(script_root, "mcp_paths.toml")):
  if os.path.exists(mcp_paths_toml):
   import tomllib
   with open(mcp_paths_toml, "rb") as f:
    config = tomllib.load(f)
   allowed = config.get("allowed_paths", {}).get("extra_dirs", [])
   for p in allowed:
    if p not in extra_dirs:
     extra_dirs.append(p)
   break

 mcp_client.configure([], extra_base_dirs=extra_dirs)
 async with stdio_server() as (read_stream, write_stream):
  await server.run(
   read_stream,
   write_stream,
   server.create_initialization_options(),
  )

if __name__ == "__main__":
 asyncio.run(main())