test(mcp): add schema/runtime conformance guard (#794)

app/mcp_tools.py

@@ -103,17 +103,18 @@ def optional_bounty_search_query_arg() -> str | None:
         return query
 
     def output_format_arg() -> str:
-        value = args.get("format", "text")
-        if value is None:
+        if "format" not in args:
             return "text"
+        value = args["format"]
+        if value is None:
+            raise ValueError("format must be a string")
         if not isinstance(value, str):
             raise ValueError("format must be a string")
         if contains_control_character(value):
             raise ValueError("format must not contain control characters")
-        normalized = value.strip().lower()
-        if normalized not in {"text", "json"}:
+        if value not in {"text", "json"}:
             raise ValueError("format must be text or json")
-        return normalized
+        return value
 
     def optional_repo_selector_arg() -> str | None:
         repo = optional_clean_str_arg("repo")

tests/mcp_conformance.py

@@ -0,0 +1,64 @@
+from __future__ import annotations
+
+from typing import Any
+
+from fastapi.testclient import TestClient
+
+
+def mcp_tools_by_name(client: TestClient) -> dict[str, dict[str, Any]]:
+    response = client.post("/mcp", json={"jsonrpc": "2.0", "id": 1, "method": "tools/list"})
+    payload = response.json()
+    return {tool["name"]: tool for tool in payload["result"]["tools"]}
+
+
+def mcp_tools_call(
+    client: TestClient,
+    *,
+    tool_name: str,
+    arguments: dict[str, Any],
+    request_id: int,
+) -> dict[str, Any]:
+    response = client.post(
+        "/mcp",
+        json={
+            "jsonrpc": "2.0",
+            "id": request_id,
+            "method": "tools/call",
+            "params": {"name": tool_name, "arguments": arguments},
+        },
+    )
+    assert response.status_code == 200
+    return response.json()
+
+
+def assert_mcp_tools_call_rejects(
+    client: TestClient,
+    *,
+    tool_name: str,
+    arguments: dict[str, Any],
+    request_id: int,
+) -> None:
+    payload = mcp_tools_call(
+        client, tool_name=tool_name, arguments=arguments, request_id=request_id
+    )
+    assert payload["jsonrpc"] == "2.0"
+    assert payload["id"] == request_id
+    error = payload["error"]
+    assert isinstance(error, dict)
+    assert error["code"] == -32602
+    assert error["message"] == "invalid tool arguments"
+
+
+def assert_mcp_tools_call_accepts(
+    client: TestClient,
+    *,
+    tool_name: str,
+    arguments: dict[str, Any],
+    request_id: int,
+) -> dict[str, Any]:
+    payload = mcp_tools_call(
+        client, tool_name=tool_name, arguments=arguments, request_id=request_id
+    )
+    assert "result" in payload
+    assert "error" not in payload
+    return payload["result"]

tests/test_mcp_schema_conformance.py

@@ -0,0 +1,204 @@
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+from fastapi.testclient import TestClient
+
+from app.db import create_schema, session_scope
+from app.ledger.service import create_bounty, ensure_genesis
+from app.main import create_app
+from app.mcp import MCP_TOOLS
+from tests.mcp_conformance import (
+    assert_mcp_tools_call_accepts,
+    assert_mcp_tools_call_rejects,
+    mcp_tools_by_name,
+)
+
+
+def _tools_with_input_schema() -> list[dict[str, Any]]:
+    return [tool for tool in MCP_TOOLS if "inputSchema" in tool]
+
+
+@pytest.mark.parametrize("tool", _tools_with_input_schema(), ids=lambda tool: tool["name"])
+def test_mcp_tools_list_input_schema_disallows_extra_properties(tool: dict[str, Any]) -> None:
+    schema = tool["inputSchema"]
+    assert schema.get("type") == "object"
+    assert schema.get("additionalProperties") is False
+
+
+@pytest.mark.parametrize(
+    ("arguments", "request_id"),
+    [
+        ({"format": "JSON"}, 101),
+        ({"format": " JSON "}, 102),
+        ({"format": "text "}, 103),
+        ({"format": " json"}, 104),
+        ({"format": None, "bounty_id": 1}, 105),
+    ],
+)
+def test_mcp_submit_work_proof_rejects_non_exact_format_enum(
+    sqlite_url: str, arguments: dict[str, object], request_id: int
+) -> None:
+    create_schema(sqlite_url)
+    with session_scope(sqlite_url) as session:
+        ensure_genesis(session)
+        create_bounty(
+            session,
+            repo="ramimbo/mergework",
+            issue_number=794,
+            issue_url="https://github.com/ramimbo/mergework/issues/794",
+            title="MCP schema conformance",
+            reward_mrwk="150",
+            acceptance="Schema/runtime conformance guard.",
+        )
+
+    client = TestClient(create_app(database_url=sqlite_url, webhook_secret="secret"))
+    assert_mcp_tools_call_rejects(
+        client,
+        tool_name="submit_work_proof",
+        arguments=arguments,
+        request_id=request_id,
+    )
+
+
+@pytest.mark.parametrize(
+    ("arguments", "request_id"),
+    [
+        ({"unexpected": "ignored"}, 110),
+        ({"format": "json", "unexpected": "ignored"}, 111),
+        ({"bounty_id": 1, "unexpected": "ignored"}, 112),
+    ],
+)
+def test_mcp_submit_work_proof_conformance_rejects_undeclared_properties(
+    sqlite_url: str, arguments: dict[str, object], request_id: int
+) -> None:
+    create_schema(sqlite_url)
+    with session_scope(sqlite_url) as session:
+        ensure_genesis(session)
+        create_bounty(
+            session,
+            repo="ramimbo/mergework",
+            issue_number=794,
+            issue_url="https://github.com/ramimbo/mergework/issues/794",
+            title="MCP schema conformance",
+            reward_mrwk="150",
+            acceptance="Schema/runtime conformance guard.",
+        )
+
+    client = TestClient(create_app(database_url=sqlite_url, webhook_secret="secret"))
+    assert_mcp_tools_call_rejects(
+        client,
+        tool_name="submit_work_proof",
+        arguments=arguments,
+        request_id=request_id,
+    )
+
+
+@pytest.mark.parametrize(
+    ("arguments", "request_id"),
+    [
+        ({"bounty_id": 1, "issue_number": 1}, 120),
+        ({"repo": "ramimbo/mergework"}, 121),
+        ({"bounty_id": 1, "repo": "ramimbo/mergework"}, 122),
+        ({"issue_number": 1, "repo": 1}, 123),
+    ],
+)
+def test_mcp_submit_work_proof_conformance_rejects_invalid_selectors(
+    sqlite_url: str, arguments: dict[str, object], request_id: int
+) -> None:
+    create_schema(sqlite_url)
+    with session_scope(sqlite_url) as session:
+        ensure_genesis(session)
+        create_bounty(
+            session,
+            repo="ramimbo/mergework",
+            issue_number=794,
+            issue_url="https://github.com/ramimbo/mergework/issues/794",
+            title="MCP schema conformance",
+            reward_mrwk="150",
+            acceptance="Schema/runtime conformance guard.",
+        )
+
+    client = TestClient(create_app(database_url=sqlite_url, webhook_secret="secret"))
+    assert_mcp_tools_call_rejects(
+        client,
+        tool_name="submit_work_proof",
+        arguments=arguments,
+        request_id=request_id,
+    )
+
+
+@pytest.mark.parametrize(
+    ("arguments", "request_id"),
+    [
+        ({"bounty_id": "099"}, 130),
+        ({"issue_number": "0656"}, 131),
+        ({"bounty_id": "+1"}, 132),
+    ],
+)
+def test_mcp_submit_work_proof_conformance_rejects_noncanonical_integers(
+    sqlite_url: str, arguments: dict[str, object], request_id: int
+) -> None:
+    create_schema(sqlite_url)
+    with session_scope(sqlite_url) as session:
+        ensure_genesis(session)
+
+    client = TestClient(create_app(database_url=sqlite_url, webhook_secret="secret"))
+    assert_mcp_tools_call_rejects(
+        client,
+        tool_name="submit_work_proof",
+        arguments=arguments,
+        request_id=request_id,
+    )
+
+
+def test_mcp_submit_work_proof_conformance_accepts_schema_valid_examples(
+    sqlite_url: str,
+) -> None:
+    create_schema(sqlite_url)
+    with session_scope(sqlite_url) as session:
+        ensure_genesis(session)
+        bounty = create_bounty(
+            session,
+            repo="ramimbo/mergework",
+            issue_number=794,
+            issue_url="https://github.com/ramimbo/mergework/issues/794",
+            title="MCP schema conformance",
+            reward_mrwk="150",
+            acceptance="Schema/runtime conformance guard.",
+        )
+
+    client = TestClient(create_app(database_url=sqlite_url, webhook_secret="secret"))
+    tools = mcp_tools_by_name(client)
+    submit_schema = tools["submit_work_proof"]["inputSchema"]
+    assert submit_schema["properties"]["format"]["enum"] == ["text", "json"]
+
+    assert_mcp_tools_call_accepts(
+        client,
+        tool_name="submit_work_proof",
+        arguments={"format": "text"},
+        request_id=140,
+    )
+    assert_mcp_tools_call_accepts(
+        client,
+        tool_name="submit_work_proof",
+        arguments={"bounty_id": bounty.id, "format": "json"},
+        request_id=141,
+    )
+    assert_mcp_tools_call_accepts(
+        client,
+        tool_name="submit_work_proof",
+        arguments={"issue_number": 794, "format": "json"},
+        request_id=142,
+    )
+    assert_mcp_tools_call_accepts(
+        client,
+        tool_name="submit_work_proof",
+        arguments={
+            "issue_number": 794,
+            "repo": "ramimbo/mergework",
+            "format": "text",
+        },
+        request_id=143,
+    )