ed
9d300537b7
Phase 1 of code_path_audit_phase_2_20260624 deleted mcp_client.MCP_TOOL_SPECS (the 778-line dict literal). This broke scripts/mcp_server.py which iterated over mcp_client.MCP_TOOL_SPECS in its list_tools() handler — the MCP server crashed on startup with AttributeError, breaking the entire manual-slop MCP. Fix: use mcp_tool_specs.get_tool_schemas() (the new ToolSpec registry) and convert via .to_dict() to the JSON-compatible dict format the MCP Tool constructor expects. Verified: 46 tools listed (45 from registry + run_powershell); tool call (get_file_summary) dispatched end-to-end correctly; 23 mcp-related unit tests pass.
121 lines
3.8 KiB
Python
121 lines
3.8 KiB
Python
"""
|
|
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.
|
|
|
|
Usage (in .claude/settings.json mcpServers):
|
|
"command": "uv", "args": ["run", "python", "scripts/mcp_server.py"]
|
|
"""
|
|
|
|
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"))
|
|
|
|
import mcp_client
|
|
import mcp_tool_specs
|
|
import shell_runner
|
|
|
|
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"]
|
|
}
|
|
}
|
|
|
|
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"],
|
|
))
|
|
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())
|
|
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())
|