feat(I): I-MCP — MCP exposure layer (6 tools: search, ask_knowledge, start_workflow, check_workflow_status, list_agents, create_task), permission-checked, 16 tests passing

tests/test_phase_i_integration.py

@@ -119,3 +119,88 @@ class TestAgentLoopApproval:
         result = ReActResult(final_content="", status="waiting_for_approval")
         assert result.status == "waiting_for_approval"
         assert result.final_content == ""
+
+
+# ─── I-MCP: MCP-Exposure ─────────────────────────────────────────────────────
+
+
+class TestMCPExposure:
+    """Test the MCP exposure layer (I-MCP)."""
+
+    def test_mcp_tools_count(self):
+        """6 MCP tools are defined."""
+        from app.ai.mcp_exposure import MCP_TOOLS
+
+        assert len(MCP_TOOLS) == 6
+
+    def test_mcp_tool_names(self):
+        """MCP tools have correct names."""
+        from app.ai.mcp_exposure import MCP_TOOLS
+
+        names = {t["name"] for t in MCP_TOOLS}
+        assert names == {"search", "ask_knowledge", "start_workflow", "check_workflow_status", "list_agents", "create_task"}
+
+    def test_get_mcp_tools_returns_schemas(self):
+        """get_mcp_tools returns tool schemas without internal fields."""
+        from app.ai.mcp_exposure import get_mcp_tools
+
+        tools = get_mcp_tools()
+        for t in tools:
+            assert "name" in t
+            assert "description" in t
+            assert "input_schema" in t
+            assert "required_permission" not in t  # Internal field not exposed
+            assert "handler" not in t  # Internal field not exposed
+
+    def test_get_mcp_tool_existing(self):
+        """get_mcp_tool returns tool definition for existing tool."""
+        from app.ai.mcp_exposure import get_mcp_tool
+
+        tool = get_mcp_tool("search")
+        assert tool is not None
+        assert tool["name"] == "search"
+        assert tool["required_permission"] == "contacts:read"
+
+    def test_get_mcp_tool_nonexistent(self):
+        """get_mcp_tool returns None for unknown tool."""
+        from app.ai.mcp_exposure import get_mcp_tool
+
+        assert get_mcp_tool("nonexistent") is None
+
+    @pytest.mark.asyncio
+    async def test_execute_mcp_tool_unknown(self):
+        """execute_mcp_tool returns error for unknown tool."""
+        from app.ai.mcp_exposure import execute_mcp_tool
+
+        result = await execute_mcp_tool(
+            db=MagicMock(), tenant_id=uuid.uuid4(), user_id=uuid.uuid4(),
+            tool_name="nonexistent", arguments={},
+        )
+        assert result["status"] == "not_found"
+
+    @pytest.mark.asyncio
+    async def test_execute_mcp_tool_permission_denied(self):
+        """execute_mcp_tool returns forbidden when permission missing."""
+        from app.ai.mcp_exposure import execute_mcp_tool
+
+        result = await execute_mcp_tool(
+            db=MagicMock(), tenant_id=uuid.uuid4(), user_id=uuid.uuid4(),
+            tool_name="start_workflow", arguments={"workflow_id": "test"},
+            user_permissions={"permissions": [], "denied_permissions": [], "is_system_admin": False},
+        )
+        assert result["status"] == "forbidden"
+
+    @pytest.mark.asyncio
+    async def test_execute_mcp_tool_search(self):
+        """execute_mcp_tool delegates to search_knowledge_tool for 'search'."""
+        from app.ai.mcp_exposure import execute_mcp_tool
+
+        with patch("app.ai.integration_tools.search_knowledge_tool", new_callable=AsyncMock) as mock_search:
+            mock_search.return_value = {"results": [], "total": 0, "query": "test"}
+            result = await execute_mcp_tool(
+                db=MagicMock(), tenant_id=uuid.uuid4(), user_id=uuid.uuid4(),
+                tool_name="search", arguments={"query": "test"},
+                user_permissions={"is_system_admin": True},
+            )
+            assert result["query"] == "test"
+            mock_search.assert_called_once()