feat: implement AI Workspace context MCP server with read-only access and add related tests

scripts/mcp/aiw-context-mcp/test_server.py

@@ -0,0 +1,87 @@
+#!/usr/bin/env python3
+
+from __future__ import annotations
+
+import importlib.util
+import json
+import sys
+import tempfile
+import unittest
+from datetime import date
+from pathlib import Path
+from unittest.mock import patch
+
+
+SERVER_PATH = Path(__file__).with_name("server.py")
+SPEC = importlib.util.spec_from_file_location("aiw_context_mcp", SERVER_PATH)
+server = importlib.util.module_from_spec(SPEC)
+assert SPEC.loader is not None
+sys.modules[SPEC.name] = server
+SPEC.loader.exec_module(server)
+
+
+class ContextMCPTests(unittest.TestCase):
+    def test_initialize_response_declares_tools(self) -> None:
+        response = server.handle_request({"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": server.PROTOCOL_VERSION}})
+
+        self.assertEqual(response["result"]["protocolVersion"], server.PROTOCOL_VERSION)
+        self.assertIn("tools", response["result"]["capabilities"])
+
+    def test_tools_list_includes_project_search(self) -> None:
+        response = server.handle_request({"jsonrpc": "2.0", "id": 1, "method": "tools/list"})
+
+        names = {tool["name"] for tool in response["result"]["tools"]}
+        self.assertIn("project_search_memory", names)
+        self.assertIn("communication_latest", names)
+
+    def test_unknown_tool_returns_protocol_error(self) -> None:
+        response = server.handle_request({"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": {"name": "missing", "arguments": {}}})
+
+        self.assertEqual(response["error"]["code"], -32602)
+
+    def test_notification_returns_none(self) -> None:
+        response = server.handle_request({"jsonrpc": "2.0", "method": "notifications/initialized"})
+
+        self.assertIsNone(response)
+
+    def test_communication_latest_reads_bounded_records(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            latest = root / "ai" / "inbox" / "mattermost-mirror" / "latest.jsonl"
+            latest.parent.mkdir(parents=True)
+            for index in range(3):
+                latest.write_text("", encoding="utf-8") if index == 0 else None
+            latest.write_text(
+                "\n".join(json.dumps({"post_id": str(index), "message": f"m{index}"}) for index in range(3)) + "\n",
+                encoding="utf-8",
+            )
+
+            with patch.object(server, "ROOT", root):
+                result = server.communication_latest({"profile": "fidelity", "limit": 2})["structuredContent"]
+
+        self.assertEqual([item["message"] for item in result["records"]], ["m1", "m2"])
+
+    def test_project_search_skips_templates(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            real = root / "project-knowledge" / "03-context" / "project.md"
+            template = root / "project-knowledge" / "09-templates" / "daily.md"
+            real.parent.mkdir(parents=True)
+            template.parent.mkdir(parents=True)
+            real.write_text("Important XFlow context", encoding="utf-8")
+            template.write_text("Important XFlow template", encoding="utf-8")
+
+            with patch.object(server, "ROOT", root):
+                result = server.project_search_memory({"profile": "fidelity", "query": "XFlow"})["structuredContent"]
+
+        self.assertEqual(len(result["matches"]), 1)
+        self.assertIn("03-context/project.md", result["matches"][0]["path"])
+
+    def test_previous_workday_skips_weekend(self) -> None:
+        monday = date(2026, 5, 18)
+
+        self.assertEqual(server.previous_workday(monday), date(2026, 5, 15))
+
+
+if __name__ == "__main__":
+    unittest.main()