Private
Public Access
0
0

fix(mcp_server): migrate from MCP_TOOL_SPECS dict to mcp_tool_specs.get_tool_schemas()

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.
This commit is contained in:
2026-06-24 20:40:20 -04:00
parent 705cb50d14
commit 9d300537b7
5 changed files with 229 additions and 1 deletions
+2 -1
View File
@@ -19,6 +19,7 @@ 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
@@ -51,7 +52,7 @@ server = Server("manual-slop-tools")
@server.list_tools()
async def list_tools() -> list[Tool]:
tools = []
for spec in mcp_client.MCP_TOOL_SPECS:
for spec in [t.to_dict() for t in mcp_tool_specs.get_tool_schemas()]:
tools.append(Tool(
name=spec["name"],
description=spec["description"],
@@ -0,0 +1,10 @@
import json
import subprocess
r = subprocess.run(["uv", "run", "python", "scripts/audit_exception_handling.py", "--json"], capture_output=True, text=True)
data = json.loads(r.stdout)
for f in data.get("files", []):
if f.get("violation_count", 0) > 0:
print(f"\n=== {f['filename']} (violations: {f['violation_count']}) ===")
for finding in f.get("findings", []):
if finding.get("category") == "INTERNAL_OPTIONAL_RETURN":
print(f" Line {finding['line']}: {finding['context']} ({finding['kind']})")
@@ -0,0 +1,19 @@
import sys
sys.path.insert(0, "src")
from src import mcp_client, mcp_tool_specs
# Check key APIs still work
print(f"TOOL_NAMES: {len(mcp_client.TOOL_NAMES)}")
print(f"tool_names(): {len(mcp_tool_specs.tool_names())}")
print(f"get_tool_schemas (no external): {len(mcp_tool_specs.get_tool_schemas())}")
print(f"get_tool_schemas: {len(mcp_client.get_tool_schemas())} (external + native)")
# Check Optional[T] removal worked
from src import ai_client
print(f"get_current_tier: {ai_client.get_current_tier_result().data}")
print(f"get_bias_profile: {ai_client.get_bias_profile_result().data}")
# Check Result[T] sentinel for parsing
from src import external_editor, session_logger, project_manager
print(f"parse_ts good: {project_manager.parse_ts_result('2026-06-24T12:00:00').data}")
print(f"parse_ts bad: {project_manager.parse_ts_result('bad').errors[0].message[:60]}")
@@ -0,0 +1,97 @@
"""Verify the MCP server can actually dispatch a tool call end-to-end.
Spawns scripts/mcp_server.py, calls get_file_summary on this test file,
and verifies the tool returned real content.
"""
import asyncio
import json
import os
import subprocess
import sys
import time
from pathlib import Path
PROJECT_ROOT = Path(__file__).resolve().parents[4]
MCP_SCRIPT = PROJECT_ROOT / "scripts" / "mcp_server.py"
def test_mcp_server_dispatches_tool():
env = {**os.environ, "PYTHONPATH": str(PROJECT_ROOT / "src")}
proc = subprocess.Popen(
["uv", "run", "python", str(MCP_SCRIPT)],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
cwd=str(PROJECT_ROOT),
env=env,
)
try:
# initialize
proc.stdin.write((json.dumps({
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {"name": "test", "version": "0.1"},
},
}) + "\n").encode())
# tools/call: get_file_summary
proc.stdin.write((json.dumps({
"jsonrpc": "2.0",
"id": 2,
"method": "tools/call",
"params": {
"name": "get_file_summary",
"arguments": {"path": str(Path(__file__))},
},
}) + "\n").encode())
proc.stdin.flush()
time.sleep(5)
proc.terminate()
stdout, stderr = proc.communicate(timeout=5)
responses = []
for line in stdout.decode("utf-8", errors="replace").strip().split("\n"):
try:
responses.append(json.loads(line))
except json.JSONDecodeError:
continue
# Find the tools/call response
call_response = None
for r in responses:
if r.get("id") == 2:
call_response = r
break
assert call_response is not None, f"No tools/call response. Got: {responses}"
assert "result" in call_response, f"Missing result in: {call_response}"
content = call_response["result"]["content"][0]["text"]
# Should mention the file
assert "test_mcp_server_starts" in content or "Python" in content, f"Unexpected content: {content[:200]}"
# No stderr errors
stderr_text = stderr.decode("utf-8", errors="replace")
assert "AttributeError" not in stderr_text
assert "ImportError" not in stderr_text
assert "ModuleNotFoundError" not in stderr_text
print(f"PASS: MCP server dispatched get_file_summary; response starts with: {content[:120]}")
return True
except Exception as e:
proc.kill()
print(f"FAIL: {e}")
return False
finally:
try:
proc.kill()
except Exception:
pass
if __name__ == "__main__":
success = test_mcp_server_dispatches_tool()
sys.exit(0 if success else 1)
@@ -0,0 +1,101 @@
"""Verify the MCP server starts and lists tools correctly.
Spawns scripts/mcp_server.py as a subprocess, sends a list_tools request,
and verifies it returns the expected number of tools.
"""
import asyncio
import json
import os
import subprocess
import sys
import time
from pathlib import Path
PROJECT_ROOT = Path(__file__).resolve().parents[4]
MCP_SCRIPT = PROJECT_ROOT / "scripts" / "mcp_server.py"
def test_mcp_server_starts_and_lists_tools():
"""Spawn the MCP server and call list_tools via JSON-RPC over stdio."""
env = {**os.environ, "PYTHONPATH": str(PROJECT_ROOT / "src")}
proc = subprocess.Popen(
["uv", "run", "python", str(MCP_SCRIPT)],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
cwd=str(PROJECT_ROOT),
env=env,
)
try:
# JSON-RPC: initialize
proc.stdin.write((json.dumps({
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {"name": "test", "version": "0.1"},
},
}) + "\n").encode())
# JSON-RPC: tools/list
proc.stdin.write((json.dumps({
"jsonrpc": "2.0",
"id": 2,
"method": "tools/list",
"params": {},
}) + "\n").encode())
proc.stdin.flush()
time.sleep(4)
proc.terminate()
stdout, stderr = proc.communicate(timeout=5)
# Parse line-delimited JSON-RPC responses
responses = []
for line in stdout.decode("utf-8", errors="replace").strip().split("\n"):
try:
responses.append(json.loads(line))
except json.JSONDecodeError:
continue
# Find the tools/list response
tools_response = None
for r in responses:
if r.get("id") == 2:
tools_response = r
break
assert tools_response is not None, f"No tools/list response. Got: {responses}"
assert "result" in tools_response, f"Missing result in: {tools_response}"
tools = tools_response["result"]["tools"]
tool_names = [t["name"] for t in tools]
# Expectations: 45 tools in mcp_tool_specs + 1 run_powershell = 46
assert len(tools) == 46, f"Expected 46 tools, got {len(tools)}: {tool_names}"
assert "run_powershell" in tool_names, f"Missing run_powershell in {tool_names}"
assert "read_file" in tool_names, f"Missing read_file in {tool_names}"
assert "py_get_skeleton" in tool_names, f"Missing py_get_skeleton in {tool_names}"
# No stderr errors
stderr_text = stderr.decode("utf-8", errors="replace")
assert "AttributeError" not in stderr_text, f"AttributeError in stderr: {stderr_text}"
assert "ImportError" not in stderr_text, f"ImportError in stderr: {stderr_text}"
assert "ModuleNotFoundError" not in stderr_text, f"ModuleNotFoundError in stderr: {stderr_text}"
print(f"PASS: MCP server listed {len(tools)} tools including run_powershell")
print(f"First 5 tools: {tool_names[:5]}")
return True
except Exception as e:
proc.kill()
print(f"FAIL: {e}")
return False
finally:
try:
proc.kill()
except Exception:
pass
if __name__ == "__main__":
success = test_mcp_server_starts_and_lists_tools()
sys.exit(0 if success else 1)